Merge branch '2411-check-copyright'

[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 657d5927d0328025eb948f6e8ee62ba215486769..4ab65d9d8774708613787b3a694f64bf876004da 100644 (file)
--- a/sdk/cwl/arvados_cwl/arvcontainer.py
+++ b/sdk/cwl/arvados_cwl/arvcontainer.py
@@ -1,7 +1,14 @@
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
 import logging
 import json
 import os
 import urllib
+import time
+import datetime
+import ciso8601
 
 import ruamel.yaml as yaml
 
@@ -42,15 +49,23 @@ class ArvadosContainer(object):
             "cwd": self.outdir,
             "priority": 1,
             "state": "Committed",
-            "properties": {}
+            "properties": {},
         }
         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"
+                "kind": "tmp",
+                "capacity": resources.get("tmpdirSize", 0) * 2**20
             }
         }
         scheduling_parameters = {}
@@ -139,11 +154,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
@@ -152,16 +162,41 @@ 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:
             scheduling_parameters["partitions"] = aslist(partition_req["partition"])
 
+        intermediate_output_req, _ = get_feature(self, "http://arvados.org/cwl#IntermediateOutput")
+        if intermediate_output_req:
+            self.output_ttl = intermediate_output_req["outputTTL"]
+        else:
+            self.output_ttl = self.arvrunner.intermediate_output_ttl
+
+        if self.output_ttl < 0:
+            raise WorkflowError("Invalid value %d for output_ttl, cannot be less than zero" % container_request["output_ttl"])
+
+        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)
@@ -187,6 +222,7 @@ class ArvadosContainer(object):
             self.output_callback({}, "permanentFail")
 
     def done(self, record):
+        outputs = {}
         try:
             container = self.arvrunner.api.containers().get(
                 uuid=record["container_uuid"]
@@ -213,7 +249,17 @@ class ArvadosContainer(object):
                                                            num_retries=self.arvrunner.num_retries)
                 done.logtail(logc, logger, "%s error log:" % self.arvrunner.label(self))
 
-            outputs = {}
+            if record["output_uuid"]:
+                if self.arvrunner.trash_intermediate or self.arvrunner.intermediate_output_ttl:
+                    # Compute the trash time to avoid requesting the collection record.
+                    trash_at = ciso8601.parse_datetime_unaware(record["modified_at"]) + datetime.timedelta(0, self.arvrunner.intermediate_output_ttl)
+                    aftertime = " at %s" % trash_at.strftime("%Y-%m-%d %H:%M:%S UTC") if self.arvrunner.intermediate_output_ttl else ""
+                    orpart = ", or" if self.arvrunner.trash_intermediate and self.arvrunner.intermediate_output_ttl else ""
+                    oncomplete = " upon successful completion of the workflow" if self.arvrunner.trash_intermediate else ""
+                    logger.info("%s Intermediate output %s (%s) will be trashed%s%s%s." % (
+                        self.arvrunner.label(self), record["output_uuid"], container["output"], aftertime, orpart, oncomplete))
+                self.arvrunner.add_intermediate_output(record["output_uuid"])
+
             if container["output"]:
                 outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep")
         except WorkflowException as e:
@@ -311,6 +357,12 @@ 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)
+
+        if self.arvrunner.trash_intermediate:
+            command.append("--trash-intermediate")
+
         command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
 
         container_req["command"] = command