standalone task event monitor app

inventory_provider/tasks/monitor.py

+import json
+import logging
+from inventory_provider.tasks.worker import app
+from inventory_provider.tasks.common import get_current_redis
+
+from threading import Thread
+
+logger = logging.getLogger(__name__)
+INFO_EVENT_TYPES = ('task-info', 'task-warning', 'task-error')
+
+
+def _monitor_proc(config):
+
+    state = app.events.State()
+
+    def _log_event(event):
+        state.event(event)
+
+        if not event['type'].startswith('task-'):
+            return
+
+        key = f'joblog:{event["uuid"]}:{event["type"]}'
+        if event['type'] in INFO_EVENT_TYPES:
+            key += f':{event["clock"]}'
+
+        r = get_current_redis(config)
+        r.set(key, json.dumps(event))
+
+        logger.debug(f'{key}: {json.dumps(event)}')
+
+    with app.connection() as connection:
+        recv = app.events.Receiver(connection, handlers={
+                '*': _log_event,
+        })
+        recv.capture(limit=None, timeout=None, wakeup=True)
+
+
+def start_monitoring():
+    thread = Thread(target=_monitor_proc, args=())
+    thread.start()
+
+
+def clear_joblog(r):
+    """
+
+    :param r:
+    :return:
+    """
+    rp = r.pipeline()
+    for key in r.scan_iter('joblog:*'):
+        rp.delete(key)
+    rp.execute()
+
+
+if __name__ == '__main__':
+    import os
+    from inventory_provider import config
+
+    logging.basicConfig(level=logging.DEBUG)
+
+    filename = os.path.join(
+        os.path.dirname(__file__),
+        '..',
+        'config-sentinel.json')
+
+    with open(filename) as f:
+        params = config.load(f)
+
+    _monitor_proc(params)