8442: Adding --submit support with --crunch2. General refactoring into more/smaller...

[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 0cb885f59b4588759bb5490f9d739650b77941a0..be1140469ade22369f5ae51fac6da24a4826ddcd 100644 (file)
--- a/sdk/cwl/arvados_cwl/arvcontainer.py
+++ b/sdk/cwl/arvados_cwl/arvcontainer.py
@@ -1,9 +1,15 @@
 import logging
+import json
+import os
+
+from cwltool.errors import WorkflowException
+from cwltool.process import get_feature, adjustFiles, UnsupportedRequirement, shortname
+
 import arvados.collection
-from cwltool.process import get_feature, adjustFiles
+
 from .arvdocker import arv_docker_get_image
 from . import done
-from cwltool.errors import WorkflowException
+from .runner import Runner
 
 logger = logging.getLogger('arvados.cwl-runner')
 
@@ -42,7 +48,9 @@ class ArvadosContainer(object):
             }
 
         if self.generatefiles:
-            vwd = arvados.collection.Collection()
+            raise UnsupportedRequirement("Generate files not supported")
+
+            vwd = arvados.collection.Collection(api_client=self.arvrunner.api_client)
             container_request["task.vwd"] = {}
             for t in self.generatefiles:
                 if isinstance(self.generatefiles[t], dict):
@@ -60,13 +68,12 @@ class ArvadosContainer(object):
         if self.environment:
             container_request["environment"].update(self.environment)
 
-        # TODO, not supported by crunchv2 yet
-        #if self.stdin:
-        #    container_request["task.stdin"] = self.pathmapper.mapper(self.stdin)[1]
+        if self.stdin:
+            raise UnsupportedRequirement("Stdin redirection currently not suppported")
 
         if self.stdout:
             mounts["stdout"] = {"kind": "file",
-                                "path": self.stdout}
+                                "path": "/var/spool/cwl/%s" % (self.stdout)}
 
         (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
         if not docker_req:
@@ -122,3 +129,78 @@ class ArvadosContainer(object):
             self.output_callback(outputs, processStatus)
         finally:
             del self.arvrunner.jobs[record["uuid"]]
+
+
+class RunnerContainer(Runner):
+    """Submit and manage a container that runs arvados-cwl-runner."""
+
+    def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
+        """Create an Arvados job specification for this workflow.
+
+        The returned dict can be used to create a job (i.e., passed as
+        the +body+ argument to jobs().create()), or as a component in
+        a pipeline template or pipeline instance.
+        """
+
+        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) 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)
+
+        workflowname = os.path.basename(self.tool.tool["id"])
+        workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
+        workflowcollection = workflowmapper.mapper(self.tool.tool["id"])[1]
+        workflowcollection = workflowcollection[5:workflowcollection.index('/')]
+        jobpath = "/var/lib/cwl/job/cwl.input.json"
+
+        container_image = arv_docker_get_image(self.arvrunner.api,
+                                               {"dockerImageId": "arvados/jobs"},
+                                               pull_image,
+                                               self.arvrunner.project_uuid)
+
+        return {
+            "command": ["arvados-cwl-runner", "--local", "--crunch2", workflowpath, jobpath],
+            "owner_uuid": self.arvrunner.project_uuid,
+            "name": self.name,
+            "output_path": "/var/spool/cwl",
+            "cwd": "/var/spool/cwl",
+            "priority": 1,
+            "state": "Committed",
+            "container_image": container_image,
+            "mounts": {
+                workflowpath: {
+                    "kind": "collection",
+                    "portable_data_hash": "%s" % workflowcollection
+                },
+                jobpath: {
+                    "kind": "collection",
+                    "portable_data_hash": "%s/cwl.input.json" % jobobj.portable_data_hash()
+                },
+                "stdout": {
+                    "kind": "file",
+                    "path": "/var/spool/cwl/cwl.output.json"
+                }
+            },
+            "runtime_constraints": {
+                "vcpus": 1,
+                "ram": 1024*1024*256
+            }
+        }
+
+    def run(self, *args, **kwargs):
+        job_spec = self.arvados_job_spec(*args, **kwargs)
+        job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
+
+        response = self.arvrunner.api.container_requests().create(
+            body=job_spec
+        ).execute(num_retries=self.arvrunner.num_retries)
+
+        self.uuid = response["uuid"]
+        self.arvrunner.jobs[response["container_uuid"]] = self
+
+        logger.info("Submitted container %s", response["uuid"])
+
+        if response["state"] in ("Complete", "Failed", "Cancelled"):
+            self.done(response)