sdk/python/arvados/events.py

   1 from ws4py.client.threadedclient import WebSocketClient
   2 import threading
   3 import json
   4 import os
   5 import time
   6 import ssl
   7 import re
   8 import config
   9 import logging
  10 import arvados
  11
  12 _logger = logging.getLogger('arvados.events')
  13
  14 class EventClient(WebSocketClient):
  15     def __init__(self, url, filters, on_event):
  16         # Prefer system's CA certificates (if available)
  17         ssl_options = {}
  18         certs_path = '/etc/ssl/certs/ca-certificates.crt'
  19         if os.path.exists(certs_path):
  20             ssl_options['ca_certs'] = certs_path
  21         if config.flag_is_true('ARVADOS_API_HOST_INSECURE'):
  22             ssl_options['cert_reqs'] = ssl.CERT_NONE
  23         else:
  24             ssl_options['cert_reqs'] = ssl.CERT_REQUIRED
  25         super(EventClient, self).__init__(url, ssl_options=ssl_options)
  26         self.filters = filters
  27         self.on_event = on_event
  28
  29     def opened(self):
  30         self.subscribe(self.filters)
  31
  32     def received_message(self, m):
  33         self.on_event(json.loads(str(m)))
  34
  35     def close_connection(self):
  36         try:
  37             self.sock.shutdown(socket.SHUT_RDWR)
  38             self.sock.close()
  39         except:
  40             pass
  41
  42     def subscribe(self, filters, last_log_id=None):
  43         m = {"method": "subscribe", "filters": filters}
  44         if last_log_id is not None:
  45             m["last_log_id"] = last_log_id
  46         self.send(json.dumps(m))
  47
  48     def unsubscribe(self, filters):
  49         self.send(json.dumps({"method": "unsubscribe", "filters": filters}))
  50
  51 class PollClient(threading.Thread):
  52     def __init__(self, api, filters, on_event, poll_time):
  53         super(PollClient, self).__init__()
  54         self.api = api
  55         if filters:
  56             self.filters = [filters]
  57         else:
  58             self.filters = [[]]
  59         self.on_event = on_event
  60         self.poll_time = poll_time
  61         self.stop = threading.Event()
  62
  63     def run(self):
  64         self.id = 0
  65         for f in self.filters:
  66             items = self.api.logs().list(limit=1, order="id desc", filters=f).execute()['items']
  67             if items:
  68                 if items[0]['id'] > self.id:
  69                     self.id = items[0]['id']
  70
  71         self.on_event({'status': 200})
  72
  73         while not self.stop.isSet():
  74             max_id = self.id
  75             for f in self.filters:
  76                 items = self.api.logs().list(order="id asc", filters=f+[["id", ">", str(self.id)]]).execute()['items']
  77                 for i in items:
  78                     if i['id'] > max_id:
  79                         max_id = i['id']
  80                     self.on_event(i)
  81             self.id = max_id
  82             self.stop.wait(self.poll_time)
  83
  84     def run_forever(self):
  85         # Have to poll here, otherwise KeyboardInterrupt will never get processed.
  86         while not self.stop.is_set():
  87             self.stop.wait(1)
  88
  89     def close(self):
  90         self.stop.set()
  91         try:
  92             self.join()
  93         except RuntimeError:
  94             # "join() raises a RuntimeError if an attempt is made to join the
  95             # current thread as that would cause a deadlock. It is also an
  96             # error to join() a thread before it has been started and attempts
  97             # to do so raises the same exception."
  98             pass
  99
 100     def subscribe(self, filters):
 101         self.on_event({'status': 200})
 102         self.filters.append(filters)
 103
 104     def unsubscribe(self, filters):
 105         del self.filters[self.filters.index(filters)]
 106
 107
 108 def subscribe(api, filters, on_event, poll_fallback=15):
 109     '''
 110     api: Must be a newly created from arvados.api(cache=False), not shared with the caller, as it may be used by a background thread.
 111     filters: Initial subscription filters.
 112     on_event: The callback when a message is received
 113     poll_fallback: 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.
 114     '''
 115     ws = None
 116     if 'websocketUrl' in api._rootDesc:
 117         try:
 118             url = "{}?api_token={}".format(api._rootDesc['websocketUrl'], api.api_token)
 119             ws = EventClient(url, filters, on_event)
 120             ws.connect()
 121             return ws
 122         except Exception as e:
 123             _logger.warn("Got exception %s trying to connect to websockets at %s" % (e, api._rootDesc['websocketUrl']))
 124             if ws:
 125                 ws.close_connection()
 126     if poll_fallback:
 127         _logger.warn("Websockets not available, falling back to log table polling")
 128         p = PollClient(api, filters, on_event, poll_fallback)
 129         p.start()
 130         return p
 131     else:
 132         _logger.error("Websockets not available")
 133         return None