From 5793384e9e15935912a32f6a5181c3735bcf6c31 Mon Sep 17 00:00:00 2001 From: Michelle Gower Date: Fri, 20 Feb 2026 11:37:40 -0600 Subject: [PATCH 01/10] Add code needed to run submit processes as jobs. Note: Limited to same cluster with shared filesystems. Includes handling of lazy subdags. --- doc/changes/DM-53494.feature.rst | 1 + doc/lsst.ctrl.bps.htcondor/userguide.rst | 16 ++ python/lsst/ctrl/bps/htcondor/common_utils.py | 3 +- .../ctrl/bps/htcondor/htcondor_service.py | 4 +- .../ctrl/bps/htcondor/htcondor_workflow.py | 17 +- python/lsst/ctrl/bps/htcondor/lssthtc.py | 63 ++++++-- .../lsst/ctrl/bps/htcondor/prepare_utils.py | 150 +++++++++++++++++- python/lsst/ctrl/bps/htcondor/report_utils.py | 4 +- tests/dag_test_utils.py | 107 +++++++++++++ tests/test_common_utils.py | 47 +++++- tests/test_htcondor_workflow.py | 50 ++++++ tests/test_lssthtc.py | 95 +++++++++-- tests/test_prepare_utils.py | 147 ++++++++++++++++- 13 files changed, 660 insertions(+), 44 deletions(-) create mode 100644 doc/changes/DM-53494.feature.rst create mode 100644 tests/dag_test_utils.py create mode 100644 tests/test_htcondor_workflow.py diff --git a/doc/changes/DM-53494.feature.rst b/doc/changes/DM-53494.feature.rst new file mode 100644 index 0000000..46562fe --- /dev/null +++ b/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. diff --git a/doc/lsst.ctrl.bps.htcondor/userguide.rst b/doc/lsst.ctrl.bps.htcondor/userguide.rst index c8d8570..5dbfc70 100644 --- a/doc/lsst.ctrl.bps.htcondor/userguide.rst +++ b/doc/lsst.ctrl.bps.htcondor/userguide.rst @@ -601,6 +601,22 @@ For more information about expressions, see HTCondor documentation: 2 held /*.nodes.log``. +.. _htc_submit_as_batch: + +Submit Stages as Batch Jobs +--------------------------- + +When BPS uses batch jobs for the submission process, there isn't much +different in how HTC handles them vs the 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 subDAG handles the payload workflow and +isn't created until the ``preparePayloadWorkflow`` job executes. + .. _htc-plugin-troubleshooting: Troubleshooting diff --git a/python/lsst/ctrl/bps/htcondor/common_utils.py b/python/lsst/ctrl/bps/htcondor/common_utils.py index d4507aa..a74042d 100644 --- a/python/lsst/ctrl/bps/htcondor/common_utils.py +++ b/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: diff --git a/python/lsst/ctrl/bps/htcondor/htcondor_service.py b/python/lsst/ctrl/bps/htcondor/htcondor_service.py index 4ef025a..46a3b23 100644 --- a/python/lsst/ctrl/bps/htcondor/htcondor_service.py +++ b/python/lsst/ctrl/bps/htcondor/htcondor_service.py @@ -180,7 +180,7 @@ 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) + write_dag_info(schedd_dag_info) _, dag_info = schedd_dag_info.popitem() _, dag_ad = dag_info.popitem() @@ -280,7 +280,7 @@ 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) + write_dag_info(schedd_dag_info) run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}" run_name = dag_ad["bps_run"] else: diff --git a/python/lsst/ctrl/bps/htcondor/htcondor_workflow.py b/python/lsst/ctrl/bps/htcondor/htcondor_workflow.py index d74655c..af9cfaf 100644 --- a/python/lsst/ctrl/bps/htcondor/htcondor_workflow.py +++ b/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"], + ) diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index 1461214..3426352 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/python/lsst/ctrl/bps/htcondor/lssthtc.py @@ -448,6 +448,7 @@ def htc_backup_files_single_path(src: str | os.PathLike, dest: str | os.PathLike for patt in [ "*.info.*", + "*.dag.dagman.log", "*.dag.metrics", "*.dag.nodes.log", "*.node_status", @@ -1000,7 +1001,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. @@ -1201,18 +1205,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: @@ -1962,12 +1970,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 @@ -2072,7 +2087,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[str, dict[Path, dict[str, Any]]]: """Read custom DAGMan job information from the file. Parameters @@ -2082,6 +2097,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. @@ -2101,24 +2118,34 @@ 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(dag_info: dict[str, dict[str, Any]], filename: str = None) -> Path: """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`]] Information about the DAGMan job. + filename : `str`, optional + Name of the file where the information will be stored. If not given, + creates a filename using bps_run value. + + Returns + ------- + filename : `pathlib.Path` + Name of the file where the information was written. """ + _LOG.debug("dag_info = %s", dag_info) schedd_name = next(iter(dag_info)) dag_id = next(iter(dag_info[schedd_name])) dag_ad = dag_info[schedd_name][dag_id] ad = {"ClusterId": dag_ad["ClusterId"], "GlobalJobId": dag_ad["GlobalJobId"]} ad.update({key: val for key, val in dag_ad.items() if key.startswith("bps")}) + if not filename: + filename = Path(f"{ad['bps_run']}.info.json") + try: with open(filename, "w") as fh: info = {schedd_name: {dag_id: ad}} @@ -2126,6 +2153,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. diff --git a/python/lsst/ctrl/bps/htcondor/prepare_utils.py b/python/lsst/ctrl/bps/htcondor/prepare_utils.py index 0273bff..44a13ae 100644 --- a/python/lsst/ctrl/bps/htcondor/prepare_utils.py +++ b/python/lsst/ctrl/bps/htcondor/prepare_utils.py @@ -51,6 +51,8 @@ _update_dicts, condor_status, htc_escape, + read_dag_info, + write_dag_info, ) _LOG = logging.getLogger(__name__) @@ -250,7 +252,10 @@ def _translate_job_cmds(cached_vals, generic_workflow, gwjob): # No point in adding periodic_release if 0 retries if gwjob.number_of_retries > 0: periodic_release = _create_periodic_release_expr( - gwjob.request_memory, gwjob.memory_multiplier, memory_max, user_release_expr + gwjob.request_memory, + gwjob.memory_multiplier, + memory_max, + user_release_expr, ) if periodic_release: jobcmds["periodic_release"] = periodic_release @@ -426,7 +431,10 @@ def _replace_cmd_vars(arguments, gwjob): try: arguments = arguments.format(**replacements) except (KeyError, TypeError) as exc: # TypeError in case None instead of {} - _LOG.error("Could not replace command variables: replacement for %s not provided", str(exc)) + _LOG.error( + "Could not replace command variables: replacement for %s not provided", + str(exc), + ) _LOG.debug("arguments: %s\ncmdvals: %s", arguments, replacements) raise return arguments @@ -917,19 +925,28 @@ def _generic_workflow_to_htcondor_dag( else: subdir_template = tmp_template + # Save list of lazy group jobs for later extra handling. + lazy_groups = [] + # Create all DAG jobs site_values = {} # Cache compute site specific values to reduce config lookups. cached_values = {} # Cache label-specific values to reduce config lookups. # Note: Can't use get_job_by_label because those only include payload jobs. for job_name in generic_workflow: gwjob = generic_workflow.get_job(job_name) - if gwjob.node_type == GenericWorkflowNodeType.PAYLOAD: + if gwjob.node_type in [ + GenericWorkflowNodeType.PAYLOAD, + GenericWorkflowNodeType.LAZY_GROUP, + ]: gwjob = cast(GenericWorkflowJob, gwjob) if gwjob.compute_site not in site_values: site_values[gwjob.compute_site] = _gather_site_values(config, gwjob.compute_site) if gwjob.label not in cached_values: cached_values[gwjob.label] = deepcopy(site_values[gwjob.compute_site]) - _update_dicts(cached_values[gwjob.label], _gather_label_values(config, gwjob.label)) + _update_dicts( + cached_values[gwjob.label], + _gather_label_values(config, gwjob.label), + ) _LOG.debug("cached: %s= %s", gwjob.label, cached_values[gwjob.label]) htc_job = _create_job( subdir_template[gwjob.label], @@ -952,11 +969,17 @@ def _generic_workflow_to_htcondor_dag( _LOG.debug("Calling adding job %s %s", htc_job.name, htc_job.label) dag.add_job(htc_job) + # Have to add the placeholder job for the workflow + if gwjob.node_type == GenericWorkflowNodeType.LAZY_GROUP: + lazy_groups.append(gwjob.name) + # Add job dependencies to the DAG (be careful with wms_ jobs) for job_name in generic_workflow: gwjob = generic_workflow.get_job(job_name) parent_name = ( - gwjob.name if gwjob.node_type == GenericWorkflowNodeType.PAYLOAD else f"wms_{gwjob.name}" + gwjob.name + if gwjob.node_type in [GenericWorkflowNodeType.PAYLOAD, GenericWorkflowNodeType.LAZY_GROUP] + else f"wms_{gwjob.name}" ) successor_jobs = [generic_workflow.get_job(j) for j in generic_workflow.successors(job_name)] children_names = [] @@ -981,13 +1004,20 @@ def _generic_workflow_to_htcondor_dag( parent_name = check_job.name else: for sjob in successor_jobs: - if sjob.node_type == GenericWorkflowNodeType.PAYLOAD: + if sjob.node_type in [ + GenericWorkflowNodeType.PAYLOAD, + GenericWorkflowNodeType.LAZY_GROUP, + ]: children_names.append(sjob.name) else: children_names.append(f"wms_{sjob.name}") dag.add_job_relationships([parent_name], children_names) + # Go back and add placeholder jobs for the lazy group dags + for lazy_group_name in lazy_groups: + _add_lazy_placeholder(lazy_group_name, generic_workflow, dag, out_prefix) + # If final job exists in generic workflow, create DAG final job final = generic_workflow.get_final() if final and isinstance(final, GenericWorkflowJob): @@ -1016,3 +1046,111 @@ def _generic_workflow_to_htcondor_dag( raise TypeError(f"Invalid type for GenericWorkflow.get_final() results ({type(final)})") return dag + + +def _add_lazy_placeholder( + prepare_job_name: str, + generic_workflow: GenericWorkflow, + dag: HTCDag, + out_prefix: str, +): + _LOG.debug("prepare_job_name = %s", prepare_job_name) + + # Make a fake job for the placeholder workflow + job1 = HTCJob("placeholder") + job1.add_job_cmds( + { + "executable": "/usr/bin/echo", + "arguments": '"BPS internal error - placeholder DAG was not replaced."', + } + ) + job1.add_job_attrs({"bps_job_name": "placeholder", "bps_job_label": "placeholder"}) + job1.add_job_attrs(generic_workflow.run_attrs) + + # TODO - Placeholder dag name needs to be the run name because that's + # currently what the bps code will name the workflow. Needs to be made + # more generic. + placeholder_dag_name = generic_workflow.name.replace("_ctrl", "") + placeholder_dag = HTCDag(name=placeholder_dag_name) + _LOG.debug("dag name = %s", placeholder_dag.graph["name"]) + placeholder_dag.add_attribs(generic_workflow.run_attrs) + placeholder_dag.add_job(job1) + + # To help ordering in bps report, save info so can figure out where + # to put jobs in lazy dag in bps_job_label + dag.add_attribs({"bps_lazy_mapping": f"{placeholder_dag.name}:{prepare_job_name}"}) + + # The subdag job to be added to the control dag + dag_job = HTCJob("wms_lazy_payload", "wms_lazy_payload") + dag_job.subfile = f"{placeholder_dag.name}.condor.sub" + dag_job.subdag = placeholder_dag + + dag.add_job(dag_job) + + # Update edges inserting between prepare job and any following jobs. + successor_jobs = list(dag.successors(prepare_job_name)) + for job in successor_jobs: + dag.add_edge(dag_job.name, job) + dag.remove_edge(prepare_job_name, job) + + dag.add_edge(prepare_job_name, dag_job.name) + + _LOG.debug("_add_lazy_placeholder: edges = %s", dag.edges) + + +def _update_job_summary(subworkflow_name: str, subworkflow_summary: str, submit_path: str) -> None: + """Add summary for jobs in given subworkflow to parent workflow's job + summary. + + Parameters + ---------- + subworkflow_name : `str` + Name of subworkflow. + subworkflow_summary : `str` + Job summary for the subworkflow. + submit_path : `str` + Directory in which to find the DAG info file. + """ + _LOG.debug("submit_path = %s", submit_path) + filename, dag_info = read_dag_info(submit_path) + _LOG.debug("dag_info = %s", dag_info) + + schedd_name = next(iter(dag_info)) + dag_values = next(iter(dag_info[schedd_name].values())) + _LOG.debug("dag_values = %s", dag_values) + + # Get lazy mapping and the job that generated this dag + generator_name = None + lazy_mapping = dag_values.get("bps_lazy_mapping", None) + if lazy_mapping: # find this workflow's generator job + for part in lazy_mapping.split(";"): + info = part.split(":") + if info[0] == subworkflow_name: + generator_name = info[1] + break + + # Update bps_job_summary + _LOG.debug( + "Before replace, name = %s, bps_job_summary = %s, add summary = %s, generator_name = %s", + subworkflow_name, + dag_values["bps_job_summary"], + subworkflow_summary, + generator_name, + ) + if generator_name: + generator_summary = f"{generator_name}:1" + dag_values["bps_job_summary"] = dag_values["bps_job_summary"].replace( + generator_summary, f"{generator_summary};{subworkflow_summary}" + ) + else: + # just append to end of bps_job_summary + dag_values["bps_job_summary"] += f";{subworkflow_summary}" + + _LOG.debug( + "After replace, name = %s, bps_job_summary = %s", + subworkflow_name, + dag_values["bps_job_summary"], + ) + + # Save updated bps_job_summary + write_dag_info(dag_info, filename) diff --git a/python/lsst/ctrl/bps/htcondor/report_utils.py b/python/lsst/ctrl/bps/htcondor/report_utils.py index 39bdf85..1327e3d 100644 --- a/python/lsst/ctrl/bps/htcondor/report_utils.py +++ b/python/lsst/ctrl/bps/htcondor/report_utils.py @@ -347,7 +347,7 @@ def _get_info_from_path(wms_path: str | os.PathLike) -> tuple[str, dict[str, dic # (instead of sneakily using HTCondor's one), the lack of that file # should be treated as seriously as lack of any other file. try: - job_info = read_dag_info(wms_path) + _, job_info = read_dag_info(wms_path) except FileNotFoundError as exc: message = f"Warn: Some information may not be available: {exc}" messages.append(message) @@ -543,7 +543,7 @@ def _summary_report(user, hist, pass_thru, schedds=None): # * bps_isjob == 'True' isn't getting set for DAG jobs that are # manually restarted. # * Any job with DAGManJobID isn't a DAG job - constraint = 'bps_isjob == "True" && JobUniverse == 7' + constraint = 'bps_isjob == "True" && JobUniverse == 7 && DAGManJobID =?= Undefined' if user: constraint += f' && (Owner == "{user}" || bps_operator == "{user}")' diff --git a/tests/dag_test_utils.py b/tests/dag_test_utils.py new file mode 100644 index 0000000..9acabf9 --- /dev/null +++ b/tests/dag_test_utils.py @@ -0,0 +1,107 @@ +# This file is part of ctrl_bps_htcondor. +# +# Developed for the LSST Data Management System. +# This product includes software developed by the LSST Project +# (https://www.lsst.org). +# See the COPYRIGHT file at the top-level directory of this distribution +# for details of code ownership. +# +# This software is dual licensed under the GNU General Public License and also +# under a 3-clause BSD license. Recipients may choose which of these licenses +# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, +# respectively. If you choose the GPL option then the following text applies +# (but note that there is still no warranty even if you opt for BSD instead): +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +"""Utility functions for ctrl_bps_htcondor tests.""" + +import logging +from pathlib import Path + +from lsst.ctrl.bps.htcondor.lssthtc import HTC_VALID_JOB_KEYS, HTCDag, HTCJob, RestrictedDict + +_LOG = logging.getLogger(__name__) + + +def make_lazy_dag(name: str, final: bool = True) -> tuple[HTCDag, list[str]]: + """Manually make a simple HTCDag with a lazy subdag for unit testing. + + Parameters + ---------- + name : `str` + Name to use for HTCDag. + final : `bool`, optional + Whether to include a FINAL job. Defaults to True. + + Returns + ------- + dag : `lsst.ctrl.bps.htcondor.lssthtc.HTCDag` + A simple HTCDag with a lazy subdag for testing. + filelist : `list` [`str`] + A list of files with relative paths that should be created + when DAG is written. + """ + dag = HTCDag(name=name) + + job_cmds = RestrictedDict(HTC_VALID_JOB_KEYS, {"executable": "/usr/bin/uptime"}) + job_attrs = {"bps_run": "test_run"} + + job = HTCJob("job1", "label1", job_cmds, initattrs=job_attrs) + subdir = Path(f"jobs/{job.label}") + job.subdir = subdir + job.subfile = f"{job.name}.sub" + job.add_dag_cmds({"dir": subdir}) + dag.add_node("job1", data=job) + + job = HTCJob("make_subdag", "label2", job_cmds, initattrs=job_attrs) + subdir = Path(f"jobs/{job.label}") + job.subdir = subdir + job.subfile = f"{job.name}.sub" + job.add_dag_cmds({"dir": subdir}) + dag.add_node(job.name, data=job) + dag.add_edge("job1", job.name) + + job = HTCJob("lazy_subdag", "lazy_label", job_cmds, initattrs=job_attrs) + job.subfile = "lazy_subdag.dag.condor.sub" + job.dag = HTCDag(name="lazy_subdag") + dag.add_node(job.name, data=job) + dag.add_edge("make_subdag", job.name) + + job = HTCJob("job4", "label4", job_cmds, initattrs=job_attrs) + subdir = Path(f"jobs/{job.label}") + job.subdir = subdir + job.subfile = f"{job.name}.sub" + job.add_dag_cmds({"dir": subdir}) + dag.add_node(job.name, data=job) + dag.add_edge("lazy_subdag", job.name) + + filelist = [ + f"{name}.dag", + "lazy_subdag.dag.condor.sub", + "jobs/label1/job1.sub", + "jobs/label2/make_subdag.sub", + "jobs/label4/job4.sub", + ] + + if final: + job = HTCJob("finalJob", "finalJob", job_cmds, initattrs=job_attrs) + subdir = Path(f"jobs/{job.label}") + job.subdir = subdir + job.subfile = f"{job.name}.sub" + job.add_dag_cmds({"dir": subdir}) + dag.graph["final_job"] = job + filelist.append("jobs/finalJob/finalJob.sub") + + return dag, filelist diff --git a/tests/test_common_utils.py b/tests/test_common_utils.py index c3bb56b..b8c66ca 100644 --- a/tests/test_common_utils.py +++ b/tests/test_common_utils.py @@ -108,12 +108,18 @@ def testDone(self): self.assertEqual(result, WmsStates.SUCCEEDED) def testErrorDagmanSuccess(self): - job = {"NodeStatus": common_utils.NodeStatus.ERROR, "StatusDetails": "DAGMAN error 0"} + job = { + "NodeStatus": common_utils.NodeStatus.ERROR, + "StatusDetails": "DAGMAN error 0", + } result = common_utils._htc_node_status_to_wms_state(job) self.assertEqual(result, WmsStates.SUCCEEDED) def testErrorDagmanFailure(self): - job = {"NodeStatus": common_utils.NodeStatus.ERROR, "StatusDetails": "DAGMAN error 1"} + job = { + "NodeStatus": common_utils.NodeStatus.ERROR, + "StatusDetails": "DAGMAN error 1", + } result = common_utils._htc_node_status_to_wms_state(job) self.assertEqual(result, WmsStates.FAILED) @@ -239,5 +245,42 @@ def testUnknownType(self): self.assertEqual(id_type, common_utils.WmsIdType.UNKNOWN) +class WmsIdToClusterTestCase(unittest.TestCase): + """Test _wms_id_to_cluster function.""" + + class _MockCollector: + def locate(self, daemon_type, schedd_name): + return """[ + CondorPlatform = "$CondorPlatform: X86_64-AlmaLinux_9.6 $"; + MyType = "Scheduler"; + Machine = "testmachine"; + Name = "testmachine"; + CondorVersion = "$CondorVersion: 24.0.10 2025-08-05 $"; + MyAddress = "<127.0.0.1:9618?addrs=127.0.0.1-9618+snip>" + ]""" + + @unittest.mock.patch("htcondor.Collector", new=_MockCollector) + @unittest.mock.patch("lsst.ctrl.bps.htcondor.common_utils.read_dag_info") + def testPath(self, mock_read): + # path must exist or _wms_id_type will assume GLOBAL string + with temporaryDirectory() as tmp_dir: + mock_read.return_value = [ + "dummy_str", + { + "testmachine": { + "1163.0": { + "testmachine": { + "ClusterId": 1163, + "GlobalJobId": "testmachine#1163.0#1722040518", + } + } + } + }, + ] + schedd_ad, cluster_id, id_type = common_utils._wms_id_to_cluster(tmp_dir) + self.assertEqual(id_type, common_utils.WmsIdType.PATH) + self.assertEqual(cluster_id, 1163) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_htcondor_workflow.py b/tests/test_htcondor_workflow.py new file mode 100644 index 0000000..6e3f094 --- /dev/null +++ b/tests/test_htcondor_workflow.py @@ -0,0 +1,50 @@ +# This file is part of ctrl_bps_htcondor. +# +# Developed for the LSST Data Management System. +# This product includes software developed by the LSST Project +# (https://www.lsst.org). +# See the COPYRIGHT file at the top-level directory of this distribution +# for details of code ownership. +# +# This software is dual licensed under the GNU General Public License and also +# under a 3-clause BSD license. Recipients may choose which of these licenses +# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, +# respectively. If you choose the GPL option then the following text applies +# (but note that there is still no warranty even if you opt for BSD instead): +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +"""Unit tests for the HTCondor WMS workflow class and related functions.""" + +import logging +import unittest + +from lsst.ctrl.bps import BpsConfig +from lsst.ctrl.bps.htcondor.htcondor_workflow import HTCondorWorkflow +from lsst.ctrl.bps.htcondor.lssthtc import HTCDag + +logger = logging.getLogger("lsst.ctrl.bps.htcondor") + + +class HTCondorWorkflowTestCase(unittest.TestCase): + """Test selected methods of the HTCondor WMS workflow class.""" + + @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_workflow._update_job_summary") + def testAddToParentWorkflow(self, mock_update): + config = BpsConfig({"bps_defined": {"submitPath": "/mock/path"}}) + workflow = HTCondorWorkflow("workflow1") + workflow.dag = HTCDag() + workflow.dag.graph["attr"]["bps_job_summary"] = "label1:5;label2:10" + workflow.add_to_parent_workflow(config) + mock_update.assert_called_once_with("workflow1", "label1:5;label2:10", "/mock/path") diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index 8318cfe..cd34cfa 100644 --- a/tests/test_lssthtc.py +++ b/tests/test_lssthtc.py @@ -37,8 +37,10 @@ from shutil import copy2, copytree, ignore_patterns, rmtree, which import htcondor +from dag_test_utils import make_lazy_dag from lsst.ctrl.bps import BpsConfig +from lsst.ctrl.bps.bps_utils import chdir from lsst.ctrl.bps.htcondor import dagman_configurator, htcondor_config, lssthtc from lsst.daf.butler import Config from lsst.utils.tests import temporaryDirectory @@ -375,6 +377,7 @@ def test_noop(self): ) def test_subdags(self): + self.maxDiff = None with temporaryDirectory() as tmp_dir: submit_dir = os.path.join(tmp_dir, "group_running_1") copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*")) @@ -835,7 +838,6 @@ def testSuccess(self): self.assertEqual( set(result_submit), { - "./tiny_success.dag.dagman.log", "./tiny_success.dag.dagman.out", "./tiny_success.dag", }, @@ -847,6 +849,7 @@ def testSuccess(self): set(result_backup), { "./tiny_success.info.json", + "./tiny_success.dag.dagman.log", "./tiny_success.dag.metrics", "./tiny_success.dag.nodes.log", "./tiny_success.node_status", @@ -877,10 +880,10 @@ def testSuccess(self): self.assertEqual( set(result_submit), { - "./tiny_success.dag.dagman.log", "./tiny_success.dag.dagman.out", "./tiny_success.dag", "000/tiny_success.info.json", + "000/tiny_success.dag.dagman.log", "000/tiny_success.dag.metrics", "000/tiny_success.dag.nodes.log", "000/tiny_success.node_status", @@ -903,11 +906,11 @@ def testDestNotInSubmitDir(self): self.assertEqual( set(result_submit), { - "./tiny_problems.dag.dagman.log", "./tiny_problems.dag.dagman.out", "./tiny_problems.dag", "./tiny_problems.dag.rescue001", "001/tiny_problems.info.json", + "001/tiny_problems.dag.dagman.log", "001/tiny_problems.dag.metrics", "001/tiny_problems.dag.nodes.log", "001/tiny_problems.node_status", @@ -928,11 +931,11 @@ def testDestInSubmitDir(self): self.assertEqual( set(result_submit), { - "./tiny_problems.dag.dagman.log", "./tiny_problems.dag.dagman.out", "./tiny_problems.dag", "./tiny_problems.dag.rescue001", "subdir/001/tiny_problems.info.json", + "subdir/001/tiny_problems.dag.dagman.log", "subdir/001/tiny_problems.dag.metrics", "subdir/001/tiny_problems.dag.nodes.log", "subdir/001/tiny_problems.node_status", @@ -952,11 +955,11 @@ def testRelativeSubdir(self): self.assertEqual( set(result_submit), { - "./tiny_problems.dag.dagman.log", "./tiny_problems.dag.dagman.out", "./tiny_problems.dag", "./tiny_problems.dag.rescue001", "reldir/001/tiny_problems.info.json", + "reldir/001/tiny_problems.dag.dagman.log", "reldir/001/tiny_problems.dag.metrics", "reldir/001/tiny_problems.dag.nodes.log", "reldir/001/tiny_problems.node_status", @@ -977,30 +980,30 @@ def testSubdags(self): set(result_submit), { "./group_failed_1.dag", - "./group_failed_1.dag.dagman.log", "./group_failed_1.dag.dagman.out", "./group_failed_1.dag.rescue001", "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_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", "subdags/wms_group_order1_val1b/group_order1_val1b.dag.rescue001", "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", "001/group_failed_1.dag.nodes.log", + "001/group_failed_1.dag.dagman.log", "001/group_failed_1.info.json", "001/group_failed_1.node_status", + "001/subdags/wms_group_order1_val1a/group_order1_val1a.dag.dagman.log", "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.dagman.log", "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.dagman.log", "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", @@ -1125,7 +1128,9 @@ def testFileMissing(self): def testSuccess(self): with temporaryDirectory() as tmp_dir: copy2(f"{TESTDIR}/data/tiny_success/tiny_success.info.json", tmp_dir) - results = lssthtc.read_dag_info(tmp_dir) + filename, results = lssthtc.read_dag_info(tmp_dir) + self.assertIn("info.json", str(filename)) + self.assertTrue(pathlib.Path(filename).is_file()) truth = { "test02": { @@ -1156,7 +1161,7 @@ def testPermissionError(self): with unittest.mock.patch("lsst.ctrl.bps.htcondor.lssthtc.open") as mocked_open: mocked_open.side_effect = PermissionError with self.assertLogs("lsst.ctrl.bps.htcondor", level="DEBUG") as cm: - results = lssthtc.read_dag_info(tmp_dir) + _, results = lssthtc.read_dag_info(tmp_dir) self.assertIn("Retrieving DAGMan job information failed:", cm.output[-1]) self.assertEqual({}, results) @@ -1384,6 +1389,74 @@ def testWriteWithoutDagConfig(self): subfile_actual = f.readlines() self.assertEqual(subfile_actual, self.subfile_expected) + def testWriteLazySubdag(self): + self.maxDiff = None + dag, truth_files = make_lazy_dag("test1", True) + with temporaryDirectory() as tmp_dir: + dag.write(tmp_dir, "", "") + + # self.assertIn("submit_path", dag.graph) + # self.assertEqual(dag.graph["submit_path"], tmp_dir) + # self.assertIn("dag_filename", dag.graph) + # self.assertEqual(dag.graph["dag_filename"], "test1.dag") + all_files = [] + for root, _, files in pathlib.Path(tmp_dir).walk(): + relroot = pathlib.Path(root).relative_to(tmp_dir) + all_files.extend([str(relroot / f) for f in files]) + self.assertEqual(sorted(all_files), sorted(truth_files)) + + +class WriteDagInfoTestCase(unittest.TestCase): + """Test for write_dag_info function.""" + + def setUp(self): + self.run = "u_testuser_DM-53494_20260220T001651Z" + self.data = { + "mycomputer": { + "24390.0": { + "ClusterId": 24390, + "GlobalJobId": "mycomputer#24390.0#1771546612", + "bps_run": self.run, + "bps_isjob": "True", + "bps_payload": "DM-53494", + "bps_project": "dev", + "bps_runsite": "site1", + "bps_campaign": "ci_rc2", + "bps_operator": "testuser", + "bps_run_quanta": "", + "bps_job_summary": "buildQuantumGraph:1;preparePayloadWorkflow:1;dummyJob:1", + "bps_wms_service": "lsst.ctrl.bps.htcondor.htcondor_service.HTCondorService", + "bps_wms_workflow": "lsst.ctrl.bps.htcondor.htcondor_workflow.HTCondorWorkflow", + "bps_wms_config_path": "dagman.conf", + } + } + } + + def testWriteWithFilename(self): + with temporaryDirectory() as tmp_dir: + with chdir(tmp_dir): + path = pathlib.Path(tmp_dir) / "test.info.json" + filename = lssthtc.write_dag_info(self.data, path) + self.assertTrue(path.is_file(), f"File not found at {path}") + self.assertEqual(filename, path) + + read_filename, read_data = lssthtc.read_dag_info(tmp_dir) + self.assertEqual(read_filename, path) + self.assertEqual(read_data, self.data) + + def testWriteWithoutFilename(self): + with temporaryDirectory() as tmp_dir: + with chdir(tmp_dir): + filename = lssthtc.write_dag_info(self.data) + self.assertTrue(filename.is_file(), f"File not found at {filename}") + + path = pathlib.Path(tmp_dir) / filename + self.assertTrue(path.is_file(), f"File not found at {path}") + + read_filename, read_data = lssthtc.read_dag_info(tmp_dir) + self.assertEqual(read_filename, path) + self.assertEqual(read_data, self.data) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_prepare_utils.py b/tests/test_prepare_utils.py index 5649bc6..d6c2d0e 100644 --- a/tests/test_prepare_utils.py +++ b/tests/test_prepare_utils.py @@ -30,6 +30,9 @@ import logging import os import unittest +from copy import deepcopy + +from networkx import is_isomorphic from lsst.ctrl.bps import ( BPS_DEFAULTS, @@ -40,8 +43,15 @@ GenericWorkflowFile, GenericWorkflowJob, ) -from lsst.ctrl.bps.htcondor import prepare_utils -from lsst.ctrl.bps.tests.gw_test_utils import make_3_label_workflow, make_3_label_workflow_groups_sort +from lsst.ctrl.bps.htcondor import lssthtc, prepare_utils +from lsst.ctrl.bps.htcondor.htcondor_config import HTC_DEFAULTS_URI +from lsst.ctrl.bps.tests.gw_test_utils import ( + make_3_label_workflow, + make_3_label_workflow_groups_sort, + make_lazy_workflow, +) +from lsst.daf.butler import Config +from lsst.utils.tests import temporaryDirectory logger = logging.getLogger("lsst.ctrl.bps.htcondor") @@ -621,5 +631,138 @@ def testUnrecognized(self): self.assertRegex(cm_log.output[0], "Unrecognized WMS placeholder: notThere") +class UpdateJobSummaryTestCase(unittest.TestCase): + """Test _update_job_summary function.""" + + def setUp(self): + self.run = "u_testuser_DM-53494_20260220T001651Z" + self.filename = f"{self.run}_ctrl.info.json" + self.data = { + "mycomputer": { + "24390.0": { + "ClusterId": 24390, + "GlobalJobId": "mycomputer#24390.0#1771546612", + "bps_run": f"{self.run}_ctrl", + "bps_isjob": "True", + "bps_payload": "DM-53494", + "bps_project": "dev", + "bps_runsite": "site1", + "bps_campaign": "ci_rc2", + "bps_operator": "testuser", + "bps_run_quanta": "", + "bps_job_summary": "buildQuantumGraph:1;preparePayloadWorkflow:1;dummyJob:1", + "bps_wms_service": "lsst.ctrl.bps.htcondor.htcondor_service.HTCondorService", + "bps_wms_workflow": "lsst.ctrl.bps.htcondor.htcondor_workflow.HTCondorWorkflow", + "bps_wms_config_path": "dagman.conf", + } + } + } + self.mapping = f"{self.run}:preparePayloadWorkflow" + self.add_summary = "pipetaskInit:1;isr:6;finalJob:1" + + def testLazyMapping(self): + dag_info = deepcopy(self.data) + dag_info["mycomputer"]["24390.0"]["bps_lazy_mapping"] = self.mapping + with temporaryDirectory() as tmp_dir: + lssthtc.write_dag_info(dag_info, f"{tmp_dir}/{self.filename}") + + prepare_utils._update_job_summary(self.run, self.add_summary, str(tmp_dir)) + + _, results = lssthtc.read_dag_info(str(tmp_dir)) + self.assertEqual( + results["mycomputer"]["24390.0"]["bps_job_summary"], + f"buildQuantumGraph:1;preparePayloadWorkflow:1;{self.add_summary};dummyJob:1", + ) + + def testNoLazyMapping(self): + # No bps_lazy_mapping at all + with temporaryDirectory() as tmp_dir: + lssthtc.write_dag_info(self.data, f"{tmp_dir}/{self.filename}") + + prepare_utils._update_job_summary(self.run, self.add_summary, str(tmp_dir)) + + _, results = lssthtc.read_dag_info(str(tmp_dir)) + self.assertEqual( + results["mycomputer"]["24390.0"]["bps_job_summary"], + f"{self.data['mycomputer']['24390.0']['bps_job_summary']};{self.add_summary}", + ) + + def testNoEntryLazyMapping(self): + # bps_lazy_mapping exists, but doesn't include this job + dag_info = deepcopy(self.data) + dag_info["mycomputer"]["24390.0"]["bps_lazy_mapping"] = "other:preparePayloadWorkflow" + with temporaryDirectory() as tmp_dir: + lssthtc.write_dag_info(dag_info, f"{tmp_dir}/{self.filename}") + + prepare_utils._update_job_summary(self.run, self.add_summary, str(tmp_dir)) + + _, results = lssthtc.read_dag_info(str(tmp_dir)) + self.assertEqual( + results["mycomputer"]["24390.0"]["bps_job_summary"], + f"{self.data['mycomputer']['24390.0']['bps_job_summary']};{self.add_summary}", + ) + + +class ReplaceCmdVarsTestCase(unittest.TestCase): + """Test _replace_cmd_vars function.""" + + def testKeyError(self): + gwjob = GenericWorkflowJob("job1", "label1") + with self.assertLogs(level="DEBUG") as cm_log: + with self.assertRaisesRegex(KeyError, ".*notthere.*"): + _ = prepare_utils._replace_cmd_vars("{notthere}", gwjob) + self.assertRegex(cm_log.output[0], ".*replacement for 'notthere' not provided.*") + + +class GenericWorkflowToHTCondorDAG(unittest.TestCase): + """Test _generic_workflow_to_htcondor_dag function.""" + + def testRegularWorkflow(self): + timestamp = "20260130T211713Z" + generic_workflow = make_3_label_workflow("test1", True) + config = BpsConfig( + { + "bpsUseShared": True, + "overwriteJobFiles": False, + "profile": {"requirements": "dummy_val == 3"}, + "attrs": {}, + "nodeset": "set1", # this shouldn't be used with auto-provisioning + "provisionResources": True, + "provisioning": {"provisioningMaxWallTime": 1200}, + "bps_defined": {"timestamp": timestamp}, + }, + defaults=Config(HTC_DEFAULTS_URI), + ) + + results = prepare_utils._generic_workflow_to_htcondor_dag(config, generic_workflow, "/mock_dir") + self.assertTrue(generic_workflow.run_attrs.items() <= results.graph["attr"].items()) + self.assertIsNotNone(results.graph["final_job"]) + self.assertTrue(is_isomorphic(results, generic_workflow)) + + def testLazyWorkflow(self): + timestamp = "20260130T211713Z" + generic_workflow = make_lazy_workflow("test1", True) + config = BpsConfig( + { + "bpsUseShared": True, + "overwriteJobFiles": False, + "profile": {"requirements": "dummy_val == 3"}, + "attrs": {}, + "nodeset": "set1", # this shouldn't be used with auto-provisioning + "provisionResources": True, + "provisioning": {"provisioningMaxWallTime": 1200}, + "bps_defined": {"timestamp": timestamp}, + }, + defaults=Config(HTC_DEFAULTS_URI), + ) + + results = prepare_utils._generic_workflow_to_htcondor_dag(config, generic_workflow, "/mock_dir") + self.assertTrue(generic_workflow.run_attrs.items() <= results.graph["attr"].items()) + self.assertIsNotNone(results.graph["final_job"]) + # Can't test isomorphic because HTCDag will have additional job for + # the lazy dagman job. + self.assertTrue(generic_workflow.nodes <= results.nodes) + + if __name__ == "__main__": unittest.main() From 586953dc98f0fc4acc293983f2246e9df703b7fd Mon Sep 17 00:00:00 2001 From: Michelle Gower Date: Fri, 20 Feb 2026 11:49:43 -0600 Subject: [PATCH 02/10] DO NOT MERGE. --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 3b61f9d..c5d69e8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ htcondor >= 8.8, == 24.0.* -lsst-ctrl-bps @ git+https://github.com/lsst/ctrl_bps.git@main +lsst-ctrl-bps @ git+https://github.com/lsst/ctrl_bps.git@tickets/DM-53494 lsst-daf-butler @ git+https://github.com/lsst/daf_butler@main lsst-pipe-base @ git+https://github.com/lsst/pipe_base@main lsst-utils @ git+https://github.com/lsst/utils@main From 557816bfac161ef1e3a56d2c834032cec306f077 Mon Sep 17 00:00:00 2001 From: Michelle Gower Date: Tue, 24 Mar 2026 20:04:59 -0500 Subject: [PATCH 03/10] Fix unportable shell command. --- python/lsst/ctrl/bps/htcondor/check_group_status.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/lsst/ctrl/bps/htcondor/check_group_status.sh b/python/lsst/ctrl/bps/htcondor/check_group_status.sh index 776175f..afd2ccc 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=$(/usr/bin/cat "$1") echo "Read status ${group_status} from group status file ($1)" >&2 exit ${group_status} From 1be07100a3582742b867b8c7e8da6415ced23f9f Mon Sep 17 00:00:00 2001 From: Michelle Gower Date: Tue, 24 Mar 2026 21:17:48 -0500 Subject: [PATCH 04/10] Make writing dot file optional. --- doc/lsst.ctrl.bps.htcondor/userguide.rst | 6 ++++++ python/lsst/ctrl/bps/htcondor/lssthtc.py | 5 ++++- python/lsst/ctrl/bps/htcondor/prepare_utils.py | 5 ++++- tests/test_lssthtc.py | 10 ++++------ tests/test_prepare_utils.py | 3 +++ 5 files changed, 21 insertions(+), 8 deletions(-) diff --git a/doc/lsst.ctrl.bps.htcondor/userguide.rst b/doc/lsst.ctrl.bps.htcondor/userguide.rst index 5dbfc70..764460d 100644 --- a/doc/lsst.ctrl.bps.htcondor/userguide.rst +++ b/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: diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index 3426352..ef00319 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/python/lsst/ctrl/bps/htcondor/lssthtc.py @@ -1229,7 +1229,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 diff --git a/python/lsst/ctrl/bps/htcondor/prepare_utils.py b/python/lsst/ctrl/bps/htcondor/prepare_utils.py index 44a13ae..90a201a 100644 --- a/python/lsst/ctrl/bps/htcondor/prepare_utils.py +++ b/python/lsst/ctrl/bps/htcondor/prepare_utils.py @@ -918,8 +918,11 @@ def _generic_workflow_to_htcondor_dag( "bps_job_summary": create_count_summary(generic_workflow.job_counts), } ) - _, tmp_template = config.search("subDirTemplate", opt={"replaceVars": False, "default": ""}) + + _, save_htc_dot = config.search("saveHTCdot", opt={"default": False}) + dag.graph["write_dot"] = save_htc_dot + if isinstance(tmp_template, str): subdir_template = defaultdict(lambda: tmp_template) else: diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index cd34cfa..7193fb4 100644 --- a/tests/test_lssthtc.py +++ b/tests/test_lssthtc.py @@ -1349,7 +1349,6 @@ def testWriteWithDagConfig(self): dagfile_expected = [ f"CONFIG {wms_config_filename}\n", f'JOB {job.name} "{job.subfile}"\n', - f"DOT {self.dag.name}.dot\n", f"NODE_STATUS_FILE {self.dag.name}.node_status\n", f'SET_JOB_ATTR bps_wms_config_path= "{wms_config_filename}"\n', ] @@ -1372,7 +1371,6 @@ def testWriteWithoutDagConfig(self): job = self.dag.nodes["test_job"]["data"] dagfile_expected = [ f'JOB {job.name} "{job.subfile}"\n', - f"DOT {self.dag.name}.dot\n", f"NODE_STATUS_FILE {self.dag.name}.node_status\n", ] @@ -1392,13 +1390,13 @@ def testWriteWithoutDagConfig(self): def testWriteLazySubdag(self): self.maxDiff = None dag, truth_files = make_lazy_dag("test1", True) + dag.graph["write_dot"] = True with temporaryDirectory() as tmp_dir: dag.write(tmp_dir, "", "") + with open(os.path.join(tmp_dir, dag.graph["dag_filename"]), encoding="utf-8") as f: + dagfile_actual = f.readlines() + self.assertIn("DOT test1.dot\n", dagfile_actual) - # self.assertIn("submit_path", dag.graph) - # self.assertEqual(dag.graph["submit_path"], tmp_dir) - # self.assertIn("dag_filename", dag.graph) - # self.assertEqual(dag.graph["dag_filename"], "test1.dag") all_files = [] for root, _, files in pathlib.Path(tmp_dir).walk(): relroot = pathlib.Path(root).relative_to(tmp_dir) diff --git a/tests/test_prepare_utils.py b/tests/test_prepare_utils.py index d6c2d0e..6938640 100644 --- a/tests/test_prepare_utils.py +++ b/tests/test_prepare_utils.py @@ -730,6 +730,7 @@ def testRegularWorkflow(self): "provisionResources": True, "provisioning": {"provisioningMaxWallTime": 1200}, "bps_defined": {"timestamp": timestamp}, + "saveHTCdot": True, }, defaults=Config(HTC_DEFAULTS_URI), ) @@ -738,6 +739,7 @@ def testRegularWorkflow(self): self.assertTrue(generic_workflow.run_attrs.items() <= results.graph["attr"].items()) self.assertIsNotNone(results.graph["final_job"]) self.assertTrue(is_isomorphic(results, generic_workflow)) + self.assertTrue(results.graph["write_dot"]) def testLazyWorkflow(self): timestamp = "20260130T211713Z" @@ -762,6 +764,7 @@ def testLazyWorkflow(self): # Can't test isomorphic because HTCDag will have additional job for # the lazy dagman job. self.assertTrue(generic_workflow.nodes <= results.nodes) + self.assertFalse(results.graph["write_dot"]) if __name__ == "__main__": From 21d1d07fb745f3ce557b2628cb762ba8b922e51b Mon Sep 17 00:00:00 2001 From: Michelle Gower Date: Thu, 16 Apr 2026 13:52:00 -0500 Subject: [PATCH 05/10] Update handling of environment and file transfers. --- .../ctrl/bps/htcondor/dagman_configurator.py | 1 + .../bps/htcondor/etc/htcondor_defaults.yaml | 40 +++ .../ctrl/bps/htcondor/htcondor_service.py | 24 ++ python/lsst/ctrl/bps/htcondor/lssthtc.py | 2 +- .../lsst/ctrl/bps/htcondor/prepare_utils.py | 239 ++++++++++++++---- tests/test_lssthtc.py | 4 +- tests/test_prepare_utils.py | 136 ++++++---- 7 files changed, 344 insertions(+), 102 deletions(-) diff --git a/python/lsst/ctrl/bps/htcondor/dagman_configurator.py b/python/lsst/ctrl/bps/htcondor/dagman_configurator.py index a9bb129..ae6210d 100644 --- a/python/lsst/ctrl/bps/htcondor/dagman_configurator.py +++ b/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, ""), } ) diff --git a/python/lsst/ctrl/bps/htcondor/etc/htcondor_defaults.yaml b/python/lsst/ctrl/bps/htcondor/etc/htcondor_defaults.yaml index 702135b..53a45a9 100644 --- a/python/lsst/ctrl/bps/htcondor/etc/htcondor_defaults.yaml +++ b/python/lsst/ctrl/bps/htcondor/etc/htcondor_defaults.yaml @@ -56,3 +56,43 @@ wmsConfig: # A boolean flag controlling whether DAGMan should generate submit files for # nested DAGs automatically. DAGMAN_GENERATE_SUBDAG_SUBMITS: true + +bpsMakeCommand: true +#payloadCommand: > +# {setupEnv} +# logDir={jobLogDir}; +# pwd; +# {gwjobCommand} & pJob=$!; +# prmon -i 5 +# -f ${logDir}/memory_monitor_output.txt +# -j ${logDir}/memory_monitor_summary.json +# -p $pJob & mJob=$!; +# wait $pJob; +# ret=$?; +# wait $mJob; +# {jobCleanup} +# exit $ret; + +payloadCommand: > + {setupEnv} + logDir={jobLogDir}; + pwd; + {gwjobCommand}; + ret=$?; + {jobCleanup} + exit $ret; + +# Job environment setup +softwarePath: "/cvmfs/sw.lsst.eu/almalinux-x86_64/lsst_distrib/{LSST_VERSION}" + +customEnvSetup: "" +setupEnv: > + unset PYTHONPATH; + source {softwarePath}/loadLSST.bash; + setup lsst_distrib; + {customEnvSetup} + +# Other job variables +jobInitDir: "`pwd`" +jobLogDir: "{jobInitDir}" +jobCleanup: "" diff --git a/python/lsst/ctrl/bps/htcondor/htcondor_service.py b/python/lsst/ctrl/bps/htcondor/htcondor_service.py index 46a3b23..9d49b13 100644 --- a/python/lsst/ctrl/bps/htcondor/htcondor_service.py +++ b/python/lsst/ctrl/bps/htcondor/htcondor_service.py @@ -583,3 +583,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") diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index ef00319..b59fe13 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/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", diff --git a/python/lsst/ctrl/bps/htcondor/prepare_utils.py b/python/lsst/ctrl/bps/htcondor/prepare_utils.py index 90a201a..b5e1a58 100644 --- a/python/lsst/ctrl/bps/htcondor/prepare_utils.py +++ b/python/lsst/ctrl/bps/htcondor/prepare_utils.py @@ -31,13 +31,13 @@ import os import re from collections import defaultdict -from copy import deepcopy from pathlib import Path from typing import Any, cast from lsst.ctrl.bps import ( BpsConfig, GenericWorkflow, + GenericWorkflowFile, GenericWorkflowGroup, GenericWorkflowJob, GenericWorkflowNodeType, @@ -48,7 +48,6 @@ from .lssthtc import ( HTCDag, HTCJob, - _update_dicts, condor_status, htc_escape, read_dag_info, @@ -83,6 +82,7 @@ def _create_job(subdir_template, cached_values, generic_workflow, gwjob, out_pre htc_job : `lsst.ctrl.bps.wms.htcondor.HTCJob` The HTCondor job equivalent to the given generic job. """ + _LOG.debug("_create_job: cached_values = %s", cached_values) htc_job = HTCJob(gwjob.name, label=gwjob.label) curvals = defaultdict(str) @@ -101,7 +101,6 @@ def _create_job(subdir_template, cached_values, generic_workflow, gwjob, out_pre "when_to_transfer_output": "ON_EXIT_OR_EVICT", "transfer_output_files": '""', # Set to empty string to disable "transfer_executable": "False", - "getenv": "True", # Exceeding memory sometimes triggers SIGBUS or SIGSEGV error. Tell # htcondor to put on hold any jobs which exited by a signal. If # executed in a bash script, like finalJob, the signals will become @@ -271,20 +270,27 @@ def _translate_job_cmds(cached_vals, generic_workflow, gwjob): jobcmds["concurrency_limit"] = gwjob.concurrency_limit # Handle command line - if gwjob.executable.transfer_executable: - jobcmds["transfer_executable"] = "True" - jobcmds["executable"] = gwjob.executable.src_uri - else: - jobcmds["executable"] = _fix_env_var_syntax(gwjob.executable.src_uri) + new_job_cmds = _translate_command_line(cached_vals, generic_workflow, gwjob) + jobcmds.update(new_job_cmds) + + # Add extra "pass-thru" job commands + if gwjob.profile: + for key, val in gwjob.profile.items(): + jobcmds[key] = val + for key, val in cached_vals["profile"].items(): + jobcmds[key] = val + + return jobcmds + - 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) - arguments = _fix_env_var_syntax(arguments) - jobcmds["arguments"] = arguments +def _translate_command_line(cached_vals, generic_workflow, gwjob): + jobcmds = {} + # Yaml environment section gets put in gwjob.environment + # which should be set by job. If user wants to set + # environment vars after setting up stack, the export + # commands should be added to the special customEnvSetup + # yaml value. if gwjob.environment: env_str = "" for name, value in gwjob.environment.items(): @@ -300,12 +306,69 @@ def _translate_job_cmds(cached_vals, generic_workflow, gwjob): # Process above added one trailing space jobcmds["environment"] = env_str.rstrip() - # Add extra "pass-thru" job commands - if gwjob.profile: - for key, val in gwjob.profile.items(): - jobcmds[key] = val - for key, val in cached_vals["profile"].items(): - jobcmds[key] = val + if cached_vals.get("bpsMakeCommand", True): + # Way to have fallback to previous behavior as well as + # a way forward to centralize logic in bps. + + jobcmds["getenv"] = "True" + + if gwjob.executable.transfer_executable: + jobcmds["transfer_executable"] = "True" + jobcmds["executable"] = gwjob.executable.src_uri + 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) + arguments = _fix_env_var_syntax(arguments) + jobcmds["arguments"] = arguments + + else: + # Instead of making a bash script, run /bin/bash -c + # HTCondor v25 has a job command called shell that can replace the + # /bin/bash when we get to that version. + + # 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) + arguments = _fix_env_var_syntax(arguments) + + if gwjob.executable.transfer_executable: + # Since replacing executable need to add this executable to the + # file transfer list. + gwfile = GenericWorkflowFile( + name=gwjob.executable.name, src_uri=gwjob.executable.src_uri, wms_transfer=True + ) + 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}" + else: + exec_name = _fix_env_var_syntax_shell(gwjob.executable.src_uri) + gwjobCommand = f"{exec_name} {arguments}" + + payloadCommand = cached_vals["payloadCommand"] + _LOG.debug("%s payloadCommand pre-format: %s", gwjob.label, payloadCommand) + payloadCommand = re.sub("{gwjobCommand}", gwjobCommand, payloadCommand) + # Remove newlines + payloadCommand = re.sub("\n", "", payloadCommand) + + _LOG.debug("%s payloadCommand post-format: %s", gwjob.label, payloadCommand) + + # jobcmds["arguments"] = htc_escape(f"-c '{payloadCommand}'") + jobcmds["arguments"] = f"-c '{payloadCommand}'" + + jobcmds["executable"] = "/bin/bash" + # Don't need to transfer /bin/bash + jobcmds["transfer_executable"] = "False" return jobcmds @@ -357,6 +420,25 @@ def _fix_env_var_syntax(oldstr): return newstr +def _fix_env_var_syntax_shell(oldstr): + """Change ENV place holders to shell var syntax. + + Parameters + ---------- + oldstr : `str` + String in which environment variable syntax is to be fixed. + + Returns + ------- + newstr : `str` + Given string with environment variable syntax fixed. + """ + newstr = oldstr + for key in re.findall(r"]+)>", oldstr): + newstr = newstr.replace(rf"", f"${{{key}}}") + return newstr + + def _replace_file_vars(use_shared, arguments, workflow, gwjob): """Replace file placeholders in command line arguments with correct physical file names. @@ -617,13 +699,17 @@ def _create_periodic_release_expr( # Never auto release a job held by user. user_expr = f"HoldReasonCode =!= 1 && {additional_expr}" + # Automatically release job if held because output file not found + # (e.g., job failed so didn't produce output file). + transfer_expr = "HoldReasonCode =?= 12" + expr = f"{is_held} && {is_retry_allowed}" if user_expr and mem_expr: - expr += f" && ({mem_expr} || {user_expr})" + expr += f" && ({transfer_expr} || {mem_expr} || {user_expr})" elif user_expr: - expr += f" && {user_expr}" + expr += f" && ({transfer_expr} || {user_expr})" elif mem_expr: - expr += f" && {mem_expr}" + expr += f" && ({transfer_expr} || {mem_expr})" return expr @@ -754,20 +840,9 @@ def _gather_site_values(config, compute_site): _LOG.debug("No execute machine in the pool matches %s", patt) if limit: config[".bps_defined.memory_limit"] = limit - - _, site_values["bpsUseShared"] = config.search("bpsUseShared", opt={"default": False}) site_values["memoryLimit"] = limit - found, value = config.search("accountingGroup", opt=search_opts) - if found: - site_values["accountingGroup"] = value - found, value = config.search("accountingUser", opt=search_opts) - if found: - site_values["accountingUser"] = value - - found, nodeset = config.search("nodeset", opt=search_opts) - if found: - site_values["nodeset"] = nodeset + _, site_values["bpsUseShared"] = config.search("bpsUseShared", opt={"default": False}) searchobj = config[f".site.{compute_site}.profile.condor"] if searchobj: @@ -781,11 +856,17 @@ def _gather_site_values(config, compute_site): _, val = config.search(key, opt=search_opts) site_values["profile"][key] = val + searchobj = config[f".site.{compute_site}"] + if searchobj: + for key, value in searchobj.items(): + if key not in site_values and key not in ["attrs", "profile"]: + site_values[key] = value + _LOG.debug("site_values = %s", site_values) return site_values -def _gather_label_values(config: BpsConfig, label: str) -> dict[str, Any]: +def _gather_label_values(config: BpsConfig, label: str, compute_site: str) -> dict[str, Any]: """Gather values specific to given job label. Parameters @@ -795,35 +876,89 @@ def _gather_label_values(config: BpsConfig, label: str) -> dict[str, Any]: information. label : `str` GenericWorkflowJob label. + compute_site : `str` + Compute site. Returns ------- values : `dict` [`str`, `~typing.Any`] Values specific to the given job label. """ + _LOG.debug("_gather_label_values: label = %s, compute_site = %s", label, compute_site) values: dict[str, Any] = {"attrs": {}, "profile": {}} search_opts = {} + if compute_site: + search_opts["curvals"] = {"curr_site": compute_site} + profile_key = "" if label == "finalJob" and "finalJob" in config: search_opts["searchobj"] = config["finalJob"] profile_key = ".finalJob.profile.condor" elif label in config["cluster"]: - search_opts["curvals"] = {"curr_cluster": label} + search_opts.setdefault("curvals", {})["curr_cluster"] = label profile_key = f".cluster.{label}.profile.condor" elif label in config["pipetask"]: - search_opts["curvals"] = {"curr_pipetask": label} + search_opts.setdefault("curvals", {})["curr_pipetask"] = label profile_key = f".pipetask.{label}.profile.condor" + if profile_key and profile_key not in config: + profile_key = f".site.{compute_site}.profile.condor" + if profile_key not in config: + profile_key = ".profile.condor" + + # Determine the hard limit for the memory requirement. + found, limit = config.search("memoryLimit", opt=search_opts) + if not found: + search_opts["default"] = DEFAULT_HTC_EXEC_PATT + _, patt = config.search("executeMachinesPattern", opt=search_opts) + del search_opts["default"] + + # To reduce the amount of data, ignore dynamic slots (if any) as, + # by definition, they cannot have more memory than + # the partitionable slot they are the part of. + constraint = f'SlotType != "Dynamic" && regexp("{patt}", Machine)' + pool_info = condor_status(constraint=constraint) + try: + limit = max(int(info["TotalSlotMemory"]) for info in pool_info.values()) + except ValueError: + _LOG.debug("No execute machine in the pool matches %s", patt) + + if limit: + config[".bps_defined.memory_limit"] = limit + values["memoryLimit"] = limit + + values["bpsUseShared"] = False + found, value = config.search("bpsUseShared", opt=search_opts) + if found: + values["bpsUseShared"] = value + found, value = config.search("releaseExpr", opt=search_opts) if found: values["releaseExpr"] = value + values["overwriteJobFiles"] = True found, value = config.search("overwriteJobFiles", opt=search_opts) if found: values["overwriteJobFiles"] = value - else: - values["overwriteJobFiles"] = True + + found, value = config.search("releaseExpr", opt=search_opts) + if found: + values["releaseExpr"] = value + + found, value = config.search("bpsMakeCommand", opt=search_opts) + values["bpsMakeCommand"] = value if found else True + if found and not value: + search_opts["skipNames"] = {"gwjobCommand"} + _LOG.debug("_gather_label_values: search_opts = %s", search_opts) + found, value = config.search("payloadCommand", opt=search_opts) + if found: + values["payloadCommand"] = value + _LOG.debug("payloadCommand = %s", value) + + found, value = config.search("nodeset", opt=search_opts) + if found: + values["nodeset"] = value if profile_key and profile_key in config: for subkey, val in config[profile_key].items(): @@ -832,6 +967,14 @@ def _gather_label_values(config: BpsConfig, label: str) -> dict[str, Any]: else: values["profile"][subkey] = val + if compute_site and compute_site in config["site"]: + site_obj = config[f".site.{compute_site}"] + for key, value in site_obj.items(): + if key not in values and key not in ["attrs", "profile"]: + values[key] = value + + _LOG.debug("_gather_label_values: label = %s, values = %s", label, values) + return values @@ -932,7 +1075,6 @@ def _generic_workflow_to_htcondor_dag( lazy_groups = [] # Create all DAG jobs - site_values = {} # Cache compute site specific values to reduce config lookups. cached_values = {} # Cache label-specific values to reduce config lookups. # Note: Can't use get_job_by_label because those only include payload jobs. for job_name in generic_workflow: @@ -942,14 +1084,8 @@ def _generic_workflow_to_htcondor_dag( GenericWorkflowNodeType.LAZY_GROUP, ]: gwjob = cast(GenericWorkflowJob, gwjob) - if gwjob.compute_site not in site_values: - site_values[gwjob.compute_site] = _gather_site_values(config, gwjob.compute_site) if gwjob.label not in cached_values: - cached_values[gwjob.label] = deepcopy(site_values[gwjob.compute_site]) - _update_dicts( - cached_values[gwjob.label], - _gather_label_values(config, gwjob.label), - ) + cached_values[gwjob.label] = _gather_label_values(config, gwjob.label, gwjob.compute_site) _LOG.debug("cached: %s= %s", gwjob.label, cached_values[gwjob.label]) htc_job = _create_job( subdir_template[gwjob.label], @@ -1024,11 +1160,8 @@ def _generic_workflow_to_htcondor_dag( # If final job exists in generic workflow, create DAG final job final = generic_workflow.get_final() if final and isinstance(final, GenericWorkflowJob): - if final.compute_site and final.compute_site not in site_values: - site_values[final.compute_site] = _gather_site_values(config, final.compute_site) if final.label not in cached_values: - cached_values[final.label] = deepcopy(site_values[final.compute_site]) - _update_dicts(cached_values[final.label], _gather_label_values(config, final.label)) + cached_values[final.label] = _gather_label_values(config, final.label, final.compute_site) final_htjob = _create_job( subdir_template[final.label], cached_values[final.label], diff --git a/tests/test_lssthtc.py b/tests/test_lssthtc.py index 7193fb4..afa594f 100644 --- a/tests/test_lssthtc.py +++ b/tests/test_lssthtc.py @@ -1187,7 +1187,7 @@ def testSuccess(self): } expected = [ "executable=$(CTRL_MPEXEC_DIR)/bin/pipetask\n", - "arguments=-a -b 2 -c\n", + 'arguments="-a -b 2 -c"\n', "request_memory=2000\n", "environment=\"one=1 two=\"2\" three='spacey 'quoted' value'\"\n", f"output={job_name}.$(Cluster).out\n", @@ -1328,7 +1328,7 @@ def setUp(self): self.subfile_expected = [ "executable=/usr/bin/echo\n", - "arguments=foo\n", + 'arguments="foo"\n', "output=test_job.$(Cluster).out\n", "error=test_job.$(Cluster).out\n", "log=test_job.$(Cluster).log\n", diff --git a/tests/test_prepare_utils.py b/tests/test_prepare_utils.py index 6938640..ff134f3 100644 --- a/tests/test_prepare_utils.py +++ b/tests/test_prepare_utils.py @@ -128,9 +128,9 @@ def testPeriodicRelease(self): htc_commands = prepare_utils._translate_job_cmds(self.cached_vals, None, gwjob) release = ( "JobStatus == 5 && NumJobStarts <= JobMaxRetries && " - "(HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " + "(HoldReasonCode =?= 12 || (HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " "HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34) && " - "min({int(2048 * pow(2, NumJobStarts - 1)), 32768}) < 32768" + "min({int(2048 * pow(2, NumJobStarts - 1)), 32768}) < 32768)" ) self.assertEqual(htc_commands["periodic_release"], release) @@ -230,32 +230,6 @@ def testNotSpecified(self): results = prepare_utils._gather_site_values(config, compute_site) self.assertEqual(results["memoryLimit"], BPS_DEFAULTS["memoryLimit"]) - def testGlobalNodeset(self): - config = BpsConfig( - {"nodeset": "global_node_set_{campaign}", "campaign": "DRP"}, - search_order=BPS_SEARCH_ORDER, - defaults=BPS_DEFAULTS, - wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", - ) - compute_site = "fr" - results = prepare_utils._gather_site_values(config, compute_site) - self.assertEqual(results["nodeset"], "global_node_set_DRP") - - def testSiteNodeset(self): - config = BpsConfig( - { - "nodeset": "global_node_set_{campaign}", - "campaign": "DRP", - "site": {"fr": {"nodeset": "fr_node_set_{campaign}"}}, - }, - search_order=BPS_SEARCH_ORDER, - defaults=BPS_DEFAULTS, - wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", - ) - compute_site = "fr" - results = prepare_utils._gather_site_values(config, compute_site) - self.assertEqual(results["nodeset"], "fr_node_set_DRP") - def testAttrsProfile(self): test_values = { "bpsNodeset": "DEVSET", @@ -286,7 +260,6 @@ class GatherLabelValuesTestCase(unittest.TestCase): def testClusterLabel(self): # Test cluster value overrides pipetask. - label = "label1" config = BpsConfig( { "cluster": { @@ -297,12 +270,13 @@ def testClusterLabel(self): } }, "pipetask": {"label1": {"releaseExpr": "pipetask_val"}}, + "site": {"site1": {}}, }, search_order=BPS_SEARCH_ORDER, defaults=BPS_DEFAULTS, wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", ) - results = prepare_utils._gather_label_values(config, label) + results = prepare_utils._gather_label_values(config, "label1", "site1") self.assertEqual( results, { @@ -310,6 +284,9 @@ def testClusterLabel(self): "profile": {"prof_val1": 3}, "releaseExpr": "cluster_val", "overwriteJobFiles": False, + "bpsMakeCommand": True, + "bpsUseShared": True, + "memoryLimit": 491520, }, ) @@ -323,58 +300,115 @@ def testPipetaskLabel(self): "overwriteJobFiles": False, "profile": {"condor": {"prof_val1": 3}}, } - } + }, + "site": {"site1": {}}, }, search_order=BPS_SEARCH_ORDER, defaults=BPS_DEFAULTS, wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", ) - results = prepare_utils._gather_label_values(config, label) + results = prepare_utils._gather_label_values(config, label, "site1") self.assertEqual( results, { "attrs": {}, + "bpsMakeCommand": True, + "bpsUseShared": True, + "memoryLimit": 491520, + "overwriteJobFiles": False, "profile": {"prof_val1": 3}, "releaseExpr": "pipetask_val", - "overwriteJobFiles": False, }, ) def testNoSection(self): label = "notThere" config = BpsConfig( - {}, + {"site": {"site1": {}}}, search_order=BPS_SEARCH_ORDER, defaults=BPS_DEFAULTS, wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", ) - results = prepare_utils._gather_label_values(config, label) - self.assertEqual(results, {"attrs": {}, "profile": {}, "overwriteJobFiles": True}) + results = prepare_utils._gather_label_values(config, label, "site1") + self.assertEqual( + results, + { + "attrs": {}, + "profile": {}, + "overwriteJobFiles": True, + "bpsMakeCommand": True, + "bpsUseShared": True, + "memoryLimit": 491520, + }, + ) def testNoOverwriteSpecified(self): label = "notthere" config = BpsConfig( - {}, + {"site": {"site1": {}}}, search_order=BPS_SEARCH_ORDER, defaults={}, wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", ) - results = prepare_utils._gather_label_values(config, label) - self.assertEqual(results, {"attrs": {}, "profile": {}, "overwriteJobFiles": True}) + results = prepare_utils._gather_label_values(config, label, "nosite") + self.assertEqual( + results, + { + "attrs": {}, + "profile": {}, + "overwriteJobFiles": True, + "bpsMakeCommand": True, + "bpsUseShared": False, + }, + ) def testFinalJob(self): label = "finalJob" config = BpsConfig( - {"finalJob": {"profile": {"condor": {"prof_val2": 6, "+attr_val1": 5}}}}, + {"site": {"site1": {}}, "finalJob": {"profile": {"condor": {"prof_val2": 6, "+attr_val1": 5}}}}, search_order=BPS_SEARCH_ORDER, defaults=BPS_DEFAULTS, wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", ) - results = prepare_utils._gather_label_values(config, label) + results = prepare_utils._gather_label_values(config, label, "site1") self.assertEqual( - results, {"attrs": {"attr_val1": 5}, "profile": {"prof_val2": 6}, "overwriteJobFiles": False} + results, + { + "attrs": {"attr_val1": 5}, + "profile": {"prof_val2": 6}, + "overwriteJobFiles": False, + "bpsMakeCommand": True, + "bpsUseShared": True, + "memoryLimit": 491520, + }, ) + def testGlobalNodeset(self): + config = BpsConfig( + {"nodeset": "global_node_set_{campaign}", "campaign": "DRP"}, + search_order=BPS_SEARCH_ORDER, + defaults=BPS_DEFAULTS, + wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", + ) + compute_site = "fr" + results = prepare_utils._gather_label_values(config, "label1", compute_site) + self.assertEqual(results["nodeset"], "global_node_set_DRP") + + def testSiteNodeset(self): + config = BpsConfig( + { + "nodeset": "global_node_set_{campaign}", + "campaign": "DRP", + "site": {"fr": {"nodeset": "fr_node_set_{campaign}"}}, + }, + search_order=BPS_SEARCH_ORDER, + defaults=BPS_DEFAULTS, + wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService", + ) + compute_site = "fr" + results = prepare_utils._gather_label_values(config, "label1", compute_site) + self.assertEqual(results["nodeset"], "fr_node_set_DRP") + class CreateCheckJobTestCase(unittest.TestCase): """Test _create_check_job function.""" @@ -391,6 +425,9 @@ def testSuccess(self): class CreatePeriodicReleaseExprTestCase(unittest.TestCase): """Test _create_periodic_release_expr function.""" + def setUp(self): + self.maxDiff = None + def testNoReleaseExpr(self): results = prepare_utils._create_periodic_release_expr(2048, 1, 32768, "") self.assertEqual(results, "") @@ -404,20 +441,27 @@ def testJustMemoryReleaseExpr(self): results = prepare_utils._create_periodic_release_expr(2048, 2, 32768, "") truth = ( "JobStatus == 5 && NumJobStarts <= JobMaxRetries && " + "(HoldReasonCode =?= 12 || " "(HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " "HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34) && " - "min({int(2048 * pow(2, NumJobStarts - 1)), 32768}) < 32768" + "min({int(2048 * pow(2, NumJobStarts - 1)), 32768}) < 32768)" ) self.assertEqual(results, truth) def testJustUserReleaseExpr(self): results = prepare_utils._create_periodic_release_expr(2048, 1, 32768, "True") - truth = "JobStatus == 5 && NumJobStarts <= JobMaxRetries && HoldReasonCode =!= 1 && True" + truth = ( + "JobStatus == 5 && NumJobStarts <= JobMaxRetries && " + "(HoldReasonCode =?= 12 || HoldReasonCode =!= 1 && True)" + ) self.assertEqual(results, truth) def testJustUserReleaseExprMultiplierNone(self): results = prepare_utils._create_periodic_release_expr(2048, None, 32768, "True") - truth = "JobStatus == 5 && NumJobStarts <= JobMaxRetries && HoldReasonCode =!= 1 && True" + truth = ( + "JobStatus == 5 && NumJobStarts <= JobMaxRetries && " + "(HoldReasonCode =?= 12 || HoldReasonCode =!= 1 && True)" + ) self.assertEqual(results, truth) def testMemoryAndUserReleaseExpr(self): @@ -425,7 +469,7 @@ def testMemoryAndUserReleaseExpr(self): results = prepare_utils._create_periodic_release_expr(2048, 2, 32768, "True") truth = ( "JobStatus == 5 && NumJobStarts <= JobMaxRetries && " - "((HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " + "(HoldReasonCode =?= 12 || (HoldReasonCode =?= 34 && HoldReasonSubCode =?= 0 || " "HoldReasonCode =?= 3 && HoldReasonSubCode =?= 34) && " "min({int(2048 * pow(2, NumJobStarts - 1)), 32768}) < 32768 || " "HoldReasonCode =!= 1 && True)" From 6e184584bd464c2dc64f0509aa5921f9269784ef Mon Sep 17 00:00:00 2001 From: Michelle Gower Date: Tue, 26 May 2026 08:31:50 -0700 Subject: [PATCH 06/10] Add timings to remote job output. --- .../bps/htcondor/etc/htcondor_defaults.yaml | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/python/lsst/ctrl/bps/htcondor/etc/htcondor_defaults.yaml b/python/lsst/ctrl/bps/htcondor/etc/htcondor_defaults.yaml index 53a45a9..e894672 100644 --- a/python/lsst/ctrl/bps/htcondor/etc/htcondor_defaults.yaml +++ b/python/lsst/ctrl/bps/htcondor/etc/htcondor_defaults.yaml @@ -74,23 +74,40 @@ bpsMakeCommand: true # exit $ret; payloadCommand: > + echo -n BPS Beginning of job commands:\ ; + /usr/bin/date +%Y-%m-%dT%H:%M:%S.%N%:z; {setupEnv} 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/{LSST_VERSION}" 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; # Other job variables jobInitDir: "`pwd`" From e650a39f125fa4253cd3c92dd6516df745d61a84 Mon Sep 17 00:00:00 2001 From: Michelle Gower Date: Mon, 1 Jun 2026 17:08:17 -0700 Subject: [PATCH 07/10] Fix reporting finalJob failure when only DAG failure. --- python/lsst/ctrl/bps/htcondor/lssthtc.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index b59fe13..5e2cdb3 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/python/lsst/ctrl/bps/htcondor/lssthtc.py @@ -2046,6 +2046,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"] From 24097569a5fd398b196359b82eab200b2fa25d98 Mon Sep 17 00:00:00 2001 From: Michelle Gower Date: Fri, 5 Jun 2026 12:52:07 -0700 Subject: [PATCH 08/10] Fix searchopt for build and prepare jobs. --- python/lsst/ctrl/bps/htcondor/prepare_utils.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/python/lsst/ctrl/bps/htcondor/prepare_utils.py b/python/lsst/ctrl/bps/htcondor/prepare_utils.py index b5e1a58..8280686 100644 --- a/python/lsst/ctrl/bps/htcondor/prepare_utils.py +++ b/python/lsst/ctrl/bps/htcondor/prepare_utils.py @@ -895,6 +895,12 @@ def _gather_label_values(config: BpsConfig, label: str, compute_site: str) -> di if label == "finalJob" and "finalJob" in config: search_opts["searchobj"] = config["finalJob"] profile_key = ".finalJob.profile.condor" + elif label == "buildQuantumGraph" and "buildQuantumGraph" in config: + search_opts["searchobj"] = config["buildQuantumGraph"] + profile_key = ".buildQuantumGraph.profile.condor" + elif label == "preparePayloadWorkflow" and "preparePayloadWorkflow" in config: + search_opts["searchobj"] = config["preparePayloadWorkflow"] + profile_key = ".preparePayloadWorkflow.profile.condor" elif label in config["cluster"]: search_opts.setdefault("curvals", {})["curr_cluster"] = label profile_key = f".cluster.{label}.profile.condor" From af801d253c4ebc9b7ee2171deaf0492d2ccf4319 Mon Sep 17 00:00:00 2001 From: Michelle Gower Date: Mon, 29 Jun 2026 19:30:49 -0500 Subject: [PATCH 09/10] Add run site to report. --- python/lsst/ctrl/bps/htcondor/report_utils.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/python/lsst/ctrl/bps/htcondor/report_utils.py b/python/lsst/ctrl/bps/htcondor/report_utils.py index 1327e3d..b97de2c 100644 --- a/python/lsst/ctrl/bps/htcondor/report_utils.py +++ b/python/lsst/ctrl/bps/htcondor/report_utils.py @@ -403,6 +403,7 @@ def _create_detailed_report_from_jobs( path=dag_ad["Iwd"], label=dag_ad.get("bps_job_label", "MISS"), run=dag_ad.get("bps_run", "MISS"), + site=dag_ad.get("bps_runsite", ""), project=dag_ad.get("bps_project", "MISS"), campaign=dag_ad.get("bps_campaign", "MISS"), payload=dag_ad.get("bps_payload", "MISS"), @@ -575,6 +576,7 @@ def _summary_report(user, hist, pass_thru, schedds=None): project=job.get("bps_project", "MISS"), campaign=job.get("bps_campaign", "MISS"), payload=job.get("bps_payload", "MISS"), + site=job.get("bps_runsite", ""), operator=_get_owner(job), run_summary=_get_run_summary(job), state=_htc_status_to_wms_state(job), From e40b7c5ce8b46e53315d2792753b6040111d85a5 Mon Sep 17 00:00:00 2001 From: Michelle Gower Date: Mon, 29 Jun 2026 19:40:29 -0500 Subject: [PATCH 10/10] Fix restart problems. Fix restart problems with backup subdir number and fix the bps_job_summary truncation in info.json. --- .../lsst/ctrl/bps/htcondor/htcondor_service.py | 16 +++++++++++++++- python/lsst/ctrl/bps/htcondor/lssthtc.py | 5 ++++- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/python/lsst/ctrl/bps/htcondor/htcondor_service.py b/python/lsst/ctrl/bps/htcondor/htcondor_service.py index 9d49b13..056bac4 100644 --- a/python/lsst/ctrl/bps/htcondor/htcondor_service.py +++ b/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, ) @@ -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_file = htc_backup_files(wms_path, subdir="backups") if (wms_path / "subdags").exists(): @@ -280,7 +289,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(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(schedd_dag_info, dag_info_filename) run_id = f"{dag_ad['ClusterId']}.{dag_ad['ProcId']}" run_name = dag_ad["bps_run"] else: diff --git a/python/lsst/ctrl/bps/htcondor/lssthtc.py b/python/lsst/ctrl/bps/htcondor/lssthtc.py index 5e2cdb3..9dcb6d4 100644 --- a/python/lsst/ctrl/bps/htcondor/lssthtc.py +++ b/python/lsst/ctrl/bps/htcondor/lssthtc.py @@ -380,7 +380,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 len(rescue_dags) == 0: + 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.