src/xfel_calibrate/finalize.py

@@ -5,7 +5,7 @@ from importlib.machinery import SourceFileLoader
 from os import chdir, listdir, path
 from pathlib import Path
 from shutil import copy, copytree, move, rmtree
-from subprocess import CalledProcessError, check_call, check_output
+from subprocess import PIPE, CalledProcessError, check_call, run
 from tempfile import TemporaryDirectory
 from textwrap import dedent
 from typing import Dict, List
@@ -129,10 +129,17 @@ def get_job_info(jobs: List[str], fmt: List[str]) -> List[List[str]]:
 
     # will use JobID to match results to jobs (duplicate field in fmt is OK)
     fmt_query = ",".join(["JobID"] + fmt)
-    sacct_out = check_output(["sacct", "--truncate", "--parsable2", "--noheader",
-                              f"--jobs={','.join(jobs)}",
-                              f"--format={fmt_query}"])
-    lines = sacct_out.decode().split("\n")
+    res = run([
+        "sacct", "--truncate", "--parsable2", "--noheader",
+        f"--jobs={','.join(jobs)}", f"--format={fmt_query}"
+    ], stdout=PIPE, text=True)
+    if res.returncode == 0:
+        lines = res.stdout.split("\n")
+    else:
+        print(f"Call to sacct failed (exit status {res.returncode})", flush=True)
+        if res.stdout:
+            print(res.stdout, flush=True)
+        lines = []
 
     missing_info = ["not-found"] * len(fmt)
     job_info = {job: missing_info for job in jobs}

webservice/job_monitor.py

@@ -8,7 +8,7 @@ import shlex
 import signal
 import time
 from bisect import insort_left
-from collections import deque
+from collections import deque, defaultdict
 from datetime import datetime, timezone
 from pathlib import Path
 from subprocess import run, PIPE
@@ -145,7 +145,7 @@ class JobsMonitor:
         self.kafka_prod = init_kafka_producer(config)
         self.kafka_topic = config['kafka']['topic']
         self.time_interval = int(config['web-service']['job-update-interval'])
-        self.instant_fails_by_host = defauldict(ExpiringEvents)
+        self.instant_fails_by_host = defaultdict(ExpiringEvents)
 
     def __enter__(self):
         return self