diff --git a/doc/changes/DM-56065.bugfix.rst b/doc/changes/DM-56065.bugfix.rst new file mode 100644 index 00000000..e86d483b --- /dev/null +++ b/doc/changes/DM-56065.bugfix.rst @@ -0,0 +1 @@ +Fixed a bug in prepare_utils._translate_command_line that caused ``submitcmd`` to throw an exception when given 0 job arguments. diff --git a/python/lsst/ctrl/bps/htcondor/htcondor_service.py b/python/lsst/ctrl/bps/htcondor/htcondor_service.py index 918f440a..302a629d 100644 --- a/python/lsst/ctrl/bps/htcondor/htcondor_service.py +++ b/python/lsst/ctrl/bps/htcondor/htcondor_service.py @@ -591,6 +591,7 @@ def ping(self, pass_thru): status = 0 message = "" _LOG.info("Not verifying that compute resources exist.") + daemon_type = htcondor.DaemonTypes.Schedd # To avoid possibly undefined error try: for daemon_type in [htcondor.DaemonTypes.Schedd, htcondor.DaemonTypes.Collector]: _ = secman.ping(coll.locate(daemon_type)) diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index b86c14c0..63768377 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/python/lsst/ctrl/bps/htcondor/lssthtc.py @@ -1538,6 +1538,7 @@ def count_jobs_in_single_dag( ) if m: job_name = m.group("jobname") + job_type = WmsNodeType.UNKNOWN name_parts = job_name.split("_") label = "" @@ -1765,6 +1766,7 @@ def read_single_node_status(filename: str | os.PathLike, init_fake_id: int) -> d # Get jobid info from other places to fill in gaps in info from node_status _, job_name_to_label, job_name_to_type = count_jobs_in_single_dag(filename.with_suffix(".dag")) loginfo: dict[str, dict[str, Any]] = {} + wms_workflow_id = MISSING_ID try: 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")) @@ -1933,7 +1935,7 @@ def read_single_dag_log(log_filename: str | os.PathLike) -> tuple[str, dict[str, FileNotFoundError If cannot find DAGMan log in given wms_path. """ - wms_workflow_id = "0" + wms_workflow_id = MISSING_ID dag_info: dict[str, dict[str, Any]] = {} filename = Path(log_filename) @@ -2266,6 +2268,11 @@ def htc_check_dagman_output(wms_path: str | os.PathLike) -> str: p = re.compile(r"^(\d\d/\d\d/\d\d \d\d:\d\d:\d\d) (Job submit try \d+/\d+ failed|Warning:.*$|ERROR:.*$)") message = "" + # Shouldn't normally see this warning. Initializing it just in case. + last_warning = ( + "Missing warning: Usually means the .dagman.out file has been truncated " + "or temporarily had a full quota." + ) try: with open(filename) as fh: last_submit_failed = "" # Since submit retries multiple times only report last one diff --git a/python/lsst/ctrl/bps/htcondor/prepare_utils.py b/python/lsst/ctrl/bps/htcondor/prepare_utils.py index ed8d6f80..dd8b4561 100644 --- a/python/lsst/ctrl/bps/htcondor/prepare_utils.py +++ b/python/lsst/ctrl/bps/htcondor/prepare_utils.py @@ -336,6 +336,13 @@ def _translate_command_line( jobcmds["environment"] = htc_envs.rstrip() _LOG.debug("_translate_command_line: saving htc environment = %s", jobcmds["environment"]) + arguments = "" + if gwjob.arguments: + arguments = gwjob.arguments + arguments = _replace_cmd_vars(arguments, gwjob) + arguments = _replace_wms_vars(arguments) + arguments = _replace_file_vars(cached_vals["bpsUseShared"], arguments, generic_workflow, gwjob) + if cached_vals.get("bpsMakeCommand", True): # Way to have fallback to previous behavior as well as # a way forward to centralize logic in bps. @@ -348,11 +355,7 @@ def _translate_command_line( else: jobcmds["executable"] = _fix_env_var_syntax(gwjob.executable.src_uri) - if gwjob.arguments: - arguments = gwjob.arguments - arguments = _replace_cmd_vars(arguments, gwjob) - arguments = _replace_wms_vars(arguments) - arguments = _replace_file_vars(cached_vals["bpsUseShared"], arguments, generic_workflow, gwjob) + if arguments: arguments = _fix_env_var_syntax(arguments) jobcmds["arguments"] = arguments @@ -364,11 +367,7 @@ def _translate_command_line( # Don't set getenv as setting up the environment is assumed to be # part of the payloadCommand. - if gwjob.arguments: - arguments = gwjob.arguments - arguments = _replace_cmd_vars(arguments, gwjob) - arguments = _replace_wms_vars(arguments) - arguments = _replace_file_vars(cached_vals["bpsUseShared"], arguments, generic_workflow, gwjob) + if arguments: arguments = _fix_env_var_syntax_shell(arguments) if gwjob.executable.transfer_executable: @@ -380,23 +379,22 @@ def _translate_command_line( generic_workflow.add_job_inputs(gwjob.name, [gwfile]) exec_name = os.path.basename(gwjob.executable.src_uri) # Ensure the executable copy is executable. - gwjobCommand = f"chmod u+x {exec_name}; ./{exec_name} {arguments}" + gwjob_command = f"chmod u+x {exec_name}; ./{exec_name} {arguments}" else: exec_name = _fix_env_var_syntax_shell(gwjob.executable.src_uri) - gwjobCommand = f"{exec_name} {arguments}" + gwjob_command = f"{exec_name} {arguments}" - payloadCommand = cached_vals["payloadCommand"] - _LOG.debug("%s payloadCommand pre-format: %s", gwjob.label, payloadCommand) - payloadCommand = re.sub("{gwjobCommand}", gwjobCommand, payloadCommand) - payloadCommand = re.sub("{gwjobExports}", job_exports, payloadCommand) + payload_command = cached_vals["payloadCommand"] + _LOG.debug("%s payload_command pre-format: %s", gwjob.label, payload_command) + payload_command = re.sub("{gwjobCommand}", gwjob_command, payload_command) + payload_command = re.sub("{gwjobExports}", job_exports, payload_command) # Remove newlines - payloadCommand = re.sub("\n", "", payloadCommand) + payload_command = re.sub("\n", "", payload_command) - _LOG.debug("%s payloadCommand post-format: %s", gwjob.label, payloadCommand) + _LOG.debug("%s payload_command post-format: %s", gwjob.label, payload_command) - # jobcmds["arguments"] = htc_escape(f"-c '{payloadCommand}'") - jobcmds["arguments"] = f"-c '{payloadCommand}'" + jobcmds["arguments"] = f"-c '{payload_command}'" jobcmds["executable"] = "/bin/bash" # Don't need to transfer /bin/bash diff --git a/python/lsst/ctrl/bps/htcondor/report_utils.py b/python/lsst/ctrl/bps/htcondor/report_utils.py index f39a10be..d97601ca 100644 --- a/python/lsst/ctrl/bps/htcondor/report_utils.py +++ b/python/lsst/ctrl/bps/htcondor/report_utils.py @@ -117,6 +117,7 @@ def _get_status_from_path(wms_path: str | os.PathLike) -> tuple[WmsStates, str]: """ wms_path = Path(wms_path).resolve() message = "" + dag_ad: dict[str, Any] = {} try: wms_workflow_id, dag_ad = read_dag_log(wms_path) except FileNotFoundError: diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index cddea744..0b39ed79 100644 --- a/tests/test_lssthtc.py +++ b/tests/test_lssthtc.py @@ -221,6 +221,16 @@ def test_no_messages(self): results = lssthtc.htc_check_dagman_output(tmp_dir) self.assertEqual("", results) + def test_missing_warning(self): + with temporaryDirectory() as tmp_dir: + with open(os.path.join(tmp_dir, "test_missing_warning.dag.dagman.out"), "w") as fh: + print( + "01/01/26 01:01:01 ERROR: Warning is fatal error because of DAGMAN_USE_STRICT setting", + file=fh, + ) + results = lssthtc.htc_check_dagman_output(tmp_dir) + self.assertIn("Missing warning", results) + class SummarizeDagTestCase(unittest.TestCase): """Test summarize_dag function.""" @@ -791,6 +801,29 @@ def testMissingLogFiles(self): self.assertEqual(len(jobs), 5) for job in jobs.values(): self.assertLess(job["ClusterId"], 0) + self.assertEqual(job["DAGManJobID"], lssthtc.MISSING_ID) + + # 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 testPermissionLogFiles(self): + # Check what happens if can't read dag log file. + self._copyFiles("tiny_success", [".dag", ".node_status", ".dag.dagman.log"]) + filename = os.path.join(self.tmpdir, "tiny_success.dag.dagman.log") + current_mode = os.stat(filename).st_mode + no_read_mode = current_mode & ~stat.S_IRUSR & ~stat.S_IRGRP & ~stat.S_IROTH + os.chmod(filename, no_read_mode) + + 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) + self.assertEqual(job["DAGManJobID"], lssthtc.MISSING_ID) # NodeStatus values from the node_status file must still be preserved. name_to_id = self._jobNameToId(jobs) diff --git a/tests/test_prepare_utils.py b/tests/test_prepare_utils.py index d65c8a28..d30a69f5 100644 --- a/tests/test_prepare_utils.py +++ b/tests/test_prepare_utils.py @@ -233,7 +233,7 @@ def testMakeCommandArguments(self): def testPayloadCommandBasic(self): # bpsMakeCommand False wraps the payloadCommand in /bin/bash -c. - gw, gwjob = self._make_job(arguments="pipetask run") + gw, gwjob = self._make_job(arguments="run -b repo") cached_vals = { "bpsUseShared": True, "bpsMakeCommand": False, @@ -243,7 +243,21 @@ def testPayloadCommandBasic(self): self.assertEqual(jobcmds["executable"], "/bin/bash") self.assertEqual(jobcmds["transfer_executable"], "False") self.assertNotIn("getenv", jobcmds) - self.assertEqual(jobcmds["arguments"], "-c 'setup; /dummy/dir/pipetask pipetask run'") + self.assertEqual(jobcmds["arguments"], "-c 'setup; /dummy/dir/pipetask run -b repo'") + + def testPayloadCommandNoArguments(self): + # No argument to the gwjob command + gw, gwjob = self._make_job(arguments="") + cached_vals = { + "bpsUseShared": True, + "bpsMakeCommand": False, + "payloadCommand": "setup; {gwjobCommand}", + } + jobcmds = prepare_utils._translate_command_line(cached_vals, gw, gwjob) + self.assertEqual(jobcmds["executable"], "/bin/bash") + self.assertEqual(jobcmds["transfer_executable"], "False") + self.assertNotIn("getenv", jobcmds) + self.assertEqual(jobcmds["arguments"], "-c 'setup; /dummy/dir/pipetask '") def testPayloadCommandStripsNewlines(self): gw, gwjob = self._make_job(arguments="run") @@ -275,7 +289,7 @@ def testPayloadCommandTransferExecutable(self): def testEnvironment(self): gw, gwjob = self._make_job() - gwjob.environment = {"TEST_INT": "1", "TEST_STR": "TWO"} + gwjob.environment = {"TEST_INT": 1, "TEST_STR": "TWO"} jobcmds = prepare_utils._translate_command_line(self.cached_vals, gw, gwjob) self.assertEqual(jobcmds["environment"], "TEST_INT='1' TEST_STR='TWO'")