Support workflow_descriptor in arvados_wes and wes-client.

Author: Peter Amstutz

wes_client/__init__.py

@@ -99,28 +99,40 @@ def fixpaths(d):
                 else:
                     d["location"] = d["path"]
                 del d["path"]
+            loc = d.get("location", "")
             if d.get("class") == "Directory":
-                loc = d.get("location", "")
                 if loc.startswith("http:") or loc.startswith("https:"):
                     logging.error("Directory inputs not supported with http references")
                     exit(33)
+            if not (loc.startswith("http:") or loc.startswith("https:")
+                    or args.job_order.startswith("http:") or args.job_order.startswith("https:")):
+                logging.error("Upload local files not supported, must use http: or https: references.")
+                exit(33)
 
     visit(input, fixpaths)
 
     workflow_url = args.workflow_url
     if not workflow_url.startswith("/") and ":" not in workflow_url:
-        workflow_url = os.path.abspath(workflow_url)
+        workflow_url = "file://" + os.path.abspath(workflow_url)
 
     if args.quiet:
         logging.basicConfig(level=logging.WARNING)
     else:
         logging.basicConfig(level=logging.INFO)
 
-    r = client.WorkflowExecutionService.RunWorkflow(body={
-        "workflow_url": workflow_url,
+    body = {
         "workflow_params": input,
         "workflow_type": "CWL",
-        "workflow_type_version": "v1.0"}).result()
+        "workflow_type_version": "v1.0"
+    }
+
+    if workflow_url.startswith("file://"):
+        with open(workflow_url[7:], "r") as f:
+            body["workflow_descriptor"] = f.read()
+    else:
+        body["workflow_url"] = workflow_url
+
+    r = client.WorkflowExecutionService.RunWorkflow(body=body).result()
 
     if args.wait:
         logging.info("Workflow id is %s", r["workflow_id"])

wes_service/arvados_wes.py

@@ -87,7 +87,7 @@ def ListWorkflows(self):
             "next_page_token": ""
         }
 
-    def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params, env):
+    def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params, env, workflow_descriptor_file):
         api = arvados.api_from_config(version="v1", apiconfig={
             "ARVADOS_API_HOST": env["ARVADOS_API_HOST"],
             "ARVADOS_API_TOKEN": env['ARVADOS_API_TOKEN'],
@@ -98,10 +98,12 @@ def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params, env):
             with tempfile.NamedTemporaryFile() as inputtemp:
                 json.dump(workflow_params, inputtemp)
                 inputtemp.flush()
+                # TODO: run submission process in a container to prevent
+                # a-c-r submission processes from seeing each other.
                 proc = subprocess.Popen(["arvados-cwl-runner", "--submit-request-uuid="+cr_uuid, # NOQA
-                                                       "--submit", "--no-wait", "--api=containers",     # NOQA
-                                                       workflow_url, inputtemp.name], env=env,
-                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE)  # NOQA
+                                         "--submit", "--no-wait", "--api=containers",     # NOQA
+                                         workflow_url, inputtemp.name], env=env,
+                                        stdout=subprocess.PIPE, stderr=subprocess.PIPE)  # NOQA
                 (stdoutdata, stderrdata) = proc.communicate()
                 if proc.returncode != 0:
                     api.container_requests().update(uuid=cr_uuid, body={"priority": 0,
@@ -111,6 +113,9 @@ def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params, env):
         except subprocess.CalledProcessError as e:
             api.container_requests().update(uuid=cr_uuid, body={"priority": 0,
                                                                 "properties": {"arvados-cwl-runner-log": str(e)}}).execute()
+        finally:
+            if workflow_descriptor_file is not None:
+                workflow_descriptor_file.close()
 
     @catch_exceptions
     def RunWorkflow(self, body):
@@ -136,7 +141,19 @@ def RunWorkflow(self, body):
                                                     "output_path": "n/a",
                                                     "priority": 500}}).execute()
 
-        threading.Thread(target=self.invoke_cwl_runner, args=(cr["uuid"], body.get("workflow_url"), body["workflow_params"], env)).start()
+        workflow_url = body.get("workflow_url")
+        workflow_descriptor_file = None
+        if body.get("workflow_descriptor"):
+            workflow_descriptor_file = tempfile.NamedTemporaryFile()
+            workflow_descriptor_file.write(body.get('workflow_descriptor'))
+            workflow_descriptor_file.flush()
+            workflow_url = workflow_descriptor_file.name
+
+        threading.Thread(target=self.invoke_cwl_runner, args=(cr["uuid"],
+                                                              workflow_url,
+                                                              body["workflow_params"],
+                                                              env,
+                                                              workflow_descriptor_file)).start()
 
         return {"workflow_id": cr["uuid"]}