Skip to content
Snippets Groups Projects

[webservice] feat: Processing calibration logs and displaying errors/warnings on myMDC

Open [webservice] feat: Processing calibration logs and displaying errors/warnings on myMDC
feat/webservice_process_cal_logs into master
Open Karim Ahmed requested to merge feat/webservice_process_cal_logs into master
Viewing commit f4555bd5
Show latest version
1 file
+ 34
− 74
Compare changes
  • Side-by-side
  • Inline
  • f4555bd5
    Karim Ahmed
    Improve log display · f4555bd5
    Karim Ahmed authored 
    - Remove unneeded text
- Show only known errors
- Show warnings for COMPLETED jobs only
webservice/job_monitor.py
+ 34
− 74
@@ -106,20 +106,6 @@ def slurm_job_status(jobid):
return "NA", "NA", "NA"
def parse_log_file(file_path):
results = []
with open(file_path, 'r') as file:
for line in file:
try:
log_entry = json.loads(line.strip())
error_message = log_entry.get('message', '')
error_class = log_entry.get('class', '')
results.append((error_message, error_class))
except json.JSONDecodeError:
log.error(f"Skipping invalid JSON: {line.strip()}")
return results
def get_report_dir(command):
args = shlex.split(command)
try:
@@ -129,56 +115,30 @@ def get_report_dir(command):
return
def process_log_file(job_id, karabo_id, report_dir, karabo_id_log, file):
if file.exists():
with open(file, 'r') as f:
for line in f:
try:
json_line = json.loads(line)
if "default" not in json_line['class'].lower():
message = json_line['message']
if message:
karabo_id_log.setdefault(
message, []).append(job_id)
except json.JSONDecodeError:
log.error(
f"Invalid JSON in errors file {file}:"
f" {line.strip()}")
return karabo_id_log
def compress_job_ids(job_ids):
"""Compress list of job IDs to a shorter representation.
Args:
job_ids (list): List of job IDs
Returns:
str: Compressed representation like "16 jobs (11498126-11498141)"
or "2 jobs (11498142, 11498143)" for non-sequential IDs
"""
if not job_ids:
return "0 jobs"
# Convert to integers and sort
ids = sorted(int(id) for id in job_ids)
# Check if they're sequential
if len(ids) > 2 and ids[-1] - ids[0] + 1 == len(ids):
return f"{len(ids)} jobs ({ids[0]}-{ids[-1]})"
if len(ids) > 4:
return f"{len(ids)} jobs (e.g. {ids[0]}, {ids[1]}...)"
return f"{len(ids)} jobs ({', '.join(str(id) for id in ids)})"
def process_log_file(job_id,karabo_id_log, file):
with open(file, 'r') as f:
for line in f:
try:
json_line = json.loads(line)
message = json_line['level'] + json_line['message']
log_class = json_line['log_class'].strip(".")
if message:
if not any(["calibration" in c.lower() for c in log_class]):
message = "Error: Uncaught error!"
karabo_id_log.setdefault(
message, []).append(job_id)
except json.JSONDecodeError:
log.error(
f"Invalid JSON in errors file {file}:"
f" {line.strip()}")
def format_log_message(errors):
def format_log_message(logs):
"""Format log messages with compressed job IDs."""
formatted = {}
for karabo_id, messages in errors.items():
for karabo_id, messages in logs.items():
formatted[karabo_id] = {
msg: compress_job_ids(job_ids)
msg: f" {len(job_ids)} jobs"
for msg, job_ids in messages.items()
}
return formatted
@@ -390,27 +350,27 @@ class JobsMonitor:
for job_id, status in job_statuses:
if status == "COMPLETED":
continue
# No need to process warnings for failed jobs.
warning_file = report_dir / f"warnings_{job_id}.log"
if warning_file.exists():
karabo_id_warn = warnings.setdefault(karabo_id, {})
process_log_file(job_id, karabo_id_warn, warning_file)
continue # No errors expected for COMPLETED jobs.
if not exec_success: # no errors expected for successful execution.
karabo_id_err = errors.setdefault(karabo_id, {})
if status == "FAILED": # process error logs
error_file = report_dir / f"errors_{job_id}.log"
process_log_file(
job_id, karabo_id, report_dir, karabo_id_err, error_file)
if error_file.exists():
process_log_file(job_id, karabo_id_err, error_file)
if len(karabo_id_err) == 0:
log.warning(f"Job {job_id} failed but no error log/messages found.")
karabo_id_err.setdefault(
"Job failed but no error logs found", []).append(job_id)
"Error: Job failed but no error logs found", []).append(job_id)
else: # Job unsucessefull with a status other than `FAILED`
karabo_id_err.setdefault(
f"SLURM job terminated with status: {status}", []).append(job_id)
# Process warning logs
warning_file = report_dir / f"warnings_{job_id}.log"
process_log_file(
job_id, karabo_id, report_dir, karabo_id_err, warning_file)
f"Error: SLURM job terminated with status: {status}", []).append(job_id)
success = (not krb_ids_failed)
@@ -440,11 +400,11 @@ class JobsMonitor:
msg = "Calibration jobs succeeded"
else:
# List success & failure by karabo_id
ok = ', '.join(sorted(krb_ids_success)) if krb_ids_success else 'none'
msg = (
f"Succeeded: {ok}; Failed: {', '.join(sorted(krb_ids_failed))} :"
f" {json.dumps(format_log_message(errors), indent=4)}"
)
ok = ', '.join(sorted(krb_ids_success)) if krb_ids_success else None
msg = ""
if ok:
msg += f"Succeeded: {ok}; "
msg += f"Failed: {json.dumps(format_log_message(errors), indent=4)}"
log.debug("Update MDC for %s, %s: %s", r['action'], r['mymdc_id'], msg)

Legal Notice | Contacts