+# 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
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
"cwd": self.outdir,
"priority": 1,
"state": "Committed",
- "properties": {}
+ "properties": {},
}
runtime_constraints = {}
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)
self.output_callback({}, "permanentFail")
def done(self, record):
+ outputs = {}
try:
container = self.arvrunner.api.containers().get(
uuid=record["container_uuid"]
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:
"""
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,
"ram": 1024*1024 * self.submit_runner_ram,
"API": True
},
+ "use_existing": self.enable_reuse,
"properties": {}
}
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")
+
+ if self.arvrunner.project_uuid:
+ command.append("--project-uuid="+self.arvrunner.project_uuid)
+
command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
container_req["command"] = command