1
2
3
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
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
37
38
39
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
51 for f in self.filters:
52 self.subscribe(f, self.last_log_id)
53
54 - def closed(self, code, reason=None):
57
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
72 self._closing = True
73
74 self._closed.wait(timeout=timeout)
75
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
83 self.send(json.dumps({"method": "unsubscribe", "filters": f}))
84
85
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
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
111
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
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
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
146
147 while not self.is_closed.wait(1):
148 pass
149
150
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
167 if self.last_log_id != None:
168
169
170 skip_old_events = [["id", ">", str(self.last_log_id)]]
171 else:
172
173
174 skip_old_events = False
175
176 self.on_event({'status': 200})
177
178 while not self._closing.is_set():
179 moreitems = False
180 for f in self.filters:
181 for tries_left in RetryLoop(num_retries=25, backoff_start=.1, max_wait=self.poll_time):
182 try:
183 if not skip_old_events:
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201 skip_old_events = [[
202 "created_at", ">=",
203 time.strftime(
204 "%Y-%m-%dT%H:%M:%SZ",
205 time.gmtime(time.time()-7200))]]
206 items = self.api.logs().list(
207 order="id desc",
208 limit=1,
209 filters=f+skip_old_events).execute()
210 if items["items"]:
211 skip_old_events = [
212 ["id", ">", str(items["items"][0]["id"])]]
213 items = {
214 "items": [],
215 "items_available": 0,
216 }
217 else:
218
219
220
221
222
223
224 items = self.api.logs().list(
225 order="id asc",
226 filters=f+skip_old_events).execute()
227 break
228 except errors.ApiError as error:
229 pass
230 else:
231 tries_left = 0
232 break
233 if tries_left == 0:
234 _logger.exception("PollClient thread could not contact API server.")
235 with self._closing_lock:
236 self._closing.set()
237 _thread.interrupt_main()
238 return
239 for i in items["items"]:
240 skip_old_events = [["id", ">", str(i["id"])]]
241 with self._closing_lock:
242 if self._closing.is_set():
243 return
244 try:
245 self.on_event(i)
246 except Exception as e:
247 _logger.exception("Unexpected exception from event callback.")
248 _thread.interrupt_main()
249 if items["items_available"] > len(items["items"]):
250 moreitems = True
251 if not moreitems:
252 self._closing.wait(self.poll_time)
253
255
256 while not self._closing.is_set():
257 self._closing.wait(1)
258
259 - def close(self, code=None, reason=None, timeout=0):
260 """Close poll client and optionally wait for it to finish.
261
262 If an :on_event: handler is running in a different thread,
263 first wait (indefinitely) for it to return.
264
265 After closing, wait up to :timeout: seconds for the thread to
266 finish the poll request in progress (if any).
267
268 :code: and :reason: are ignored. They are present for
269 interface compatibility with EventClient.
270 """
271
272 with self._closing_lock:
273 self._closing.set()
274 try:
275 self.join(timeout=timeout)
276 except RuntimeError:
277
278
279
280
281 pass
282
286
288 del self.filters[self.filters.index(f)]
289
290
292 endpoint = api._rootDesc.get('websocketUrl', None)
293 if not endpoint:
294 raise errors.FeatureNotEnabledError(
295 "Server does not advertise a websocket endpoint")
296 uri_with_token = "{}?api_token={}".format(endpoint, api.api_token)
297 try:
298 client = EventClient(uri_with_token, filters, on_event, last_log_id)
299 except Exception:
300 _logger.warning("Failed to connect to websockets on %s" % endpoint)
301 raise
302 else:
303 return client
304
305
306 -def subscribe(api, filters, on_event, poll_fallback=15, last_log_id=None):
307 """
308 :api:
309 a client object retrieved from arvados.api(). The caller should not use this client object for anything else after calling subscribe().
310 :filters:
311 Initial subscription filters.
312 :on_event:
313 The callback when a message is received.
314 :poll_fallback:
315 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.
316 :last_log_id:
317 Log rows that are newer than the log id
318 """
319
320 if not poll_fallback:
321 return _subscribe_websocket(api, filters, on_event, last_log_id)
322
323 try:
324 if not config.flag_is_true('ARVADOS_DISABLE_WEBSOCKETS'):
325 return _subscribe_websocket(api, filters, on_event, last_log_id)
326 else:
327 _logger.info("Using polling because ARVADOS_DISABLE_WEBSOCKETS is true")
328 except Exception as e:
329 _logger.warning("Falling back to polling after websocket error: %s" % e)
330 p = PollClient(api, filters, on_event, poll_fallback, last_log_id)
331 p.start()
332 return p
333