services/nodemanager/tests/test_daemon.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import absolute_import, print_function
   4
   5 import time
   6 import unittest
   7
   8 import mock
   9 import pykka
  10
  11 import arvnodeman.daemon as nmdaemon
  12 from arvnodeman.jobqueue import ServerCalculator
  13 from arvnodeman.computenode.dispatch import ComputeNodeMonitorActor
  14 from . import testutil
  15 import logging
  16
  17 class NodeManagerDaemonActorTestCase(testutil.ActorTestMixin,
  18                                      unittest.TestCase):
  19     maxDiff = None
  20     def mock_node_start(self, **kwargs):
  21         # Make sure that every time the daemon starts a setup actor,
  22         # it gets a new mock object back.
  23         get_cloud_size = mock.MagicMock()
  24         get_cloud_size.get.return_value = kwargs["cloud_size"]
  25         mock_actor = mock.MagicMock()
  26         mock_proxy = mock.NonCallableMock(name='setup_mock_proxy',
  27                                           cloud_size=get_cloud_size,
  28                                           actor_ref=mock_actor)
  29         mock_actor.proxy.return_value = mock_proxy
  30         mock_actor.tell_proxy.return_value = mock_proxy
  31
  32         self.last_setup = mock_proxy
  33         return mock_actor
  34
  35     def mock_node_shutdown(self, **kwargs):
  36         # Make sure that every time the daemon starts a shutdown actor,
  37         # it gets a new mock object back.
  38         get_cloud_node = mock.MagicMock()
  39         if "node_monitor" in kwargs:
  40             get_cloud_node.get.return_value = kwargs["node_monitor"].proxy().cloud_node.get()
  41         mock_actor = mock.MagicMock()
  42         mock_proxy = mock.NonCallableMock(name='shutdown_mock_proxy',
  43                                           cloud_node=get_cloud_node,
  44                                           actor_ref=mock_actor)
  45
  46         mock_actor.proxy.return_value = mock_proxy
  47         self.last_shutdown = mock_proxy
  48
  49         return mock_actor
  50
  51     def make_daemon(self, cloud_nodes=[], arvados_nodes=[], want_sizes=[],
  52                     avail_sizes=[(testutil.MockSize(1), {"cores": 1})],
  53                     min_nodes=0, max_nodes=8,
  54                     shutdown_windows=[54, 5, 1],
  55                     max_total_price=None):
  56         for name in ['cloud_nodes', 'arvados_nodes', 'server_wishlist']:
  57             setattr(self, name + '_poller', mock.MagicMock(name=name + '_mock'))
  58         self.arv_factory = mock.MagicMock(name='arvados_mock')
  59         self.cloud_factory = mock.MagicMock(name='cloud_mock')
  60         self.cloud_factory().node_start_time.return_value = time.time()
  61         self.cloud_updates = mock.MagicMock(name='updates_mock')
  62         self.timer = testutil.MockTimer(deliver_immediately=False)
  63         self.cloud_factory().node_id.side_effect = lambda node: node.id
  64
  65         self.node_setup = mock.MagicMock(name='setup_mock')
  66         self.node_setup.start.side_effect = self.mock_node_start
  67         self.node_setup.reset_mock()
  68
  69         self.node_shutdown = mock.MagicMock(name='shutdown_mock')
  70         self.node_shutdown.start.side_effect = self.mock_node_shutdown
  71
  72         self.daemon = nmdaemon.NodeManagerDaemonActor.start(
  73             self.server_wishlist_poller, self.arvados_nodes_poller,
  74             self.cloud_nodes_poller, self.cloud_updates, self.timer,
  75             self.arv_factory, self.cloud_factory,
  76             shutdown_windows, ServerCalculator(avail_sizes),
  77             min_nodes, max_nodes, 600, 1800, 3600,
  78             self.node_setup, self.node_shutdown,
  79             max_total_price=max_total_price).proxy()
  80         if cloud_nodes is not None:
  81             self.daemon.update_cloud_nodes(cloud_nodes).get(self.TIMEOUT)
  82         if arvados_nodes is not None:
  83             self.daemon.update_arvados_nodes(arvados_nodes).get(self.TIMEOUT)
  84         if want_sizes is not None:
  85             self.daemon.update_server_wishlist(want_sizes).get(self.TIMEOUT)
  86
  87     def monitor_list(self):
  88         return pykka.ActorRegistry.get_by_class(ComputeNodeMonitorActor)
  89
  90     def monitored_arvados_nodes(self):
  91         pairings = []
  92         for future in [actor.proxy().arvados_node
  93                        for actor in self.monitor_list()]:
  94             try:
  95                 pairings.append(future.get(self.TIMEOUT))
  96             except pykka.ActorDeadError:
  97                 pass
  98         return pairings
  99
 100     def alive_monitor_count(self):
 101         return len(self.monitored_arvados_nodes())
 102
 103     def assertShutdownCancellable(self, expected=True):
 104         self.assertTrue(self.node_shutdown.start.called)
 105         self.assertIs(expected,
 106                       self.node_shutdown.start.call_args[1]['cancellable'],
 107                       "ComputeNodeShutdownActor incorrectly cancellable")
 108
 109     def test_easy_node_creation(self):
 110         size = testutil.MockSize(1)
 111         self.make_daemon(want_sizes=[size])
 112         self.stop_proxy(self.daemon)
 113         self.assertTrue(self.node_setup.start.called)
 114
 115     def check_monitors_arvados_nodes(self, *arv_nodes):
 116         self.assertItemsEqual(arv_nodes, self.monitored_arvados_nodes())
 117
 118     def test_node_pairing(self):
 119         cloud_node = testutil.cloud_node_mock(1)
 120         arv_node = testutil.arvados_node_mock(1)
 121         self.make_daemon([cloud_node], [arv_node])
 122         self.stop_proxy(self.daemon)
 123         self.check_monitors_arvados_nodes(arv_node)
 124
 125     def test_node_pairing_after_arvados_update(self):
 126         cloud_node = testutil.cloud_node_mock(2)
 127         self.make_daemon([cloud_node],
 128                          [testutil.arvados_node_mock(1, ip_address=None)])
 129         arv_node = testutil.arvados_node_mock(2)
 130         self.daemon.update_arvados_nodes([arv_node]).get(self.TIMEOUT)
 131         self.stop_proxy(self.daemon)
 132         self.check_monitors_arvados_nodes(arv_node)
 133
 134     def test_arvados_node_un_and_re_paired(self):
 135         # We need to create the Arvados node mock after spinning up the daemon
 136         # to make sure it's new enough to pair with the cloud node.
 137         self.make_daemon([testutil.cloud_node_mock(3)], arvados_nodes=None)
 138         arv_node = testutil.arvados_node_mock(3)
 139         self.daemon.update_arvados_nodes([arv_node]).get(self.TIMEOUT)
 140         self.check_monitors_arvados_nodes(arv_node)
 141         self.daemon.update_cloud_nodes([]).get(self.TIMEOUT)
 142         self.assertEqual(0, self.alive_monitor_count())
 143         self.daemon.update_cloud_nodes([testutil.cloud_node_mock(3)])
 144         self.stop_proxy(self.daemon)
 145         self.check_monitors_arvados_nodes(arv_node)
 146
 147     def test_old_arvados_node_not_double_assigned(self):
 148         arv_node = testutil.arvados_node_mock(3, age=9000)
 149         size = testutil.MockSize(3)
 150         self.make_daemon(arvados_nodes=[arv_node], avail_sizes=[(size, {"cores":1})])
 151         self.daemon.update_server_wishlist([size]).get(self.TIMEOUT)
 152         self.daemon.update_server_wishlist([size, size]).get(self.TIMEOUT)
 153         self.stop_proxy(self.daemon)
 154         used_nodes = [call[1].get('arvados_node')
 155                       for call in self.node_setup.start.call_args_list]
 156         self.assertEqual(2, len(used_nodes))
 157         self.assertIn(arv_node, used_nodes)
 158         self.assertIn(None, used_nodes)
 159
 160     def test_node_count_satisfied(self):
 161         self.make_daemon([testutil.cloud_node_mock()],
 162                          want_sizes=[testutil.MockSize(1)])
 163         self.stop_proxy(self.daemon)
 164         self.assertFalse(self.node_setup.start.called)
 165
 166     def test_dont_count_missing_as_busy(self):
 167         size = testutil.MockSize(1)
 168         self.make_daemon(cloud_nodes=[testutil.cloud_node_mock(1),
 169                                       testutil.cloud_node_mock(2)],
 170                          arvados_nodes=[testutil.arvados_node_mock(1),
 171                                       testutil.arvados_node_mock(2, last_ping_at='1970-01-01T01:02:03.04050607Z')],
 172                          want_sizes=[size, size])
 173         self.stop_proxy(self.daemon)
 174         self.assertTrue(self.node_setup.start.called)
 175
 176     def test_missing_counts_towards_max(self):
 177         size = testutil.MockSize(1)
 178         self.make_daemon(cloud_nodes=[testutil.cloud_node_mock(1),
 179                                       testutil.cloud_node_mock(2)],
 180                          arvados_nodes=[testutil.arvados_node_mock(1),
 181                                         testutil.arvados_node_mock(2, last_ping_at='1970-01-01T01:02:03.04050607Z')],
 182                          want_sizes=[size, size],
 183                          max_nodes=2)
 184         self.stop_proxy(self.daemon)
 185         self.assertFalse(self.node_setup.start.called)
 186
 187     def test_excess_counts_missing(self):
 188         size = testutil.MockSize(1)
 189         cloud_nodes = [testutil.cloud_node_mock(1), testutil.cloud_node_mock(2)]
 190         self.make_daemon(cloud_nodes=cloud_nodes,
 191                          arvados_nodes=[testutil.arvados_node_mock(1),
 192                                         testutil.arvados_node_mock(2, last_ping_at='1970-01-01T01:02:03.04050607Z')],
 193                          want_sizes=[size])
 194         self.assertEqual(2, self.alive_monitor_count())
 195         for mon_ref in self.monitor_list():
 196             self.daemon.node_can_shutdown(mon_ref.proxy()).get(self.TIMEOUT)
 197         self.assertEqual(1, self.node_shutdown.start.call_count)
 198
 199     def test_missing_shutdown_not_excess(self):
 200         size = testutil.MockSize(1)
 201         cloud_nodes = [testutil.cloud_node_mock(1), testutil.cloud_node_mock(2)]
 202         self.make_daemon(cloud_nodes=cloud_nodes,
 203                          arvados_nodes=[testutil.arvados_node_mock(1),
 204                                         testutil.arvados_node_mock(2, last_ping_at='1970-01-01T01:02:03.04050607Z')],
 205                          want_sizes=[size])
 206
 207         get_cloud_node = mock.MagicMock(name="get_cloud_node")
 208         get_cloud_node.get.return_value = cloud_nodes[1]
 209         mock_node_monitor = mock.MagicMock()
 210         mock_node_monitor.proxy.return_value = mock.NonCallableMock(cloud_node=get_cloud_node)
 211         mock_shutdown = self.node_shutdown.start(node_monitor=mock_node_monitor)
 212
 213         self.daemon.shutdowns.get()[cloud_nodes[1].id] = mock_shutdown.proxy()
 214         self.daemon.sizes_booting_shutdown.get()[cloud_nodes[1].id] = size
 215
 216         self.assertEqual(2, self.alive_monitor_count())
 217         for mon_ref in self.monitor_list():
 218             self.daemon.node_can_shutdown(mon_ref.proxy()).get(self.TIMEOUT)
 219         self.assertEqual(1, self.node_shutdown.start.call_count)
 220
 221     def test_booting_nodes_counted(self):
 222         cloud_node = testutil.cloud_node_mock(1)
 223         arv_node = testutil.arvados_node_mock(1)
 224         server_wishlist = [testutil.MockSize(1)] * 2
 225         self.make_daemon([cloud_node], [arv_node], server_wishlist)
 226         self.daemon.max_nodes.get(self.TIMEOUT)
 227         self.assertTrue(self.node_setup.start.called)
 228         self.daemon.update_server_wishlist(server_wishlist).get(self.TIMEOUT)
 229         self.stop_proxy(self.daemon)
 230         self.assertEqual(1, self.node_setup.start.call_count)
 231
 232     def test_boot_new_node_when_all_nodes_busy(self):
 233         size = testutil.MockSize(2)
 234         arv_node = testutil.arvados_node_mock(2, job_uuid=True)
 235         self.make_daemon([testutil.cloud_node_mock(2, size=size)], [arv_node],
 236                          [size], avail_sizes=[(size, {"cores":1})])
 237         self.stop_proxy(self.daemon)
 238         self.assertTrue(self.node_setup.start.called)
 239
 240     def test_boot_new_node_below_min_nodes(self):
 241         min_size = testutil.MockSize(1)
 242         wish_size = testutil.MockSize(3)
 243         avail_sizes = [(min_size, {"cores": 1}),
 244                        (wish_size, {"cores": 3})]
 245         self.make_daemon([], [], None, avail_sizes=avail_sizes, min_nodes=2)
 246         self.daemon.update_server_wishlist([wish_size]).get(self.TIMEOUT)
 247         self.daemon.update_cloud_nodes([]).get(self.TIMEOUT)
 248         self.daemon.update_server_wishlist([wish_size]).get(self.TIMEOUT)
 249         self.stop_proxy(self.daemon)
 250         self.assertEqual([wish_size, min_size],
 251                          [call[1].get('cloud_size')
 252                           for call in self.node_setup.start.call_args_list])
 253
 254     def test_no_new_node_when_ge_min_nodes_busy(self):
 255         cloud_nodes = [testutil.cloud_node_mock(n) for n in range(1, 4)]
 256         arv_nodes = [testutil.arvados_node_mock(n, job_uuid=True)
 257                      for n in range(1, 4)]
 258         self.make_daemon(cloud_nodes, arv_nodes, [], min_nodes=2)
 259         self.stop_proxy(self.daemon)
 260         self.assertEqual(0, self.node_setup.start.call_count)
 261
 262     def test_no_new_node_when_max_nodes_busy(self):
 263         self.make_daemon([testutil.cloud_node_mock(3)],
 264                          [testutil.arvados_node_mock(3, job_uuid=True)],
 265                          [testutil.MockSize(3)],
 266                          max_nodes=1)
 267         self.stop_proxy(self.daemon)
 268         self.assertFalse(self.node_setup.start.called)
 269
 270     def start_node_boot(self, cloud_node=None, arv_node=None, id_num=1):
 271         if cloud_node is None:
 272             cloud_node = testutil.cloud_node_mock(id_num)
 273         if arv_node is None:
 274             arv_node = testutil.arvados_node_mock(id_num)
 275         self.make_daemon(want_sizes=[testutil.MockSize(id_num)],
 276                          avail_sizes=[(testutil.MockSize(id_num), {"cores":1})])
 277         self.daemon.max_nodes.get(self.TIMEOUT)
 278         self.assertEqual(1, self.node_setup.start.call_count)
 279         self.last_setup.cloud_node.get.return_value = cloud_node
 280         self.last_setup.arvados_node.get.return_value = arv_node
 281         return self.last_setup
 282
 283     def test_no_new_node_when_booted_node_not_usable(self):
 284         cloud_node = testutil.cloud_node_mock(4)
 285         arv_node = testutil.arvados_node_mock(4, crunch_worker_state='down')
 286         setup = self.start_node_boot(cloud_node, arv_node)
 287         self.daemon.node_up(setup).get(self.TIMEOUT)
 288         self.assertEqual(1, self.alive_monitor_count())
 289         self.daemon.update_cloud_nodes([cloud_node])
 290         self.daemon.update_arvados_nodes([arv_node])
 291         self.daemon.update_server_wishlist(
 292             [testutil.MockSize(1)]).get(self.TIMEOUT)
 293         self.stop_proxy(self.daemon)
 294         self.assertEqual(1, self.node_setup.start.call_count)
 295
 296     def test_no_duplication_when_booting_node_listed_fast(self):
 297         # Test that we don't start two ComputeNodeMonitorActors when
 298         # we learn about a booting node through a listing before we
 299         # get the "node up" message from CloudNodeSetupActor.
 300         cloud_node = testutil.cloud_node_mock(1)
 301         setup = self.start_node_boot(cloud_node)
 302         self.daemon.update_cloud_nodes([cloud_node])
 303         self.daemon.node_up(setup).get(self.TIMEOUT)
 304         self.assertEqual(1, self.alive_monitor_count())
 305
 306     def test_no_duplication_when_booted_node_listed(self):
 307         cloud_node = testutil.cloud_node_mock(2)
 308         setup = self.start_node_boot(cloud_node, id_num=2)
 309         self.daemon.node_up(setup)
 310         self.daemon.update_cloud_nodes([cloud_node]).get(self.TIMEOUT)
 311         self.assertEqual(1, self.alive_monitor_count())
 312
 313     def test_node_counted_after_boot_with_slow_listing(self):
 314         # Test that, after we boot a compute node, we assume it exists
 315         # even it doesn't appear in the listing (e.g., because of delays
 316         # propagating tags).
 317         setup = self.start_node_boot()
 318         self.daemon.node_up(setup).get(self.TIMEOUT)
 319         self.assertEqual(1, self.alive_monitor_count())
 320         self.daemon.update_cloud_nodes([]).get(self.TIMEOUT)
 321         self.assertEqual(1, self.alive_monitor_count())
 322
 323     def test_booted_unlisted_node_counted(self):
 324         setup = self.start_node_boot(id_num=1)
 325         self.daemon.node_up(setup)
 326         self.daemon.update_server_wishlist(
 327             [testutil.MockSize(1)]).get(self.TIMEOUT)
 328         self.stop_proxy(self.daemon)
 329         self.assertEqual(1, self.node_setup.start.call_count)
 330
 331     def test_booted_node_can_shutdown(self):
 332         setup = self.start_node_boot()
 333         self.daemon.node_up(setup).get(self.TIMEOUT)
 334         self.assertEqual(1, self.alive_monitor_count())
 335         monitor = self.monitor_list()[0].proxy()
 336         self.daemon.update_server_wishlist([])
 337         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 338         self.stop_proxy(self.daemon)
 339         self.assertTrue(self.node_shutdown.start.called,
 340                         "daemon did not shut down booted node on offer")
 341
 342     def test_booted_node_lifecycle(self):
 343         cloud_node = testutil.cloud_node_mock(6)
 344         setup = self.start_node_boot(cloud_node, id_num=6)
 345         self.daemon.node_up(setup).get(self.TIMEOUT)
 346         self.assertEqual(1, self.alive_monitor_count())
 347         monitor = self.monitor_list()[0].proxy()
 348         self.daemon.update_server_wishlist([])
 349         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 350         self.assertShutdownCancellable(True)
 351         shutdown = self.node_shutdown.start().proxy()
 352         shutdown.cloud_node.get.return_value = cloud_node
 353         self.daemon.node_finished_shutdown(shutdown).get(self.TIMEOUT)
 354         self.assertTrue(shutdown.stop.called,
 355                         "shutdown actor not stopped after finishing")
 356         self.assertTrue(monitor.actor_ref.actor_stopped.wait(self.TIMEOUT),
 357                         "monitor for booted node not stopped after shutdown")
 358         self.daemon.update_server_wishlist(
 359             [testutil.MockSize(2)]).get(self.TIMEOUT)
 360         self.stop_proxy(self.daemon)
 361         self.assertTrue(self.node_setup.start.called,
 362                         "second node not started after booted node stopped")
 363
 364     def test_booted_node_shut_down_when_never_listed(self):
 365         setup = self.start_node_boot()
 366         self.daemon.node_up(setup).get(self.TIMEOUT)
 367         self.assertEqual(1, self.alive_monitor_count())
 368         self.assertFalse(self.node_shutdown.start.called)
 369         self.timer.deliver()
 370         self.stop_proxy(self.daemon)
 371         self.assertShutdownCancellable(False)
 372
 373     def test_booted_node_shut_down_when_never_paired(self):
 374         cloud_node = testutil.cloud_node_mock(2)
 375         setup = self.start_node_boot(cloud_node)
 376         self.daemon.node_up(setup).get(self.TIMEOUT)
 377         self.assertEqual(1, self.alive_monitor_count())
 378         self.daemon.update_cloud_nodes([cloud_node])
 379         self.timer.deliver()
 380         self.stop_proxy(self.daemon)
 381         self.assertShutdownCancellable(False)
 382
 383     def test_booted_node_shut_down_when_never_working(self):
 384         cloud_node = testutil.cloud_node_mock(4)
 385         arv_node = testutil.arvados_node_mock(4, crunch_worker_state='down')
 386         setup = self.start_node_boot(cloud_node, arv_node)
 387         self.daemon.node_up(setup).get(self.TIMEOUT)
 388         self.assertEqual(1, self.alive_monitor_count())
 389         self.daemon.update_cloud_nodes([cloud_node])
 390         self.daemon.update_arvados_nodes([arv_node]).get(self.TIMEOUT)
 391         self.timer.deliver()
 392         self.stop_proxy(self.daemon)
 393         self.assertShutdownCancellable(False)
 394
 395     def test_node_that_pairs_not_considered_failed_boot(self):
 396         cloud_node = testutil.cloud_node_mock(3)
 397         arv_node = testutil.arvados_node_mock(3)
 398         setup = self.start_node_boot(cloud_node, arv_node)
 399         self.daemon.node_up(setup).get(self.TIMEOUT)
 400         self.assertEqual(1, self.alive_monitor_count())
 401         self.daemon.update_cloud_nodes([cloud_node])
 402         self.daemon.update_arvados_nodes([arv_node]).get(self.TIMEOUT)
 403         self.timer.deliver()
 404         self.stop_proxy(self.daemon)
 405         self.assertFalse(self.node_shutdown.start.called)
 406
 407     def test_node_that_pairs_busy_not_considered_failed_boot(self):
 408         cloud_node = testutil.cloud_node_mock(5)
 409         arv_node = testutil.arvados_node_mock(5, job_uuid=True)
 410         setup = self.start_node_boot(cloud_node, arv_node)
 411         self.daemon.node_up(setup).get(self.TIMEOUT)
 412         self.assertEqual(1, self.alive_monitor_count())
 413         self.daemon.update_cloud_nodes([cloud_node])
 414         self.daemon.update_arvados_nodes([arv_node]).get(self.TIMEOUT)
 415         self.timer.deliver()
 416         self.stop_proxy(self.daemon)
 417         self.assertFalse(self.node_shutdown.start.called)
 418
 419     def test_booting_nodes_shut_down(self):
 420         self.make_daemon(want_sizes=[testutil.MockSize(1)])
 421         self.daemon.update_server_wishlist([]).get(self.TIMEOUT)
 422         self.stop_proxy(self.daemon)
 423         self.assertTrue(self.last_setup.stop_if_no_cloud_node.called)
 424
 425     def test_all_booting_nodes_tried_to_shut_down(self):
 426         size = testutil.MockSize(2)
 427         self.make_daemon(want_sizes=[size], avail_sizes=[(size, {"cores":1})])
 428         self.daemon.max_nodes.get(self.TIMEOUT)
 429         setup1 = self.last_setup
 430         setup1.stop_if_no_cloud_node().get.return_value = False
 431         setup1.stop_if_no_cloud_node.reset_mock()
 432         self.daemon.update_server_wishlist([size, size]).get(self.TIMEOUT)
 433         self.daemon.max_nodes.get(self.TIMEOUT)
 434         self.assertIsNot(setup1, self.last_setup)
 435         self.last_setup.stop_if_no_cloud_node().get.return_value = True
 436         self.last_setup.stop_if_no_cloud_node.reset_mock()
 437         self.daemon.update_server_wishlist([]).get(self.TIMEOUT)
 438         self.daemon.max_nodes.get(self.TIMEOUT)
 439         self.stop_proxy(self.daemon)
 440         self.assertEqual(1, self.last_setup.stop_if_no_cloud_node.call_count)
 441         self.assertTrue(setup1.stop_if_no_cloud_node.called)
 442
 443     def test_shutdown_declined_at_wishlist_capacity(self):
 444         cloud_node = testutil.cloud_node_mock(1)
 445         size = testutil.MockSize(1)
 446         self.make_daemon(cloud_nodes=[cloud_node], want_sizes=[size])
 447         self.assertEqual(1, self.alive_monitor_count())
 448         monitor = self.monitor_list()[0].proxy()
 449         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 450         self.stop_proxy(self.daemon)
 451         self.assertFalse(self.node_shutdown.start.called)
 452
 453     def test_shutdown_declined_below_min_nodes(self):
 454         cloud_node = testutil.cloud_node_mock(1)
 455         self.make_daemon(cloud_nodes=[cloud_node], min_nodes=1)
 456         self.assertEqual(1, self.alive_monitor_count())
 457         monitor = self.monitor_list()[0].proxy()
 458         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 459         self.stop_proxy(self.daemon)
 460         self.assertFalse(self.node_shutdown.start.called)
 461
 462     def test_shutdown_accepted_below_capacity(self):
 463         self.make_daemon(cloud_nodes=[testutil.cloud_node_mock()])
 464         self.assertEqual(1, self.alive_monitor_count())
 465         monitor = self.monitor_list()[0].proxy()
 466         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 467         self.stop_proxy(self.daemon)
 468         self.assertTrue(self.node_shutdown.start.called)
 469
 470     def test_shutdown_declined_when_idle_and_job_queued(self):
 471         cloud_nodes = [testutil.cloud_node_mock(n) for n in [3, 4]]
 472         arv_nodes = [testutil.arvados_node_mock(3, job_uuid=True),
 473                      testutil.arvados_node_mock(4, job_uuid=None)]
 474         self.make_daemon(cloud_nodes, arv_nodes, [testutil.MockSize(1)])
 475         self.assertEqual(2, self.alive_monitor_count())
 476         for mon_ref in self.monitor_list():
 477             monitor = mon_ref.proxy()
 478             if monitor.cloud_node.get(self.TIMEOUT) is cloud_nodes[-1]:
 479                 break
 480         else:
 481             self.fail("monitor for idle node not found")
 482         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 483         self.stop_proxy(self.daemon)
 484         self.assertFalse(self.node_shutdown.start.called)
 485
 486     def test_node_shutdown_after_cancelled_shutdown(self):
 487         cloud_node = testutil.cloud_node_mock(5)
 488         self.make_daemon([cloud_node], [testutil.arvados_node_mock(5)])
 489         self.assertEqual(1, self.alive_monitor_count())
 490         monitor = self.monitor_list()[0].proxy()
 491         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 492         self.last_shutdown.success.get.return_value = False
 493         self.daemon.node_finished_shutdown(self.last_shutdown).get(self.TIMEOUT)
 494         self.assertEqual(1, self.alive_monitor_count())
 495
 496         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 497         self.last_shutdown.success.get.return_value = True
 498         self.last_shutdown.stop.side_effect = lambda: monitor.stop()
 499         self.daemon.node_finished_shutdown(self.last_shutdown).get(self.TIMEOUT)
 500         self.assertEqual(0, self.alive_monitor_count())
 501
 502     def test_broken_node_blackholed_after_cancelled_shutdown(self):
 503         size = testutil.MockSize(8)
 504         cloud_node = testutil.cloud_node_mock(8, size=size)
 505         wishlist = [size]
 506         self.make_daemon([cloud_node], [testutil.arvados_node_mock(8)],
 507                          wishlist, avail_sizes=[(size, {"cores":1})])
 508         self.assertEqual(1, self.alive_monitor_count())
 509         self.assertFalse(self.node_setup.start.called)
 510         monitor = self.monitor_list()[0].proxy()
 511         shutdown_proxy = self.node_shutdown.start().proxy
 512         shutdown_proxy().cloud_node.get.return_value = cloud_node
 513         shutdown_proxy().success.get.return_value = False
 514         shutdown_proxy().cancel_reason.get.return_value = self.node_shutdown.NODE_BROKEN
 515         self.daemon.update_server_wishlist([]).get(self.TIMEOUT)
 516         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 517         self.daemon.node_finished_shutdown(shutdown_proxy()).get(self.TIMEOUT)
 518         self.daemon.update_cloud_nodes([cloud_node]).get(self.TIMEOUT)
 519         self.daemon.update_server_wishlist(wishlist).get(self.TIMEOUT)
 520         self.stop_proxy(self.daemon)
 521         self.assertEqual(1, self.node_setup.start.call_count)
 522
 523     def test_nodes_shutting_down_replaced_below_max_nodes(self):
 524         size = testutil.MockSize(6)
 525         cloud_node = testutil.cloud_node_mock(6, size=size)
 526         self.make_daemon([cloud_node], [testutil.arvados_node_mock(6)],
 527                          avail_sizes=[(size, {"cores":1})])
 528         self.assertEqual(1, self.alive_monitor_count())
 529         monitor = self.monitor_list()[0].proxy()
 530         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 531         self.assertTrue(self.node_shutdown.start.called)
 532         self.daemon.update_server_wishlist(
 533             [testutil.MockSize(6)]).get(self.TIMEOUT)
 534         self.stop_proxy(self.daemon)
 535         self.assertTrue(self.node_setup.start.called)
 536
 537     def test_nodes_shutting_down_not_replaced_at_max_nodes(self):
 538         cloud_node = testutil.cloud_node_mock(7)
 539         self.make_daemon([cloud_node], [testutil.arvados_node_mock(7)],
 540                          max_nodes=1)
 541         self.assertEqual(1, self.alive_monitor_count())
 542         monitor = self.monitor_list()[0].proxy()
 543         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 544         self.assertTrue(self.node_shutdown.start.called)
 545         self.daemon.update_server_wishlist(
 546             [testutil.MockSize(7)]).get(self.TIMEOUT)
 547         self.stop_proxy(self.daemon)
 548         self.assertFalse(self.node_setup.start.called)
 549
 550     def test_nodes_shutting_down_count_against_excess(self):
 551         size = testutil.MockSize(8)
 552         cloud_nodes = [testutil.cloud_node_mock(n, size=size) for n in [8, 9]]
 553         arv_nodes = [testutil.arvados_node_mock(n, size=size) for n in [8, 9]]
 554         self.make_daemon(cloud_nodes, arv_nodes, [size],
 555                          avail_sizes=[(size, {"cores":1})])
 556         self.assertEqual(2, self.alive_monitor_count())
 557         for mon_ref in self.monitor_list():
 558             self.daemon.node_can_shutdown(mon_ref.proxy()).get(self.TIMEOUT)
 559         self.assertEqual(1, self.node_shutdown.start.call_count)
 560
 561     def test_clean_shutdown_waits_for_node_setup_finish(self):
 562         new_node = self.start_node_boot()
 563         new_node.stop_if_no_cloud_node().get.return_value = False
 564         new_node.stop_if_no_cloud_node.reset_mock()
 565         self.daemon.shutdown().get(self.TIMEOUT)
 566         self.assertTrue(new_node.stop_if_no_cloud_node.called)
 567         self.daemon.node_up(new_node).get(self.TIMEOUT)
 568         self.assertTrue(new_node.stop.called)
 569         self.timer.deliver()
 570         self.assertTrue(
 571             self.daemon.actor_ref.actor_stopped.wait(self.TIMEOUT))
 572
 573     def test_wishlist_ignored_after_shutdown(self):
 574         new_node = self.start_node_boot()
 575         new_node.stop_if_no_cloud_node().get.return_value = False
 576         new_node.stop_if_no_cloud_node.reset_mock()
 577         self.daemon.shutdown().get(self.TIMEOUT)
 578         size = testutil.MockSize(2)
 579         self.daemon.update_server_wishlist([size] * 2).get(self.TIMEOUT)
 580         self.timer.deliver()
 581         self.stop_proxy(self.daemon)
 582         self.assertEqual(1, self.node_setup.start.call_count)
 583
 584     def test_shutdown_actor_stopped_when_cloud_node_delisted(self):
 585         self.make_daemon(cloud_nodes=[testutil.cloud_node_mock()])
 586         self.assertEqual(1, self.alive_monitor_count())
 587         monitor = self.monitor_list()[0].proxy()
 588         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 589         self.daemon.update_cloud_nodes([]).get(self.TIMEOUT)
 590         self.stop_proxy(self.daemon)
 591         self.assertEqual(
 592             1, self.last_shutdown.stop.call_count)
 593
 594     def test_shutdown_actor_cleanup_copes_with_dead_actors(self):
 595         self.make_daemon(cloud_nodes=[testutil.cloud_node_mock()])
 596         self.assertEqual(1, self.alive_monitor_count())
 597         monitor = self.monitor_list()[0].proxy()
 598         self.daemon.node_can_shutdown(monitor).get(self.TIMEOUT)
 599         # We're mainly testing that update_cloud_nodes catches and handles
 600         # the ActorDeadError.
 601         self.last_shutdown.stop.side_effect = pykka.ActorDeadError
 602         self.daemon.update_cloud_nodes([]).get(self.TIMEOUT)
 603         self.stop_proxy(self.daemon)
 604         self.assertEqual(1, self.last_shutdown.stop.call_count)
 605
 606     def busywait(self, f):
 607         n = 0
 608         while not f() and n < 10:
 609             time.sleep(.1)
 610             n += 1
 611         self.assertTrue(f())
 612
 613     def test_node_create_two_sizes(self):
 614         small = testutil.MockSize(1)
 615         big = testutil.MockSize(2)
 616         avail_sizes = [(testutil.MockSize(1), {"cores":1}),
 617                         (testutil.MockSize(2), {"cores":2})]
 618         self.make_daemon(want_sizes=[small, small, small, big],
 619                          avail_sizes=avail_sizes, max_nodes=4)
 620
 621         # the daemon runs in another thread, so we need to wait and see
 622         # if it does all the work we're expecting it to do before stopping it.
 623         self.busywait(lambda: self.node_setup.start.call_count == 4)
 624         booting = self.daemon.booting.get(self.TIMEOUT)
 625         self.stop_proxy(self.daemon)
 626         sizecounts = {a[0].id: 0 for a in avail_sizes}
 627         for b in booting.itervalues():
 628             sizecounts[b.cloud_size.get().id] += 1
 629         logging.info(sizecounts)
 630         self.assertEqual(3, sizecounts[small.id])
 631         self.assertEqual(1, sizecounts[big.id])
 632
 633     def test_node_max_nodes_two_sizes(self):
 634         small = testutil.MockSize(1)
 635         big = testutil.MockSize(2)
 636         avail_sizes = [(testutil.MockSize(1), {"cores":1}),
 637                         (testutil.MockSize(2), {"cores":2})]
 638         self.make_daemon(want_sizes=[small, small, small, big],
 639                          avail_sizes=avail_sizes, max_nodes=3)
 640
 641         # the daemon runs in another thread, so we need to wait and see
 642         # if it does all the work we're expecting it to do before stopping it.
 643         self.busywait(lambda: self.node_setup.start.call_count == 3)
 644         booting = self.daemon.booting.get(self.TIMEOUT)
 645         self.stop_proxy(self.daemon)
 646         sizecounts = {a[0].id: 0 for a in avail_sizes}
 647         for b in booting.itervalues():
 648             sizecounts[b.cloud_size.get().id] += 1
 649         self.assertEqual(2, sizecounts[small.id])
 650         self.assertEqual(1, sizecounts[big.id])
 651
 652     def test_wishlist_reconfigure(self):
 653         small = testutil.MockSize(1)
 654         big = testutil.MockSize(2)
 655         avail_sizes = [(small, {"cores":1}), (big, {"cores":2})]
 656
 657         self.make_daemon(cloud_nodes=[testutil.cloud_node_mock(1, small),
 658                                       testutil.cloud_node_mock(2, small),
 659                                       testutil.cloud_node_mock(3, big)],
 660                          arvados_nodes=[testutil.arvados_node_mock(1),
 661                                         testutil.arvados_node_mock(2),
 662                                         testutil.arvados_node_mock(3)],
 663                          want_sizes=[small, small, big],
 664                          avail_sizes=avail_sizes)
 665
 666         self.daemon.update_server_wishlist([small, big, big]).get(self.TIMEOUT)
 667
 668         self.assertEqual(0, self.node_shutdown.start.call_count)
 669
 670         for c in self.daemon.cloud_nodes.get().nodes.itervalues():
 671             self.daemon.node_can_shutdown(c.actor)
 672
 673         booting = self.daemon.booting.get()
 674         shutdowns = self.daemon.shutdowns.get()
 675
 676         self.stop_proxy(self.daemon)
 677
 678         self.assertEqual(1, self.node_setup.start.call_count)
 679         self.assertEqual(1, self.node_shutdown.start.call_count)
 680
 681         # booting a new big node
 682         sizecounts = {a[0].id: 0 for a in avail_sizes}
 683         for b in booting.itervalues():
 684             sizecounts[b.cloud_size.get().id] += 1
 685         self.assertEqual(0, sizecounts[small.id])
 686         self.assertEqual(1, sizecounts[big.id])
 687
 688         # shutting down a small node
 689         sizecounts = {a[0].id: 0 for a in avail_sizes}
 690         for b in shutdowns.itervalues():
 691             sizecounts[b.cloud_node.get().size.id] += 1
 692         self.assertEqual(1, sizecounts[small.id])
 693         self.assertEqual(0, sizecounts[big.id])
 694
 695     def test_node_max_price(self):
 696         small = testutil.MockSize(1)
 697         big = testutil.MockSize(2)
 698         avail_sizes = [(testutil.MockSize(1), {"cores":1, "price":1}),
 699                         (testutil.MockSize(2), {"cores":2, "price":2})]
 700         self.make_daemon(want_sizes=[small, small, small, big],
 701                          avail_sizes=avail_sizes,
 702                          max_nodes=4,
 703                          max_total_price=4)
 704         # the daemon runs in another thread, so we need to wait and see
 705         # if it does all the work we're expecting it to do before stopping it.
 706         self.busywait(lambda: self.node_setup.start.call_count == 3)
 707         booting = self.daemon.booting.get()
 708         self.stop_proxy(self.daemon)
 709
 710         sizecounts = {a[0].id: 0 for a in avail_sizes}
 711         for b in booting.itervalues():
 712             sizecounts[b.cloud_size.get().id] += 1
 713         logging.info(sizecounts)
 714
 715         # Booting 3 small nodes and not booting a big node would also partially
 716         # satisfy the wishlist and come in under the price cap, however the way
 717         # the update_server_wishlist() currently works effectively results in a
 718         # round-robin creation of one node of each size in the wishlist, so
 719         # test for that.
 720         self.assertEqual(2, sizecounts[small.id])
 721         self.assertEqual(1, sizecounts[big.id])