10576: Add ability to run pipeline templates containing CWL from command line.

[arvados.git] / sdk / cwl / arvados_cwl / arvcontainer.py

diff --git a/sdk/cwl/arvados_cwl/arvcontainer.py b/sdk/cwl/arvados_cwl/arvcontainer.py
index cea2b1091d400be70014cc512029db8d2b872955..dbbd83da2efd7d8f609dcdbfd90d10f414e6c1ac 100644 (file)
--- a/sdk/cwl/arvados_cwl/arvcontainer.py
+++ b/sdk/cwl/arvados_cwl/arvcontainer.py
@@ -37,7 +37,8 @@ class ArvadosContainer(object):
             "output_path": self.outdir,
             "cwd": self.outdir,
             "priority": 1,
-            "state": "Committed"
+            "state": "Committed",
+            "properties": {}
         }
         runtime_constraints = {}
         mounts = {
@@ -113,12 +114,20 @@ class ArvadosContainer(object):
         container_request["use_existing"] = kwargs.get("enable_reuse", True)
         container_request["scheduling_parameters"] = scheduling_parameters
 
+        if kwargs.get("runnerjob", "").startswith("arvwf:"):
+            wfuuid = kwargs["runnerjob"][6:kwargs["runnerjob"].index("#")]
+            wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
+            if container_request["name"] == "main":
+                container_request["name"] = wfrecord["name"]
+            container_request["properties"]["template_uuid"] = wfuuid
+
         try:
             response = self.arvrunner.api.container_requests().create(
                 body=container_request
             ).execute(num_retries=self.arvrunner.num_retries)
 
-            self.arvrunner.processes[response["uuid"]] = self
+            self.uuid = response["uuid"]
+            self.arvrunner.processes[self.uuid] = self
 
             logger.info("Container request %s (%s) state is %s", self.name, response["uuid"], response["state"])
 
@@ -173,15 +182,6 @@ class RunnerContainer(Runner):
 
         workflowmapper = super(RunnerContainer, self).arvados_job_spec(dry_run=dry_run, pull_image=pull_image, **kwargs)
 
-        with arvados.collection.Collection(api_client=self.arvrunner.api,
-                                           keep_client=self.arvrunner.keep_client,
-                                           num_retries=self.arvrunner.num_retries) as jobobj:
-            with jobobj.open("cwl.input.json", "w") as f:
-                json.dump(self.job_order, f, sort_keys=True, indent=4)
-            jobobj.save_new(owner_uuid=self.arvrunner.project_uuid)
-
-        jobpath = "/var/lib/cwl/job/cwl.input.json"
-
         container_req = {
             "owner_uuid": self.arvrunner.project_uuid,
             "name": self.name,
@@ -191,9 +191,9 @@ class RunnerContainer(Runner):
             "state": "Committed",
             "container_image": arvados_jobs_image(self.arvrunner),
             "mounts": {
-                jobpath: {
-                    "kind": "collection",
-                    "portable_data_hash": "%s/cwl.input.json" % jobobj.portable_data_hash()
+                "/var/lib/cwl/cwl.input.json": {
+                    "kind": "json",
+                    "content": self.job_order
                 },
                 "stdout": {
                     "kind": "file",
@@ -208,7 +208,8 @@ class RunnerContainer(Runner):
                 "vcpus": 1,
                 "ram": 1024*1024 * self.submit_runner_ram,
                 "API": True
-            }
+            },
+            "properties": {}
         }
 
         workflowcollection = workflowmapper.mapper(self.tool.tool["id"])[1]
@@ -222,14 +223,16 @@ class RunnerContainer(Runner):
                 }
         elif workflowcollection.startswith("arvwf:"):
             workflowpath = "/var/lib/cwl/workflow.json#main"
-            fetcher = CollectionFetcher({}, None,
-                                        api_client=self.arvrunner.api,
-                                        keep_client=self.arvrunner.keep_client)
-            wfobj = yaml.safe_load(fetcher.fetch_text(workflowcollection))
+            wfuuid = workflowcollection[6:workflowcollection.index("#")]
+            wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
+            wfobj = yaml.safe_load(wfrecord["definition"])
+            if container_req["name"].startswith("arvwf:"):
+                container_req["name"] = wfrecord["name"]
             container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
                 "kind": "json",
                 "json": wfobj
             }
+            container_req["properties"]["template_uuid"] = wfuuid
 
         command = ["arvados-cwl-runner", "--local", "--api=containers"]
         if self.output_name:
@@ -243,7 +246,7 @@ class RunnerContainer(Runner):
         else:
             command.append("--disable-reuse")
 
-        command.extend([workflowpath, jobpath])
+        command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
 
         container_req["command"] = command
 
@@ -260,7 +263,7 @@ class RunnerContainer(Runner):
         ).execute(num_retries=self.arvrunner.num_retries)
 
         self.uuid = response["uuid"]
-        self.arvrunner.processes[response["uuid"]] = self
+        self.arvrunner.processes[self.uuid] = self
 
         logger.info("Submitted container %s", response["uuid"])