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. 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/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 1461214..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 ------ @@ -411,16 +413,20 @@ 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*")}: + 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. @@ -428,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. @@ -1746,7 +1752,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 @@ -2265,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 @@ -2277,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) @@ -2358,28 +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: - """Update the subdag failures in the main rescue file - and backup the failed subdag dirs. +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) - for failed_subdag in failed_subdags: - htc_backup_files( - rescue_file.parent / "subdags" / failed_subdag, subdir=f"backups/subdags/{failed_subdag}" - ) + return failed_subdags def _update_dicts(dict1, dict2): diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index 8318cfe..e612698 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 @@ -644,6 +643,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 _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 _jobNameToId(self, jobs): + return {info["DAGNodeName"]: id_ for id_, info in jobs.items()} + + def testAllDone(self): + self._copyFiles( + "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._jobNameToId(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 testMixedStatuses(self): + self._copyFiles( + "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._jobNameToId(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 testRunningWorkflow(self): + self._copyFiles( + "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._jobNameToId(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 testMissingNodeStatusFile(self): + # Omit the .node_status file; jobs must be built from the event log + # and dag. + self._copyFiles( + "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._jobNameToId(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 testMissingLogFiles(self): + # Omit both log files; every job should get a fake negative ID. + self._copyFiles("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._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 testInitFakeId(self): + # Verify fake IDs count down from the given starting value. + 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) + + 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 testFromDagJobAttribute(self): + self._copyFiles( + "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 testServiceJobPlaceholder(self): + self._copyFiles( + "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.""" @@ -862,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]) @@ -893,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]) @@ -920,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]) @@ -944,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]) @@ -968,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]) @@ -983,6 +1179,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", @@ -990,21 +1190,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", }, ) @@ -1019,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) @@ -1058,11 +1254,109 @@ 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) +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."""