services/nodemanager/arvnodeman/computenode/dispatch/__init__.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 functools
   9 import logging
  10 import time
  11 import re
  12
  13 import libcloud.common.types as cloud_types
  14 from libcloud.common.exceptions import BaseHTTPError
  15
  16 import pykka
  17
  18 from .. import \
  19     arvados_node_fqdn, arvados_node_mtime, arvados_timestamp, timestamp_fresh, \
  20     arvados_node_missing, RetryMixin
  21 from ...clientactor import _notify_subscribers
  22 from ... import config
  23 from ... import status
  24 from .transitions import transitions
  25
  26 QuotaExceeded = "QuotaExceeded"
  27
  28 class ComputeNodeStateChangeBase(config.actor_class, RetryMixin):
  29     """Base class for actors that change a compute node's state.
  30
  31     This base class takes care of retrying changes and notifying
  32     subscribers when the change is finished.
  33     """
  34     def __init__(self, cloud_client, arvados_client, timer_actor,
  35                  retry_wait, max_retry_wait):
  36         super(ComputeNodeStateChangeBase, self).__init__()
  37         RetryMixin.__init__(self, retry_wait, max_retry_wait,
  38                             None, cloud_client, timer_actor)
  39         self._later = self.actor_ref.tell_proxy()
  40         self._arvados = arvados_client
  41         self.subscribers = set()
  42
  43     def _set_logger(self):
  44         self._logger = logging.getLogger("%s.%s" % (self.__class__.__name__, self.actor_urn[33:]))
  45
  46     def on_start(self):
  47         self._set_logger()
  48
  49     def _finished(self):
  50         if self.subscribers is None:
  51             raise Exception("Actor tried to finish twice")
  52         _notify_subscribers(self.actor_ref.proxy(), self.subscribers)
  53         self.subscribers = None
  54         self._logger.info("finished")
  55
  56     def subscribe(self, subscriber):
  57         if self.subscribers is None:
  58             try:
  59                 subscriber(self.actor_ref.proxy())
  60             except pykka.ActorDeadError:
  61                 pass
  62         else:
  63             self.subscribers.add(subscriber)
  64
  65     def _clean_arvados_node(self, arvados_node, explanation):
  66         return self._arvados.nodes().update(
  67             uuid=arvados_node['uuid'],
  68             body={'hostname': None,
  69                   'ip_address': None,
  70                   'slot_number': None,
  71                   'first_ping_at': None,
  72                   'last_ping_at': None,
  73                   'properties': {},
  74                   'info': {'ec2_instance_id': None,
  75                            'last_action': explanation}},
  76             ).execute()
  77
  78     @staticmethod
  79     def _finish_on_exception(orig_func):
  80         @functools.wraps(orig_func)
  81         def finish_wrapper(self, *args, **kwargs):
  82             try:
  83                 return orig_func(self, *args, **kwargs)
  84             except Exception as error:
  85                 self._logger.error("Actor error %s", error)
  86                 self._finished()
  87         return finish_wrapper
  88
  89
  90 class ComputeNodeSetupActor(ComputeNodeStateChangeBase):
  91     """Actor to create and set up a cloud compute node.
  92
  93     This actor prepares an Arvados node record for a new compute node
  94     (either creating one or cleaning one passed in), then boots the
  95     actual compute node.  It notifies subscribers when the cloud node
  96     is successfully created (the last step in the process for Node
  97     Manager to handle).
  98     """
  99     def __init__(self, timer_actor, arvados_client, cloud_client,
 100                  cloud_size, arvados_node=None,
 101                  retry_wait=1, max_retry_wait=180):
 102         super(ComputeNodeSetupActor, self).__init__(
 103             cloud_client, arvados_client, timer_actor,
 104             retry_wait, max_retry_wait)
 105         self.cloud_size = cloud_size
 106         self.arvados_node = None
 107         self.cloud_node = None
 108         self.error = None
 109         if arvados_node is None:
 110             self._later.create_arvados_node()
 111         else:
 112             self._later.prepare_arvados_node(arvados_node)
 113
 114     @ComputeNodeStateChangeBase._finish_on_exception
 115     @RetryMixin._retry(config.ARVADOS_ERRORS)
 116     def create_arvados_node(self):
 117         self.arvados_node = self._arvados.nodes().create(
 118             body={}, assign_slot=True).execute()
 119         self._later.create_cloud_node()
 120
 121     @ComputeNodeStateChangeBase._finish_on_exception
 122     @RetryMixin._retry(config.ARVADOS_ERRORS)
 123     def prepare_arvados_node(self, node):
 124         self._clean_arvados_node(node, "Prepared by Node Manager")
 125         self.arvados_node = self._arvados.nodes().update(
 126             uuid=node['uuid'], body={}, assign_slot=True).execute()
 127         self._later.create_cloud_node()
 128
 129     @ComputeNodeStateChangeBase._finish_on_exception
 130     @RetryMixin._retry()
 131     def create_cloud_node(self):
 132         self._logger.info("Sending create_node request for node size %s.",
 133                           self.cloud_size.name)
 134         try:
 135             self.cloud_node = self._cloud.create_node(self.cloud_size,
 136                                                       self.arvados_node)
 137         except BaseHTTPError as e:
 138             if e.code == 429 or "RequestLimitExceeded" in e.message:
 139                 # Don't consider API rate limits to be quota errors.
 140                 # re-raise so the Retry logic applies.
 141                 raise
 142
 143             # The set of possible error codes / messages isn't documented for
 144             # all clouds, so use a keyword heuristic to determine if the
 145             # failure is likely due to a quota.
 146             if re.search(r'(exceed|quota|limit)', e.message, re.I):
 147                 self.error = QuotaExceeded
 148                 self._logger.warning("Quota exceeded: %s", e)
 149                 self._finished()
 150                 return
 151             else:
 152                 # Something else happened, re-raise so the Retry logic applies.
 153                 raise
 154         except Exception as e:
 155             raise
 156
 157         # The information included in the node size object we get from libcloud
 158         # is inconsistent between cloud drivers.  Replace libcloud NodeSize
 159         # object with compatible CloudSizeWrapper object which merges the size
 160         # info reported from the cloud with size information from the
 161         # configuration file.
 162         self.cloud_node.size = self.cloud_size
 163
 164         self._logger.info("Cloud node %s created.", self.cloud_node.id)
 165         self._later.update_arvados_node_properties()
 166
 167     @ComputeNodeStateChangeBase._finish_on_exception
 168     @RetryMixin._retry(config.ARVADOS_ERRORS)
 169     def update_arvados_node_properties(self):
 170         """Tell Arvados some details about the cloud node.
 171
 172         Currently we only include size/price from our request, which
 173         we already knew before create_cloud_node(), but doing it here
 174         gives us an opportunity to provide more detail from
 175         self.cloud_node, too.
 176         """
 177         self.arvados_node['properties']['cloud_node'] = {
 178             # Note this 'size' is the node size we asked the cloud
 179             # driver to create -- not necessarily equal to the size
 180             # reported by the cloud driver for the node that was
 181             # created.
 182             'size': self.cloud_size.id,
 183             'price': self.cloud_size.price,
 184         }
 185         self.arvados_node = self._arvados.nodes().update(
 186             uuid=self.arvados_node['uuid'],
 187             body={'properties': self.arvados_node['properties']},
 188         ).execute()
 189         self._logger.info("%s updated properties.", self.arvados_node['uuid'])
 190         self._later.post_create()
 191
 192     @RetryMixin._retry()
 193     def post_create(self):
 194         self._cloud.post_create_node(self.cloud_node)
 195         self._logger.info("%s post-create work done.", self.cloud_node.id)
 196         self._finished()
 197
 198     def stop_if_no_cloud_node(self):
 199         if self.cloud_node is not None:
 200             return False
 201         self.stop()
 202         return True
 203
 204
 205 class ComputeNodeShutdownActor(ComputeNodeStateChangeBase):
 206     """Actor to shut down a compute node.
 207
 208     This actor simply destroys a cloud node, retrying as needed.
 209     """
 210     # Reasons for a shutdown to be cancelled.
 211     WINDOW_CLOSED = "shutdown window closed"
 212     DESTROY_FAILED = "destroy_node failed"
 213
 214     def __init__(self, timer_actor, cloud_client, arvados_client, node_monitor,
 215                  cancellable=True, retry_wait=1, max_retry_wait=180):
 216         # If a ShutdownActor is cancellable, it will ask the
 217         # ComputeNodeMonitorActor if it's still eligible before taking each
 218         # action, and stop the shutdown process if the node is no longer
 219         # eligible.  Normal shutdowns based on job demand should be
 220         # cancellable; shutdowns based on node misbehavior should not.
 221         super(ComputeNodeShutdownActor, self).__init__(
 222             cloud_client, arvados_client, timer_actor,
 223             retry_wait, max_retry_wait)
 224         self._monitor = node_monitor.proxy()
 225         self.cloud_node = self._monitor.cloud_node.get()
 226         self.cancellable = cancellable
 227         self.cancel_reason = None
 228         self.success = None
 229
 230     def _set_logger(self):
 231         self._logger = logging.getLogger("%s.%s.%s" % (self.__class__.__name__, self.actor_urn[33:], self.cloud_node.name))
 232
 233     def on_start(self):
 234         super(ComputeNodeShutdownActor, self).on_start()
 235         self._later.shutdown_node()
 236
 237     def _arvados_node(self):
 238         return self._monitor.arvados_node.get()
 239
 240     def _finished(self, success_flag=None):
 241         if success_flag is not None:
 242             self.success = success_flag
 243         return super(ComputeNodeShutdownActor, self)._finished()
 244
 245     def cancel_shutdown(self, reason, **kwargs):
 246         if not self.cancellable:
 247             return False
 248         if self.cancel_reason is not None:
 249             # already cancelled
 250             return False
 251         self.cancel_reason = reason
 252         self._logger.info("Shutdown cancelled: %s.", reason)
 253         self._finished(success_flag=False)
 254         return True
 255
 256     def _cancel_on_exception(orig_func):
 257         @functools.wraps(orig_func)
 258         def finish_wrapper(self, *args, **kwargs):
 259             try:
 260                 return orig_func(self, *args, **kwargs)
 261             except Exception as error:
 262                 self._logger.error("Actor error %s", error)
 263                 self._logger.debug("", exc_info=True)
 264                 self._later.cancel_shutdown("Unhandled exception %s" % error, try_resume=False)
 265         return finish_wrapper
 266
 267     @_cancel_on_exception
 268     def shutdown_node(self):
 269         if self.cancel_reason is not None:
 270             # already cancelled
 271             return
 272         if self.cancellable:
 273             self._logger.info("Checking that node is still eligible for shutdown")
 274             eligible, reason = self._monitor.shutdown_eligible().get()
 275             if not eligible:
 276                 self.cancel_shutdown("No longer eligible for shut down because %s" % reason,
 277                                      try_resume=True)
 278                 return
 279         # If boot failed, count the event
 280         if self._monitor.get_state().get() == 'unpaired':
 281             status.tracker.counter_add('boot_failures')
 282         self._destroy_node()
 283
 284     def _destroy_node(self):
 285         self._logger.info("Starting shutdown")
 286         arv_node = self._arvados_node()
 287         if self._cloud.destroy_node(self.cloud_node):
 288             self.cancellable = False
 289             self._logger.info("Shutdown success")
 290             if arv_node:
 291                 self._later.clean_arvados_node(arv_node)
 292             else:
 293                 self._finished(success_flag=True)
 294         else:
 295             self.cancel_shutdown(self.DESTROY_FAILED, try_resume=False)
 296
 297     @ComputeNodeStateChangeBase._finish_on_exception
 298     @RetryMixin._retry(config.ARVADOS_ERRORS)
 299     def clean_arvados_node(self, arvados_node):
 300         self._clean_arvados_node(arvados_node, "Shut down by Node Manager")
 301         self._finished(success_flag=True)
 302
 303
 304 class ComputeNodeUpdateActor(config.actor_class, RetryMixin):
 305     """Actor to dispatch one-off cloud management requests.
 306
 307     This actor receives requests for small cloud updates, and
 308     dispatches them to a real driver.  ComputeNodeMonitorActors use
 309     this to perform maintenance tasks on themselves.  Having a
 310     dedicated actor for this gives us the opportunity to control the
 311     flow of requests; e.g., by backing off when errors occur.
 312     """
 313     def __init__(self, cloud_factory, timer_actor, max_retry_wait=180):
 314         super(ComputeNodeUpdateActor, self).__init__()
 315         RetryMixin.__init__(self, 1, max_retry_wait,
 316                             None, cloud_factory(), timer_actor)
 317         self._cloud = cloud_factory()
 318         self._later = self.actor_ref.tell_proxy()
 319
 320     def _set_logger(self):
 321         self._logger = logging.getLogger("%s.%s" % (self.__class__.__name__, self.actor_urn[33:]))
 322
 323     def on_start(self):
 324         self._set_logger()
 325
 326     @RetryMixin._retry()
 327     def sync_node(self, cloud_node, arvados_node):
 328         if self._cloud.node_fqdn(cloud_node) != arvados_node_fqdn(arvados_node):
 329             return self._cloud.sync_node(cloud_node, arvados_node)
 330
 331
 332 class ComputeNodeMonitorActor(config.actor_class):
 333     """Actor to manage a running compute node.
 334
 335     This actor gets updates about a compute node's cloud and Arvados records.
 336     It uses this information to notify subscribers when the node is eligible
 337     for shutdown.
 338     """
 339     def __init__(self, cloud_node, cloud_node_start_time, shutdown_timer,
 340                  timer_actor, update_actor, cloud_client,
 341                  arvados_node=None, poll_stale_after=600, node_stale_after=3600,
 342                  boot_fail_after=1800
 343     ):
 344         super(ComputeNodeMonitorActor, self).__init__()
 345         self._later = self.actor_ref.tell_proxy()
 346         self._shutdowns = shutdown_timer
 347         self._timer = timer_actor
 348         self._update = update_actor
 349         self._cloud = cloud_client
 350         self.cloud_node = cloud_node
 351         self.cloud_node_start_time = cloud_node_start_time
 352         self.poll_stale_after = poll_stale_after
 353         self.node_stale_after = node_stale_after
 354         self.boot_fail_after = boot_fail_after
 355         self.subscribers = set()
 356         self.arvados_node = None
 357         self.consecutive_idle = 0
 358         self._later.update_arvados_node(arvados_node)
 359         self.last_shutdown_opening = None
 360         self._later.consider_shutdown()
 361
 362     def _set_logger(self):
 363         self._logger = logging.getLogger("%s.%s.%s" % (self.__class__.__name__, self.actor_urn[33:], self.cloud_node.name))
 364
 365     def on_start(self):
 366         self._set_logger()
 367         self._timer.schedule(self.cloud_node_start_time + self.boot_fail_after, self._later.consider_shutdown)
 368
 369     def subscribe(self, subscriber):
 370         self.subscribers.add(subscriber)
 371
 372     def _debug(self, msg, *args):
 373         self._logger.debug(msg, *args)
 374
 375     def get_state(self):
 376         """Get node state, one of ['unpaired', 'busy', 'idle', 'down']."""
 377
 378         # If this node is not associated with an Arvados node, return
 379         # 'unpaired' if we're in the boot grace period, and 'down' if not,
 380         # so it isn't counted towards usable nodes.
 381         if self.arvados_node is None:
 382             if timestamp_fresh(self.cloud_node_start_time,
 383                                self.boot_fail_after):
 384                 return 'unpaired'
 385             else:
 386                 return 'down'
 387
 388         state = self.arvados_node['crunch_worker_state']
 389
 390         # If state information is not available because it is missing or the
 391         # record is stale, return 'down'.
 392         if not state or not timestamp_fresh(arvados_node_mtime(self.arvados_node),
 393                                             self.node_stale_after):
 394             state = 'down'
 395
 396         # There's a window between when a node pings for the first time and the
 397         # value of 'slurm_state' is synchronized by crunch-dispatch.  In this
 398         # window, the node will still report as 'down'.  Check that
 399         # first_ping_at is truthy and consider the node 'idle' during the
 400         # initial boot grace period.
 401         if (state == 'down' and
 402             self.arvados_node['first_ping_at'] and
 403             timestamp_fresh(self.cloud_node_start_time,
 404                             self.boot_fail_after) and
 405             not self._cloud.broken(self.cloud_node)):
 406             state = 'idle'
 407
 408         # "missing" means last_ping_at is stale, this should be
 409         # considered "down"
 410         if arvados_node_missing(self.arvados_node, self.node_stale_after):
 411             state = 'down'
 412
 413         # Turns out using 'job_uuid' this way is a bad idea.  The node record
 414         # is assigned the job_uuid before the job is locked (which removes it
 415         # from the queue) which means the job will be double-counted as both in
 416         # the wishlist and but also keeping a node busy.  This end result is
 417         # excess nodes being booted.
 418         #if state == 'idle' and self.arvados_node['job_uuid']:
 419         #    state = 'busy'
 420
 421         # Update idle node times tracker
 422         if state == 'idle':
 423             status.tracker.idle_in(self.arvados_node['hostname'])
 424         else:
 425             status.tracker.idle_out(self.arvados_node['hostname'])
 426
 427         return state
 428
 429     def in_state(self, *states):
 430         return self.get_state() in states
 431
 432     def shutdown_eligible(self):
 433         """Determine if node is candidate for shut down.
 434
 435         Returns a tuple of (boolean, string) where the first value is whether
 436         the node is candidate for shut down, and the second value is the
 437         reason for the decision.
 438         """
 439
 440         # Collect states and then consult state transition table whether we
 441         # should shut down.  Possible states are:
 442         # crunch_worker_state = ['unpaired', 'busy', 'idle', 'down']
 443         # window = ["open", "closed"]
 444         # boot_grace = ["boot wait", "boot exceeded"]
 445         # idle_grace = ["not idle", "idle wait", "idle exceeded"]
 446
 447         if self.arvados_node and not timestamp_fresh(arvados_node_mtime(self.arvados_node), self.node_stale_after):
 448             return (False, "node state is stale")
 449
 450         crunch_worker_state = self.get_state()
 451
 452         window = "open" if self._shutdowns.window_open() else "closed"
 453
 454         if timestamp_fresh(self.cloud_node_start_time, self.boot_fail_after):
 455             boot_grace = "boot wait"
 456         else:
 457             boot_grace = "boot exceeded"
 458
 459         if crunch_worker_state == "idle":
 460             # Must report as "idle" at least two consecutive times
 461             if self.consecutive_idle < 2:
 462                 idle_grace = 'idle wait'
 463             else:
 464                 idle_grace = 'idle exceeded'
 465         else:
 466             idle_grace = 'not idle'
 467
 468         node_state = (crunch_worker_state, window, boot_grace, idle_grace)
 469         t = transitions[node_state]
 470         if t is not None:
 471             # yes, shutdown eligible
 472             return (True, "node state is %s" % (node_state,))
 473         else:
 474             # no, return a reason
 475             return (False, "node state is %s" % (node_state,))
 476
 477     def consider_shutdown(self):
 478         try:
 479             eligible, reason = self.shutdown_eligible()
 480             next_opening = self._shutdowns.next_opening()
 481             if eligible:
 482                 self._debug("Suggesting shutdown because %s", reason)
 483                 _notify_subscribers(self.actor_ref.proxy(), self.subscribers)
 484             else:
 485                 self._debug("Not eligible for shut down because %s", reason)
 486
 487                 if self.last_shutdown_opening != next_opening:
 488                     self._debug("Shutdown window closed.  Next at %s.",
 489                                 time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(next_opening)))
 490                     self._timer.schedule(next_opening, self._later.consider_shutdown)
 491                     self.last_shutdown_opening = next_opening
 492         except Exception:
 493             self._logger.exception("Unexpected exception")
 494
 495     def offer_arvados_pair(self, arvados_node):
 496         first_ping_s = arvados_node.get('first_ping_at')
 497         if (self.arvados_node is not None) or (not first_ping_s):
 498             return None
 499         elif ((arvados_node['info'].get('ec2_instance_id') == self._cloud.node_id(self.cloud_node)) and
 500               (arvados_timestamp(first_ping_s) >= self.cloud_node_start_time)):
 501             self._later.update_arvados_node(arvados_node)
 502             return self.cloud_node.id
 503         else:
 504             return None
 505
 506     def update_cloud_node(self, cloud_node):
 507         if cloud_node is not None:
 508             self.cloud_node = cloud_node
 509             self._later.consider_shutdown()
 510
 511     def update_arvados_node(self, arvados_node):
 512         """Called when the latest Arvados node record is retrieved.
 513
 514         Calls the updater's sync_node() method.
 515
 516         """
 517         # This method is a little unusual in the way it just fires off the
 518         # request without checking the result or retrying errors.  That's
 519         # because this update happens every time we reload the Arvados node
 520         # list: if a previous sync attempt failed, we'll see that the names
 521         # are out of sync and just try again.  ComputeNodeUpdateActor has
 522         # the logic to throttle those effective retries when there's trouble.
 523         if arvados_node is not None:
 524             self.arvados_node = arvados_node
 525             self._update.sync_node(self.cloud_node, self.arvados_node)
 526             if self.arvados_node['crunch_worker_state'] == "idle":
 527                 self.consecutive_idle += 1
 528             else:
 529                 self.consecutive_idle = 0
 530             self._later.consider_shutdown()