DM-42127: Add htcondor support for bps report --return-exit-codes

.pre-commit-config.yaml

@@ -9,7 +9,7 @@ repos:
       - id: trailing-whitespace
       - id: check-toml
   - repo: https://github.com/psf/black
-    rev: 23.12.0
+    rev: 24.1.1
     hooks:
       - id: black
         # It is recommended to specify the latest version of Python
@@ -24,7 +24,7 @@ repos:
         name: isort (python)
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: v0.1.8
+    rev: v0.1.15
     hooks:
       - id: ruff
   - repo: https://github.com/numpy/numpydoc

doc/changes/DM-42127.feature.rst

@@ -0,0 +1 @@
+Added plugin support for reporting error exit codes with ``bps report``.

python/lsst/ctrl/bps/htcondor/htcondor_service.py

@@ -365,7 +365,15 @@
         _LOG.debug("job_ids = %s", job_ids)
         return job_ids
 
-    def report(self, wms_workflow_id=None, user=None, hist=0, pass_thru=None, is_global=False):
+    def report(
+        self,
+        wms_workflow_id=None,
+        user=None,
+        hist=0,
+        pass_thru=None,
+        is_global=False,
+        return_exit_codes=False,
+    ):
         """Return run information based upon given constraints.
 
         Parameters
@@ -382,6 +390,13 @@
             If set, all job queues (and their histories) will be queried for
             job information. Defaults to False which means that only the local
             job queue will be queried.
+        return_exit_codes : `bool`, optional
+            If set, return exit codes related to jobs with a
+            non-success status. Defaults to False, which means that only
+            the summary state is returned.
+
+            Only applicable in the context of a WMS with associated
+            handlers to return exit codes from jobs.
 
         Returns
         -------
@@ -541,9 +556,9 @@
                 out_prefix,
             )
             if "post" not in final_htjob.dagcmds:
-                final_htjob.dagcmds[
-                    "post"
-                ] = f"{os.path.dirname(__file__)}/final_post.sh {final.name} $DAG_STATUS $RETURN"
+                final_htjob.dagcmds["post"] = (
+                    f"{os.path.dirname(__file__)}/final_post.sh {final.name} $DAG_STATUS $RETURN"
+                )
             htc_workflow.dag.add_final_job(final_htjob)
         elif final and isinstance(final, GenericWorkflow):
             raise NotImplementedError("HTCondor plugin does not support a workflow as the final job")
@@ -1192,6 +1207,7 @@
     """
     _LOG.debug("_create_detailed_report: id = %s, job = %s", wms_workflow_id, jobs[wms_workflow_id])
     dag_job = jobs[wms_workflow_id]
+    task_jobs = {job_id: job_ad for job_id, job_ad in jobs.items() if job_id != wms_workflow_id}
     report = WmsRunReport(
         wms_id=f"{dag_job['ClusterId']}.{dag_job['ProcId']}",
         global_wms_id=dag_job.get("GlobalJobId", "MISS"),
@@ -1207,20 +1223,20 @@
         jobs=[],
         total_number_jobs=dag_job["total_jobs"],
         job_state_counts=dag_job["state_counts"],
+        exit_code_summary=_get_exit_code_summary(task_jobs),
     )
 
-    for job_id, job_info in jobs.items():
+    for job_id, job_info in task_jobs.items():
         try:
-            if job_info["ClusterId"] != int(float(wms_workflow_id)):
-                job_report = WmsJobReport(
-                    wms_id=job_id,
-                    name=job_info.get("DAGNodeName", job_id),
-                    label=job_info.get("bps_job_label", pegasus_name_to_label(job_info["DAGNodeName"])),
-                    state=_htc_status_to_wms_state(job_info),
-                )
-                if job_report.label == "init":
-                    job_report.label = "pipetaskInit"
-                report.jobs.append(job_report)
+            job_report = WmsJobReport(
+                wms_id=job_id,
+                name=job_info.get("DAGNodeName", job_id),
+                label=job_info.get("bps_job_label", pegasus_name_to_label(job_info["DAGNodeName"])),
+                state=_htc_status_to_wms_state(job_info),
+            )
+            if job_report.label == "init":
+                job_report.label = "pipetaskInit"
+            report.jobs.append(job_report)
         except KeyError as ex:
             _LOG.error("Job missing key '%s': %s", str(ex), job_info)
             raise
@@ -1392,6 +1408,36 @@
     return summary
 
 
+def _get_exit_code_summary(jobs):
+    """Get the exit code summary for a run.
+
+    Parameters
+    ----------
+    jobs : `dict` [`str`, `dict` [`str`, Any]]
+        Mapping HTCondor job id to job information.
+
+    Returns
+    -------
+    summary : `dict` [`str`, `list` [`int`]]
+        Jobs' exit codes per job label.
+    """
+    summary = {}
+    for job_id, job_ad in jobs.items():
+        job_label = job_ad["bps_job_label"]
+        summary.setdefault(job_label, [])
+        try:
+            if job_ad["JobStatus"] in {JobStatus.COMPLETED, JobStatus.HELD}:
+                try:
+                    exit_code = job_ad["ExitSignal"] if job_ad["ExitBySignal"] else job_ad["ExitCode"]
+                except KeyError:
+                    _LOG.debug("Cannot determine exit code for job '%s'", job_id)
+                else:
+                    summary[job_label].append(exit_code)
+        except KeyError:
+            _LOG.debug("Attribute 'JobStatus' not found in classad for job '%s'", job_id)
+    return summary
+
+
 def _get_state_counts_from_jobs(wms_workflow_id, jobs):
     """Count number of jobs per WMS state.