services/nodemanager/tests/test_jobqueue.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import absolute_import, print_function
   4
   5 import unittest
   6 import mock
   7
   8 import arvnodeman.jobqueue as jobqueue
   9 from . import testutil
  10
  11 class ServerCalculatorTestCase(unittest.TestCase):
  12     def make_calculator(self, factors, **kwargs):
  13         return jobqueue.ServerCalculator(
  14             [(testutil.MockSize(n), {'cores': n}) for n in factors], **kwargs)
  15
  16     def calculate(self, servcalc, *constraints):
  17         return servcalc.servers_for_queue(
  18             [{'uuid': 'zzzzz-jjjjj-{:015x}'.format(index),
  19               'runtime_constraints': cdict}
  20              for index, cdict in enumerate(constraints)])
  21
  22     def test_empty_queue_needs_no_servers(self):
  23         servcalc = self.make_calculator([1])
  24         self.assertEqual([], servcalc.servers_for_queue([]))
  25
  26     def test_easy_server_count(self):
  27         servcalc = self.make_calculator([1])
  28         servlist = self.calculate(servcalc, {'min_nodes': 3})
  29         self.assertEqual(3, len(servlist))
  30
  31     def test_default_5pct_ram_value_decrease(self):
  32         servcalc = self.make_calculator([1])
  33         servlist = self.calculate(servcalc, {'min_ram_mb_per_node': 128})
  34         self.assertEqual(0, len(servlist))
  35         servlist = self.calculate(servcalc, {'min_ram_mb_per_node': 121})
  36         self.assertEqual(1, len(servlist))
  37
  38     def test_custom_node_mem_scaling_factor(self):
  39         # Simulate a custom 'node_mem_scaling' config parameter by passing
  40         # the value to ServerCalculator
  41         servcalc = self.make_calculator([1], node_mem_scaling=0.5)
  42         servlist = self.calculate(servcalc, {'min_ram_mb_per_node': 128})
  43         self.assertEqual(0, len(servlist))
  44         servlist = self.calculate(servcalc, {'min_ram_mb_per_node': 64})
  45         self.assertEqual(1, len(servlist))
  46
  47     def test_implicit_server_count(self):
  48         servcalc = self.make_calculator([1])
  49         servlist = self.calculate(servcalc, {}, {'min_nodes': 3})
  50         self.assertEqual(4, len(servlist))
  51
  52     def test_bad_min_nodes_override(self):
  53         servcalc = self.make_calculator([1])
  54         servlist = self.calculate(servcalc,
  55                                   {'min_nodes': -2}, {'min_nodes': 'foo'})
  56         self.assertEqual(2, len(servlist))
  57
  58     def test_ignore_unsatisfiable_jobs(self):
  59         servcalc = self.make_calculator([1], max_nodes=9)
  60         servlist = self.calculate(servcalc,
  61                                   {'min_cores_per_node': 2},
  62                                   {'min_ram_mb_per_node': 256},
  63                                   {'min_nodes': 6},
  64                                   {'min_nodes': 12},
  65                                   {'min_scratch_mb_per_node': 300000})
  66         self.assertEqual(6, len(servlist))
  67
  68     def test_ignore_too_expensive_jobs(self):
  69         servcalc = self.make_calculator([1, 2], max_nodes=12, max_price=6)
  70         servlist = self.calculate(servcalc,
  71                                   {'min_cores_per_node': 1, 'min_nodes': 6})
  72         self.assertEqual(6, len(servlist))
  73
  74         servlist = self.calculate(servcalc,
  75                                   {'min_cores_per_node': 2, 'min_nodes': 6})
  76         self.assertEqual(0, len(servlist))
  77
  78     def test_job_requesting_max_nodes_accepted(self):
  79         servcalc = self.make_calculator([1], max_nodes=4)
  80         servlist = self.calculate(servcalc, {'min_nodes': 4})
  81         self.assertEqual(4, len(servlist))
  82
  83     def test_cheapest_size(self):
  84         servcalc = self.make_calculator([2, 4, 1, 3])
  85         self.assertEqual(testutil.MockSize(1), servcalc.cheapest_size())
  86
  87     def test_next_biggest(self):
  88         servcalc = self.make_calculator([1, 2, 4, 8])
  89         servlist = self.calculate(servcalc,
  90                                   {'min_cores_per_node': 3},
  91                                   {'min_cores_per_node': 6})
  92         self.assertEqual([servcalc.cloud_sizes[2].id,
  93                           servcalc.cloud_sizes[3].id],
  94                          [s.id for s in servlist])
  95
  96     def test_multiple_sizes(self):
  97         servcalc = self.make_calculator([1, 2])
  98         servlist = self.calculate(servcalc,
  99                                   {'min_cores_per_node': 2},
 100                                   {'min_cores_per_node': 1},
 101                                   {'min_cores_per_node': 1})
 102         self.assertEqual([servcalc.cloud_sizes[1].id,
 103                           servcalc.cloud_sizes[0].id,
 104                           servcalc.cloud_sizes[0].id],
 105                          [s.id for s in servlist])
 106
 107         servlist = self.calculate(servcalc,
 108                                   {'min_cores_per_node': 1},
 109                                   {'min_cores_per_node': 2},
 110                                   {'min_cores_per_node': 1})
 111         self.assertEqual([servcalc.cloud_sizes[0].id,
 112                           servcalc.cloud_sizes[1].id,
 113                           servcalc.cloud_sizes[0].id],
 114                          [s.id for s in servlist])
 115
 116         servlist = self.calculate(servcalc,
 117                                   {'min_cores_per_node': 1},
 118                                   {'min_cores_per_node': 1},
 119                                   {'min_cores_per_node': 2})
 120         self.assertEqual([servcalc.cloud_sizes[0].id,
 121                           servcalc.cloud_sizes[0].id,
 122                           servcalc.cloud_sizes[1].id],
 123                          [s.id for s in servlist])
 124
 125
 126
 127 class JobQueueMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
 128                                    unittest.TestCase):
 129     TEST_CLASS = jobqueue.JobQueueMonitorActor
 130
 131     class MockCalculator(object):
 132         @staticmethod
 133         def servers_for_queue(queue):
 134             return [testutil.MockSize(n) for n in queue]
 135
 136
 137     def build_monitor(self, side_effect, *args, **kwargs):
 138         super(JobQueueMonitorActorTestCase, self).build_monitor(*args, **kwargs)
 139         self.client.jobs().queue().execute.side_effect = side_effect
 140
 141     @mock.patch("subprocess.check_output")
 142     def test_subscribers_get_server_lists(self, mock_squeue):
 143         mock_squeue.return_value = ""
 144
 145         self.build_monitor([{'items': [1, 2]}], self.MockCalculator(), True, True)
 146         self.monitor.subscribe(self.subscriber).get(self.TIMEOUT)
 147         self.stop_proxy(self.monitor)
 148         self.subscriber.assert_called_with([testutil.MockSize(1),
 149                                             testutil.MockSize(2)])
 150
 151     @mock.patch("subprocess.check_output")
 152     def test_squeue_server_list(self, mock_squeue):
 153         mock_squeue.return_value = """1|1024|0|Resources|zzzzz-zzzzz-zzzzzzzzzzzzzzy
 154 2|1024|0|Resources|zzzzz-zzzzz-zzzzzzzzzzzzzzz
 155 """
 156
 157         super(JobQueueMonitorActorTestCase, self).build_monitor(jobqueue.ServerCalculator(
 158             [(testutil.MockSize(n), {'cores': n, 'ram': n*1024, 'scratch': n}) for n in range(1, 3)]),
 159                                                                 True, True)
 160         self.monitor.subscribe(self.subscriber).get(self.TIMEOUT)
 161         self.stop_proxy(self.monitor)
 162         self.subscriber.assert_called_with([testutil.MockSize(1),
 163                                             testutil.MockSize(2)])
 164
 165     @mock.patch("subprocess.check_output")
 166     def test_squeue_server_list_suffix(self, mock_squeue):
 167         mock_squeue.return_value = """1|1024M|0|ReqNodeNotAvail, UnavailableNod|zzzzz-zzzzz-zzzzzzzzzzzzzzy
 168 1|2G|0|ReqNodeNotAvail, UnavailableNod|zzzzz-zzzzz-zzzzzzzzzzzzzzz
 169 """
 170
 171         super(JobQueueMonitorActorTestCase, self).build_monitor(jobqueue.ServerCalculator(
 172             [(testutil.MockSize(n), {'cores': n, 'ram': n*1024, 'scratch': n}) for n in range(1, 3)]),
 173                                                                 True, True)
 174         self.monitor.subscribe(self.subscriber).get(self.TIMEOUT)
 175         self.stop_proxy(self.monitor)
 176         self.subscriber.assert_called_with([testutil.MockSize(1),
 177                                             testutil.MockSize(2)])
 178
 179     def test_coerce_to_mb(self):
 180         self.assertEqual(1, jobqueue.JobQueueMonitorActor.coerce_to_mb("1"))
 181         self.assertEqual(512, jobqueue.JobQueueMonitorActor.coerce_to_mb("512"))
 182         self.assertEqual(512, jobqueue.JobQueueMonitorActor.coerce_to_mb("512M"))
 183         self.assertEqual(1024, jobqueue.JobQueueMonitorActor.coerce_to_mb("1024M"))
 184         self.assertEqual(1024, jobqueue.JobQueueMonitorActor.coerce_to_mb("1G"))
 185         self.assertEqual(1536, jobqueue.JobQueueMonitorActor.coerce_to_mb("1.5G"))
 186         self.assertEqual(2048, jobqueue.JobQueueMonitorActor.coerce_to_mb("2G"))
 187         self.assertEqual(1025, jobqueue.JobQueueMonitorActor.coerce_to_mb("1025M"))
 188         self.assertEqual(1048576, jobqueue.JobQueueMonitorActor.coerce_to_mb("1T"))
 189         self.assertEqual(1572864, jobqueue.JobQueueMonitorActor.coerce_to_mb("1.5T"))
 190         self.assertEqual(1073741824, jobqueue.JobQueueMonitorActor.coerce_to_mb("1P"))
 191         self.assertEqual(1610612736, jobqueue.JobQueueMonitorActor.coerce_to_mb("1.5P"))
 192         self.assertEqual(0, jobqueue.JobQueueMonitorActor.coerce_to_mb("0"))
 193         self.assertEqual(0, jobqueue.JobQueueMonitorActor.coerce_to_mb("0M"))
 194         self.assertEqual(0, jobqueue.JobQueueMonitorActor.coerce_to_mb("0G"))
 195
 196
 197 if __name__ == '__main__':
 198     unittest.main()