From 2a7855ced2d71b0aacce8f463037762ed832e6ae Mon Sep 17 00:00:00 2001 From: Mikolaj Kowalik Date: Mon, 8 Jun 2026 16:51:08 -0400 Subject: [PATCH 1/8] Avoid injecting condor_dagman info to jobs info A bug was causing mixing up condor_dagman job data with those related to DAGMan jobs if `dag.nodes.log` was missing. Fixed that. --- python/lsst/ctrl/bps/htcondor/lssthtc.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index 1461214..9f5e8b2 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/python/lsst/ctrl/bps/htcondor/lssthtc.py @@ -1746,7 +1746,7 @@ def read_single_node_status(filename: str | os.PathLike, init_fake_id: int) -> d _, job_name_to_label, job_name_to_type = count_jobs_in_single_dag(filename.with_suffix(".dag")) loginfo: dict[str, dict[str, Any]] = {} try: - wms_workflow_id, loginfo = read_single_dag_log(filename.with_suffix(".dag.dagman.log")) + wms_workflow_id, _ = read_single_dag_log(filename.with_suffix(".dag.dagman.log")) loginfo = read_single_dag_nodes_log(filename.with_suffix(".dag.nodes.log")) except (OSError, PermissionError): pass From d11f91f87bf116784a640bf8531440f693e22b3a Mon Sep 17 00:00:00 2001 From: Mikolaj Kowalik Date: Tue, 9 Jun 2026 11:43:32 -0400 Subject: [PATCH 2/8] Add unit tests covering read_single_node_status() --- tests/test_lssthtc.py | 202 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 202 insertions(+) diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index 8318cfe..7b7d7ba 100644 --- a/tests/test_lssthtc.py +++ b/tests/test_lssthtc.py @@ -644,6 +644,208 @@ def testSubdagsFailed(self): ) +class ReadSingleNodeStatusTestCase(unittest.TestCase): + """Test read_single_node_status function.""" + + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + + def tearDown(self): + rmtree(self.tmpdir, ignore_errors=True) + + def _copy_files(self, data_subdir, suffixes): + """Copy files with given suffixes from tests/data//.""" + for suffix in suffixes: + copy2(f"{TESTDIR}/data/{data_subdir}/{data_subdir}{suffix}", self.tmpdir) + + def _job_name_to_id(self, jobs): + return {info["DAGNodeName"]: id_ for id_, info in jobs.items()} + + def test_all_done(self): + self._copy_files( + "tiny_success", + [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"], + ) + filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status" + jobs = lssthtc.read_single_node_status(filename, -1) + + self.assertEqual(len(jobs), 5) + name_to_id = self._job_name_to_id(jobs) + + # All four submitted nodes are marked DONE. + for name in [ + "pipetaskInit", + "5bba27bd-8df7-4668-a9c5-e911192c5cdb_label1_val1_val2", + "0b225f1f-6edf-4380-b546-76c97947a88f_label2_val1_val2", + "finalJob", + ]: + self.assertIn(name, name_to_id, msg=f"Missing job {name}") + self.assertEqual( + jobs[name_to_id[name]]["NodeStatus"], + lssthtc.NodeStatus.DONE, + msg=f"Expected DONE for {name}", + ) + + # Service job not tracked by node_status; it came from the event log so + # it has a real positive ClusterId but no NodeStatus field. + self.assertIn("provisioningJob", name_to_id) + self.assertGreater(jobs[name_to_id["provisioningJob"]]["ClusterId"], 0) + + # Spot-check labels and types. + self.assertEqual(jobs[name_to_id["pipetaskInit"]]["bps_job_label"], "pipetaskInit") + self.assertEqual(jobs[name_to_id["pipetaskInit"]]["wms_node_type"], lssthtc.WmsNodeType.PAYLOAD) + self.assertEqual(jobs[name_to_id["finalJob"]]["wms_node_type"], lssthtc.WmsNodeType.FINAL) + self.assertEqual(jobs[name_to_id["provisioningJob"]]["wms_node_type"], lssthtc.WmsNodeType.SERVICE) + + # DAGManJobID is populated from the dagman log for every job. + for job in jobs.values(): + self.assertIn("DAGManJobID", job) + + def test_mixed_statuses(self): + self._copy_files( + "tiny_problems", + [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"], + ) + filename = pathlib.Path(self.tmpdir) / "tiny_problems.node_status" + jobs = lssthtc.read_single_node_status(filename, -1) + + self.assertEqual(len(jobs), 7) + name_to_id = self._job_name_to_id(jobs) + + self.assertEqual(jobs[name_to_id["pipetaskInit"]]["NodeStatus"], lssthtc.NodeStatus.DONE) + self.assertEqual( + jobs[name_to_id["057c8caf-66f6-4612-abf7-cdea5b666b1b_label1_val1a_val2b"]]["NodeStatus"], + lssthtc.NodeStatus.ERROR, + ) + self.assertEqual( + jobs[name_to_id["4a7f478b-2e9b-435c-a730-afac3f621658_label1_val1a_val2a"]]["NodeStatus"], + lssthtc.NodeStatus.DONE, + ) + self.assertEqual( + jobs[name_to_id["40040b97-606d-4997-98d3-e0493055fe7e_label2_val1a_val2b"]]["NodeStatus"], + lssthtc.NodeStatus.FUTILE, + ) + self.assertEqual(jobs[name_to_id["finalJob"]]["NodeStatus"], lssthtc.NodeStatus.ERROR) + # Service job not tracked by node_status; came from event log so + # it has a real positive ClusterId but no NodeStatus field. + self.assertIn("provisioningJob", name_to_id) + self.assertGreater(jobs[name_to_id["provisioningJob"]]["ClusterId"], 0) + + def test_running_workflow(self): + self._copy_files( + "tiny_running", + [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"], + ) + filename = pathlib.Path(self.tmpdir) / "tiny_running.node_status" + jobs = lssthtc.read_single_node_status(filename, -1) + + self.assertEqual(len(jobs), 5) + name_to_id = self._job_name_to_id(jobs) + + self.assertEqual(jobs[name_to_id["pipetaskInit"]]["NodeStatus"], lssthtc.NodeStatus.DONE) + self.assertEqual( + jobs[name_to_id["ca27ea57-c014-44c1-838a-78c06bc3ec1b_label1_val1_val2"]]["NodeStatus"], + lssthtc.NodeStatus.SUBMITTED, + ) + self.assertEqual( + jobs[name_to_id["dbf919fa-5453-4b05-8806-ad6390fda0a3_label2_val1_val2"]]["NodeStatus"], + lssthtc.NodeStatus.NOT_READY, + ) + self.assertEqual(jobs[name_to_id["finalJob"]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY) + # Service job appeared in the event log; has a real positive ClusterId. + self.assertIn("provisioningJob", name_to_id) + self.assertGreater(jobs[name_to_id["provisioningJob"]]["ClusterId"], 0) + + def test_missing_node_status_file(self): + # Omit the .node_status file; jobs must be built from the event log + # and dag. + self._copy_files( + "tiny_problems", + [".dag", ".dag.dagman.log", ".dag.nodes.log"], + ) + filename = pathlib.Path(self.tmpdir) / "tiny_problems.node_status" + jobs = lssthtc.read_single_node_status(filename, -1) + + self.assertEqual(len(jobs), 7) + name_to_id = self._job_name_to_id(jobs) + + # Jobs that appeared in the event log have real (positive) cluster IDs. + self.assertEqual(jobs[name_to_id["pipetaskInit"]]["DAGNodeName"], "pipetaskInit") + self.assertGreater(jobs[name_to_id["pipetaskInit"]]["ClusterId"], 0) + + # The service job appeared in the event log and has a real positive ID. + self.assertGreater(jobs[name_to_id["provisioningJob"]]["ClusterId"], 0) + + # All jobs carry the correct label and type from the dag file. + self.assertEqual(jobs[name_to_id["pipetaskInit"]]["wms_node_type"], lssthtc.WmsNodeType.PAYLOAD) + self.assertEqual(jobs[name_to_id["provisioningJob"]]["wms_node_type"], lssthtc.WmsNodeType.SERVICE) + + def test_missing_log_files(self): + # Omit both log files; every job should get a fake negative ID. + self._copy_files("tiny_success", [".dag", ".node_status"]) + filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status" + jobs = lssthtc.read_single_node_status(filename, -1) + + self.assertEqual(len(jobs), 5) + for job in jobs.values(): + self.assertLess(job["ClusterId"], 0) + + # NodeStatus values from the node_status file must still be preserved. + name_to_id = self._job_name_to_id(jobs) + self.assertEqual(jobs[name_to_id["pipetaskInit"]]["NodeStatus"], lssthtc.NodeStatus.DONE) + self.assertEqual(jobs[name_to_id["finalJob"]]["NodeStatus"], lssthtc.NodeStatus.DONE) + self.assertEqual(jobs[name_to_id["provisioningJob"]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY) + + def test_init_fake_id(self): + # Verify fake IDs count down from the given starting value. + self._copy_files("tiny_success", [".dag", ".node_status"]) + filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status" + init_fake_id = -10 + jobs = lssthtc.read_single_node_status(filename, init_fake_id) + + self.assertEqual(len(jobs), 5) + cluster_ids = [job["ClusterId"] for job in jobs.values()] + # All IDs must be at most init_fake_id (i.e., -10 or lower). + for cid in cluster_ids: + self.assertLessEqual(cid, init_fake_id) + # All IDs must be unique. + self.assertEqual(len(set(cluster_ids)), len(cluster_ids)) + + def test_from_dag_job_attribute(self): + self._copy_files( + "tiny_success", + [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"], + ) + filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status" + jobs = lssthtc.read_single_node_status(filename, -1) + + for id_, job in jobs.items(): + self.assertEqual( + job["from_dag_job"], + "wms_tiny_success", + msg=f"Job {id_} has wrong from_dag_job", + ) + + def test_service_job_placeholder(self): + self._copy_files( + "tiny_prov_no_submit", + [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"], + ) + filename = pathlib.Path(self.tmpdir) / "tiny_prov_no_submit.node_status" + jobs = lssthtc.read_single_node_status(filename, -1) + + service_jobs = [ + (id_, info) + for id_, info in jobs.items() + if info.get("wms_node_type") == lssthtc.WmsNodeType.SERVICE + ] + self.assertEqual(len(service_jobs), 1) + service_id, service_job = service_jobs[0] + self.assertEqual(service_job["DAGNodeName"], "provisioningJob") + self.assertEqual(service_job["NodeStatus"], lssthtc.NodeStatus.NOT_READY) + self.assertLess(service_job["ClusterId"], 0) + + class HTCJobTestCase(unittest.TestCase): """Test HTCJob methods.""" From 9c052013b696e9ff0df1c5985eba46b3909baa4a Mon Sep 17 00:00:00 2001 From: Mikolaj Kowalik Date: Fri, 19 Jun 2026 11:05:08 -0400 Subject: [PATCH 3/8] Do not back up files of subdags that succeeded During workflow restarts, selected files are being moved to a dedicated backup directory as HTCondor recreates them with the fresh content once a workflow is restarted. However, the backup function was also moving out such files for subdags that succeeded. HTCondor does not rerun DAGs that completed successfully. Their absence led to reporting incorrect job status counts as the information in these files is used by the plugin's reporting mechanism. Made changes to ensure that only files of the failed subdags are backed up. --- .../lsst/ctrl/bps/htcondor/check_group_status.sh | 4 ++-- python/lsst/ctrl/bps/htcondor/lssthtc.py | 10 ++++++++-- tests/test_lssthtc.py | 16 ++++++++-------- 3 files changed, 18 insertions(+), 12 deletions(-) diff --git a/python/lsst/ctrl/bps/htcondor/check_group_status.sh b/python/lsst/ctrl/bps/htcondor/check_group_status.sh index 776175f..df2e71e 100755 --- a/python/lsst/ctrl/bps/htcondor/check_group_status.sh +++ b/python/lsst/ctrl/bps/htcondor/check_group_status.sh @@ -12,6 +12,6 @@ if ! [ -f "$1" ]; then exit 1 fi -group_status=$(<"$1") +group_status=$(cat "$1") echo "Read status ${group_status} from group status file ($1)" >&2 -exit ${group_status} +exit "${group_status}" diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index 9f5e8b2..c5e5d48 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/python/lsst/ctrl/bps/htcondor/lssthtc.py @@ -411,8 +411,14 @@ def htc_backup_files( else: htc_backup_files_single_path(path, dest) - # also back up any subdag info - for subdag_dir in path.glob("subdags/*"): + # Back up selected files for failed subdags as well. + # + # Do NOT back up files of subdags that succeeded! These files need to stay + # in their respective directories as they will not be recreated by HTCondor + # after the run is restarted. As HTCondorService.report() uses information + # in these files to determine job statuses, their absence may lead to + # reporting incorrect job status counts. + for subdag_dir in {file.parent for file in path.glob("subdags/*/*.rescue*")}: subdag_dest = dest / subdag_dir.relative_to(path) subdag_dest.mkdir(parents=True, exist_ok=False) htc_backup_files_single_path(subdag_dir, subdag_dest) diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index 7b7d7ba..3937100 100644 --- a/tests/test_lssthtc.py +++ b/tests/test_lssthtc.py @@ -1185,6 +1185,10 @@ def testSubdags(self): "subdags/wms_group_order1_val1a/group_order1_val1a.dag", "subdags/wms_group_order1_val1a/group_order1_val1a.dag.dagman.log", "subdags/wms_group_order1_val1a/group_order1_val1a.dag.dagman.out", + "subdags/wms_group_order1_val1a/group_order1_val1a.dag.nodes.log", + "subdags/wms_group_order1_val1a/group_order1_val1a.node_status", + "subdags/wms_group_order1_val1a/wms_group_order1_val1a.dag.post.out", + "subdags/wms_group_order1_val1a/wms_group_order1_val1a.status.txt", "subdags/wms_group_order1_val1b/group_order1_val1b.dag", "subdags/wms_group_order1_val1b/group_order1_val1b.dag.dagman.log", "subdags/wms_group_order1_val1b/group_order1_val1b.dag.dagman.out", @@ -1192,21 +1196,17 @@ def testSubdags(self): "subdags/wms_group_order1_val1c/group_order1_val1c.dag", "subdags/wms_group_order1_val1c/group_order1_val1c.dag.dagman.log", "subdags/wms_group_order1_val1c/group_order1_val1c.dag.dagman.out", + "subdags/wms_group_order1_val1c/group_order1_val1c.dag.nodes.log", + "subdags/wms_group_order1_val1c/group_order1_val1c.node_status", + "subdags/wms_group_order1_val1c/wms_group_order1_val1c.dag.post.out", + "subdags/wms_group_order1_val1c/wms_group_order1_val1c.status.txt", "001/group_failed_1.dag.nodes.log", "001/group_failed_1.info.json", "001/group_failed_1.node_status", - "001/subdags/wms_group_order1_val1a/group_order1_val1a.dag.nodes.log", - "001/subdags/wms_group_order1_val1a/group_order1_val1a.node_status", - "001/subdags/wms_group_order1_val1a/wms_group_order1_val1a.dag.post.out", - "001/subdags/wms_group_order1_val1a/wms_group_order1_val1a.status.txt", "001/subdags/wms_group_order1_val1b/group_order1_val1b.dag.nodes.log", "001/subdags/wms_group_order1_val1b/group_order1_val1b.node_status", "001/subdags/wms_group_order1_val1b/wms_group_order1_val1b.status.txt", "001/subdags/wms_group_order1_val1b/wms_group_order1_val1b.dag.post.out", - "001/subdags/wms_group_order1_val1c/group_order1_val1c.dag.nodes.log", - "001/subdags/wms_group_order1_val1c/group_order1_val1c.node_status", - "001/subdags/wms_group_order1_val1c/wms_group_order1_val1c.dag.post.out", - "001/subdags/wms_group_order1_val1c/wms_group_order1_val1c.status.txt", }, ) From 7df8dda657c12046ae1908e5b124e5c40674dca0 Mon Sep 17 00:00:00 2001 From: Mikolaj Kowalik Date: Fri, 19 Jun 2026 11:29:25 -0400 Subject: [PATCH 4/8] Do not back up files in _update_rescue_file() _udpate_rescue_file() was trying to back up files for each failed subdags However, at the time of its execution these files were already backed up. Also, the function, in its current implementation, doesn't have access to the correct location of the backup directory. As a result it was creating an empty directory tree in the directory with the failed subdag. Modified it so it has a single responsibility -- updating the rescue file of the main DAG. --- python/lsst/ctrl/bps/htcondor/lssthtc.py | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index c5e5d48..ca82a6a 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/python/lsst/ctrl/bps/htcondor/lssthtc.py @@ -2365,8 +2365,7 @@ def _copy_done_lines(failed_subdags: list[str], infh: TextIO, outfh: TextIO) -> def _update_rescue_file(rescue_file: Path) -> None: - """Update the subdag failures in the main rescue file - and backup the failed subdag dirs. + """Update the subdag failures in the main rescue file. Parameters ---------- @@ -2382,10 +2381,6 @@ def _update_rescue_file(rescue_file: Path) -> None: _copy_done_lines(failed_subdags, infh, outfh) rescue_file.unlink() rescue_tmp.rename(rescue_file) - for failed_subdag in failed_subdags: - htc_backup_files( - rescue_file.parent / "subdags" / failed_subdag, subdir=f"backups/subdags/{failed_subdag}" - ) def _update_dicts(dict1, dict2): From 9f13c7ceb898c9ad013e3bacd655e9c7310cea5a Mon Sep 17 00:00:00 2001 From: Mikolaj Kowalik Date: Mon, 13 Jul 2026 13:10:03 -0400 Subject: [PATCH 5/8] Make sure only failed subdags are backed up The function responsible for backing up files between restarts was using a very simple strategy to determine if files in a subdag directory needs a backup -- if a subdag directory contained a rescue file, the files were backed up. This strategy turned out to be too simplistic. It erroneously was backing up files for subdags that completed successfully after a restart as rescue files from previous failures were still present. To address this limitation, now the subdags that fails (if any) are being retrieved directly from the main DAG rescue file. --- .../ctrl/bps/htcondor/htcondor_service.py | 10 +- python/lsst/ctrl/bps/htcondor/lssthtc.py | 124 ++++++++++-------- tests/test_lssthtc.py | 26 ++-- 3 files changed, 86 insertions(+), 74 deletions(-) diff --git a/python/lsst/ctrl/bps/htcondor/htcondor_service.py b/python/lsst/ctrl/bps/htcondor/htcondor_service.py index 4ef025a..d2e227f 100644 --- a/python/lsst/ctrl/bps/htcondor/htcondor_service.py +++ b/python/lsst/ctrl/bps/htcondor/htcondor_service.py @@ -257,9 +257,13 @@ def restart(self, wms_workflow_id): ) _LOG.info("Backing up select HTCondor files from previous run attempt") - rescue_file = htc_backup_files(wms_path, subdir="backups") - if (wms_path / "subdags").exists(): - _update_rescue_file(rescue_file) + rescue_files = sorted(wms_path.glob("*.rescue[0-9][0-9][0-9]")) + last_rescue_file = Path(rescue_files[-1]) if rescue_files else None + has_subdags = (wms_path / "subdags").exists() + failed_subdags = None + if last_rescue_file and has_subdags: + failed_subdags = set(_update_rescue_file(last_rescue_file)) + htc_backup_files(wms_path, subdir="backups", failed_subdags=failed_subdags) # For workflow portability, internal paths are all relative. Hence # the DAG needs to be resubmitted to HTCondor from inside the submit diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index ca82a6a..6932cbd 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/python/lsst/ctrl/bps/htcondor/lssthtc.py @@ -325,8 +325,11 @@ def __str__(self): def htc_backup_files( - wms_path: str | os.PathLike, subdir: str | os.PathLike | None = None, limit: int = 100 -) -> Path | None: + wms_path: str | os.PathLike, + subdir: str | os.PathLike | None = None, + limit: int = 100, + failed_subdags: set[str] | None = None, +) -> None: """Backup select HTCondor files in the submit directory. Files will be saved in separate subdirectories which will be created in @@ -352,11 +355,10 @@ def htc_backup_files( the last backup files will be overwritten. The default value is 100 to match the default value of HTCondor's DAGMAN_MAX_RESCUE_NUM in version 8.8+. - - Returns - ------- - last_rescue_file : `pathlib.Path` or None - Path to the latest rescue file or None if doesn't exist. + failed_subdags : `set` [`str`], optional + Names of subdag jobs that failed. Only files of these subdags will be + backed up. If None (the default), files of all subdags with rescue + files are backed up. Raises ------ @@ -419,14 +421,12 @@ def htc_backup_files( # in these files to determine job statuses, their absence may lead to # reporting incorrect job status counts. for subdag_dir in {file.parent for file in path.glob("subdags/*/*.rescue*")}: + if failed_subdags is not None and subdag_dir.name not in failed_subdags: + continue subdag_dest = dest / subdag_dir.relative_to(path) subdag_dest.mkdir(parents=True, exist_ok=False) htc_backup_files_single_path(subdag_dir, subdag_dest) - last_rescue_file = rescue_dags[-1] if rescue_dags else None - _LOG.debug("last_rescue_file = %s", last_rescue_file) - return last_rescue_file - def htc_backup_files_single_path(src: str | os.PathLike, dest: str | os.PathLike) -> None: """Move particular htc files to a different directory for later debugging. @@ -434,7 +434,7 @@ def htc_backup_files_single_path(src: str | os.PathLike, dest: str | os.PathLike Parameters ---------- src : `str` or `os.PathLike` - Directory from which to backup particular files. + Directory from which to back up particular files. dest : `str` or `os.PathLike` Directory to which particular files are moved. @@ -2271,7 +2271,7 @@ def htc_check_dagman_output(wms_path: str | os.PathLike) -> str: return message -def _read_rescue_headers(infh: TextIO) -> tuple[list[str], list[str]]: +def _read_rescue_headers(infh: TextIO) -> list[str]: """Read header lines from a rescue file. Parameters @@ -2283,60 +2283,69 @@ def _read_rescue_headers(infh: TextIO) -> tuple[list[str], list[str]]: ------- header_lines : `list` [`str`] Header lines read from the rescue file. + """ + header_lines = [] + for line in infh: + line = line.strip() + if not line.startswith("#"): + break + header_lines.append(line) + return header_lines + + +def _update_rescue_headers(header_lines: list[str]) -> list[str]: + """Update rescue header lines in place. + + Replaces ``wms_check_status`` node name prefixes with the corresponding + group node names and adjusts the count of nodes premarked DONE. + + Parameters + ---------- + header_lines : `list` [`str`] + Header lines to update in place. + + Returns + ------- failed_subdags : `list` [`str`] Names of failed subdag jobs. """ - header_lines: list[str] = [] - failed = False - failed_subdags: list[str] = [] + failed_subdags = [] + + # Relies on HTCondor writing all failed node names on a single line + # (see src/condor_dagman/dag.cpp in https://github.com/htcondor/htcondor). + for i, line in enumerate(header_lines): + if line.startswith("# Nodes that failed:"): + nodes = header_lines[i + 1][1:].strip().split(",") + new_nodes = [] + for node in nodes: + if node.startswith("wms_check_status"): + node = node[17:] + failed_subdags.append(node) + new_nodes.append(node) + header_lines[i + 1] = f"# {','.join(new_nodes)}" + break - for line in infh: - line = line.strip() - if line.startswith("#"): - if line.startswith("# Nodes that failed:"): - failed = True - header_lines.append(line) - elif failed: - orig_failed_nodes = line[1:].strip().split(",") - new_failed_nodes = [] - for node in orig_failed_nodes: - if node.startswith("wms_check_status"): - group_node = node[17:] - failed_subdags.append(group_node) - new_failed_nodes.append(group_node) - else: - new_failed_nodes.append(node) - header_lines.append(f"# {','.join(new_failed_nodes)}") - if orig_failed_nodes[-1] == "": - failed = False - else: - header_lines.append(line) - elif line.strip() == "": # end of headers + for i, line in enumerate(header_lines): + m = re.match(r"^(# Nodes premarked DONE:)\s+(\d+)", line) + if m: + header_lines[i] = f"{m.group(1)} {int(m.group(2)) - len(failed_subdags)}" break - return header_lines, failed_subdags + + return failed_subdags -def _write_rescue_headers(header_lines: list[str], failed_subdags: list[str], outfh: TextIO) -> None: +def _write_rescue_headers(header_lines: list[str], outfh: TextIO) -> None: """Write the header lines to the new rescue file. Parameters ---------- header_lines : `list` [`str`] Header lines to write to the new rescue file. - failed_subdags : `list` [`str`] - Job names of the failed subdags. outfh : `TextIO` New rescue file. """ - done_str = "# Nodes premarked DONE" - pattern = f"^{done_str}:\\s+(\\d+)" - for header_line in header_lines: - m = re.match(pattern, header_line) - if m: - print(f"{done_str}: {int(m.group(1)) - len(failed_subdags)}", file=outfh) - else: - print(header_line, file=outfh) - + for line in header_lines: + print(line, file=outfh) print("", file=outfh) @@ -2364,23 +2373,30 @@ def _copy_done_lines(failed_subdags: list[str], infh: TextIO, outfh: TextIO) -> print(line, file=outfh) -def _update_rescue_file(rescue_file: Path) -> None: +def _update_rescue_file(rescue_file: Path) -> list[str]: """Update the subdag failures in the main rescue file. Parameters ---------- rescue_file : `pathlib.Path` The main rescue file that needs to be updated. + + Returns + ------- + failed_subdags : `list` [`str`] + Names of failed subdag jobs found in the DAGMan rescue file. """ # To reduce memory requirements, not reading entire file into memory. rescue_tmp = rescue_file.with_suffix(rescue_file.suffix + ".tmp") with open(rescue_file) as infh: - header_lines, failed_subdags = _read_rescue_headers(infh) + header_lines = _read_rescue_headers(infh) + failed_subdags = _update_rescue_headers(header_lines) with open(rescue_tmp, "w") as outfh: - _write_rescue_headers(header_lines, failed_subdags, outfh) + _write_rescue_headers(header_lines, outfh) _copy_done_lines(failed_subdags, infh, outfh) rescue_file.unlink() rescue_tmp.rename(rescue_file) + return failed_subdags def _update_dicts(dict1, dict2): diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index 3937100..caefdb3 100644 --- a/tests/test_lssthtc.py +++ b/tests/test_lssthtc.py @@ -31,7 +31,6 @@ import os import pathlib import stat -import sys import tempfile import unittest from shutil import copy2, copytree, ignore_patterns, rmtree, which @@ -1064,15 +1063,14 @@ def testDirectoryNotFound(self): test_tmp_dir = pathlib.Path(tmp_dir) submit_dir = test_tmp_dir / "submit" with self.assertRaises(FileNotFoundError): - _ = lssthtc.htc_backup_files(submit_dir) + lssthtc.htc_backup_files(submit_dir) def testSuccess(self): with temporaryDirectory() as tmp_dir: test_tmp_dir = pathlib.Path(tmp_dir) submit_dir = test_tmp_dir / "submit" copytree(f"{TESTDIR}/data/tiny_success", submit_dir, ignore=ignore_patterns("*~", ".???*")) - result_rescue = lssthtc.htc_backup_files(submit_dir) - self.assertIsNone(result_rescue) + lssthtc.htc_backup_files(submit_dir) result_submit = [] for root, _, files in os.walk(submit_dir): result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files]) @@ -1095,10 +1093,9 @@ def testDestNotInSubmitDir(self): submit_dir = test_tmp_dir / "submit" copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*")) with self.assertLogs("lsst.ctrl.bps.htcondor", level="WARNING") as cm: - result_rescue = lssthtc.htc_backup_files(submit_dir, test_tmp_dir / "backup") + lssthtc.htc_backup_files(submit_dir, test_tmp_dir / "backup") self.assertIn("Invalid backup location:", cm.output[-1]) - result_rescue = lssthtc.htc_backup_files(submit_dir) - self.assertTrue((submit_dir / "tiny_problems.dag.rescue001").samefile(result_rescue)) + lssthtc.htc_backup_files(submit_dir) result_submit = [] for root, _, files in os.walk(submit_dir): result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files]) @@ -1122,8 +1119,7 @@ def testDestInSubmitDir(self): submit_dir = test_tmp_dir / "submit" backup_dir = submit_dir / "subdir" copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*")) - result_rescue = lssthtc.htc_backup_files(submit_dir, backup_dir) - self.assertTrue((submit_dir / "tiny_problems.dag.rescue001").samefile(result_rescue)) + lssthtc.htc_backup_files(submit_dir, backup_dir) result_submit = [] for root, _, files in os.walk(submit_dir): result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files]) @@ -1146,8 +1142,7 @@ def testRelativeSubdir(self): test_tmp_dir = pathlib.Path(tmp_dir) submit_dir = test_tmp_dir / "submit" copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*")) - result_rescue = lssthtc.htc_backup_files(submit_dir, "reldir") - self.assertTrue((submit_dir / "tiny_problems.dag.rescue001").samefile(result_rescue)) + lssthtc.htc_backup_files(submit_dir, "reldir") result_submit = [] for root, _, files in os.walk(submit_dir): result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files]) @@ -1170,8 +1165,7 @@ def testSubdags(self): test_tmp_dir = pathlib.Path(tmp_dir) submit_dir = test_tmp_dir / "submit" copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*")) - result_rescue = lssthtc.htc_backup_files(submit_dir) - self.assertTrue(result_rescue.samefile(submit_dir / "group_failed_1.dag.rescue001")) + lssthtc.htc_backup_files(submit_dir) result_submit = [] for root, _, files in os.walk(submit_dir): result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files]) @@ -1221,8 +1215,8 @@ def testSuccess(self): submit_dir = test_tmp_dir / "submit" copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*")) rescue_file = submit_dir / "group_failed_1.dag.rescue001" - lssthtc._update_rescue_file(rescue_file) - + failed_subdags = lssthtc._update_rescue_file(rescue_file) + self.assertEqual(set(failed_subdags), {"wms_group_order1_val1b"}) with open(rescue_file) as fh: lines = fh.readlines() results = "".join(lines) @@ -1260,8 +1254,6 @@ def testSuccess(self): DONE wms_check_status_wms_group_order1_val1c """ - print("results = ", results, file=sys.stderr) - print("truth = ", truth, file=sys.stderr) self.assertEqual(results, truth) From 1b55e4f7e109e28c5f35ff9670876eba7a54d1a0 Mon Sep 17 00:00:00 2001 From: Mikolaj Kowalik Date: Mon, 13 Jul 2026 13:11:37 -0400 Subject: [PATCH 6/8] Add unit tests covering new helper functions --- tests/test_lssthtc.py | 100 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 100 insertions(+) diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index caefdb3..d850f7f 100644 --- a/tests/test_lssthtc.py +++ b/tests/test_lssthtc.py @@ -1257,6 +1257,106 @@ def testSuccess(self): self.assertEqual(results, truth) +class ReadRescueHeadersTestCase(unittest.TestCase): + """Test _read_rescue_headers function.""" + + def testTypical(self): + content = "# Header line 1\n# Header line 2\n\nDONE somenode\n" + result = lssthtc._read_rescue_headers(io.StringIO(content)) + self.assertEqual(result, ["# Header line 1", "# Header line 2"]) + + def testEmptyFile(self): + result = lssthtc._read_rescue_headers(io.StringIO("")) + self.assertEqual(result, []) + + def testOnlyHeaderLines(self): + content = "# Line 1\n# Line 2\n# Line 3\n" + result = lssthtc._read_rescue_headers(io.StringIO(content)) + self.assertEqual(result, ["# Line 1", "# Line 2", "# Line 3"]) + + def testFirstLineNotComment(self): + content = "DONE somenode\n# Header\n" + result = lssthtc._read_rescue_headers(io.StringIO(content)) + self.assertEqual(result, []) + + def testWhitespaceStripped(self): + content = " # Header line 1 \n # Header line 2 \n\n" + result = lssthtc._read_rescue_headers(io.StringIO(content)) + self.assertEqual(result, ["# Header line 1", "# Header line 2"]) + + +class WriteRescueHeadersTestCase(unittest.TestCase): + """Test _write_rescue_headers function.""" + + def testTypical(self): + header_lines = ["# Header line 1", "# Header line 2", "# Header line 3"] + outfh = io.StringIO() + lssthtc._write_rescue_headers(header_lines, outfh) + self.assertEqual(outfh.getvalue(), "# Header line 1\n# Header line 2\n# Header line 3\n\n") + + def testEmptyList(self): + outfh = io.StringIO() + lssthtc._write_rescue_headers([], outfh) + self.assertEqual(outfh.getvalue(), "\n") + + def testSingleLine(self): + outfh = io.StringIO() + lssthtc._write_rescue_headers(["# Only line"], outfh) + self.assertEqual(outfh.getvalue(), "# Only line\n\n") + + +class UpdateRescueHeadersTestCase(unittest.TestCase): + """Test _update_rescue_headers function.""" + + def testWithFailedSubdag(self): + header_lines = [ + "# Total number of Nodes: 26", + "# Nodes premarked DONE: 22", + "# Nodes that failed: 2", + "# wms_check_status_wms_group_order1_val1b,finalJob,", + ] + result = lssthtc._update_rescue_headers(header_lines) + self.assertEqual(result, ["wms_group_order1_val1b"]) + self.assertEqual(header_lines[1], "# Nodes premarked DONE: 21") + self.assertEqual(header_lines[3], "# wms_group_order1_val1b,finalJob,") + + def testNoSubdagFailures(self): + header_lines = [ + "# Nodes premarked DONE: 5", + "# Nodes that failed: 1", + "# finalJob,", + ] + result = lssthtc._update_rescue_headers(header_lines) + self.assertEqual(result, []) + self.assertEqual(header_lines[0], "# Nodes premarked DONE: 5") + self.assertEqual(header_lines[2], "# finalJob,") + + def testMultipleFailedSubdags(self): + header_lines = [ + "# Nodes premarked DONE: 10", + "# Nodes that failed: 3", + "# wms_check_status_subdag_a,wms_check_status_subdag_b,finalJob,", + ] + result = lssthtc._update_rescue_headers(header_lines) + self.assertEqual(result, ["subdag_a", "subdag_b"]) + self.assertEqual(header_lines[0], "# Nodes premarked DONE: 8") + self.assertEqual(header_lines[2], "# subdag_a,subdag_b,finalJob,") + + def testNoFailedNodesLine(self): + header_lines = [ + "# Total number of Nodes: 5", + "# Nodes premarked DONE: 5", + ] + original = list(header_lines) + result = lssthtc._update_rescue_headers(header_lines) + self.assertEqual(result, []) + self.assertEqual(header_lines, original) + + def testEmptyHeader(self): + result = lssthtc._update_rescue_headers([]) + self.assertEqual(result, []) + + class ReadDagStatusTestCase(unittest.TestCase): """Test read_dag_status function and read_single_dag_status.""" From ef10a7b4c56a462f0b5c7d9731b04c4272cd5109 Mon Sep 17 00:00:00 2001 From: Mikolaj Kowalik Date: Mon, 13 Jul 2026 18:51:42 -0400 Subject: [PATCH 7/8] Use camelCase in ReadSingleNodeStatusTestCase Contrary to the names of the methods in all other test cases, the names of the methods in the ReadSingleNodeStatusTestCase were using snake_case instead of camelCase. Fixed that. --- tests/test_lssthtc.py | 46 +++++++++++++++++++++---------------------- 1 file changed, 23 insertions(+), 23 deletions(-) diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index d850f7f..e612698 100644 --- a/tests/test_lssthtc.py +++ b/tests/test_lssthtc.py @@ -652,16 +652,16 @@ def setUp(self): def tearDown(self): rmtree(self.tmpdir, ignore_errors=True) - def _copy_files(self, data_subdir, suffixes): + def _copyFiles(self, data_subdir, suffixes): """Copy files with given suffixes from tests/data//.""" for suffix in suffixes: copy2(f"{TESTDIR}/data/{data_subdir}/{data_subdir}{suffix}", self.tmpdir) - def _job_name_to_id(self, jobs): + def _jobNameToId(self, jobs): return {info["DAGNodeName"]: id_ for id_, info in jobs.items()} - def test_all_done(self): - self._copy_files( + def testAllDone(self): + self._copyFiles( "tiny_success", [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"], ) @@ -669,7 +669,7 @@ def test_all_done(self): jobs = lssthtc.read_single_node_status(filename, -1) self.assertEqual(len(jobs), 5) - name_to_id = self._job_name_to_id(jobs) + name_to_id = self._jobNameToId(jobs) # All four submitted nodes are marked DONE. for name in [ @@ -700,8 +700,8 @@ def test_all_done(self): for job in jobs.values(): self.assertIn("DAGManJobID", job) - def test_mixed_statuses(self): - self._copy_files( + def testMixedStatuses(self): + self._copyFiles( "tiny_problems", [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"], ) @@ -709,7 +709,7 @@ def test_mixed_statuses(self): jobs = lssthtc.read_single_node_status(filename, -1) self.assertEqual(len(jobs), 7) - name_to_id = self._job_name_to_id(jobs) + name_to_id = self._jobNameToId(jobs) self.assertEqual(jobs[name_to_id["pipetaskInit"]]["NodeStatus"], lssthtc.NodeStatus.DONE) self.assertEqual( @@ -730,8 +730,8 @@ def test_mixed_statuses(self): self.assertIn("provisioningJob", name_to_id) self.assertGreater(jobs[name_to_id["provisioningJob"]]["ClusterId"], 0) - def test_running_workflow(self): - self._copy_files( + def testRunningWorkflow(self): + self._copyFiles( "tiny_running", [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"], ) @@ -739,7 +739,7 @@ def test_running_workflow(self): jobs = lssthtc.read_single_node_status(filename, -1) self.assertEqual(len(jobs), 5) - name_to_id = self._job_name_to_id(jobs) + name_to_id = self._jobNameToId(jobs) self.assertEqual(jobs[name_to_id["pipetaskInit"]]["NodeStatus"], lssthtc.NodeStatus.DONE) self.assertEqual( @@ -755,10 +755,10 @@ def test_running_workflow(self): self.assertIn("provisioningJob", name_to_id) self.assertGreater(jobs[name_to_id["provisioningJob"]]["ClusterId"], 0) - def test_missing_node_status_file(self): + def testMissingNodeStatusFile(self): # Omit the .node_status file; jobs must be built from the event log # and dag. - self._copy_files( + self._copyFiles( "tiny_problems", [".dag", ".dag.dagman.log", ".dag.nodes.log"], ) @@ -766,7 +766,7 @@ def test_missing_node_status_file(self): jobs = lssthtc.read_single_node_status(filename, -1) self.assertEqual(len(jobs), 7) - name_to_id = self._job_name_to_id(jobs) + name_to_id = self._jobNameToId(jobs) # Jobs that appeared in the event log have real (positive) cluster IDs. self.assertEqual(jobs[name_to_id["pipetaskInit"]]["DAGNodeName"], "pipetaskInit") @@ -779,9 +779,9 @@ def test_missing_node_status_file(self): self.assertEqual(jobs[name_to_id["pipetaskInit"]]["wms_node_type"], lssthtc.WmsNodeType.PAYLOAD) self.assertEqual(jobs[name_to_id["provisioningJob"]]["wms_node_type"], lssthtc.WmsNodeType.SERVICE) - def test_missing_log_files(self): + def testMissingLogFiles(self): # Omit both log files; every job should get a fake negative ID. - self._copy_files("tiny_success", [".dag", ".node_status"]) + self._copyFiles("tiny_success", [".dag", ".node_status"]) filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status" jobs = lssthtc.read_single_node_status(filename, -1) @@ -790,14 +790,14 @@ def test_missing_log_files(self): self.assertLess(job["ClusterId"], 0) # NodeStatus values from the node_status file must still be preserved. - name_to_id = self._job_name_to_id(jobs) + name_to_id = self._jobNameToId(jobs) self.assertEqual(jobs[name_to_id["pipetaskInit"]]["NodeStatus"], lssthtc.NodeStatus.DONE) self.assertEqual(jobs[name_to_id["finalJob"]]["NodeStatus"], lssthtc.NodeStatus.DONE) self.assertEqual(jobs[name_to_id["provisioningJob"]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY) - def test_init_fake_id(self): + def testInitFakeId(self): # Verify fake IDs count down from the given starting value. - self._copy_files("tiny_success", [".dag", ".node_status"]) + self._copyFiles("tiny_success", [".dag", ".node_status"]) filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status" init_fake_id = -10 jobs = lssthtc.read_single_node_status(filename, init_fake_id) @@ -810,8 +810,8 @@ def test_init_fake_id(self): # All IDs must be unique. self.assertEqual(len(set(cluster_ids)), len(cluster_ids)) - def test_from_dag_job_attribute(self): - self._copy_files( + def testFromDagJobAttribute(self): + self._copyFiles( "tiny_success", [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"], ) @@ -825,8 +825,8 @@ def test_from_dag_job_attribute(self): msg=f"Job {id_} has wrong from_dag_job", ) - def test_service_job_placeholder(self): - self._copy_files( + def testServiceJobPlaceholder(self): + self._copyFiles( "tiny_prov_no_submit", [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"], ) From 1b4296e3f86ba134e37eb2a3dc462d10cfa4c5f8 Mon Sep 17 00:00:00 2001 From: Mikolaj Kowalik Date: Tue, 9 Jun 2026 11:06:40 -0400 Subject: [PATCH 8/8] Add a news item describing the changes --- doc/changes/DM-54540.bugfix.rst | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 doc/changes/DM-54540.bugfix.rst diff --git a/doc/changes/DM-54540.bugfix.rst b/doc/changes/DM-54540.bugfix.rst new file mode 100644 index 0000000..6e510a2 --- /dev/null +++ b/doc/changes/DM-54540.bugfix.rst @@ -0,0 +1,2 @@ +- Fix bug causing ``bps report`` to emit confusing error due to mixing information specific to a ``condor_dagman`` job with the information related to its jobs. +- Fix the issue with ``bps report`` providing incorrect job counts after restarts.