services/nodemanager/arvnodeman/daemon.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import absolute_import, print_function
   4
   5 import functools
   6 import logging
   7 import time
   8
   9 import pykka
  10
  11 from . import computenode as cnode
  12 from .computenode import dispatch
  13 from .config import actor_class
  14
  15 class _ComputeNodeRecord(object):
  16     def __init__(self, actor=None, cloud_node=None, arvados_node=None,
  17                  assignment_time=float('-inf')):
  18         self.actor = actor
  19         self.cloud_node = cloud_node
  20         self.arvados_node = arvados_node
  21         self.assignment_time = assignment_time
  22
  23
  24 class _BaseNodeTracker(object):
  25     def __init__(self):
  26         self.nodes = {}
  27         self.orphans = {}
  28         self._blacklist = set()
  29
  30     # Proxy the methods listed below to self.nodes.
  31     def _proxy_method(name):
  32         method = getattr(dict, name)
  33         @functools.wraps(method, ('__name__', '__doc__'))
  34         def wrapper(self, *args, **kwargs):
  35             return method(self.nodes, *args, **kwargs)
  36         return wrapper
  37
  38     for _method_name in ['__contains__', '__getitem__', '__len__', 'get']:
  39         locals()[_method_name] = _proxy_method(_method_name)
  40
  41     def record_key(self, record):
  42         return self.item_key(getattr(record, self.RECORD_ATTR))
  43
  44     def add(self, record):
  45         self.nodes[self.record_key(record)] = record
  46
  47     def blacklist(self, key):
  48         self._blacklist.add(key)
  49
  50     def update_record(self, key, item):
  51         setattr(self.nodes[key], self.RECORD_ATTR, item)
  52
  53     def update_from(self, response):
  54         unseen = set(self.nodes.iterkeys())
  55         for item in response:
  56             key = self.item_key(item)
  57             if key in self._blacklist:
  58                 continue
  59             elif key in unseen:
  60                 unseen.remove(key)
  61                 self.update_record(key, item)
  62             else:
  63                 yield key, item
  64         self.orphans = {key: self.nodes.pop(key) for key in unseen}
  65
  66     def unpaired(self):
  67         return (record for record in self.nodes.itervalues()
  68                 if getattr(record, self.PAIR_ATTR) is None)
  69
  70
  71 class _CloudNodeTracker(_BaseNodeTracker):
  72     RECORD_ATTR = 'cloud_node'
  73     PAIR_ATTR = 'arvados_node'
  74     item_key = staticmethod(lambda cloud_node: cloud_node.id)
  75
  76
  77 class _ArvadosNodeTracker(_BaseNodeTracker):
  78     RECORD_ATTR = 'arvados_node'
  79     PAIR_ATTR = 'cloud_node'
  80     item_key = staticmethod(lambda arvados_node: arvados_node['uuid'])
  81
  82     def find_stale_node(self, stale_time):
  83         for record in self.nodes.itervalues():
  84             node = record.arvados_node
  85             if (not cnode.timestamp_fresh(cnode.arvados_node_mtime(node),
  86                                           stale_time) and
  87                   not cnode.timestamp_fresh(record.assignment_time,
  88                                             stale_time)):
  89                 return node
  90         return None
  91
  92
  93 class NodeManagerDaemonActor(actor_class):
  94     """Node Manager daemon.
  95
  96     This actor subscribes to all information polls about cloud nodes,
  97     Arvados nodes, and the job queue.  It creates a ComputeNodeMonitorActor
  98     for every cloud node, subscribing them to poll updates
  99     appropriately.  It creates and destroys cloud nodes based on job queue
 100     demand, and stops the corresponding ComputeNode actors when their work
 101     is done.
 102     """
 103     def __init__(self, server_wishlist_actor, arvados_nodes_actor,
 104                  cloud_nodes_actor, cloud_update_actor, timer_actor,
 105                  arvados_factory, cloud_factory,
 106                  shutdown_windows, server_calculator,
 107                  min_nodes, max_nodes,
 108                  poll_stale_after=600,
 109                  boot_fail_after=1800,
 110                  node_stale_after=7200,
 111                  node_setup_class=dispatch.ComputeNodeSetupActor,
 112                  node_shutdown_class=dispatch.ComputeNodeShutdownActor,
 113                  node_actor_class=dispatch.ComputeNodeMonitorActor,
 114                  max_total_price=0):
 115         super(NodeManagerDaemonActor, self).__init__()
 116         self._node_setup = node_setup_class
 117         self._node_shutdown = node_shutdown_class
 118         self._node_actor = node_actor_class
 119         self._cloud_updater = cloud_update_actor
 120         self._timer = timer_actor
 121         self._new_arvados = arvados_factory
 122         self._new_cloud = cloud_factory
 123         self._cloud_driver = self._new_cloud()
 124         self._logger = logging.getLogger('arvnodeman.daemon')
 125         self._later = self.actor_ref.proxy()
 126         self.shutdown_windows = shutdown_windows
 127         self.server_calculator = server_calculator
 128         self.min_cloud_size = self.server_calculator.cheapest_size()
 129         self.min_nodes = min_nodes
 130         self.max_nodes = max_nodes
 131         self.max_total_price = max_total_price
 132         self.poll_stale_after = poll_stale_after
 133         self.boot_fail_after = boot_fail_after
 134         self.node_stale_after = node_stale_after
 135         self.last_polls = {}
 136         for poll_name in ['server_wishlist', 'arvados_nodes', 'cloud_nodes']:
 137             poll_actor = locals()[poll_name + '_actor']
 138             poll_actor.subscribe(getattr(self._later, 'update_' + poll_name))
 139             setattr(self, '_{}_actor'.format(poll_name), poll_actor)
 140             self.last_polls[poll_name] = -self.poll_stale_after
 141         self.cloud_nodes = _CloudNodeTracker()
 142         self.arvados_nodes = _ArvadosNodeTracker()
 143         self.booting = {}       # Actor IDs to ComputeNodeSetupActors
 144         self.booted = {}        # Cloud node IDs to _ComputeNodeRecords
 145         self.shutdowns = {}     # Cloud node IDs to ComputeNodeShutdownActors
 146         self._logger.debug("Daemon initialized")
 147
 148     def _update_poll_time(self, poll_key):
 149         self.last_polls[poll_key] = time.time()
 150
 151     def _pair_nodes(self, node_record, arvados_node):
 152         self._logger.info("Cloud node %s has associated with Arvados node %s",
 153                           node_record.cloud_node.id, arvados_node['uuid'])
 154         self._arvados_nodes_actor.subscribe_to(
 155             arvados_node['uuid'], node_record.actor.update_arvados_node)
 156         node_record.arvados_node = arvados_node
 157         self.arvados_nodes.add(node_record)
 158
 159     def _new_node(self, cloud_node):
 160         start_time = self._cloud_driver.node_start_time(cloud_node)
 161         shutdown_timer = cnode.ShutdownTimer(start_time,
 162                                              self.shutdown_windows)
 163         actor = self._node_actor.start(
 164             cloud_node=cloud_node,
 165             cloud_node_start_time=start_time,
 166             shutdown_timer=shutdown_timer,
 167             cloud_fqdn_func=self._cloud_driver.node_fqdn,
 168             update_actor=self._cloud_updater,
 169             timer_actor=self._timer,
 170             arvados_node=None,
 171             poll_stale_after=self.poll_stale_after,
 172             node_stale_after=self.node_stale_after,
 173             cloud_client=self._cloud_driver,
 174             boot_fail_after=self.boot_fail_after).proxy()
 175         actor.subscribe(self._later.node_can_shutdown)
 176         self._cloud_nodes_actor.subscribe_to(cloud_node.id,
 177                                              actor.update_cloud_node)
 178         record = _ComputeNodeRecord(actor, cloud_node)
 179         return record
 180
 181     def update_cloud_nodes(self, nodelist):
 182         self._update_poll_time('cloud_nodes')
 183         for key, node in self.cloud_nodes.update_from(nodelist):
 184             self._logger.info("Registering new cloud node %s", key)
 185             if key in self.booted:
 186                 record = self.booted.pop(key)
 187             else:
 188                 record = self._new_node(node)
 189             self.cloud_nodes.add(record)
 190             for arv_rec in self.arvados_nodes.unpaired():
 191                 if record.actor.offer_arvados_pair(arv_rec.arvados_node).get():
 192                     self._pair_nodes(record, arv_rec.arvados_node)
 193                     break
 194         for key, record in self.cloud_nodes.orphans.iteritems():
 195             if key in self.shutdowns:
 196                 try:
 197                     self.shutdowns[key].stop().get()
 198                 except pykka.ActorDeadError:
 199                     pass
 200                 del self.shutdowns[key]
 201             record.actor.stop()
 202             record.cloud_node = None
 203
 204     def update_arvados_nodes(self, nodelist):
 205         self._update_poll_time('arvados_nodes')
 206         for key, node in self.arvados_nodes.update_from(nodelist):
 207             self._logger.info("Registering new Arvados node %s", key)
 208             record = _ComputeNodeRecord(arvados_node=node)
 209             self.arvados_nodes.add(record)
 210         for arv_rec in self.arvados_nodes.unpaired():
 211             arv_node = arv_rec.arvados_node
 212             for cloud_rec in self.cloud_nodes.unpaired():
 213                 if cloud_rec.actor.offer_arvados_pair(arv_node).get():
 214                     self._pair_nodes(cloud_rec, arv_node)
 215                     break
 216
 217     def _nodes_up(self, size):
 218         up = 0
 219         up += sum(1
 220                   for c in self.booting.itervalues()
 221                   if size is None or c.cloud_size.get().id == size.id)
 222         up += sum(1
 223                   for i in (self.booted, self.cloud_nodes.nodes)
 224                   for c in i.itervalues()
 225                   if size is None or c.cloud_node.size.id == size.id)
 226         return up
 227
 228     def _total_price(self):
 229         cost = 0
 230         cost += sum(c.cloud_size.get().price
 231                   for c in self.booting.itervalues())
 232         cost += sum(c.cloud_node.size.price
 233                     for i in (self.booted, self.cloud_nodes.nodes)
 234                     for c in i.itervalues())
 235         return cost
 236
 237     def _nodes_busy(self, size):
 238         return sum(1 for busy in
 239                    pykka.get_all(rec.actor.in_state('busy') for rec in
 240                                  self.cloud_nodes.nodes.itervalues()
 241                                  if rec.cloud_node.size.id == size.id)
 242                    if busy)
 243
 244     def _nodes_missing(self, size):
 245         return sum(1 for arv_node in
 246                    pykka.get_all(rec.actor.arvados_node for rec in
 247                                  self.cloud_nodes.nodes.itervalues()
 248                                  if rec.cloud_node.size.id == size.id and rec.actor.cloud_node.get().id not in self.shutdowns)
 249                    if arv_node and cnode.arvados_node_missing(arv_node, self.node_stale_after))
 250
 251     def _size_wishlist(self, size):
 252         return sum(1 for c in self.last_wishlist if c.id == size.id)
 253
 254     def _size_shutdowns(self, size):
 255         return sum(1 for c in self.shutdowns.itervalues()
 256                    if c.cloud_node.get().size.id == size.id)
 257
 258     def _nodes_wanted(self, size):
 259         total_up_count = self._nodes_up(None)
 260         under_min = self.min_nodes - total_up_count
 261         over_max = total_up_count - self.max_nodes
 262         total_price = self._total_price()
 263
 264         if over_max >= 0:
 265             return -over_max
 266         elif under_min > 0 and size.id == self.min_cloud_size.id:
 267             return under_min
 268
 269         up_count = self._nodes_up(size) - (self._size_shutdowns(size) +
 270                                            self._nodes_busy(size) +
 271                                            self._nodes_missing(size))
 272
 273         wanted = self._size_wishlist(size) - up_count
 274         if wanted > 0 and self.max_total_price and ((total_price + (size.price*wanted)) > self.max_total_price):
 275             can_boot = int((self.max_total_price - total_price) / size.price)
 276             if can_boot == 0:
 277                 self._logger.info("Not booting %s (price %s) because with it would exceed max_total_price of %s (current total_price is %s)",
 278                                   size.name, size.price, self.max_total_price, total_price)
 279             return can_boot
 280         else:
 281             return wanted
 282
 283     def _nodes_excess(self, size):
 284         up_count = self._nodes_up(size) - self._size_shutdowns(size)
 285         if size.id == self.min_cloud_size.id:
 286             up_count -= self.min_nodes
 287         return up_count - self._nodes_busy(size) - self._size_wishlist(size)
 288
 289     def update_server_wishlist(self, wishlist):
 290         self._update_poll_time('server_wishlist')
 291         self.last_wishlist = wishlist
 292         for sz in reversed(self.server_calculator.cloud_sizes):
 293             size = sz.real
 294             nodes_wanted = self._nodes_wanted(size)
 295             if nodes_wanted > 0:
 296                 self._later.start_node(size)
 297             elif (nodes_wanted < 0) and self.booting:
 298                 self._later.stop_booting_node(size)
 299
 300     def _check_poll_freshness(orig_func):
 301         """Decorator to inhibit a method when poll information is stale.
 302
 303         This decorator checks the timestamps of all the poll information the
 304         daemon has received.  The decorated method is only called if none
 305         of the timestamps are considered stale.
 306         """
 307         @functools.wraps(orig_func)
 308         def wrapper(self, *args, **kwargs):
 309             now = time.time()
 310             if all(now - t < self.poll_stale_after
 311                    for t in self.last_polls.itervalues()):
 312                 return orig_func(self, *args, **kwargs)
 313             else:
 314                 return None
 315         return wrapper
 316
 317     @_check_poll_freshness
 318     def start_node(self, cloud_size):
 319         nodes_wanted = self._nodes_wanted(cloud_size)
 320         if nodes_wanted < 1:
 321             return None
 322         arvados_node = self.arvados_nodes.find_stale_node(self.node_stale_after)
 323         self._logger.info("Want %s more nodes.  Booting a %s node.",
 324                           nodes_wanted, cloud_size.name)
 325         new_setup = self._node_setup.start(
 326             timer_actor=self._timer,
 327             arvados_client=self._new_arvados(),
 328             arvados_node=arvados_node,
 329             cloud_client=self._new_cloud(),
 330             cloud_size=cloud_size).proxy()
 331         self.booting[new_setup.actor_ref.actor_urn] = new_setup
 332         if arvados_node is not None:
 333             self.arvados_nodes[arvados_node['uuid']].assignment_time = (
 334                 time.time())
 335         new_setup.subscribe(self._later.node_up)
 336         if nodes_wanted > 1:
 337             self._later.start_node(cloud_size)
 338
 339     def _get_actor_attrs(self, actor, *attr_names):
 340         return pykka.get_all([getattr(actor, name) for name in attr_names])
 341
 342     def node_up(self, setup_proxy):
 343         cloud_node = setup_proxy.cloud_node.get()
 344         del self.booting[setup_proxy.actor_ref.actor_urn]
 345         setup_proxy.stop()
 346         record = self.cloud_nodes.get(cloud_node.id)
 347         if record is None:
 348             record = self._new_node(cloud_node)
 349             self.booted[cloud_node.id] = record
 350         self._timer.schedule(time.time() + self.boot_fail_after,
 351                              self._later.shutdown_unpaired_node, cloud_node.id)
 352
 353     @_check_poll_freshness
 354     def stop_booting_node(self, size):
 355         nodes_excess = self._nodes_excess(size)
 356         if (nodes_excess < 1) or not self.booting:
 357             return None
 358         for key, node in self.booting.iteritems():
 359             if node.cloud_size.get().id == size.id and node.stop_if_no_cloud_node().get():
 360                 del self.booting[key]
 361                 if nodes_excess > 1:
 362                     self._later.stop_booting_node(size)
 363                 break
 364
 365     def _begin_node_shutdown(self, node_actor, cancellable):
 366         cloud_node_id = node_actor.cloud_node.get().id
 367         if cloud_node_id in self.shutdowns:
 368             return None
 369         shutdown = self._node_shutdown.start(
 370             timer_actor=self._timer, cloud_client=self._new_cloud(),
 371             arvados_client=self._new_arvados(),
 372             node_monitor=node_actor.actor_ref, cancellable=cancellable).proxy()
 373         self.shutdowns[cloud_node_id] = shutdown
 374         shutdown.subscribe(self._later.node_finished_shutdown)
 375
 376     @_check_poll_freshness
 377     def node_can_shutdown(self, node_actor):
 378         if self._nodes_excess(node_actor.cloud_node.get().size) > 0:
 379             self._begin_node_shutdown(node_actor, cancellable=True)
 380
 381     def shutdown_unpaired_node(self, cloud_node_id):
 382         for record_dict in [self.cloud_nodes, self.booted]:
 383             if cloud_node_id in record_dict:
 384                 record = record_dict[cloud_node_id]
 385                 break
 386         else:
 387             return None
 388         if not record.actor.in_state('idle', 'busy').get():
 389             self._begin_node_shutdown(record.actor, cancellable=False)
 390
 391     def node_finished_shutdown(self, shutdown_actor):
 392         cloud_node, success, cancel_reason = self._get_actor_attrs(
 393             shutdown_actor, 'cloud_node', 'success', 'cancel_reason')
 394         shutdown_actor.stop()
 395         cloud_node_id = cloud_node.id
 396         if not success:
 397             if cancel_reason == self._node_shutdown.NODE_BROKEN:
 398                 self.cloud_nodes.blacklist(cloud_node_id)
 399             del self.shutdowns[cloud_node_id]
 400         elif cloud_node_id in self.booted:
 401             self.booted.pop(cloud_node_id).actor.stop()
 402             del self.shutdowns[cloud_node_id]
 403
 404     def shutdown(self):
 405         self._logger.info("Shutting down after signal.")
 406         self.poll_stale_after = -1  # Inhibit starting/stopping nodes
 407         setup_stops = {key: node.stop_if_no_cloud_node()
 408                        for key, node in self.booting.iteritems()}
 409         self.booting = {key: self.booting[key]
 410                         for key in setup_stops if not setup_stops[key].get()}
 411         self._later.await_shutdown()
 412
 413     def await_shutdown(self):
 414         if self.booting:
 415             self._timer.schedule(time.time() + 1, self._later.await_shutdown)
 416         else:
 417             self.stop()