DM-53494: Add code needed to run submit processes as jobs.

doc/changes/DM-53494.feature.rst

@@ -0,0 +1 @@
+Added capabilities needed to run submit processes as batch jobs on same cluster with shared filesystems.  This included handling of lazy subdags.

doc/lsst.ctrl.bps.htcondor/userguide.rst

@@ -195,6 +195,12 @@ at the root level or inside a ``site`` section, but not inside a ``pipetask``,
 If your main workflow contains sub-workflow defined in individual DAG
 description files, they will use the same configuration as the main workflow.
 
+Miscellaneous
+^^^^^^^^^^^^^
+
+* ``saveHTCdot`` - true/false.  Whether condor_dagman outputs a DOT
+  representation of the workflow DAG.
+
 .. __: https://htcondor.readthedocs.io/en/latest/admin-manual/configuration-macros.html#dagman-configuration-file-entries
 
 .. .. _htc-plugin-authenticating:
@@ -628,6 +634,22 @@ For more information about expressions, see HTCondor documentation:
    2 held <submit dir>/*.nodes.log``.
 
 
+.. _htc_submit_as_batch:
+
+Submit Stages as Batch Jobs
+---------------------------
+
+When BPS uses batch jobs for submission, HTC handles them much the same
+way it handles payload jobs.  There will be a ``jobs`` subdir for each of
+them which will contain the standard output/error and HTCondor job log.
+If you have to provision resources for the payload jobs, these new jobs
+will also need resources provisioned.
+
+Noticable differences in the HTCondor-level details include a second
+``*.dag`` file in the submit directory and a second ``condor_dagman``
+job in the queue.  This is sub-DAG handles the payload workflow and
+isn't created until the ``preparePayloadWorkflow`` job executes.
+
 .. _htc-plugin-troubleshooting:
 
 Troubleshooting

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

@@ -229,6 +229,7 @@ def _wms_id_to_cluster(wms_id):
     schedd_ad = None
     cluster_id = None
     id_type = _wms_id_type(wms_id)
+    _LOG.debug("id_type = %s", id_type.name)
     if id_type == WmsIdType.LOCAL:
         schedd_ad = coll.locate(htcondor.DaemonTypes.Schedd)
         cluster_id = int(float(wms_id))
@@ -244,7 +245,7 @@ def _wms_id_to_cluster(wms_id):
             cluster_id = int(float(job_id))
     elif id_type == WmsIdType.PATH:
         try:
-            job_info = read_dag_info(wms_id)
+            _, job_info = read_dag_info(wms_id)
         except (FileNotFoundError, PermissionError, OSError):
             pass
         else:

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

@@ -66,6 +66,7 @@
         "dagman_debug": (str, ""),
         "dagman_node_record_info": (str, ""),
         "dagman_record_machine_attrs": (str, ""),
+        "dagman_manager_job_append_getenv": (str, ""),
     }
 )
 

python/lsst/ctrl/bps/htcondor/etc/htcondor_defaults.yaml

@@ -83,3 +83,58 @@ wmsConfig:
   # A boolean flag controlling whether DAGMan should generate submit files for
   # nested DAGs automatically.
   DAGMAN_GENERATE_SUBDAG_SUBMITS: true
+
+# Needs to be true for plain SLAC submissions.
+bpsMakeCommand: true
+
+payloadCommand: >
+  echo -n BPS Beginning of job commands:\ ;
+  /usr/bin/date +%Y-%m-%dT%H:%M:%S.%N%:z;
+  {setupEnv}
+  {gwjobExports}
+  {postEnvSetup}
+  logDir={jobLogDir};
+  pwd;
+  echo -n BPS Before running command:\ ;
+  /usr/bin/date +%Y-%m-%dT%H:%M:%S.%N%:z;
+  echo ========================================;
+  {gwjobCommand};
+  ret=$?;
+  echo ========================================;
+  echo -n BPS After running command:\ ;
+  /usr/bin/date +%Y-%m-%dT%H:%M:%S.%N%:z;
+  echo Command exited with code: $ret;
+  {jobCleanup}
+  echo -n BPS End of job commands:\ ;
+  /usr/bin/date +%Y-%m-%dT%H:%M:%S.%N%:z;
+  exit $ret;
+
+# Job environment setup
+softwarePath: "/cvmfs/sw.lsst.eu/almalinux-x86_64/lsst_distrib/{lsstVersion}"
+customEnvSetup: ""
+setupEnv: >
+  unset PYTHONPATH;
+  echo Using stack in {softwarePath};
+  source {softwarePath}/loadLSST.bash;
+  setup lsst_distrib;
+  echo -n BPS After setup of lsst_distrib:\ ;
+  /usr/bin/date +%Y-%m-%dT%H:%M:%S.%N%:z;
+  {customEnvSetup}
+  echo -n BPS After custom env setup:\ ;
+  /usr/bin/date +%Y-%m-%dT%H:%M:%S.%N%:z;
+
+gwjobExports: ""
+postEnvSetup: ""
+
+# Other job variables
+jobInitDir: "`pwd`"
+jobLogDir: "{jobInitDir}"
+jobCleanup: ""
+
+# Special slot on K8 AP node for running preparePayloadWorkflow
+site:
+  BPS_SUBMIT:
+    nodeset: "BPS_SUBMIT"
+    bpsUseShared: true
+    bpsUseRunTempSpace: false
+    softwarePath: "/sdf/group/rubin/sw/w_latest"

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

@@ -59,6 +59,7 @@
     htc_create_submit_from_file,
     htc_submit_dag,
     htc_version,
+    read_dag_info,
     read_dag_status,
     write_dag_info,
 )
@@ -180,10 +181,10 @@ def submit(self, workflow, **kwargs):
             _LOG.info("Submitting from directory: %s", os.getcwd())
             schedd_dag_info = htc_submit_dag(sub)
             if schedd_dag_info:
-                write_dag_info(f"{dag.name}.info.json", schedd_dag_info)
+                _, dag_info = next(iter(schedd_dag_info.items()))
+                dag_id, dag_ad = next(iter(dag_info.items()))
 
-                _, dag_info = schedd_dag_info.popitem()
-                _, dag_ad = dag_info.popitem()
+                write_dag_info(f"{dag_ad['bps_run']}.info.json", schedd_dag_info)
 
                 dag.run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}"
                 workflow.run_id = dag.run_id
@@ -256,6 +257,14 @@ def restart(self, wms_workflow_id):
                 "Cannot determine the execution status of the workflow, continuing with restart regardless"
             )
 
+        # In the case of lazy DAGs, workflow summaries can change at
+        # runtime.  So read the workflow's info.json file before moving
+        # it to backup dir and use to update the summaries later before
+        # writing the new info.json file.
+        dag_info_filename, old_dag_schedd_info = read_dag_info(wms_path)
+        old_dag_info = next(iter(old_dag_schedd_info.values()))
+        old_dag_ad = next(iter(old_dag_info.values()))
+
         _LOG.info("Backing up select HTCondor files from previous run attempt")
         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
@@ -284,7 +293,12 @@ def restart(self, wms_workflow_id):
                 if schedd_dag_info:
                     dag_info = next(iter(schedd_dag_info.values()))
                     dag_ad = next(iter(dag_info.values()))
-                    write_dag_info(f"{dag_ad['bps_run']}.info.json", schedd_dag_info)
+
+                    # Just in case lazy DAGs, update the summaries.
+                    dag_ad["bps_job_summary"] = old_dag_ad["bps_job_summary"]
+                    dag_ad["bps_run_quanta"] = old_dag_ad["bps_run_quanta"]
+
+                    write_dag_info(dag_info_filename, schedd_dag_info)
                     run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}"
                     run_name = dag_ad["bps_run"]
                 else:
@@ -587,3 +601,27 @@ def ping(self, pass_thru):
             status = 1
             message = f"Permission problem with {daemon_type} service."
         return status, message
+
+    def run_submission_checks(self):
+        """Check to run at start if running WMS specific submission steps.
+
+        Any exception other than NotImplementedError will halt submission.
+        Submit directory may not yet exist when this is called.
+        """
+        # Some early config sanity checks
+        found, value = self.config.search("bpsMakeCommand")
+        bps_make_command = value if found else True
+        if not bps_make_command:
+            found, value = self.config.search("payloadCommand", opt={"replaceVars": False})
+            if not found:
+                raise KeyError("Missing 'payloadCommand' in config while bpsMakeCommand=True")
+
+            if "setupEnv" in value:
+                found, value = self.config.search("setupEnv", opt={"replaceVars": False})
+                if not found:
+                    raise KeyError("Missing 'setupEnv' in config, but appears in payloadCommand")
+
+                if "lsstVersion" in value:
+                    found, value = self.config.search("lsstVersion", opt={"replaceVars": False})
+                    if not found:
+                        raise KeyError("Missing 'lsstVersion' in config, but appears in setupEnv in config")

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

@@ -35,9 +35,10 @@
 
 from lsst.ctrl.bps import (
     BaseWmsWorkflow,
+    BpsConfig,
 )
 
-from .prepare_utils import _generic_workflow_to_htcondor_dag
+from .prepare_utils import _generic_workflow_to_htcondor_dag, _update_job_summary
 
 _LOG = logging.getLogger(__name__)
 
@@ -87,3 +88,17 @@ def write(self, out_prefix):
 
         # Write down the workflow in HTCondor format.
         self.dag.write(out_prefix, job_subdir="jobs/{self.label}")
+
+    def add_to_parent_workflow(self, config: BpsConfig) -> None:
+        """Add self to parent workflow.
+
+        Parameters
+        ----------
+        config : `lsst.ctrl.bps.BpsConfig`
+            Configuration.
+        """
+        _update_job_summary(
+            self.name,
+            self.dag.graph["attr"]["bps_job_summary"],
+            config[".bps_defined.submitPath"],
+        )

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

@@ -188,7 +188,7 @@ class WmsNodeType(IntEnum):
     """Job used to correctly prune jobs after a subdag."""
 
 
-HTC_QUOTE_KEYS = {"environment"}
+HTC_QUOTE_KEYS = {"environment", "arguments"}
 HTC_VALID_JOB_KEYS = {
     "universe",
     "executable",
@@ -383,7 +383,10 @@ def htc_backup_files(
         raise FileNotFoundError(f"Directory {path} not found")
 
     # Initialize the backup counter.
-    rescue_dags = list(path.glob("*.rescue[0-9][0-9][0-9]"))
+    # If using control DAG, don't want to include nested DAGs.
+    rescue_dags = list(path.glob("*_ctrl.dag.rescue[0-9][0-9][0-9]"))
+    if not rescue_dags:
+        rescue_dags = list(path.glob("*.rescue[0-9][0-9][0-9]"))
     counter = min(len(rescue_dags), limit)
 
     # Create the backup directory and move select files there.
@@ -1007,7 +1010,10 @@ def write_submit_file(self, submit_path: str | os.PathLike) -> None:
         if not subfile.is_absolute():
             subfile = Path(submit_path) / subfile
         if not subfile.exists():
+            _LOG.debug("Writing subfile: %s", subfile)
             htc_write_condor_file(subfile, self.name, self.cmds, self.attrs)
+        else:
+            _LOG.debug("Using existing subfile: %s", subfile)
 
     def write_dag_commands(self, stream, dag_rel_path, command_name="JOB"):
         """Write DAG commands for single job to output stream.
@@ -1208,18 +1214,22 @@ def write(self, submit_path, job_subdir="", dag_subdir="", dag_rel_path=""):
                     _LOG.error("Job %s doesn't have data (keys: %s).", name, nodeval.keys())
                     raise
                 if job.subdag:
-                    dag_subdir = f"subdags/{job.name}"
+                    if job.subfile:
+                        this_dag_rel_path = ""
+                    else:
+                        this_dag_rel_path = "../.."
+                        dag_subdir = f"subdags/{job.name}"
                     if "dir" in job.dagcmds:
                         subdir = job.dagcmds["dir"]
                     else:
                         subdir = job_subdir
                     if dagman_config_path is not None:
                         job.subdag.add_attribs({"bps_wms_config_path": str(dagman_config_path)})
-                    job.subdag.write(submit_path, subdir, dag_subdir, "../..")
-                    fh.write(
-                        f"SUBDAG EXTERNAL {job.name} {Path(job.subdag.graph['dag_filename']).name} "
-                        f"DIR {dag_subdir}\n"
-                    )
+                    job.subdag.write(submit_path, subdir, dag_subdir, this_dag_rel_path)
+                    fh.write(f"SUBDAG EXTERNAL {job.name} {Path(job.subdag.graph['dag_filename']).name}")
+                    if dag_subdir:
+                        fh.write(f" DIR {dag_subdir}")
+                    fh.write("\n")
                     if job.dagcmds:
                         _htc_write_job_commands(fh, job.name, job.dagcmds)
                 else:
@@ -1228,7 +1238,10 @@ def write(self, submit_path, job_subdir="", dag_subdir="", dag_rel_path=""):
 
             for edge in self.edges():
                 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
-            print(f"DOT {self.name}.dot", file=fh)
+
+            if self.graph.get("write_dot", False):
+                print(f"DOT {self.name}.dot", file=fh)
+
             print(f"NODE_STATUS_FILE {self.name}.node_status", file=fh)
 
             # Add bps attributes to dag submission
@@ -1969,12 +1982,19 @@ def read_dag_log(wms_path: str | os.PathLike) -> tuple[str, dict[str, Any]]:
 
     path = Path(wms_path)
     if path.exists():
-        try:
-            filename = next(path.glob("*.dag.dagman.log"))
-        except StopIteration as exc:
-            raise FileNotFoundError(f"DAGMan log not found in {wms_path}") from exc
-        _LOG.debug("dag node log filename: %s", filename)
-        wms_workflow_id, dag_info = read_single_dag_log(filename)
+        # Can be more than one dag file in directory.  Assume one
+        # with lowest ID is main DAG
+        ids = []
+        for filename in path.glob("*.dag.dagman.log"):
+            _LOG.debug("dag log filename: %s", filename)
+            single_id, single_dag_info = read_single_dag_log(filename)
+            _update_dicts(dag_info, single_dag_info)
+            ids.append(single_id)
+        if ids:
+            wms_workflow_id = min(ids)
+
+    if wms_workflow_id == MISSING_ID:
+        raise FileNotFoundError(f"DAGMan log not found in {wms_path}")
 
     return wms_workflow_id, dag_info
 
@@ -2035,6 +2055,11 @@ def read_single_dag_nodes_log(filename: str | os.PathLike) -> dict[str, dict[str
             # plus subdags.
             if event["EventTypeNumber"] == 9 and info[id_].get("EventTypeNumber", -1) == 5:
                 _LOG.debug("Skipping spurious JobAbortedEvent: %s", dict(event))
+            elif event["EventTypeNumber"] == 16 and event["DAGNodeName"] == "finalJob":
+                # FINAL job's post script exit code is special and indicates
+                # status of DAG instead of just the FINAL job.  Save the
+                # information separately.
+                info[id_]["post"] = dict(event)
             else:
                 _update_dicts(info[id_], event)
                 info[id_][f"{event.type.name.lower()}_time"] = event["EventTime"]
@@ -2079,7 +2104,7 @@ def read_dag_nodes_log(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]
     return info
 
 
-def read_dag_info(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
+def read_dag_info(wms_path: str | os.PathLike) -> tuple[Path, dict[str, dict[str, Any]]]:
     """Read custom DAGMan job information from the file.
 
     Parameters
@@ -2089,6 +2114,8 @@ def read_dag_info(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
 
     Returns
     -------
+    filename : `pathlib.Path`
+        Name of file containing the dag information.
     dag_info : `dict` [`str`, `dict` [`str`, `~typing.Any`]]
         HTCondor job information.
 
@@ -2108,22 +2135,23 @@ def read_dag_info(wms_path: str | os.PathLike) -> dict[str, dict[str, Any]]:
             dag_info = json.load(fh)
     except (OSError, PermissionError) as exc:
         _LOG.debug("Retrieving DAGMan job information failed: %s", exc)
-    return dag_info
+    return filename, dag_info
 
 
-def write_dag_info(filename, dag_info):
+def write_dag_info(filename: str, schedd_dag_info: dict[str, dict[str, Any]]):
     """Write custom job information about DAGMan job.
 
     Parameters
     ----------
     filename : `str`
-        Name of the file where the information will be stored.
-    dag_info : `dict` [`str` `dict` [`str`, `~typing.Any`]]
+        Name of the file where the information will be stored.  If not given,
+        creates a filename using bps_run value.
+    schedd_dag_info : `dict` [`str` `dict` [`str`, `~typing.Any`]]
         Information about the DAGMan job.
     """
-    schedd_name = next(iter(dag_info))
-    dag_id = next(iter(dag_info[schedd_name]))
-    dag_ad = dag_info[schedd_name][dag_id]
+    _LOG.debug("schedd_dag_info = %s", schedd_dag_info)
+    schedd_name, dag_info = next(iter(schedd_dag_info.items()))
+    dag_id, dag_ad = next(iter(dag_info.items()))
     ad = {"ClusterId": dag_ad["ClusterId"], "GlobalJobId": dag_ad["GlobalJobId"]}
     ad.update({key: val for key, val in dag_ad.items() if key.startswith("bps")})
     try:
@@ -2133,6 +2161,8 @@ def write_dag_info(filename, dag_info):
     except (KeyError, OSError, PermissionError) as exc:
         _LOG.debug("Persisting DAGMan job information failed: %s", exc)
 
+    return filename
+
 
 def htc_tweak_log_info(wms_path: str | Path, job: dict[str, Any]) -> None:
     """Massage the given job info has same structure as if came from condor_q.