+++ /dev/null
-# Copyright (C) The Arvados Authors. All rights reserved.
-#
-# SPDX-License-Identifier: AGPL-3.0
-
-from __future__ import absolute_import, print_function
-
-import errno
-import logging
-import os
-import signal
-import time
-import threading
-import traceback
-
-import pykka
-
-from .status import tracker
-
-class _TellCallableProxy(object):
- """Internal helper class for proxying callables."""
-
- def __init__(self, ref, attr_path):
- self.actor_ref = ref
- self._attr_path = attr_path
-
- def __call__(self, *args, **kwargs):
- message = {
- 'command': 'pykka_call',
- 'attr_path': self._attr_path,
- 'args': args,
- 'kwargs': kwargs,
- }
- self.actor_ref.tell(message)
-
-
-class TellActorProxy(pykka.ActorProxy):
- """ActorProxy in which all calls are implemented as using tell().
-
- The standard pykka.ActorProxy always uses ask() and returns a Future. If
- the target method raises an exception, it is placed in the Future object
- and re-raised when get() is called on the Future. Unfortunately, most
- messaging in Node Manager is asynchronous and the caller does not store the
- Future object returned by the call to ActorProxy. As a result, exceptions
- resulting from these calls end up in limbo, neither reported in the logs
- nor handled by on_failure().
-
- The TellActorProxy uses tell() instead of ask() and does not return a
- Future object. As a result, if the target method raises an exception, it
- will be logged and on_failure() will be called as intended.
-
- """
-
- def __repr__(self):
- return '<ActorProxy for %s, attr_path=%s>' % (
- self.actor_ref, self._attr_path)
-
- def __getattr__(self, name):
- """Get a callable from the actor."""
- attr_path = self._attr_path + (name,)
- if attr_path not in self._known_attrs:
- self._known_attrs = self._get_attributes()
- attr_info = self._known_attrs.get(attr_path)
- if attr_info is None:
- raise AttributeError('%s has no attribute "%s"' % (self, name))
- if attr_info['callable']:
- if attr_path not in self._callable_proxies:
- self._callable_proxies[attr_path] = _TellCallableProxy(
- self.actor_ref, attr_path)
- return self._callable_proxies[attr_path]
- else:
- raise AttributeError('attribute "%s" is not a callable on %s' % (name, self))
-
-class TellableActorRef(pykka.ActorRef):
- """ActorRef adding the tell_proxy() method to get TellActorProxy."""
-
- def tell_proxy(self):
- return TellActorProxy(self)
-
-class BaseNodeManagerActor(pykka.ThreadingActor):
- """Base class for actors in node manager, redefining actor_ref as a
- TellableActorRef and providing a default on_failure handler.
- """
-
- def __init__(self, *args, **kwargs):
- super(pykka.ThreadingActor, self).__init__(*args, **kwargs)
- self.actor_ref = TellableActorRef(self)
- self._killfunc = kwargs.get("killfunc", os.kill)
-
- def on_failure(self, exception_type, exception_value, tb):
- lg = getattr(self, "_logger", logging)
- if (exception_type in (threading.ThreadError, MemoryError) or
- exception_type is OSError and exception_value.errno == errno.ENOMEM):
- lg.critical("Unhandled exception is a fatal error, killing Node Manager")
- self._killfunc(os.getpid(), signal.SIGKILL)
- tracker.counter_add('actor_exceptions')
-
- def ping(self):
- return True
-
- def get_thread(self):
- return threading.current_thread()
-
-class WatchdogActor(pykka.ThreadingActor):
- def __init__(self, timeout, *args, **kwargs):
- super(pykka.ThreadingActor, self).__init__(*args, **kwargs)
- self.timeout = timeout
- self.actors = [a.proxy() for a in args]
- self.actor_ref = TellableActorRef(self)
- self._later = self.actor_ref.tell_proxy()
- self._killfunc = kwargs.get("killfunc", os.kill)
-
- def kill_self(self, e, act):
- lg = getattr(self, "_logger", logging)
- lg.critical("Watchdog exception", exc_info=e)
- lg.critical("Actor %s watchdog ping time out, killing Node Manager", act)
- self._killfunc(os.getpid(), signal.SIGKILL)
-
- def on_start(self):
- self._later.run()
-
- def run(self):
- a = None
- try:
- for a in self.actors:
- a.ping().get(self.timeout)
- time.sleep(20)
- self._later.run()
- except Exception as e:
- self.kill_self(e, a)
projects / arvados.git / blobdiff