services/nodemanager/arvnodeman/baseactor.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4
   5 from __future__ import absolute_import, print_function
   6
   7 import errno
   8 import logging
   9 import os
  10 import signal
  11 import time
  12 import threading
  13 import traceback
  14
  15 import pykka
  16
  17 class _TellCallableProxy(object):
  18     """Internal helper class for proxying callables."""
  19
  20     def __init__(self, ref, attr_path):
  21         self.actor_ref = ref
  22         self._attr_path = attr_path
  23
  24     def __call__(self, *args, **kwargs):
  25         message = {
  26             'command': 'pykka_call',
  27             'attr_path': self._attr_path,
  28             'args': args,
  29             'kwargs': kwargs,
  30         }
  31         self.actor_ref.tell(message)
  32
  33
  34 class TellActorProxy(pykka.ActorProxy):
  35     """ActorProxy in which all calls are implemented as using tell().
  36
  37     The standard pykka.ActorProxy always uses ask() and returns a Future.  If
  38     the target method raises an exception, it is placed in the Future object
  39     and re-raised when get() is called on the Future.  Unfortunately, most
  40     messaging in Node Manager is asynchronous and the caller does not store the
  41     Future object returned by the call to ActorProxy.  As a result, exceptions
  42     resulting from these calls end up in limbo, neither reported in the logs
  43     nor handled by on_failure().
  44
  45     The TellActorProxy uses tell() instead of ask() and does not return a
  46     Future object.  As a result, if the target method raises an exception, it
  47     will be logged and on_failure() will be called as intended.
  48
  49     """
  50
  51     def __repr__(self):
  52         return '<ActorProxy for %s, attr_path=%s>' % (
  53             self.actor_ref, self._attr_path)
  54
  55     def __getattr__(self, name):
  56         """Get a callable from the actor."""
  57         attr_path = self._attr_path + (name,)
  58         if attr_path not in self._known_attrs:
  59             self._known_attrs = self._get_attributes()
  60         attr_info = self._known_attrs.get(attr_path)
  61         if attr_info is None:
  62             raise AttributeError('%s has no attribute "%s"' % (self, name))
  63         if attr_info['callable']:
  64             if attr_path not in self._callable_proxies:
  65                 self._callable_proxies[attr_path] = _TellCallableProxy(
  66                     self.actor_ref, attr_path)
  67             return self._callable_proxies[attr_path]
  68         else:
  69             raise AttributeError('attribute "%s" is not a callable on %s' % (name, self))
  70
  71 class TellableActorRef(pykka.ActorRef):
  72     """ActorRef adding the tell_proxy() method to get TellActorProxy."""
  73
  74     def tell_proxy(self):
  75         return TellActorProxy(self)
  76
  77 class BaseNodeManagerActor(pykka.ThreadingActor):
  78     """Base class for actors in node manager, redefining actor_ref as a
  79     TellableActorRef and providing a default on_failure handler.
  80     """
  81
  82     def __init__(self, *args, **kwargs):
  83          super(pykka.ThreadingActor, self).__init__(*args, **kwargs)
  84          self.actor_ref = TellableActorRef(self)
  85          self._killfunc = kwargs.get("killfunc", os.kill)
  86
  87     def on_failure(self, exception_type, exception_value, tb):
  88         lg = getattr(self, "_logger", logging)
  89         if (exception_type in (threading.ThreadError, MemoryError) or
  90             exception_type is OSError and exception_value.errno == errno.ENOMEM):
  91             lg.critical("Unhandled exception is a fatal error, killing Node Manager")
  92             self._killfunc(os.getpid(), signal.SIGKILL)
  93
  94     def ping(self):
  95         return True
  96
  97     def get_thread(self):
  98         return threading.current_thread()
  99
 100 class WatchdogActor(pykka.ThreadingActor):
 101     def __init__(self, timeout, *args, **kwargs):
 102          super(pykka.ThreadingActor, self).__init__(*args, **kwargs)
 103          self.timeout = timeout
 104          self.actors = [a.proxy() for a in args]
 105          self.actor_ref = TellableActorRef(self)
 106          self._later = self.actor_ref.tell_proxy()
 107          self._killfunc = kwargs.get("killfunc", os.kill)
 108
 109     def kill_self(self, e, act):
 110         lg = getattr(self, "_logger", logging)
 111         lg.critical("Watchdog exception", exc_info=e)
 112         lg.critical("Actor %s watchdog ping time out, killing Node Manager", act)
 113         self._killfunc(os.getpid(), signal.SIGKILL)
 114
 115     def on_start(self):
 116         self._later.run()
 117
 118     def run(self):
 119         a = None
 120         try:
 121             for a in self.actors:
 122                 a.ping().get(self.timeout)
 123             time.sleep(20)
 124             self._later.run()
 125         except Exception as e:
 126             self.kill_self(e, a)