11100: Propagate through to runner. Use intermediate_output_ttl consistently.

[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 9e76cf711ecaaf81c6e5e42131ea3d717da00c01..b195ce21b184976c7d5c44723c06700dfb0c4dd9 100644 (file)
--- a/sdk/cwl/arvados_cwl/arvcontainer.py
+++ b/sdk/cwl/arvados_cwl/arvcontainer.py
@@ -7,16 +7,16 @@ import ruamel.yaml as yaml
 
 from cwltool.errors import WorkflowException
 from cwltool.process import get_feature, UnsupportedRequirement, shortname
-from cwltool.pathmapper import adjustFiles, adjustDirObjs
+from cwltool.pathmapper import adjustFileObjs, adjustDirObjs
 from cwltool.utils import aslist
 
 import arvados.collection
 
 from .arvdocker import arv_docker_get_image
 from . import done
-from .runner import Runner, arvados_jobs_image, packed_workflow, trim_listing
+from .runner import Runner, arvados_jobs_image, packed_workflow, trim_anonymous_location
 from .fsaccess import CollectionFetcher
-from .pathmapper import NoFollowPathMapper
+from .pathmapper import NoFollowPathMapper, trim_listing
 from .perf import Perf
 
 logger = logging.getLogger('arvados.cwl-runner')
@@ -42,43 +42,54 @@ class ArvadosContainer(object):
             "cwd": self.outdir,
             "priority": 1,
             "state": "Committed",
-            "properties": {}
+            "properties": {},
+            "output_ttl", self.arvrunner.intermediate_output_ttl
         }
         runtime_constraints = {}
+
+        resources = self.builder.resources
+        if resources is not None:
+            runtime_constraints["vcpus"] = resources.get("cores", 1)
+            runtime_constraints["ram"] = resources.get("ram") * 2**20
+
         mounts = {
             self.outdir: {
-                "kind": "tmp"
+                "kind": "tmp",
+                "capacity": resources.get("outdirSize", 0) * 2**20
+            },
+            self.tmpdir: {
+                "kind": "tmp",
+                "capacity": resources.get("tmpdirSize", 0) * 2**20
             }
         }
         scheduling_parameters = {}
 
-        dirs = set()
-        for f in self.pathmapper.files():
-            pdh, p, tp, stg = self.pathmapper.mapper(f)
-            if tp == "Directory" and '/' not in pdh:
-                mounts[p] = {
-                    "kind": "collection",
-                    "portable_data_hash": pdh[5:]
-                }
-                dirs.add(pdh)
-
-        for f in self.pathmapper.files():
-            res, p, tp, stg = self.pathmapper.mapper(f)
-            if res.startswith("keep:"):
-                res = res[5:]
-            elif res.startswith("/keep/"):
-                res = res[6:]
-            else:
+        rf = [self.pathmapper.mapper(f) for f in self.pathmapper.referenced_files]
+        rf.sort(key=lambda k: k.resolved)
+        prevdir = None
+        for resolved, target, tp, stg in rf:
+            if not stg:
                 continue
-            sp = res.split("/", 1)
-            pdh = sp[0]
-            if pdh not in dirs:
-                mounts[p] = {
-                    "kind": "collection",
-                    "portable_data_hash": pdh
-                }
-                if len(sp) == 2:
-                    mounts[p]["path"] = urllib.unquote(sp[1])
+            if prevdir and target.startswith(prevdir):
+                continue
+            if tp == "Directory":
+                targetdir = target
+            else:
+                targetdir = os.path.dirname(target)
+            sp = resolved.split("/", 1)
+            pdh = sp[0][5:]   # remove "keep:"
+            mounts[targetdir] = {
+                "kind": "collection",
+                "portable_data_hash": pdh
+            }
+            if len(sp) == 2:
+                if tp == "Directory":
+                    path = sp[1]
+                else:
+                    path = os.path.dirname(sp[1])
+                if path and path != "/":
+                    mounts[targetdir]["path"] = path
+            prevdir = targetdir + "/"
 
         with Perf(metrics, "generatefiles %s" % self.name):
             if self.generatefiles["listing"]:
@@ -137,11 +148,6 @@ class ArvadosContainer(object):
                                                                      pull_image,
                                                                      self.arvrunner.project_uuid)
 
-        resources = self.builder.resources
-        if resources is not None:
-            runtime_constraints["vcpus"] = resources.get("cores", 1)
-            runtime_constraints["ram"] = resources.get("ram") * 2**20
-
         api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement")
         if api_req:
             runtime_constraints["API"] = True
@@ -150,6 +156,15 @@ class ArvadosContainer(object):
         if runtime_req:
             if "keep_cache" in runtime_req:
                 runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"] * 2**20
+            if "outputDirType" in runtime_req:
+                if runtime_req["outputDirType"] == "local_output_dir":
+                    # Currently the default behavior.
+                    pass
+                elif runtime_req["outputDirType"] == "keep_output_dir":
+                    mounts[self.outdir]= {
+                        "kind": "collection",
+                        "writable": True
+                    }
 
         partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
         if partition_req:
@@ -186,6 +201,8 @@ class ArvadosContainer(object):
 
     def done(self, record):
         try:
+            self.arvrunner.add_intermediate_output(record["output_uuid"])
+
             container = self.arvrunner.api.containers().get(
                 uuid=record["container_uuid"]
             ).execute(num_retries=self.arvrunner.num_retries)
@@ -238,6 +255,8 @@ class RunnerContainer(Runner):
         """
 
         adjustDirObjs(self.job_order, trim_listing)
+        adjustFileObjs(self.job_order, trim_anonymous_location)
+        adjustDirObjs(self.job_order, trim_anonymous_location)
 
         container_req = {
             "owner_uuid": self.arvrunner.project_uuid,
@@ -307,6 +326,9 @@ class RunnerContainer(Runner):
         if self.on_error:
             command.append("--on-error=" + self.on_error)
 
+        if self.intermediate_output_ttl:
+            command.append("--intermediate-output-ttl=%d" % self.intermediate_output_ttl)
+
         command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
 
         container_req["command"] = command