WS-677: Multi-stage workflow research & development

Changes

• Updated workflow_service to handle DAG submission
• Added new three-stage DAG-ified null_dag workflow and templates

Testing

• Unit tests are written and pass
• Submitting normal jobs still works
• Submitting DAGs works

Completes WS-677

shared/workspaces/workspaces/workflow/services/workflow_service.py

@@ -447,10 +447,10 @@ class WorkflowService(WorkflowServiceIF):
         logger.info("executing on folder %s", folder)
 
         # some file in here should end in .dag; that file is our dagman input
-        dag_file = list(folder.glob("*.dag"))[0]
-        if dag_file:
-            logger.info("dagman file %s exists.", dag_file)
-            return self._submit_dag(folder, dag_file)
+        dag_files = list(folder.glob("*.dag"))
+        if dag_files:
+            logger.info("dagman file %s exists.", dag_files[0])
+            return self._submit_dag(folder, dag_files[0])
         else:
             job_file = list(folder.glob("*.condor"))[0]
             logger.info("condor file %s exists.", job_file)
@@ -496,7 +496,7 @@ class WorkflowService(WorkflowServiceIF):
         # submit
         logger.info("submitting DAG to condor...")
         subprocess.run(
-            ["condor_submit_dag", str(dag_file)],
+            ["condor_submit_dag", "-f", str(dag_file)],
             cwd=str(folder.absolute()),
             preexec_fn=self._switch_to_submituser,
         )