Fix arvados-cwl-runner --version argument (again) refs #12526

[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 b58a858c8d40775c227038ddb3fe43665b851b3c..9f8382291ccc82c76f5cd03824cb42ced718b0ca 100644 (file)
--- a/sdk/cwl/arvados_cwl/arvcontainer.py
+++ b/sdk/cwl/arvados_cwl/arvcontainer.py
@@ -1,3 +1,7 @@
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
 import logging
 import json
 import os
@@ -10,14 +14,14 @@ import ruamel.yaml as yaml
 
 from cwltool.errors import WorkflowException
 from cwltool.process import get_feature, UnsupportedRequirement, shortname
-from cwltool.pathmapper import adjustFileObjs, adjustDirObjs
+from cwltool.pathmapper import adjustFileObjs, adjustDirObjs, visit_class
 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_anonymous_location
+from .runner import Runner, arvados_jobs_image, packed_workflow, trim_anonymous_location, remove_redundant_fields
 from .fsaccess import CollectionFetcher
 from .pathmapper import NoFollowPathMapper, trim_listing
 from .perf import Perf
@@ -43,7 +47,7 @@ class ArvadosContainer(object):
             "name": self.name,
             "output_path": self.outdir,
             "cwd": self.outdir,
-            "priority": 1,
+            "priority": kwargs.get("priority"),
             "state": "Committed",
             "properties": {},
         }
@@ -101,17 +105,32 @@ class ArvadosContainer(object):
                 generatemapper = NoFollowPathMapper([self.generatefiles], "", "",
                                                     separateDirs=False)
 
+                logger.debug("generatemapper is %s", generatemapper._pathmap)
+
                 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 in ("File", "Directory", "WritableFile", "WritableDirectory"):
+                            if p.resolved.startswith("_:"):
+                                vwd.mkdirs(p.target)
+                            else:
+                                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"))
 
+                def keepemptydirs(p):
+                    if isinstance(p, arvados.collection.RichCollectionBase):
+                        if len(p) == 0:
+                            p.open(".keep", "w").close()
+                        else:
+                            for c in p:
+                                keepemptydirs(p[c])
+
+                keepemptydirs(vwd)
+
                 with Perf(metrics, "generatefiles.save_new %s" % self.name):
                     vwd.save_new()
 
@@ -122,6 +141,8 @@ class ArvadosContainer(object):
                     mounts[mountpoint] = {"kind": "collection",
                                           "portable_data_hash": vwd.portable_data_hash(),
                                           "path": p.target}
+                    if p.type.startswith("Writable"):
+                        mounts[mountpoint]["writable"] = True
 
         container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
         if self.environment:
@@ -184,9 +205,15 @@ class ArvadosContainer(object):
         container_request["output_ttl"] = self.output_ttl
         container_request["mounts"] = mounts
         container_request["runtime_constraints"] = runtime_constraints
-        container_request["use_existing"] = kwargs.get("enable_reuse", True)
         container_request["scheduling_parameters"] = scheduling_parameters
 
+        enable_reuse = kwargs.get("enable_reuse", True)
+        if enable_reuse:
+            reuse_req, _ = get_feature(self, "http://arvados.org/cwl#ReuseRequirement")
+            if reuse_req:
+                enable_reuse = reuse_req["enableReuse"]
+        container_request["use_existing"] = enable_reuse
+
         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)
@@ -276,15 +303,15 @@ class RunnerContainer(Runner):
         """
 
         adjustDirObjs(self.job_order, trim_listing)
-        adjustFileObjs(self.job_order, trim_anonymous_location)
-        adjustDirObjs(self.job_order, trim_anonymous_location)
+        visit_class(self.job_order, ("File", "Directory"), trim_anonymous_location)
+        visit_class(self.job_order, ("File", "Directory"), remove_redundant_fields)
 
         container_req = {
             "owner_uuid": self.arvrunner.project_uuid,
             "name": self.name,
             "output_path": "/var/spool/cwl",
             "cwd": "/var/spool/cwl",
-            "priority": 1,
+            "priority": self.priority,
             "state": "Committed",
             "container_image": arvados_jobs_image(self.arvrunner, self.jobs_image),
             "mounts": {
@@ -306,6 +333,7 @@ class RunnerContainer(Runner):
                 "ram": 1024*1024 * self.submit_runner_ram,
                 "API": True
             },
+            "use_existing": self.enable_reuse,
             "properties": {}
         }
 
@@ -319,7 +347,7 @@ class RunnerContainer(Runner):
                 "portable_data_hash": "%s" % workflowcollection
             }
         else:
-            packed = packed_workflow(self.arvrunner, self.tool)
+            packed = packed_workflow(self.arvrunner, self.tool, self.merged_map)
             workflowpath = "/var/lib/cwl/workflow.json#main"
             container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
                 "kind": "json",
@@ -328,6 +356,7 @@ class RunnerContainer(Runner):
             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:
             command.append("--output-name=" + self.output_name)
@@ -353,6 +382,11 @@ class RunnerContainer(Runner):
         if self.arvrunner.trash_intermediate:
             command.append("--trash-intermediate")
 
+        if self.arvrunner.project_uuid:
+            command.append("--project-uuid="+self.arvrunner.project_uuid)
+
+        command.append("--eval-timeout=%s" % self.arvrunner.eval_timeout)
+
         command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
 
         container_req["command"] = command