Merge branch '9397-cwl-initialworkdir-crunchv2' closes #9397

[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 14513c7c32a70dfbc75bb14873c1a3f6d19790b4..3090936121daf32d9ffa61744d1f5b932b73590a 100644 (file)
--- a/sdk/cwl/arvados_cwl/arvcontainer.py
+++ b/sdk/cwl/arvados_cwl/arvcontainer.py
@@ -15,8 +15,11 @@ from .arvdocker import arv_docker_get_image
 from . import done
 from .runner import Runner, arvados_jobs_image, packed_workflow, trim_listing
 from .fsaccess import CollectionFetcher
+from .pathmapper import NoFollowPathMapper
+from .perf import Perf
 
 logger = logging.getLogger('arvados.cwl-runner')
+metrics = logging.getLogger('arvados.cwl-runner.metrics')
 
 class ArvadosContainer(object):
     """Submit and manage a Crunch container request for executing a CWL CommandLineTool."""
@@ -50,23 +53,61 @@ class ArvadosContainer(object):
 
         dirs = set()
         for f in self.pathmapper.files():
-            _, p, tp = self.pathmapper.mapper(f)
-            if tp == "Directory" and '/' not in p[6:]:
+            pdh, p, tp = self.pathmapper.mapper(f)
+            if tp == "Directory" and '/' not in pdh:
                 mounts[p] = {
                     "kind": "collection",
-                    "portable_data_hash": p[6:]
+                    "portable_data_hash": pdh[5:]
                 }
-                dirs.add(p[6:])
+                dirs.add(pdh)
+
         for f in self.pathmapper.files():
-            _, p, tp = self.pathmapper.mapper(f)
-            if p[6:].split("/")[0] not in dirs:
+            res, p, tp = self.pathmapper.mapper(f)
+            if res.startswith("keep:"):
+                res = res[5:]
+            elif res.startswith("/keep/"):
+                res = res[6:]
+            else:
+                continue
+            sp = res.split("/", 1)
+            pdh = sp[0]
+            if pdh not in dirs:
                 mounts[p] = {
                     "kind": "collection",
-                    "portable_data_hash": p[6:]
+                    "portable_data_hash": pdh
                 }
-
-        if self.generatefiles["listing"]:
-            raise UnsupportedRequirement("InitialWorkDirRequirement not supported with --api=containers")
+                if len(sp) == 2:
+                    mounts[p]["path"] = sp[1]
+
+        with Perf(metrics, "generatefiles %s" % self.name):
+            if self.generatefiles["listing"]:
+                vwd = arvados.collection.Collection(api_client=self.arvrunner.api,
+                                                    keep_client=self.arvrunner.keep_client,
+                                                    num_retries=self.arvrunner.num_retries)
+                generatemapper = NoFollowPathMapper([self.generatefiles], "", "",
+                                                    separateDirs=False)
+
+                with Perf(metrics, "createfiles %s" % self.name):
+                    for f, p in generatemapper.items():
+                        if not p.target:
+                            pass
+                        elif p.type in ("File", "Directory"):
+                            source, path = self.arvrunner.fs_access.get_collection(p.resolved)
+                            vwd.copy(path, p.target, source_collection=source)
+                        elif p.type == "CreateFile":
+                            with vwd.open(p.target, "w") as n:
+                                n.write(p.resolved.encode("utf-8"))
+
+                with Perf(metrics, "generatefiles.save_new %s" % self.name):
+                    vwd.save_new()
+
+                for f, p in generatemapper.items():
+                    if not p.target:
+                        continue
+                    mountpoint = "%s/%s" % (self.outdir, p.target)
+                    mounts[mountpoint] = {"kind": "collection",
+                                          "portable_data_hash": vwd.portable_data_hash(),
+                                          "path": p.target}
 
         container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
         if self.environment:
@@ -103,7 +144,7 @@ class ArvadosContainer(object):
         runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
         if runtime_req:
             if "keep_cache" in runtime_req:
-                runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"]
+                runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"] * 2**20
 
         partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
         if partition_req:
@@ -129,10 +170,11 @@ class ArvadosContainer(object):
             self.uuid = response["uuid"]
             self.arvrunner.processes[self.uuid] = self
 
-            logger.info("%s %s state is %s", self.arvrunner.label(self), response["uuid"], response["state"])
-
             if response["state"] == "Final":
+                logger.info("%s reused container %s", self.arvrunner.label(self), response["container_uuid"])
                 self.done(response)
+            else:
+                logger.info("%s %s state is %s", self.arvrunner.label(self), response["uuid"], response["state"])
         except Exception as e:
             logger.error("%s got error %s" % (self.arvrunner.label(self), str(e)))
             self.output_callback({}, "permanentFail")
@@ -190,8 +232,6 @@ class RunnerContainer(Runner):
         the +body+ argument to container_requests().create().
         """
 
-        packed = packed_workflow(self.arvrunner, self.tool)
-
         adjustDirObjs(self.job_order, trim_listing)
 
         container_req = {
@@ -224,13 +264,24 @@ class RunnerContainer(Runner):
             "properties": {}
         }
 
-        workflowpath = "/var/lib/cwl/workflow.json#main"
-        container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
-            "kind": "json",
-            "json": packed
-        }
-        if self.tool.tool.get("id", "").startswith("arvwf:"):
-            container_req["properties"]["template_uuid"] = self.tool.tool["id"][6:33]
+        if self.tool.tool.get("id", "").startswith("keep:"):
+            sp = self.tool.tool["id"].split('/')
+            workflowcollection = sp[0][5:]
+            workflowname = "/".join(sp[1:])
+            workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
+            container_req["mounts"]["/var/lib/cwl/workflow"] = {
+                "kind": "collection",
+                "portable_data_hash": "%s" % workflowcollection
+            }
+        else:
+            packed = packed_workflow(self.arvrunner, self.tool)
+            workflowpath = "/var/lib/cwl/workflow.json#main"
+            container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
+                "kind": "json",
+                "content": packed
+            }
+            if self.tool.tool.get("id", "").startswith("arvwf:"):
+                container_req["properties"]["template_uuid"] = self.tool.tool["id"][6:33]
 
         command = ["arvados-cwl-runner", "--local", "--api=containers", "--no-log-timestamps"]
         if self.output_name: