DM-56065: Fix exception-causing bug when 0 job arguments.

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.

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))

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

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

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:

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)

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'")