9 import ruamel.yaml as yaml
11 from cwltool.errors import WorkflowException
12 from cwltool.process import get_feature, UnsupportedRequirement, shortname
13 from cwltool.pathmapper import adjustFileObjs, adjustDirObjs
14 from cwltool.utils import aslist
16 import arvados.collection
18 from .arvdocker import arv_docker_get_image
20 from .runner import Runner, arvados_jobs_image, packed_workflow, trim_anonymous_location
21 from .fsaccess import CollectionFetcher
22 from .pathmapper import NoFollowPathMapper, trim_listing
23 from .perf import Perf
25 logger = logging.getLogger('arvados.cwl-runner')
26 metrics = logging.getLogger('arvados.cwl-runner.metrics')
28 class ArvadosContainer(object):
29 """Submit and manage a Crunch container request for executing a CWL CommandLineTool."""
31 def __init__(self, runner):
32 self.arvrunner = runner
36 def update_pipeline_component(self, r):
39 def run(self, dry_run=False, pull_image=True, **kwargs):
41 "command": self.command_line,
42 "owner_uuid": self.arvrunner.project_uuid,
44 "output_path": self.outdir,
50 runtime_constraints = {}
52 resources = self.builder.resources
53 if resources is not None:
54 runtime_constraints["vcpus"] = resources.get("cores", 1)
55 runtime_constraints["ram"] = resources.get("ram") * 2**20
60 "capacity": resources.get("outdirSize", 0) * 2**20
64 "capacity": resources.get("tmpdirSize", 0) * 2**20
67 scheduling_parameters = {}
69 rf = [self.pathmapper.mapper(f) for f in self.pathmapper.referenced_files]
70 rf.sort(key=lambda k: k.resolved)
72 for resolved, target, tp, stg in rf:
75 if prevdir and target.startswith(prevdir):
80 targetdir = os.path.dirname(target)
81 sp = resolved.split("/", 1)
82 pdh = sp[0][5:] # remove "keep:"
85 "portable_data_hash": pdh
91 path = os.path.dirname(sp[1])
92 if path and path != "/":
93 mounts[targetdir]["path"] = path
94 prevdir = targetdir + "/"
96 with Perf(metrics, "generatefiles %s" % self.name):
97 if self.generatefiles["listing"]:
98 vwd = arvados.collection.Collection(api_client=self.arvrunner.api,
99 keep_client=self.arvrunner.keep_client,
100 num_retries=self.arvrunner.num_retries)
101 generatemapper = NoFollowPathMapper([self.generatefiles], "", "",
104 with Perf(metrics, "createfiles %s" % self.name):
105 for f, p in generatemapper.items():
108 elif p.type in ("File", "Directory"):
109 source, path = self.arvrunner.fs_access.get_collection(p.resolved)
110 vwd.copy(path, p.target, source_collection=source)
111 elif p.type == "CreateFile":
112 with vwd.open(p.target, "w") as n:
113 n.write(p.resolved.encode("utf-8"))
115 with Perf(metrics, "generatefiles.save_new %s" % self.name):
118 for f, p in generatemapper.items():
121 mountpoint = "%s/%s" % (self.outdir, p.target)
122 mounts[mountpoint] = {"kind": "collection",
123 "portable_data_hash": vwd.portable_data_hash(),
126 container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
128 container_request["environment"].update(self.environment)
131 sp = self.stdin[6:].split("/", 1)
132 mounts["stdin"] = {"kind": "collection",
133 "portable_data_hash": sp[0],
137 mounts["stderr"] = {"kind": "file",
138 "path": "%s/%s" % (self.outdir, self.stderr)}
141 mounts["stdout"] = {"kind": "file",
142 "path": "%s/%s" % (self.outdir, self.stdout)}
144 (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
146 docker_req = {"dockerImageId": "arvados/jobs"}
148 container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
151 self.arvrunner.project_uuid)
153 api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement")
155 runtime_constraints["API"] = True
157 runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
159 if "keep_cache" in runtime_req:
160 runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"] * 2**20
161 if "outputDirType" in runtime_req:
162 if runtime_req["outputDirType"] == "local_output_dir":
163 # Currently the default behavior.
165 elif runtime_req["outputDirType"] == "keep_output_dir":
166 mounts[self.outdir]= {
167 "kind": "collection",
171 partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
173 scheduling_parameters["partitions"] = aslist(partition_req["partition"])
175 intermediate_output_req, _ = get_feature(self, "http://arvados.org/cwl#IntermediateOutput")
176 if intermediate_output_req:
177 self.output_ttl = intermediate_output_req["outputTTL"]
179 self.output_ttl = self.arvrunner.intermediate_output_ttl
181 if self.output_ttl < 0:
182 raise WorkflowError("Invalid value %d for output_ttl, cannot be less than zero" % container_request["output_ttl"])
184 container_request["output_ttl"] = self.output_ttl
185 container_request["mounts"] = mounts
186 container_request["runtime_constraints"] = runtime_constraints
187 container_request["scheduling_parameters"] = scheduling_parameters
189 enable_reuse = kwargs.get("enable_reuse", True)
191 reuse_req, _ = get_feature(self, "http://arvados.org/cwl#ReuseRequirement")
193 enable_reuse = reuse_req["enableReuse"]
194 container_request["use_existing"] = enable_reuse
196 if kwargs.get("runnerjob", "").startswith("arvwf:"):
197 wfuuid = kwargs["runnerjob"][6:kwargs["runnerjob"].index("#")]
198 wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
199 if container_request["name"] == "main":
200 container_request["name"] = wfrecord["name"]
201 container_request["properties"]["template_uuid"] = wfuuid
204 response = self.arvrunner.api.container_requests().create(
205 body=container_request
206 ).execute(num_retries=self.arvrunner.num_retries)
208 self.uuid = response["uuid"]
209 self.arvrunner.processes[self.uuid] = self
211 if response["state"] == "Final":
212 logger.info("%s reused container %s", self.arvrunner.label(self), response["container_uuid"])
215 logger.info("%s %s state is %s", self.arvrunner.label(self), response["uuid"], response["state"])
216 except Exception as e:
217 logger.error("%s got error %s" % (self.arvrunner.label(self), str(e)))
218 self.output_callback({}, "permanentFail")
220 def done(self, record):
223 container = self.arvrunner.api.containers().get(
224 uuid=record["container_uuid"]
225 ).execute(num_retries=self.arvrunner.num_retries)
226 if container["state"] == "Complete":
227 rcode = container["exit_code"]
228 if self.successCodes and rcode in self.successCodes:
229 processStatus = "success"
230 elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
231 processStatus = "temporaryFail"
232 elif self.permanentFailCodes and rcode in self.permanentFailCodes:
233 processStatus = "permanentFail"
235 processStatus = "success"
237 processStatus = "permanentFail"
239 processStatus = "permanentFail"
241 if processStatus == "permanentFail":
242 logc = arvados.collection.CollectionReader(container["log"],
243 api_client=self.arvrunner.api,
244 keep_client=self.arvrunner.keep_client,
245 num_retries=self.arvrunner.num_retries)
246 done.logtail(logc, logger, "%s error log:" % self.arvrunner.label(self))
248 if record["output_uuid"]:
249 if self.arvrunner.trash_intermediate or self.arvrunner.intermediate_output_ttl:
250 # Compute the trash time to avoid requesting the collection record.
251 trash_at = ciso8601.parse_datetime_unaware(record["modified_at"]) + datetime.timedelta(0, self.arvrunner.intermediate_output_ttl)
252 aftertime = " at %s" % trash_at.strftime("%Y-%m-%d %H:%M:%S UTC") if self.arvrunner.intermediate_output_ttl else ""
253 orpart = ", or" if self.arvrunner.trash_intermediate and self.arvrunner.intermediate_output_ttl else ""
254 oncomplete = " upon successful completion of the workflow" if self.arvrunner.trash_intermediate else ""
255 logger.info("%s Intermediate output %s (%s) will be trashed%s%s%s." % (
256 self.arvrunner.label(self), record["output_uuid"], container["output"], aftertime, orpart, oncomplete))
257 self.arvrunner.add_intermediate_output(record["output_uuid"])
259 if container["output"]:
260 outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep")
261 except WorkflowException as e:
262 logger.error("%s unable to collect output from %s:\n%s",
263 self.arvrunner.label(self), container["output"], e, exc_info=(e if self.arvrunner.debug else False))
264 processStatus = "permanentFail"
265 except Exception as e:
266 logger.exception("%s while getting output object: %s", self.arvrunner.label(self), e)
267 processStatus = "permanentFail"
269 self.output_callback(outputs, processStatus)
270 if record["uuid"] in self.arvrunner.processes:
271 del self.arvrunner.processes[record["uuid"]]
274 class RunnerContainer(Runner):
275 """Submit and manage a container that runs arvados-cwl-runner."""
277 def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
278 """Create an Arvados container request for this workflow.
280 The returned dict can be used to create a container passed as
281 the +body+ argument to container_requests().create().
284 adjustDirObjs(self.job_order, trim_listing)
285 adjustFileObjs(self.job_order, trim_anonymous_location)
286 adjustDirObjs(self.job_order, trim_anonymous_location)
289 "owner_uuid": self.arvrunner.project_uuid,
291 "output_path": "/var/spool/cwl",
292 "cwd": "/var/spool/cwl",
294 "state": "Committed",
295 "container_image": arvados_jobs_image(self.arvrunner, self.jobs_image),
297 "/var/lib/cwl/cwl.input.json": {
299 "content": self.job_order
303 "path": "/var/spool/cwl/cwl.output.json"
306 "kind": "collection",
310 "runtime_constraints": {
312 "ram": 1024*1024 * self.submit_runner_ram,
318 if self.tool.tool.get("id", "").startswith("keep:"):
319 sp = self.tool.tool["id"].split('/')
320 workflowcollection = sp[0][5:]
321 workflowname = "/".join(sp[1:])
322 workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
323 container_req["mounts"]["/var/lib/cwl/workflow"] = {
324 "kind": "collection",
325 "portable_data_hash": "%s" % workflowcollection
328 packed = packed_workflow(self.arvrunner, self.tool)
329 workflowpath = "/var/lib/cwl/workflow.json#main"
330 container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
334 if self.tool.tool.get("id", "").startswith("arvwf:"):
335 container_req["properties"]["template_uuid"] = self.tool.tool["id"][6:33]
337 command = ["arvados-cwl-runner", "--local", "--api=containers", "--no-log-timestamps"]
339 command.append("--output-name=" + self.output_name)
340 container_req["output_name"] = self.output_name
343 command.append("--output-tags=" + self.output_tags)
345 if kwargs.get("debug"):
346 command.append("--debug")
348 if self.enable_reuse:
349 command.append("--enable-reuse")
351 command.append("--disable-reuse")
354 command.append("--on-error=" + self.on_error)
356 if self.intermediate_output_ttl:
357 command.append("--intermediate-output-ttl=%d" % self.intermediate_output_ttl)
359 if self.arvrunner.trash_intermediate:
360 command.append("--trash-intermediate")
362 command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
364 container_req["command"] = command
369 def run(self, *args, **kwargs):
370 kwargs["keepprefix"] = "keep:"
371 job_spec = self.arvados_job_spec(*args, **kwargs)
372 job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
374 response = self.arvrunner.api.container_requests().create(
376 ).execute(num_retries=self.arvrunner.num_retries)
378 self.uuid = response["uuid"]
379 self.arvrunner.processes[self.uuid] = self
381 logger.info("%s submitted container %s", self.arvrunner.label(self), response["uuid"])
383 if response["state"] == "Final":
386 def done(self, record):
388 container = self.arvrunner.api.containers().get(
389 uuid=record["container_uuid"]
390 ).execute(num_retries=self.arvrunner.num_retries)
391 except Exception as e:
392 logger.exception("%s while getting runner container: %s", self.arvrunner.label(self), e)
393 self.arvrunner.output_callback({}, "permanentFail")
395 super(RunnerContainer, self).done(container)
397 if record["uuid"] in self.arvrunner.processes:
398 del self.arvrunner.processes[record["uuid"]]