Send Kafka notifications when correction is complete

Description

We're experimenting with the migration system sending Kafka notifications, including when migration has completed and a raw run is available on Maxwell ('event': 'migration_complete'). It would be nice to have similar notifications when a corrected run is available, and the calibration webservice is the obvious place to send those.

This is very minimal at the moment. It would be nice to send more information (migration events also have run directory, instrument & cycle), but this isn't easy without changing the sqlite database, which is a bigger change than I wanted to make at the moment. It could be extended to add events when correction starts, and events for characterisation (e.g. dark processing).

It will also need some work to handle multiple kinds of processing running on the same run. We probably want events when each piece of processing completes (with a robust way to identify which is which), and another event when everything that writes to proc/r0123/ has finished. But we shouldn't let the perfect be the enemy of the good, IMO.

How Has This Been Tested?

Run on max-exfl017, re-corrected a CALLAB run while listening for the event. This is the result I got:

{'event': 'correction_complete',
 'proposal': '900113',
 'run': '9998',
 'detector': 'lpd',
 'success': True}

Types of changes

• New feature (non-breaking change which adds functionality)

Checklist:

• My code follows the code style of this project.

Reviewers

@danilevc @ahmedk

webservice/listen_kafka.py

+"""Print Kafka events sent by the webservice.
+"""
+import json
+import os.path as osp
+
+import yaml
+from kafka import KafkaConsumer
+
+conf_file = osp.join(osp.dirname(__file__), 'webservice.yaml')
+
+with open(conf_file, "r") as f:
+    config = yaml.safe_load(f)
+
+topic = config['kafka']['topic']
+brokers = config['kafka']['brokers']
+print(f"topic: {topic!r}")
+print(f"brokers: {brokers!r}")
+
+kc = KafkaConsumer(topic, bootstrap_servers=brokers)
+
+print("Waiting for messages (Ctrl-C to quit)...")
+try:
+    for record in kc:
+        msg = json.loads(record.value.decode())
+        print(msg)
+except KeyboardInterrupt:
+    print("Exiting")
+    kc.close()