DM-31887: Execution Butler bug fixes plus making it the default behavior.

config/no_execution_butler.yaml

@@ -0,0 +1,18 @@
+# THIS FILE MAY BE DEPRECATED IN FUTURE VERSIONS
+#
+# This turns off the default bps use of execution butler
+# Execution butler is used to limit database connections.
+# Turning off execution butler should only be done after
+# reporting problems with it turned on.
+
+# Turn off Execution Butler
+executionButler:
+  whenCreate: "NEVER"
+  whenMerge: "NEVER"
+
+# The pipetaskInit command is different when not using execution butler
+pipetask:
+  pipetaskInit:
+    # Notes: cannot have --extend-run, should have both -o and --output-run
+    # because this is where the chaining is currently done
+    runQuantumCommand: "${CTRL_MPEXEC_DIR}/bin/pipetask {initPreCmdOpts} run -b {butlerConfig} -i {inCollection} -o {output} --output-run {outCollection} --qgraph {qgraphFile} --qgraph-id {qgraphId} --qgraph-node-id {qgraphNodeId} --clobber-outputs --init-only --register-dataset-types {extraInitOptions}"

doc/changes/DM-31887.bugfix.rst

@@ -0,0 +1,3 @@
+* Fix variable substitution in merge job commands.
+* Fix bug where final job doesn't appear in report.
+* Fix bug in HTCondor plugin for reporting final job status when --id <path>.

doc/changes/DM-31887.feature.rst

@@ -0,0 +1 @@
+Make using an execution butler the default.

python/lsst/ctrl/bps/bps_config.py

@@ -257,6 +257,12 @@ def search(self, key, opt=None):
 
         _LOG.debug("curvals = %s", curvals)
 
+        # There's a problem with the searchobj being a BpsConfig
+        # and its handling of __getitem__.  Until that part of
+        # BpsConfig is rewritten, force the searchobj to a Config.
+        if "searchobj" in opt:
+            opt["searchobj"] = Config(opt["searchobj"])
+
         if key in curvals:
             _LOG.debug("found %s in curvals", key)
             found = True

python/lsst/ctrl/bps/etc/bps_defaults.yaml

@@ -34,9 +34,9 @@ payload:
 
 pipetask:
   pipetaskInit:
-    # Notes: cannot have --extend-run, should have both -o and --output-run
-    # because this is where the chaining is currently done
-    runQuantumCommand: "${CTRL_MPEXEC_DIR}/bin/pipetask {initPreCmdOpts} run -b {butlerConfig} -i {inCollection} -o {output} --output-run {outCollection} --qgraph {qgraphFile} --qgraph-id {qgraphId} --qgraph-node-id {qgraphNodeId} --clobber-outputs --init-only --register-dataset-types {extraInitOptions}"
+    # Notes:  Declaring and chaining now happen within execution butler steps.
+    # This command no longer needs -o and must have # --extend-run.
+    runQuantumCommand: "${CTRL_MPEXEC_DIR}/bin/pipetask {initPreCmdOpts} run -b {butlerConfig} -i {inCollection} -o {output} --output-run {outCollection} --qgraph {qgraphFile} --qgraph-id {qgraphId} --qgraph-node-id {qgraphNodeId} --clobber-outputs --init-only --extend-run {extraInitOptions}"
   #OPT myTask:
   #OPT   requestCpus:
   #OPT   requestMemory:
@@ -74,7 +74,16 @@ saveClusteredQgraph: False
 useLazyCommands: True
 bpsUseShared: False
 
-# Execution Butler not yet the default
 executionButler:
-  whenCreate: "NEVER"
-  whenMerge: "NEVER"
+  whenCreate: "SUBMIT"
+  #OPT executionButlerDir: "/my/exec/butler/dir"  # User-provided or defaults to executionButlerTemplate
+  createCommand: "${CTRL_MPEXEC_DIR}/bin/pipetask qgraph -b {butlerConfig} -i {inCollection} -o {output} --output-run {outCollection} --save-execution-butler {executionButlerDir} -g {qgraphFile}"
+  whenMerge: "ALWAYS"
+  implementation: JOB  # Added for future flexibility, e.g., if prefer workflow instead of shell script.
+  concurrencyLimit: db_limit
+
+  mergePreCmdOpts: "{defaultPreCmdOpts}"
+  command1: "${DAF_BUTLER_DIR}/bin/butler {mergePreCmdOpts} transfer-datasets {executionButlerDir} {butlerConfig} --collections {outCollection}"
+  command2: "${DAF_BUTLER_DIR}/bin/butler {mergePreCmdOpts} collection-chain {butlerConfig} {output} --flatten --mode=extend {inCollection}"
+  command3: "${DAF_BUTLER_DIR}/bin/butler {mergePreCmdOpts} collection-chain {butlerConfig} {output} --flatten --mode=prepend {outCollection}"
+

python/lsst/ctrl/bps/generic_workflow.py

@@ -343,6 +343,16 @@ def job_counts(self):
         for job_name in self:
             gwjob = self.get_job(job_name)
             jcounts[gwjob.label] += 1
+
+        # Final is separate
+        final = self.get_final()
+        if isinstance(final, GenericWorkflow):
+            for job_name in final:
+                gwjob = final.get_job(job_name)
+                jcounts[gwjob.label] += 1
+        else:
+            jcounts[final.label] += 1
+
         return jcounts
 
     def __iter__(self):

python/lsst/ctrl/bps/transform.py

@@ -762,7 +762,8 @@ def _create_final_command(config, prefix):
     arguments : `str`
         Command line needed to call the final script.
     """
-    search_opt = {'replaceVars': False, 'replaceEnvVars': False, 'expandEnvVars': False}
+    search_opt = {'replaceVars': False, 'replaceEnvVars': False, 'expandEnvVars': False,
+                  'searchobj': config['executionButler']}
 
     script_file = os.path.join(prefix, "final_job.bash")
     with open(script_file, "w") as fh:

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

@@ -990,7 +990,7 @@ def _get_run_summary(job):
         Number of jobs per PipelineTask label in approximate pipeline order.
         Format: <label>:<count>[;<label>:<count>]+
     """
-    summary = job.get("bps_run_summary", None)
+    summary = job.get("bps_job_summary", None)
     if not summary:
         summary, _ = summary_from_dag(job["Iwd"])
         if not summary:
@@ -1186,7 +1186,11 @@ def _htc_node_status_to_wms_state(job):
     elif status == NodeStatus.DONE:
         wms_state = WmsStates.SUCCEEDED
     elif status == NodeStatus.ERROR:
-        wms_state = WmsStates.FAILED
+        # Use job exist instead of post script exit
+        if "DAGMAN error 0" in job["StatusDetails"]:
+            wms_state = WmsStates.SUCCEEDED
+        else:
+            wms_state = WmsStates.FAILED
 
     return wms_state
 

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

@@ -640,6 +640,9 @@ def add_final_job(self, job):
         job : `HTCJob`
             HTCJob to add to the HTCDag as a FINAL job.
         """
+        # Add dag level attributes to each job
+        job.add_job_attrs(self.graph["attr"])
+
         self.graph['final_job'] = job
 
     def del_job(self, job_name):
@@ -859,6 +862,13 @@ def summary_from_dag(dir_name):
                             counts[label] += 1
                         else:
                             _LOG.warning("Parse DAG: unmatched job line: %s", line)
+                elif line.startswith("FINAL"):
+                    m = re.match(r"FINAL ([^\s]+) jobs/([^/]+)/", line)
+                    if m:
+                        label = m.group(2)
+                        job_name_to_pipetask[m.group(1)] = label
+                        counts[label] += 1
+
     except (OSError, PermissionError, StopIteration):
         pass