self.cloud_node = cloud_node
self.arvados_node = arvados_node
self.assignment_time = assignment_time
-
+ self.shutdown_actor = None
class _BaseNodeTracker(object):
def __init__(self):
self.nodes = {}
self.orphans = {}
- self._blacklist = set()
# Proxy the methods listed below to self.nodes.
def _proxy_method(name):
def add(self, record):
self.nodes[self.record_key(record)] = record
- def blacklist(self, key):
- self._blacklist.add(key)
-
def update_record(self, key, item):
setattr(self.nodes[key], self.RECORD_ATTR, item)
unseen = set(self.nodes.iterkeys())
for item in response:
key = self.item_key(item)
- if key in self._blacklist:
- continue
- elif key in unseen:
+ if key in unseen:
unseen.remove(key)
self.update_record(key, item)
else:
node_stale_after=7200,
node_setup_class=dispatch.ComputeNodeSetupActor,
node_shutdown_class=dispatch.ComputeNodeShutdownActor,
- node_actor_class=dispatch.ComputeNodeMonitorActor):
+ node_actor_class=dispatch.ComputeNodeMonitorActor,
+ max_total_price=0):
super(NodeManagerDaemonActor, self).__init__()
self._node_setup = node_setup_class
self._node_shutdown = node_shutdown_class
self._new_arvados = arvados_factory
self._new_cloud = cloud_factory
self._cloud_driver = self._new_cloud()
- self._logger = logging.getLogger('arvnodeman.daemon')
- self._later = self.actor_ref.proxy()
+ self._later = self.actor_ref.tell_proxy()
self.shutdown_windows = shutdown_windows
self.server_calculator = server_calculator
self.min_cloud_size = self.server_calculator.cheapest_size()
self.min_nodes = min_nodes
self.max_nodes = max_nodes
+ self.max_total_price = max_total_price
self.poll_stale_after = poll_stale_after
self.boot_fail_after = boot_fail_after
self.node_stale_after = node_stale_after
self.cloud_nodes = _CloudNodeTracker()
self.arvados_nodes = _ArvadosNodeTracker()
self.booting = {} # Actor IDs to ComputeNodeSetupActors
- self.booted = {} # Cloud node IDs to _ComputeNodeRecords
- self.shutdowns = {} # Cloud node IDs to ComputeNodeShutdownActors
- self._logger.debug("Daemon initialized")
+ self.sizes_booting = {} # Actor IDs to node size
+
+ def on_start(self):
+ self._logger = logging.getLogger("%s.%s" % (self.__class__.__name__, self.actor_urn[33:]))
+ self._logger.debug("Daemon started")
def _update_poll_time(self, poll_key):
self.last_polls[poll_key] = time.time()
def _pair_nodes(self, node_record, arvados_node):
- self._logger.info("Cloud node %s has associated with Arvados node %s",
- node_record.cloud_node.id, arvados_node['uuid'])
+ self._logger.info("Cloud node %s is now paired with Arvados node %s",
+ node_record.cloud_node.name, arvados_node['uuid'])
self._arvados_nodes_actor.subscribe_to(
arvados_node['uuid'], node_record.actor.update_arvados_node)
node_record.arvados_node = arvados_node
poll_stale_after=self.poll_stale_after,
node_stale_after=self.node_stale_after,
cloud_client=self._cloud_driver,
- boot_fail_after=self.boot_fail_after).proxy()
- actor.subscribe(self._later.node_can_shutdown)
+ boot_fail_after=self.boot_fail_after)
+ actorTell = actor.tell_proxy()
+ actorTell.subscribe(self._later.node_can_shutdown)
self._cloud_nodes_actor.subscribe_to(cloud_node.id,
- actor.update_cloud_node)
- record = _ComputeNodeRecord(actor, cloud_node)
+ actorTell.update_cloud_node)
+ record = _ComputeNodeRecord(actor.proxy(), cloud_node)
return record
+ def _register_cloud_node(self, node):
+ rec = self.cloud_nodes.get(node.id)
+ if rec is None:
+ self._logger.info("Registering new cloud node %s", node.id)
+ record = self._new_node(node)
+ self.cloud_nodes.add(record)
+ else:
+ rec.cloud_node = node
+
def update_cloud_nodes(self, nodelist):
self._update_poll_time('cloud_nodes')
- for key, node in self.cloud_nodes.update_from(nodelist):
- self._logger.info("Registering new cloud node %s", key)
- if key in self.booted:
- record = self.booted.pop(key)
- else:
- record = self._new_node(node)
- self.cloud_nodes.add(record)
- for arv_rec in self.arvados_nodes.unpaired():
- if record.actor.offer_arvados_pair(arv_rec.arvados_node).get():
- self._pair_nodes(record, arv_rec.arvados_node)
- break
- for key, record in self.cloud_nodes.orphans.iteritems():
- if key in self.shutdowns:
+ for _, node in self.cloud_nodes.update_from(nodelist):
+ self._register_cloud_node(node)
+
+ self.try_pairing()
+
+ for record in self.cloud_nodes.orphans.itervalues():
+ if record.shutdown_actor:
try:
- self.shutdowns[key].stop().get()
+ record.shutdown_actor.stop()
except pykka.ActorDeadError:
pass
- del self.shutdowns[key]
- record.actor.stop()
- record.cloud_node = None
+ record.shutdown_actor = None
+
+ # A recently booted node is a node that successfully completed the
+ # setup actor but has not yet appeared in the cloud node list.
+ # This will have the tag _nodemanager_recently_booted on it, which
+ # means (if we're not shutting it down) we want to put it back into
+ # the cloud node list. Once it really appears in the cloud list,
+ # the object in record.cloud_node will be replaced by a new one
+ # that lacks the "_nodemanager_recently_booted" tag.
+ if hasattr(record.cloud_node, "_nodemanager_recently_booted"):
+ self.cloud_nodes.add(record)
+ else:
+ # Node disappeared from the cloud node list. Stop the monitor
+ # actor if necessary and forget about the node.
+ if record.actor:
+ try:
+ record.actor.stop()
+ except pykka.ActorDeadError:
+ pass
+ record.actor = None
+ record.cloud_node = None
+
+ def _register_arvados_node(self, key, arv_node):
+ self._logger.info("Registering new Arvados node %s", key)
+ record = _ComputeNodeRecord(arvados_node=arv_node)
+ self.arvados_nodes.add(record)
def update_arvados_nodes(self, nodelist):
self._update_poll_time('arvados_nodes')
for key, node in self.arvados_nodes.update_from(nodelist):
- self._logger.info("Registering new Arvados node %s", key)
- record = _ComputeNodeRecord(arvados_node=node)
- self.arvados_nodes.add(record)
- for arv_rec in self.arvados_nodes.unpaired():
- arv_node = arv_rec.arvados_node
- for cloud_rec in self.cloud_nodes.unpaired():
- if cloud_rec.actor.offer_arvados_pair(arv_node).get():
- self._pair_nodes(cloud_rec, arv_node)
+ self._register_arvados_node(key, node)
+ self.try_pairing()
+
+ def try_pairing(self):
+ for record in self.cloud_nodes.unpaired():
+ for arv_rec in self.arvados_nodes.unpaired():
+ if record.actor.offer_arvados_pair(arv_rec.arvados_node).get():
+ self._pair_nodes(record, arv_rec.arvados_node)
break
- def _nodes_up(self):
- return sum(len(nodelist) for nodelist in
- [self.cloud_nodes, self.booted, self.booting])
-
- def _nodes_busy(self):
- return sum(1 for busy in
- pykka.get_all(rec.actor.in_state('busy') for rec in
- self.cloud_nodes.nodes.itervalues())
- if busy)
-
- def _nodes_missing(self):
- return sum(1 for arv_node in
- pykka.get_all(rec.actor.arvados_node for rec in
- self.cloud_nodes.nodes.itervalues()
- if rec.actor.cloud_node.get().id not in self.shutdowns)
- if arv_node and cnode.arvados_node_missing(arv_node, self.node_stale_after))
-
- def _nodes_wanted(self):
- up_count = self._nodes_up()
- under_min = self.min_nodes - up_count
- over_max = up_count - self.max_nodes
+ def _nodes_booting(self, size):
+ s = sum(1
+ for c in self.booting.iterkeys()
+ if size is None or self.sizes_booting[c].id == size.id)
+ return s
+
+ def _node_states(self, size):
+ proxy_states = []
+ states = []
+ for rec in self.cloud_nodes.nodes.itervalues():
+ if size is None or rec.cloud_node.size.id == size.id:
+ if rec.shutdown_actor is None and rec.actor is not None:
+ proxy_states.append(rec.actor.get_state())
+ else:
+ states.append("shutdown")
+ return states + pykka.get_all(proxy_states)
+
+ def _state_counts(self, size):
+ states = self._node_states(size)
+ counts = {
+ "booting": self._nodes_booting(size),
+ "unpaired": 0,
+ "busy": 0,
+ "idle": 0,
+ "down": 0,
+ "shutdown": 0
+ }
+ for s in states:
+ counts[s] = counts[s] + 1
+ return counts
+
+ def _nodes_up(self, counts):
+ up = counts["booting"] + counts["unpaired"] + counts["idle"] + counts["busy"]
+ return up
+
+ def _total_price(self):
+ cost = 0
+ cost += sum(self.sizes_booting[c].price
+ for c in self.booting.iterkeys())
+ cost += sum(c.cloud_node.size.price
+ for c in self.cloud_nodes.nodes.itervalues())
+ return cost
+
+ def _size_wishlist(self, size):
+ return sum(1 for c in self.last_wishlist if c.id == size.id)
+
+ def _nodes_wanted(self, size):
+ total_node_count = self._nodes_booting(None) + len(self.cloud_nodes)
+ under_min = self.min_nodes - total_node_count
+ over_max = total_node_count - self.max_nodes
+ total_price = self._total_price()
+
+ counts = self._state_counts(size)
+
+ up_count = self._nodes_up(counts)
+ busy_count = counts["busy"]
+
+ self._logger.info("%s: wishlist %i, up %i (booting %i, unpaired %i, idle %i, busy %i), down %i, shutdown %i", size.name,
+ self._size_wishlist(size),
+ up_count,
+ counts["booting"],
+ counts["unpaired"],
+ counts["idle"],
+ busy_count,
+ counts["down"],
+ counts["shutdown"])
+
if over_max >= 0:
return -over_max
- elif under_min > 0:
+ elif under_min > 0 and size.id == self.min_cloud_size.id:
return under_min
+
+ wanted = self._size_wishlist(size) - (up_count - busy_count)
+ if wanted > 0 and self.max_total_price and ((total_price + (size.price*wanted)) > self.max_total_price):
+ can_boot = int((self.max_total_price - total_price) / size.price)
+ if can_boot == 0:
+ self._logger.info("Not booting %s (price %s) because with it would exceed max_total_price of %s (current total_price is %s)",
+ size.name, size.price, self.max_total_price, total_price)
+ return can_boot
else:
- up_count -= len(self.shutdowns) + self._nodes_busy() + self._nodes_missing()
- return len(self.last_wishlist) - up_count
-
- def _nodes_excess(self):
- up_count = self._nodes_up() - len(self.shutdowns)
- over_min = up_count - self.min_nodes
- if over_min <= 0:
- return over_min
- else:
- return up_count - self._nodes_busy() - len(self.last_wishlist)
+ return wanted
+
+ def _nodes_excess(self, size):
+ counts = self._state_counts(size)
+ up_count = self._nodes_up(counts)
+ if size.id == self.min_cloud_size.id:
+ up_count -= self.min_nodes
+ return up_count - (counts["busy"] + self._size_wishlist(size))
def update_server_wishlist(self, wishlist):
self._update_poll_time('server_wishlist')
self.last_wishlist = wishlist
- nodes_wanted = self._nodes_wanted()
- if nodes_wanted > 0:
- self._later.start_node()
- elif (nodes_wanted < 0) and self.booting:
- self._later.stop_booting_node()
+ for size in reversed(self.server_calculator.cloud_sizes):
+ try:
+ nodes_wanted = self._nodes_wanted(size)
+ if nodes_wanted > 0:
+ self._later.start_node(size)
+ elif (nodes_wanted < 0) and self.booting:
+ self._later.stop_booting_node(size)
+ except Exception as e:
+ self._logger.exception("while calculating nodes wanted for size %s", size)
def _check_poll_freshness(orig_func):
"""Decorator to inhibit a method when poll information is stale.
return wrapper
@_check_poll_freshness
- def start_node(self):
- nodes_wanted = self._nodes_wanted()
+ def start_node(self, cloud_size):
+ nodes_wanted = self._nodes_wanted(cloud_size)
if nodes_wanted < 1:
return None
arvados_node = self.arvados_nodes.find_stale_node(self.node_stale_after)
- try:
- cloud_size = self.last_wishlist[self._nodes_up()]
- except IndexError:
- cloud_size = self.min_cloud_size
- self._logger.info("Want %s more nodes. Booting a %s node.",
+ self._logger.info("Want %i more %s nodes. Booting a node.",
nodes_wanted, cloud_size.name)
new_setup = self._node_setup.start(
timer_actor=self._timer,
cloud_client=self._new_cloud(),
cloud_size=cloud_size).proxy()
self.booting[new_setup.actor_ref.actor_urn] = new_setup
+ self.sizes_booting[new_setup.actor_ref.actor_urn] = cloud_size
+
if arvados_node is not None:
self.arvados_nodes[arvados_node['uuid']].assignment_time = (
time.time())
new_setup.subscribe(self._later.node_up)
if nodes_wanted > 1:
- self._later.start_node()
+ self._later.start_node(cloud_size)
def _get_actor_attrs(self, actor, *attr_names):
return pykka.get_all([getattr(actor, name) for name in attr_names])
def node_up(self, setup_proxy):
- cloud_node = setup_proxy.cloud_node.get()
- del self.booting[setup_proxy.actor_ref.actor_urn]
+ # Called when a SetupActor has completed.
+ cloud_node, arvados_node = self._get_actor_attrs(
+ setup_proxy, 'cloud_node', 'arvados_node')
setup_proxy.stop()
- record = self.cloud_nodes.get(cloud_node.id)
- if record is None:
- record = self._new_node(cloud_node)
- self.booted[cloud_node.id] = record
- self._timer.schedule(time.time() + self.boot_fail_after,
- self._later.shutdown_unpaired_node, cloud_node.id)
+
+ # If cloud_node is None then the node create wasn't
+ # successful and so there isn't anything to do.
+ if cloud_node is not None:
+ # Node creation succeeded. Update cloud node list.
+ cloud_node._nodemanager_recently_booted = True
+ self._register_cloud_node(cloud_node)
+ del self.booting[setup_proxy.actor_ref.actor_urn]
+ del self.sizes_booting[setup_proxy.actor_ref.actor_urn]
@_check_poll_freshness
- def stop_booting_node(self):
- nodes_excess = self._nodes_excess()
+ def stop_booting_node(self, size):
+ nodes_excess = self._nodes_excess(size)
if (nodes_excess < 1) or not self.booting:
return None
for key, node in self.booting.iteritems():
- if node.stop_if_no_cloud_node().get():
+ if node and node.cloud_size.get().id == size.id and node.stop_if_no_cloud_node().get():
del self.booting[key]
+ del self.sizes_booting[key]
+
if nodes_excess > 1:
- self._later.stop_booting_node()
+ self._later.stop_booting_node(size)
break
def _begin_node_shutdown(self, node_actor, cancellable):
- cloud_node_id = node_actor.cloud_node.get().id
- if cloud_node_id in self.shutdowns:
+ cloud_node_obj = node_actor.cloud_node.get()
+ cloud_node_id = cloud_node_obj.id
+ record = self.cloud_nodes[cloud_node_id]
+ if record.shutdown_actor is not None:
return None
shutdown = self._node_shutdown.start(
timer_actor=self._timer, cloud_client=self._new_cloud(),
arvados_client=self._new_arvados(),
- node_monitor=node_actor.actor_ref, cancellable=cancellable).proxy()
- self.shutdowns[cloud_node_id] = shutdown
- shutdown.subscribe(self._later.node_finished_shutdown)
+ node_monitor=node_actor.actor_ref, cancellable=cancellable)
+ record.shutdown_actor = shutdown.proxy()
+ shutdown.tell_proxy().subscribe(self._later.node_finished_shutdown)
@_check_poll_freshness
def node_can_shutdown(self, node_actor):
- if self._nodes_excess() > 0:
+ if self._nodes_excess(node_actor.cloud_node.get().size) > 0:
self._begin_node_shutdown(node_actor, cancellable=True)
-
- def shutdown_unpaired_node(self, cloud_node_id):
- for record_dict in [self.cloud_nodes, self.booted]:
- if cloud_node_id in record_dict:
- record = record_dict[cloud_node_id]
- break
- else:
- return None
- if not record.actor.in_state('idle', 'busy').get():
- self._begin_node_shutdown(record.actor, cancellable=False)
+ elif self.cloud_nodes.nodes.get(node_actor.cloud_node.get().id).arvados_node is None:
+ # Node is unpaired, which means it probably exceeded its booting
+ # grace period without a ping, so shut it down so we can boot a new
+ # node in its place.
+ self._begin_node_shutdown(node_actor, cancellable=False)
+ elif node_actor.in_state('down').get():
+ # Node is down and unlikely to come back.
+ self._begin_node_shutdown(node_actor, cancellable=False)
def node_finished_shutdown(self, shutdown_actor):
- cloud_node, success, cancel_reason = self._get_actor_attrs(
- shutdown_actor, 'cloud_node', 'success', 'cancel_reason')
- shutdown_actor.stop()
+ try:
+ cloud_node, success = self._get_actor_attrs(
+ shutdown_actor, 'cloud_node', 'success')
+ except pykka.ActorDeadError:
+ return
cloud_node_id = cloud_node.id
+ record = self.cloud_nodes[cloud_node_id]
+ shutdown_actor.stop()
+ record.shutdown_actor = None
+
if not success:
- if cancel_reason == self._node_shutdown.NODE_BROKEN:
- self.cloud_nodes.blacklist(cloud_node_id)
- del self.shutdowns[cloud_node_id]
- elif cloud_node_id in self.booted:
- self.booted.pop(cloud_node_id).actor.stop()
- del self.shutdowns[cloud_node_id]
+ return
+
+ # Shutdown was successful, so stop the monitor actor, otherwise it
+ # will keep offering the node as a candidate for shutdown.
+ record.actor.stop()
+ record.actor = None
+
+ # If the node went from being booted to being shut down without ever
+ # appearing in the cloud node list, it will have the
+ # _nodemanager_recently_booted tag, so get rid of it so that the node
+ # can be forgotten completely.
+ if hasattr(record.cloud_node, "_nodemanager_recently_booted"):
+ del record.cloud_node._nodemanager_recently_booted
def shutdown(self):
self._logger.info("Shutting down after signal.")
projects / arvados.git / blobdiff