create a build task for remote submission

lsst · Sep 19, 2023 · 14aa1ff · 14aa1ff
1 parent d68f275
commit 14aa1ff
Show file tree

Hide file tree

Showing 2 changed files with 48 additions and 12 deletions.
diff --git a/python/lsst/ctrl/bps/drivers.py b/python/lsst/ctrl/bps/drivers.py
@@ -396,6 +396,30 @@ def submit_driver(config_file, **kwargs):
         "not accurately reflect actual memory usage by the bps process."
     )
 
+    remote_build = {}
+    config = BpsConfig(config_file, BPS_SEARCH_ORDER)
+    _, remote_build = config.search("remoteBuild", opt={"default": {}})
+    if remote_build:
+        if config["wmsServiceClass"] == "lsst.ctrl.bps.panda.PanDAService":
+            if not remote_build.search("enabled", opt={"default": False})[1]:
+                remote_build = {}
+                _LOG.info("The workflow is sumitted to the local Data Facility.")
+            else:
+                _LOG.info("Remote submission is enabled. The workflow is sumitted to a remote Data Facility.")
+                _LOG.info("Initializing execution environment")
+                with time_this(
+                    log=_LOG,
+                    level=logging.INFO,
+                    prefix=None,
+                    msg="Initializing execution environment completed",
+                    mem_usage=True,
+                    mem_unit=DEFAULT_MEM_UNIT,
+                    mem_fmt=DEFAULT_MEM_FMT,
+                ):
+                    config = _init_submission_driver(config_file, **kwargs)
+    else:
+        _LOG.info("The workflow is sumitted to the local Data Facility.")
+
     _LOG.info("Starting submission process")
     with time_this(
         log=_LOG,
@@ -406,7 +430,8 @@ def submit_driver(config_file, **kwargs):
         mem_unit=DEFAULT_MEM_UNIT,
         mem_fmt=DEFAULT_MEM_FMT,
     ):
-        wms_workflow_config, wms_workflow = prepare_driver(config_file, **kwargs)
+        if not remote_build:
+            wms_workflow_config, wms_workflow = prepare_driver(config_file, **kwargs)
 
         _LOG.info("Starting submit stage")
         with time_this(
@@ -418,7 +443,10 @@ def submit_driver(config_file, **kwargs):
             mem_unit=DEFAULT_MEM_UNIT,
             mem_fmt=DEFAULT_MEM_FMT,
         ):
-            submit(wms_workflow_config, wms_workflow)
+            if remote_build:
+                wms_workflow = submit(config, None, remote_build=remote_build, config_file=config_file)
+            else:
+                submit(wms_workflow_config, wms_workflow)
             _LOG.info("Run '%s' submitted for execution with id '%s'", wms_workflow.name, wms_workflow.run_id)
     if _LOG.isEnabledFor(logging.INFO):
         _LOG.info(

diff --git a/python/lsst/ctrl/bps/submit.py b/python/lsst/ctrl/bps/submit.py
@@ -38,7 +38,7 @@
 
 
 @timeMethod(logger=_LOG, logLevel=VERBOSE)
-def submit(config, wms_workflow, wms_service=None):
+def submit(config, wms_workflow, wms_service=None, remote_build=None, config_file=None):
     """Convert generic workflow to a workflow for a particular WMS.
 
     Parameters
@@ -60,18 +60,26 @@ def submit(config, wms_workflow, wms_service=None):
         wms_service_class = doImport(config["wmsServiceClass"])
         wms_service = wms_service_class(config)
 
-    _, when_create = config.search(".executionButler.whenCreate")
-    if when_create.upper() == "SUBMIT":
-        _, execution_butler_dir = config.search(".bps_defined.executionButlerDir")
-        _LOG.info("Creating execution butler in '%s'", execution_butler_dir)
-        with time_this(log=_LOG, level=logging.INFO, prefix=None, msg="Completed creating execution butler"):
-            _create_execution_butler(
-                config, config["runQgraphFile"], execution_butler_dir, config["submitPath"]
-            )
+    if not remote_build:
+        _, when_create = config.search(".executionButler.whenCreate")
+        if when_create.upper() == "SUBMIT":
+            _, execution_butler_dir = config.search(".bps_defined.executionButlerDir")
+            _LOG.info("Creating execution butler in '%s'", execution_butler_dir)
+            with time_this(
+                log=_LOG, level=logging.INFO, prefix=None, msg="Completed creating execution butler"
+            ):
+                _create_execution_butler(
+                    config, config["runQgraphFile"], execution_butler_dir, config["submitPath"]
+                )
 
     _LOG.info("Submitting run to a workflow management system for execution")
     with time_this(
         log=_LOG, level=logging.INFO, prefix=None, msg="Completed submitting to a workflow management system"
     ):
-        workflow = wms_service.submit(wms_workflow)
+        if remote_build:
+            workflow = wms_service.submit(
+                wms_workflow, config=config, remote_build=remote_build, config_file=config_file
+            )
+        else:
+            workflow = wms_service.submit(wms_workflow)
     return workflow