[Webservice] Fix database lock timeouts

Description

We have been seeing errors in the webservice where it tries to write to the database and times out on a lock. I found 12 instances in the log, with tracebacks pointing to 3 different INSERT statements.

Traceback (most recent call last):                                                                                                                                                                        
  File "/home/xcal/deployments/development/git.xfel.eu/detectors/pycalibration/pycalibration-2023-03-22-3.9.2-9e88f31/webservice/webservice.py", line 1034, in _continue
    ret, _ = await self.launch_jobs(                                                                                                                                                                      
  File "/home/xcal/deployments/development/git.xfel.eu/detectors/pycalibration/pycalibration-2023-03-22-3.9.2-9e88f31/webservice/webservice.py", line 1397, in launch_jobs
    cur = self.job_db.execute(                                                                                                                                                                            
sqlite3.OperationalError: database is locked

This indicates that something running concurrently with the webservice is keeping the database locked (i.e. holding a transaction open) for several seconds. The jobs monitor and the web overview (http://max-exfl016.desy.de:8008/) are the only other things that routinely access this database.

I think I've found the culprit in the job monitor code. When updating job statuses, it was issuing SQL updates in a loop which also runs sacct to get job statuses, and then committing (finishing the transaction) outside the loop. If a lot of jobs had finished since the last update, or sacct was slow for some reason, this would mean the database was locked until it finished. I've addressed this by batching the updates needed and writing to the database outside the loop.

I also added some timing code wherever we write to the database, so we can see if it's taking longer than expected.

How Has This Been Tested?

TBD

Types of changes

• Bug fix (non-breaking change which fixes an issue)

Checklist:

• My code follows the code style of this project.

Reviewers

@ahmedk @schmidtp

webservice/job_monitor.py

@@ -15,11 +15,11 @@ from kafka.errors import KafkaError
 try:
     from .config import webservice as config
     from .messages import MDC, Errors, MigrationError, Success
-    from .webservice import init_job_db, init_md_client
+    from .webservice import init_job_db, init_md_client, time_db_transaction
 except ImportError:
     from config import webservice as config
     from messages import MDC, Errors, MigrationError, Success
-    from webservice import init_job_db, init_md_client
+    from webservice import init_job_db, init_md_client, time_db_transaction
 
 log = logging.getLogger(__name__)
 
@@ -148,17 +148,18 @@ class JobsMonitor:
 
         Newly completed executions are present with an empty list.
         """
-        c = self.job_db.cursor()
-        c.execute("SELECT job_id, exec_id FROM slurm_jobs WHERE finished = 0")
-
         statii = slurm_status()
         # Check that slurm is giving proper feedback
         if statii is None:
             return {}
         log.debug(f"SLURM info {statii}")
 
+        jobs_to_check = self.job_db.execute(
+            "SELECT job_id, exec_id FROM slurm_jobs WHERE finished = 0"
+        ).fetchall()
         ongoing_jobs_by_exn = {}
-        for r in c.fetchall():
+        updates = []
+        for r in jobs_to_check:
             log.debug(f"Job in DB before update: %s", tuple(r))
             execn_ongoing_jobs = ongoing_jobs_by_exn.setdefault(r['exec_id'], [])
 
@@ -173,12 +174,14 @@ class JobsMonitor:
                 _, runtime, slstatus = slurm_job_status(r['job_id'])
                 finished = True
 
-            c.execute(
+            updates.append((finished, runtime, slstatus, r['job_id']))
+
+        with time_db_transaction(self.job_db, 'Update jobs'):
+            self.job_db.executemany(
                 "UPDATE slurm_jobs SET finished=?, elapsed=?, status=? WHERE job_id = ?",
-                (finished, runtime, slstatus, r['job_id'])
+                updates
             )
 
-        self.job_db.commit()
         return ongoing_jobs_by_exn
 
     def process_request_still_going(self, req_id, running_jobs_info):
@@ -216,7 +219,7 @@ class JobsMonitor:
             "WHERE exec_id = ?",
             (exec_id,)
         ).fetchone()
-        with self.job_db:
+        with time_db_transaction(self.job_db, 'Update execution'):
             self.job_db.execute(
                 "UPDATE executions SET success = ? WHERE exec_id = ?",
                 (success, exec_id)