Merge branch '4751-node-manager-stricter-node-pairing-wip'
[arvados.git] / 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)