[Webservice] Restructure database to give more meaningful success/failure information

Description

At present, the success/failure information we send to myMdC is unreliable in various important ways - see calibration/planning#104 . In addition, the correction_complete events sent via Kafka are sent per detector, but they don't actually tell you which detector they relate to (they give a detector type, such as JUNGFRAU, but not the Karabo ID of a specific detector). Replacing the webservice still looks like a big enough job that it seemed worth spending a bit of time improving the old one. I hope this might also ease the transition to the new one when it's ready.

The jobs database is completely redesigned, with three tables instead of one: each request from myMdC can have several executions, each of which can have several slurm_jobs. I think these terms line up with what Robert is using in Orca. I haven't tried to do any migration of existing data - we're only using it to track active jobs, so I'm assuming we'll just start with an empty database when deploying this change.

The jobs monitor code is also largely refactored, with the loop broken up into several methods in a JobsMonitor class.

How Has This Been Tested?

Deployed on max-exfl017, and submitted correction & dark requests from the test myMdC instance for the CALLAB proposal.

This revealed a number of previously hidden issues which are (hopefully) specific to CALLAB. The presence of JNGFR01 in the filenames triggers an execution for both FXE_XAD_JF1M and SPB_IRDA_JF4M, at least one of which will fail on any given run (as it can't find its data). Previously, the webservice would hide this, because the failing jobs would fail quickly, and the status would only come from the last jobs running, but with these changes the failures are visible.

Relevant Documents (optional)

https://in.xfel.eu/test_metadata/proposals/259/runs/63612

https://in.xfel.eu/test_metadata/proposals/259/runs/86193

https://in.xfel.eu/test_metadata/proposals/259#proposal-calibration

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Refactor (refactoring code with no functionality changes)

Checklist:

• My code follows the code style of this project.

Reviewers

@ahmedk @roscar @schmidtp

webservice/job_monitor.py

@@ -3,6 +3,7 @@ import argparse
 import json
 import locale
 import logging
+import signal
 import time
 from datetime import datetime, timezone
 from pathlib import Path
@@ -93,6 +94,13 @@ class JobsMonitor:
         self.kafka_topic = config['kafka']['topic']
         self.time_interval = int(config['web-service']['job-update-interval'])
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.job_db.close()
+        self.kafka_prod.close(timeout=5)
+
     def run(self):
         while True:
             try:
@@ -218,9 +226,9 @@ class JobsMonitor:
                 self.kafka_prod.send(self.kafka_topic, {
                     'event': 'correction_complete',
                     'proposal': r['proposal'],
-                    'run': r['run'],
+                    'run': str(r['run']),
                     'detector': r['det_type'],
-                    'karabo_id': r['karabo_id'],
+                    'detector_identifier': r['karabo_id'],
                     'success': success,
                 })
             except KafkaError:
@@ -249,7 +257,7 @@ class JobsMonitor:
                 self.kafka_prod.send(self.kafka_topic, {
                     'event': 'run_corrections_complete',
                     'proposal': r['proposal'],
-                    'run': r['run'],
+                    'run': str(r['run']),
                     'success': success,
                 })
             except KafkaError:
@@ -293,6 +301,9 @@ class JobsMonitor:
             log.error("Failed to update MDC dark run id %s", dark_run_id)
             log.error(Errors.MDC_RESPONSE.format(response))
 
+def interrupted(signum, frame):
+    raise KeyboardInterrupt
+
 def main(argv=None):
     # Ensure files are opened as UTF-8 by default, regardless of environment.
     locale.setlocale(locale.LC_CTYPE, ('en_US', 'UTF-8'))
@@ -318,7 +329,15 @@ def main(argv=None):
     )
     # DEBUG logs from kafka-python are very verbose, so we'll turn them off
     logging.getLogger('kafka').setLevel(logging.INFO)
-    JobsMonitor(config).run()
+
+    # Treat SIGTERM like SIGINT (Ctrl-C) & do a clean shutdown
+    signal.signal(signal.SIGTERM, interrupted)
+
+    with JobsMonitor(config) as jm:
+        try:
+            jm.run()
+        except KeyboardInterrupt:
+            logging.info("Shutting down on SIGINT/SIGTERM")
 
 
 if __name__ == "__main__":