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_and_return_unsatisfiable_jobs(self):
  62         servcalc = self.make_calculator([1], max_nodes=9)
  63         servlist, u_jobs = 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         # Only unsatisfiable jobs are returned on u_jobs
  71         self.assertIn('zzzzz-jjjjj-000000000000000', u_jobs.keys())
  72         self.assertIn('zzzzz-jjjjj-000000000000001', u_jobs.keys())
  73         self.assertNotIn('zzzzz-jjjjj-000000000000002', u_jobs.keys())
  74         self.assertIn('zzzzz-jjjjj-000000000000003', u_jobs.keys())
  75         self.assertIn('zzzzz-jjjjj-000000000000004', u_jobs.keys())
  76
  77     def test_ignore_too_expensive_jobs(self):
  78         servcalc = self.make_calculator([1, 2], max_nodes=12, max_price=6)
  79         servlist, _ = self.calculate(servcalc,
  80                                      {'min_cores_per_node': 1, 'min_nodes': 6})
  81         self.assertEqual(6, len(servlist))
  82
  83         servlist, _ = self.calculate(servcalc,
  84                                      {'min_cores_per_node': 2, 'min_nodes': 6})
  85         self.assertEqual(0, len(servlist))
  86
  87     def test_job_requesting_max_nodes_accepted(self):
  88         servcalc = self.make_calculator([1], max_nodes=4)
  89         servlist, _ = self.calculate(servcalc, {'min_nodes': 4})
  90         self.assertEqual(4, len(servlist))
  91
  92     def test_cheapest_size(self):
  93         servcalc = self.make_calculator([2, 4, 1, 3])
  94         self.assertEqual(testutil.MockSize(1), servcalc.cheapest_size())
  95
  96     def test_next_biggest(self):
  97         servcalc = self.make_calculator([1, 2, 4, 8])
  98         servlist, _ = self.calculate(servcalc,
  99                                      {'min_cores_per_node': 3},
 100                                      {'min_cores_per_node': 6})
 101         self.assertEqual([servcalc.cloud_sizes[2].id,
 102                           servcalc.cloud_sizes[3].id],
 103                          [s.id for s in servlist])
 104
 105     def test_multiple_sizes(self):
 106         servcalc = self.make_calculator([1, 2])
 107         servlist, _ = self.calculate(servcalc,
 108                                      {'min_cores_per_node': 2},
 109                                      {'min_cores_per_node': 1},
 110                                      {'min_cores_per_node': 1})
 111         self.assertEqual([servcalc.cloud_sizes[1].id,
 112                           servcalc.cloud_sizes[0].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': 2},
 119                                      {'min_cores_per_node': 1})
 120         self.assertEqual([servcalc.cloud_sizes[0].id,
 121                           servcalc.cloud_sizes[1].id,
 122                           servcalc.cloud_sizes[0].id],
 123                          [s.id for s in servlist])
 124
 125         servlist, _ = self.calculate(servcalc,
 126                                      {'min_cores_per_node': 1},
 127                                      {'min_cores_per_node': 1},
 128                                      {'min_cores_per_node': 2})
 129         self.assertEqual([servcalc.cloud_sizes[0].id,
 130                           servcalc.cloud_sizes[0].id,
 131                           servcalc.cloud_sizes[1].id],
 132                          [s.id for s in servlist])
 133
 134
 135
 136 class JobQueueMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
 137                                    unittest.TestCase):
 138     TEST_CLASS = jobqueue.JobQueueMonitorActor
 139
 140
 141     class MockCalculator(object):
 142         @staticmethod
 143         def servers_for_queue(queue):
 144             return ([testutil.MockSize(n) for n in queue], {})
 145
 146
 147     class MockCalculatorUnsatisfiableJobs(object):
 148         @staticmethod
 149         def servers_for_queue(queue):
 150             return ([], {k["uuid"]: "Unsatisfiable job mock" for k in queue})
 151
 152
 153     def build_monitor(self, side_effect, *args, **kwargs):
 154         super(JobQueueMonitorActorTestCase, self).build_monitor(*args, **kwargs)
 155         self.client.jobs().queue().execute.side_effect = side_effect
 156
 157     @mock.patch("subprocess.check_call")
 158     @mock.patch("subprocess.check_output")
 159     def test_unsatisfiable_jobs(self, mock_squeue, mock_scancel):
 160         #mock_scancel.return_value = ""
 161         job_uuid = 'zzzzz-8i9sb-zzzzzzzzzzzzzzz'
 162         container_uuid = 'yyyyy-dz642-yyyyyyyyyyyyyyy'
 163         mock_squeue.return_value = "1|1024|0|Resources|" + container_uuid + "\n"
 164
 165         self.build_monitor([{'items': [{'uuid': job_uuid}]}],
 166                            self.MockCalculatorUnsatisfiableJobs(), True, True)
 167         self.monitor.subscribe(self.subscriber).get(self.TIMEOUT)
 168         self.stop_proxy(self.monitor)
 169         self.client.jobs().cancel.assert_called_with(uuid=job_uuid)
 170         mock_scancel.assert_called_with(['scancel', '--name='+container_uuid])
 171
 172     @mock.patch("subprocess.check_output")
 173     def test_subscribers_get_server_lists(self, mock_squeue):
 174         mock_squeue.return_value = ""
 175
 176         self.build_monitor([{'items': [1, 2]}], self.MockCalculator(), 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     @mock.patch("subprocess.check_output")
 183     def test_squeue_server_list(self, mock_squeue):
 184         mock_squeue.return_value = """1|1024|0|Resources|zzzzz-zzzzz-zzzzzzzzzzzzzzy
 185 2|1024|0|Resources|zzzzz-zzzzz-zzzzzzzzzzzzzzz
 186 """
 187
 188         super(JobQueueMonitorActorTestCase, self).build_monitor(jobqueue.ServerCalculator(
 189             [(testutil.MockSize(n), {'cores': n, 'ram': n*1024, 'scratch': n}) for n in range(1, 3)]),
 190                                                                 True, True)
 191         self.monitor.subscribe(self.subscriber).get(self.TIMEOUT)
 192         self.stop_proxy(self.monitor)
 193         self.subscriber.assert_called_with([testutil.MockSize(1),
 194                                             testutil.MockSize(2)])
 195
 196     @mock.patch("subprocess.check_output")
 197     def test_squeue_server_list_suffix(self, mock_squeue):
 198         mock_squeue.return_value = """1|1024M|0|ReqNodeNotAvail, UnavailableNod|zzzzz-zzzzz-zzzzzzzzzzzzzzy
 199 1|2G|0|ReqNodeNotAvail, UnavailableNod|zzzzz-zzzzz-zzzzzzzzzzzzzzz
 200 """
 201
 202         super(JobQueueMonitorActorTestCase, self).build_monitor(jobqueue.ServerCalculator(
 203             [(testutil.MockSize(n), {'cores': n, 'ram': n*1024, 'scratch': n}) for n in range(1, 3)]),
 204                                                                 True, True)
 205         self.monitor.subscribe(self.subscriber).get(self.TIMEOUT)
 206         self.stop_proxy(self.monitor)
 207         self.subscriber.assert_called_with([testutil.MockSize(1),
 208                                             testutil.MockSize(2)])
 209
 210     def test_coerce_to_mb(self):
 211         self.assertEqual(1, jobqueue.JobQueueMonitorActor.coerce_to_mb("1"))
 212         self.assertEqual(512, jobqueue.JobQueueMonitorActor.coerce_to_mb("512"))
 213         self.assertEqual(512, jobqueue.JobQueueMonitorActor.coerce_to_mb("512M"))
 214         self.assertEqual(1024, jobqueue.JobQueueMonitorActor.coerce_to_mb("1024M"))
 215         self.assertEqual(1024, jobqueue.JobQueueMonitorActor.coerce_to_mb("1G"))
 216         self.assertEqual(1536, jobqueue.JobQueueMonitorActor.coerce_to_mb("1.5G"))
 217         self.assertEqual(2048, jobqueue.JobQueueMonitorActor.coerce_to_mb("2G"))
 218         self.assertEqual(1025, jobqueue.JobQueueMonitorActor.coerce_to_mb("1025M"))
 219         self.assertEqual(1048576, jobqueue.JobQueueMonitorActor.coerce_to_mb("1T"))
 220         self.assertEqual(1572864, jobqueue.JobQueueMonitorActor.coerce_to_mb("1.5T"))
 221         self.assertEqual(1073741824, jobqueue.JobQueueMonitorActor.coerce_to_mb("1P"))
 222         self.assertEqual(1610612736, jobqueue.JobQueueMonitorActor.coerce_to_mb("1.5P"))
 223         self.assertEqual(0, jobqueue.JobQueueMonitorActor.coerce_to_mb("0"))
 224         self.assertEqual(0, jobqueue.JobQueueMonitorActor.coerce_to_mb("0M"))
 225         self.assertEqual(0, jobqueue.JobQueueMonitorActor.coerce_to_mb("0G"))
 226
 227
 228 if __name__ == '__main__':
 229     unittest.main()