Merge branch 'tickets/DM-31944'

lsst · Sep 29, 2021 · aaf56d1 · aaf56d1
2 parents 679a4d4 + 188ff0d
commit aaf56d1
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 7 deletions.
diff --git a/doc/changes/DM-31944.bugfix.rst b/doc/changes/DM-31944.bugfix.rst
@@ -0,0 +1 @@
+Fix single concurrency limit splitting.
diff --git a/doc/changes/DM-31944.misc.rst b/doc/changes/DM-31944.misc.rst
@@ -0,0 +1 @@
+Persist bps DAG attributes across manual restarts.
diff --git a/python/lsst/ctrl/bps/generic_workflow.py b/python/lsst/ctrl/bps/generic_workflow.py
@@ -206,7 +206,7 @@ class GenericWorkflowJob:
     throttling jobs within a single workflow).
     """
 
-    concurrency_limit: Optional[list]
+    concurrency_limit: Optional[str]
     """Names of concurrency limits that the WMS plugin can appropriately
     translate to limit the number of this job across all running workflows.
     """
@@ -268,7 +268,7 @@ def __init__(self, name: str):
         self.abort_return_value = None
         self.priority = None
         self.category = None
-        self.concurrency_limit = []
+        self.concurrency_limit = None
         self.queue = None
         self.pre_cmdline = None
         self.post_cmdline = None

diff --git a/python/lsst/ctrl/bps/wms/htcondor/htcondor_service.py b/python/lsst/ctrl/bps/wms/htcondor/htcondor_service.py
@@ -506,7 +506,7 @@ def _translate_job_cmds(config, generic_workflow, gwjob):
     # May need to move to special site-specific implementation if sites use
     # other mechanisms.
     if gwjob.concurrency_limit:
-        jobcmds["concurrency_limit"] = ",".join(gwjob.concurrency_limit)
+        jobcmds["concurrency_limit"] = gwjob.concurrency_limit
 
     # Handle command line
     if gwjob.executable.transfer_executable:

diff --git a/python/lsst/ctrl/bps/wms/htcondor/lssthtc.py b/python/lsst/ctrl/bps/wms/htcondor/lssthtc.py
@@ -359,10 +359,6 @@ def htc_submit_dag(htc_dag, submit_options=None):
     else:
         sub = _htc_submit_dag_old(htc_dag.graph["dag_filename"], submit_options)
 
-    # add attributes to dag submission
-    for key, value in htc_dag.graph["attr"].items():
-        sub[f"+{key}"] = f'"{htc_escape(value)}"'
-
     # submit DAG to HTCondor's schedd
     schedd = htcondor.Schedd()
     with schedd.transaction() as txn:
@@ -684,6 +680,11 @@ def write(self, submit_path, job_subdir=""):
                 print(f"PARENT {edge[0]} CHILD {edge[1]}", file=fh)
             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
+            for key, value in self.graph["attr"].items():
+                print(f'SET_JOB_ATTR {key}= "{htc_escape(value)}"', file=fh)
+
             if self.graph["final_job"]:
                 job = self.graph["final_job"]
                 job.write_submit_file(submit_path, job_subdir)