16169: Monkey patch load_tool.resolve_and_validate_document to fix bug
[arvados.git] / sdk / python / arvados / events.py
1 # Copyright (C) The Arvados Authors. All rights reserved.
2 #
3 # SPDX-License-Identifier: Apache-2.0
4
5 from __future__ import absolute_import
6 from future import standard_library
7 standard_library.install_aliases()
8 from builtins import str
9 from builtins import object
10 import arvados
11 from . import config
12 from . import errors
13 from .retry import RetryLoop
14
15 import logging
16 import json
17 import _thread
18 import threading
19 import time
20 import os
21 import re
22 import ssl
23 from ws4py.client.threadedclient import WebSocketClient
24
25 _logger = logging.getLogger('arvados.events')
26
27
28 class _EventClient(WebSocketClient):
29     def __init__(self, url, filters, on_event, last_log_id, on_closed):
30         ssl_options = {'ca_certs': arvados.util.ca_certs_path()}
31         if config.flag_is_true('ARVADOS_API_HOST_INSECURE'):
32             ssl_options['cert_reqs'] = ssl.CERT_NONE
33         else:
34             ssl_options['cert_reqs'] = ssl.CERT_REQUIRED
35
36         # Warning: If the host part of url resolves to both IPv6 and
37         # IPv4 addresses (common with "localhost"), only one of them
38         # will be attempted -- and it might not be the right one. See
39         # ws4py's WebSocketBaseClient.__init__.
40         super(_EventClient, self).__init__(url, ssl_options=ssl_options)
41
42         self.filters = filters
43         self.on_event = on_event
44         self.last_log_id = last_log_id
45         self._closing_lock = threading.RLock()
46         self._closing = False
47         self._closed = threading.Event()
48         self.on_closed = on_closed
49
50     def opened(self):
51         for f in self.filters:
52             self.subscribe(f, self.last_log_id)
53
54     def closed(self, code, reason=None):
55         self._closed.set()
56         self.on_closed()
57
58     def received_message(self, m):
59         with self._closing_lock:
60             if not self._closing:
61                 self.on_event(json.loads(str(m)))
62
63     def close(self, code=1000, reason='', timeout=0):
64         """Close event client and optionally wait for it to finish.
65
66         :timeout: is the number of seconds to wait for ws4py to
67         indicate that the connection has closed.
68         """
69         super(_EventClient, self).close(code, reason)
70         with self._closing_lock:
71             # make sure we don't process any more messages.
72             self._closing = True
73         # wait for ws4py to tell us the connection is closed.
74         self._closed.wait(timeout=timeout)
75
76     def subscribe(self, f, last_log_id=None):
77         m = {"method": "subscribe", "filters": f}
78         if last_log_id is not None:
79             m["last_log_id"] = last_log_id
80         self.send(json.dumps(m))
81
82     def unsubscribe(self, f):
83         self.send(json.dumps({"method": "unsubscribe", "filters": f}))
84
85
86 class EventClient(object):
87     def __init__(self, url, filters, on_event_cb, last_log_id):
88         self.url = url
89         if filters:
90             self.filters = [filters]
91         else:
92             self.filters = [[]]
93         self.on_event_cb = on_event_cb
94         self.last_log_id = last_log_id
95         self.is_closed = threading.Event()
96         self._setup_event_client()
97
98     def _setup_event_client(self):
99         self.ec = _EventClient(self.url, self.filters, self.on_event,
100                                self.last_log_id, self.on_closed)
101         self.ec.daemon = True
102         try:
103             self.ec.connect()
104         except Exception:
105             self.ec.close_connection()
106             raise
107
108     def subscribe(self, f, last_log_id=None):
109         self.filters.append(f)
110         self.ec.subscribe(f, last_log_id)
111
112     def unsubscribe(self, f):
113         del self.filters[self.filters.index(f)]
114         self.ec.unsubscribe(f)
115
116     def close(self, code=1000, reason='', timeout=0):
117         self.is_closed.set()
118         self.ec.close(code, reason, timeout)
119
120     def on_event(self, m):
121         if m.get('id') != None:
122             self.last_log_id = m.get('id')
123         try:
124             self.on_event_cb(m)
125         except Exception as e:
126             _logger.exception("Unexpected exception from event callback.")
127             _thread.interrupt_main()
128
129     def on_closed(self):
130         if not self.is_closed.is_set():
131             _logger.warning("Unexpected close. Reconnecting.")
132             for tries_left in RetryLoop(num_retries=25, backoff_start=.1, max_wait=15):
133                 try:
134                     self._setup_event_client()
135                     _logger.warning("Reconnect successful.")
136                     break
137                 except Exception as e:
138                     _logger.warning("Error '%s' during websocket reconnect.", e)
139             if tries_left == 0:
140                 _logger.exception("EventClient thread could not contact websocket server.")
141                 self.is_closed.set()
142                 _thread.interrupt_main()
143                 return
144
145     def run_forever(self):
146         # Have to poll here to let KeyboardInterrupt get raised.
147         while not self.is_closed.wait(1):
148             pass
149
150
151 class PollClient(threading.Thread):
152     def __init__(self, api, filters, on_event, poll_time, last_log_id):
153         super(PollClient, self).__init__()
154         self.api = api
155         if filters:
156             self.filters = [filters]
157         else:
158             self.filters = [[]]
159         self.on_event = on_event
160         self.poll_time = poll_time
161         self.daemon = True
162         self.last_log_id = last_log_id
163         self._closing = threading.Event()
164         self._closing_lock = threading.RLock()
165
166         if self.last_log_id != None:
167             # Caller supplied the last-seen event ID from a previous
168             # connection.
169             self._skip_old_events = [["id", ">", str(self.last_log_id)]]
170         else:
171             # We need to do a reverse-order query to find the most
172             # recent event ID (see "if not self._skip_old_events"
173             # in run()).
174             self._skip_old_events = False
175
176     def run(self):
177         self.on_event({'status': 200})
178
179         while not self._closing.is_set():
180             moreitems = False
181             for f in self.filters:
182                 for tries_left in RetryLoop(num_retries=25, backoff_start=.1, max_wait=self.poll_time):
183                     try:
184                         if not self._skip_old_events:
185                             # If the caller didn't provide a known
186                             # recent ID, our first request will ask
187                             # for the single most recent event from
188                             # the last 2 hours (the time restriction
189                             # avoids doing an expensive database
190                             # query, and leaves a big enough margin to
191                             # account for clock skew). If we do find a
192                             # recent event, we remember its ID but
193                             # then discard it (we are supposed to be
194                             # returning new/current events, not old
195                             # ones).
196                             #
197                             # Subsequent requests will get multiple
198                             # events in chronological order, and
199                             # filter on that same cutoff time, or
200                             # (once we see our first matching event)
201                             # the ID of the last-seen event.
202                             #
203                             # Note: self._skip_old_events must not be
204                             # set until the threshold is decided.
205                             # Otherwise, tests will be unreliable.
206                             filter_by_time = [[
207                                 "created_at", ">=",
208                                 time.strftime(
209                                     "%Y-%m-%dT%H:%M:%SZ",
210                                     time.gmtime(time.time()-7200))]]
211                             items = self.api.logs().list(
212                                 order="id desc",
213                                 limit=1,
214                                 filters=f+filter_by_time).execute()
215                             if items["items"]:
216                                 self._skip_old_events = [
217                                     ["id", ">", str(items["items"][0]["id"])]]
218                                 items = {
219                                     "items": [],
220                                     "items_available": 0,
221                                 }
222                             else:
223                                 # No recent events. We can keep using
224                                 # the same timestamp threshold until
225                                 # we receive our first new event.
226                                 self._skip_old_events = filter_by_time
227                         else:
228                             # In this case, either we know the most
229                             # recent matching ID, or we know there
230                             # were no matching events in the 2-hour
231                             # window before subscribing. Either way we
232                             # can safely ask for events in ascending
233                             # order.
234                             items = self.api.logs().list(
235                                 order="id asc",
236                                 filters=f+self._skip_old_events).execute()
237                         break
238                     except errors.ApiError as error:
239                         pass
240                     else:
241                         tries_left = 0
242                         break
243                 if tries_left == 0:
244                     _logger.exception("PollClient thread could not contact API server.")
245                     with self._closing_lock:
246                         self._closing.set()
247                     _thread.interrupt_main()
248                     return
249                 for i in items["items"]:
250                     self._skip_old_events = [["id", ">", str(i["id"])]]
251                     with self._closing_lock:
252                         if self._closing.is_set():
253                             return
254                         try:
255                             self.on_event(i)
256                         except Exception as e:
257                             _logger.exception("Unexpected exception from event callback.")
258                             _thread.interrupt_main()
259                 if items["items_available"] > len(items["items"]):
260                     moreitems = True
261             if not moreitems:
262                 self._closing.wait(self.poll_time)
263
264     def run_forever(self):
265         # Have to poll here, otherwise KeyboardInterrupt will never get processed.
266         while not self._closing.is_set():
267             self._closing.wait(1)
268
269     def close(self, code=None, reason=None, timeout=0):
270         """Close poll client and optionally wait for it to finish.
271
272         If an :on_event: handler is running in a different thread,
273         first wait (indefinitely) for it to return.
274
275         After closing, wait up to :timeout: seconds for the thread to
276         finish the poll request in progress (if any).
277
278         :code: and :reason: are ignored. They are present for
279         interface compatibility with EventClient.
280         """
281
282         with self._closing_lock:
283             self._closing.set()
284         try:
285             self.join(timeout=timeout)
286         except RuntimeError:
287             # "join() raises a RuntimeError if an attempt is made to join the
288             # current thread as that would cause a deadlock. It is also an
289             # error to join() a thread before it has been started and attempts
290             # to do so raises the same exception."
291             pass
292
293     def subscribe(self, f):
294         self.on_event({'status': 200})
295         self.filters.append(f)
296
297     def unsubscribe(self, f):
298         del self.filters[self.filters.index(f)]
299
300
301 def _subscribe_websocket(api, filters, on_event, last_log_id=None):
302     endpoint = api._rootDesc.get('websocketUrl', None)
303     if not endpoint:
304         raise errors.FeatureNotEnabledError(
305             "Server does not advertise a websocket endpoint")
306     uri_with_token = "{}?api_token={}".format(endpoint, api.api_token)
307     try:
308         client = EventClient(uri_with_token, filters, on_event, last_log_id)
309     except Exception:
310         _logger.warning("Failed to connect to websockets on %s" % endpoint)
311         raise
312     else:
313         return client
314
315
316 def subscribe(api, filters, on_event, poll_fallback=15, last_log_id=None):
317     """
318     :api:
319       a client object retrieved from arvados.api(). The caller should not use this client object for anything else after calling subscribe().
320     :filters:
321       Initial subscription filters.
322     :on_event:
323       The callback when a message is received.
324     :poll_fallback:
325       If websockets are not available, fall back to polling every N seconds.  If poll_fallback=False, this will return None if websockets are not available.
326     :last_log_id:
327       Log rows that are newer than the log id
328     """
329
330     if not poll_fallback:
331         return _subscribe_websocket(api, filters, on_event, last_log_id)
332
333     try:
334         if not config.flag_is_true('ARVADOS_DISABLE_WEBSOCKETS'):
335             return _subscribe_websocket(api, filters, on_event, last_log_id)
336         else:
337             _logger.info("Using polling because ARVADOS_DISABLE_WEBSOCKETS is true")
338     except Exception as e:
339         _logger.warning("Falling back to polling after websocket error: %s" % e)
340     p = PollClient(api, filters, on_event, poll_fallback, last_log_id)
341     p.start()
342     return p