services/nodemanager/arvnodeman/clientactor.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import absolute_import, print_function
   4
   5 import logging
   6 import time
   7
   8 import pykka
   9
  10 from .config import actor_class
  11
  12 def _notify_subscribers(response, subscribers):
  13     """Send the response to all the subscriber methods.
  14
  15     If any of the subscriber actors have stopped, remove them from the
  16     subscriber set.
  17     """
  18     dead_subscribers = set()
  19     for subscriber in subscribers:
  20         try:
  21             subscriber(response)
  22         except pykka.ActorDeadError:
  23             dead_subscribers.add(subscriber)
  24     subscribers.difference_update(dead_subscribers)
  25
  26 class RemotePollLoopActor(actor_class):
  27     """Abstract actor class to regularly poll a remote service.
  28
  29     This actor sends regular requests to a remote service, and sends each
  30     response to subscribers.  It takes care of error handling, and retrying
  31     requests with exponential backoff.
  32
  33     To use this actor, define the _send_request method.  If you also
  34     define an _item_key method, this class will support subscribing to
  35     a specific item by key in responses.
  36     """
  37     def __init__(self, client, timer_actor, poll_wait=60, max_poll_wait=180):
  38         super(RemotePollLoopActor, self).__init__()
  39         self._client = client
  40         self._timer = timer_actor
  41         self._logger = logging.getLogger(self.LOGGER_NAME)
  42         self._later = self.actor_ref.proxy()
  43         self._polling_started = False
  44         self.log_prefix = "{} (at {})".format(self.__class__.__name__, id(self))
  45         self.min_poll_wait = poll_wait
  46         self.max_poll_wait = max_poll_wait
  47         self.poll_wait = self.min_poll_wait
  48         self.all_subscribers = set()
  49         self.key_subscribers = {}
  50         if hasattr(self, '_item_key'):
  51             self.subscribe_to = self._subscribe_to
  52
  53     def _start_polling(self):
  54         if not self._polling_started:
  55             self._polling_started = True
  56             self._later.poll()
  57
  58     def subscribe(self, subscriber):
  59         self.all_subscribers.add(subscriber)
  60         self._logger.debug("%r subscribed to all events", subscriber)
  61         self._start_polling()
  62
  63     # __init__ exposes this method to the proxy if the subclass defines
  64     # _item_key.
  65     def _subscribe_to(self, key, subscriber):
  66         self.key_subscribers.setdefault(key, set()).add(subscriber)
  67         self._logger.debug("%r subscribed to events for '%s'", subscriber, key)
  68         self._start_polling()
  69
  70     def _send_request(self):
  71         raise NotImplementedError("subclasses must implement request method")
  72
  73     def _got_response(self, response):
  74         self._logger.debug("%s got response with %d items",
  75                            self.log_prefix, len(response))
  76         self.poll_wait = self.min_poll_wait
  77         _notify_subscribers(response, self.all_subscribers)
  78         if hasattr(self, '_item_key'):
  79             items = {self._item_key(x): x for x in response}
  80             for key, subscribers in self.key_subscribers.iteritems():
  81                 _notify_subscribers(items.get(key), subscribers)
  82
  83     def _got_error(self, error):
  84         self.poll_wait = min(self.poll_wait * 2, self.max_poll_wait)
  85         return "{} got error: {} - waiting {} seconds".format(
  86             self.log_prefix, error, self.poll_wait)
  87
  88     def is_common_error(self, exception):
  89         return False
  90
  91     def poll(self, scheduled_start=None):
  92         self._logger.debug("%s sending poll", self.log_prefix)
  93         start_time = time.time()
  94         if scheduled_start is None:
  95             scheduled_start = start_time
  96         try:
  97             response = self._send_request()
  98         except Exception as error:
  99             errmsg = self._got_error(error)
 100             if self.is_common_error(error):
 101                 self._logger.warning(errmsg)
 102             else:
 103                 self._logger.exception(errmsg)
 104             next_poll = start_time + self.poll_wait
 105         else:
 106             self._got_response(response)
 107             next_poll = scheduled_start + self.poll_wait
 108         end_time = time.time()
 109         if next_poll < end_time:  # We've drifted too much; start fresh.
 110             next_poll = end_time + self.poll_wait
 111         self._timer.schedule(next_poll, self._later.poll, next_poll)