Store CalCat requests/responses for reproducibility

Description

This directs CalCat queries through a proxy server which records the responses and replays them if the same queries are made again. This aims to improve reproducibility, by avoiding the need to talk to an external service when reproducing corrections, without duplicating all of the query code in a 'pre' notebook.

The proxy server code is in the calparrot repository, which should be reviewed alongside this.

The proxy server is launched on each allocated correction node, using a shared SQLite database (relying on SQLite's file locking to coordinate access). The first node to make a given request adds a row with a response status of -1 before making the request; other nodes that see this will wait up to 15 seconds for a response in the database before making their own request.

The proxy server picks a random port to listen on, and writes this in a file which we read back (I've tried to come up with a more elegant way to do this, but I haven't found any I like so far) EDIT: now the proxy server runs princess as a subprocess. We use environment variables to modify the cal_tools.restful_config settings so that notebooks will use the proxy server.

Potential risk: if we one day rearrange the constant HDF5 files, updating CalCat, we won't be able to reproduce corrections from before, because the paths are saved. However, changing the root directory (currently /gpfs/exfel/d/cal/caldb_store) is relatively easy, as this is not part of the CalCat responses.

How Has This Been Tested?

I have run an LPD correction using this; the results are under

/gpfs/exfel/data/scratch/kluyvert/lpd-corr-p2808-r136-parrot

The metadata directory (slurm_out_...) includes calparrot.sqlite, with 32 entries - light & dark conditions for each of 16 LPD modules.

For now, this will only affect LPD correction, since the other detector corrections all use cal_db_remote, which isn't affected by this. !738 (merged) switches ePix100 over, and we plan to switch all detectors eventually.

Types of changes

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

Checklist:

• My code follows the code style of this project.
• I have updated the documentation accordingly.
• I added tests where appropriate.

Reviewers

src/xfel_calibrate/calibrate.py

@@ -385,11 +385,14 @@ class JobArgs:
         """Run this job in a local process, return exit status"""
         return call(self.format_cmd(python), cwd=work_dir)
 
-    def submit_job(self, work_dir, python, slurm_opts, after_ok=(), after_any=()):
+    def submit_job(
+            self, work_dir, python, slurm_opts, after_ok=(), after_any=(), env=None
+    ):
         """Submit this job to Slurm, return its job ID"""
         cmd = slurm_opts.get_launcher_command(work_dir, after_ok, after_any)
         cmd += self.format_cmd(python)
-        output = check_output(cmd, cwd=work_dir).decode('utf-8')
+        # sbatch propagates environment variables into the job by default
+        output = check_output(cmd, cwd=work_dir, env=env).decode('utf-8')
         return output.partition(';')[0].strip()  # job ID
 
 
@@ -440,7 +443,7 @@ class JobChain:
                 'steps': [step.to_dict() for step in self.steps]
             }, f, indent=2)
 
-    def submit_jobs(self, slurm_opts: SlurmOptions):
+    def submit_jobs(self, slurm_opts: SlurmOptions, env=None):
         """Submit these jobs to Slurm, return a list of job IDs
 
         Slurm dependencies are used to manage the sequence of jobs.
@@ -453,7 +456,9 @@ class JobChain:
             step_job_ids = []
             kw = {('after_any' if step.after_error else 'after_ok'): dep_job_ids}
             for job_desc in step.jobs:
-                jid = job_desc.submit_job(self.work_dir, self.python, slurm_opts, **kw)
+                jid = job_desc.submit_job(
+                    self.work_dir, self.python, slurm_opts, env=env, **kw
+                )
                 step_job_ids.append(jid)
             dep_job_ids = step_job_ids
             all_job_ids.extend(step_job_ids)