services/nodemanager/tests/test_jobqueue.py

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