X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/7cdfe579a8ad314cf303a280b0de68b026244748..5c4d9d38dcee73a7ffb6221c80f707c3924da64f:/sdk/cwl/arvados_cwl/arvjob.py diff --git a/sdk/cwl/arvados_cwl/arvjob.py b/sdk/cwl/arvados_cwl/arvjob.py index a620a20749..2d112c87a7 100644 --- a/sdk/cwl/arvados_cwl/arvjob.py +++ b/sdk/cwl/arvados_cwl/arvjob.py @@ -1,24 +1,31 @@ +# Copyright (C) The Arvados Authors. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + import logging import re import copy import json +import datetime import time -from cwltool.process import get_feature, shortname, UnsupportedRequirement +from cwltool.process import shortname, UnsupportedRequirement from cwltool.errors import WorkflowException -from cwltool.draft2tool import revmap_file, CommandLineTool +from cwltool.command_line_tool import revmap_file, CommandLineTool from cwltool.load_tool import fetch_document from cwltool.builder import Builder -from cwltool.pathmapper import adjustFileObjs, adjustDirObjs +from cwltool.pathmapper import adjustFileObjs, adjustDirObjs, visit_class +from cwltool.job import JobBase from schema_salad.sourceline import SourceLine import ruamel.yaml as yaml import arvados.collection +from arvados.errors import ApiError from .arvdocker import arv_docker_get_image -from .runner import Runner, arvados_jobs_image, packed_workflow, upload_workflow_collection, trim_anonymous_location +from .runner import Runner, arvados_jobs_image, packed_workflow, upload_workflow_collection, trim_anonymous_location, remove_redundant_fields from .pathmapper import VwdPathMapper, trim_listing from .perf import Perf from . import done @@ -27,19 +34,27 @@ from ._version import __version__ logger = logging.getLogger('arvados.cwl-runner') metrics = logging.getLogger('arvados.cwl-runner.metrics') -crunchrunner_re = re.compile(r"^\S+ \S+ \d+ \d+ stderr \S+ \S+ crunchrunner: \$\(task\.(tmpdir|outdir|keep)\)=(.*)") +crunchrunner_re = re.compile(r"^.*crunchrunner: \$\(task\.(tmpdir|outdir|keep)\)=(.*)$") crunchrunner_git_commit = 'a3f2cb186e437bfce0031b024b2157b73ed2717d' -class ArvadosJob(object): +class ArvadosJob(JobBase): """Submit and manage a Crunch job for executing a CWL CommandLineTool.""" - def __init__(self, runner): + def __init__(self, runner, + builder, # type: Builder + joborder, # type: Dict[Text, Union[Dict[Text, Any], List, Text]] + make_path_mapper, # type: Callable[..., PathMapper] + requirements, # type: List[Dict[Text, Text]] + hints, # type: List[Dict[Text, Text]] + name # type: Text + ): + super(ArvadosJob, self).__init__(builder, joborder, make_path_mapper, requirements, hints, name) self.arvrunner = runner self.running = False self.uuid = None - def run(self, dry_run=False, pull_image=True, **kwargs): + def run(self, runtimeContext): script_parameters = { "command": self.command_line } @@ -62,7 +77,11 @@ class ArvadosJob(object): if vwd: with Perf(metrics, "generatefiles.save_new %s" % self.name): - vwd.save_new() + info = self._get_intermediate_collection_info() + vwd.save_new(name=info["name"], + ensure_unique_name=True, + trash_at=info["trash_at"], + properties=info["properties"]) for f, p in generatemapper.items(): if p.type == "File": @@ -91,12 +110,15 @@ class ArvadosJob(object): script_parameters["task.permanentFailCodes"] = self.permanentFailCodes with Perf(metrics, "arv_docker_get_image %s" % self.name): - (docker_req, docker_is_req) = get_feature(self, "DockerRequirement") - if docker_req and kwargs.get("use_container") is not False: + (docker_req, docker_is_req) = self.get_requirement("DockerRequirement") + if docker_req and runtimeContext.use_container is not False: if docker_req.get("dockerOutputDirectory"): raise SourceLine(docker_req, "dockerOutputDirectory", UnsupportedRequirement).makeError( "Option 'dockerOutputDirectory' of DockerRequirement not supported.") - runtime_constraints["docker_image"] = arv_docker_get_image(self.arvrunner.api, docker_req, pull_image, self.arvrunner.project_uuid) + runtime_constraints["docker_image"] = arv_docker_get_image(self.arvrunner.api, + docker_req, + runtimeContext.pull_image, + self.arvrunner.project_uuid) else: runtime_constraints["docker_image"] = "arvados/jobs" @@ -106,7 +128,7 @@ class ArvadosJob(object): runtime_constraints["min_ram_mb_per_node"] = resources.get("ram") runtime_constraints["min_scratch_mb_per_node"] = resources.get("tmpdirSize", 0) + resources.get("outdirSize", 0) - runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints") + runtime_req, _ = self.get_requirement("http://arvados.org/cwl#RuntimeConstraints") if runtime_req: if "keep_cache" in runtime_req: runtime_constraints["keep_cache_mb_per_task"] = runtime_req["keep_cache"] @@ -123,6 +145,14 @@ class ArvadosJob(object): if not self.arvrunner.ignore_docker_for_reuse: filters.append(["docker_image_locator", "in docker", runtime_constraints["docker_image"]]) + enable_reuse = runtimeContext.enable_reuse + if enable_reuse: + reuse_req, _ = self.get_requirement("http://arvados.org/cwl#ReuseRequirement") + if reuse_req: + enable_reuse = reuse_req["enableReuse"] + + self.output_callback = self.arvrunner.get_wrapped_callback(self.output_callback) + try: with Perf(metrics, "create %s" % self.name): response = self.arvrunner.api.jobs().create( @@ -136,17 +166,31 @@ class ArvadosJob(object): "runtime_constraints": runtime_constraints }, filters=filters, - find_or_create=kwargs.get("enable_reuse", True) + find_or_create=enable_reuse ).execute(num_retries=self.arvrunner.num_retries) - self.arvrunner.processes[response["uuid"]] = self + self.uuid = response["uuid"] + self.arvrunner.process_submitted(self) self.update_pipeline_component(response) if response["state"] == "Complete": logger.info("%s reused job %s", self.arvrunner.label(self), response["uuid"]) - with Perf(metrics, "done %s" % self.name): - self.done(response) + # Give read permission to the desired project on reused jobs + if response["owner_uuid"] != self.arvrunner.project_uuid: + try: + self.arvrunner.api.links().create(body={ + 'link_class': 'permission', + 'name': 'can_read', + 'tail_uuid': self.arvrunner.project_uuid, + 'head_uuid': response["uuid"], + }).execute(num_retries=self.arvrunner.num_retries) + except ApiError as e: + # The user might not have "manage" access on the job: log + # a message and continue. + logger.info("Creating read permission on job %s: %s", + response["uuid"], + e) else: logger.info("%s %s is %s", self.arvrunner.label(self), response["uuid"], response["state"]) except Exception as e: @@ -154,25 +198,28 @@ class ArvadosJob(object): self.output_callback({}, "permanentFail") def update_pipeline_component(self, record): - if self.arvrunner.pipeline: - self.arvrunner.pipeline["components"][self.name] = {"job": record} - with Perf(metrics, "update_pipeline_component %s" % self.name): - self.arvrunner.pipeline = self.arvrunner.api.pipeline_instances().update(uuid=self.arvrunner.pipeline["uuid"], - body={ - "components": self.arvrunner.pipeline["components"] - }).execute(num_retries=self.arvrunner.num_retries) - if self.arvrunner.uuid: - try: - job = self.arvrunner.api.jobs().get(uuid=self.arvrunner.uuid).execute() - if job: - components = job["components"] - components[self.name] = record["uuid"] - self.arvrunner.api.jobs().update(uuid=self.arvrunner.uuid, + with self.arvrunner.workflow_eval_lock: + if self.arvrunner.pipeline: + self.arvrunner.pipeline["components"][self.name] = {"job": record} + with Perf(metrics, "update_pipeline_component %s" % self.name): + self.arvrunner.pipeline = self.arvrunner.api.pipeline_instances().update( + uuid=self.arvrunner.pipeline["uuid"], body={ - "components": components + "components": self.arvrunner.pipeline["components"] }).execute(num_retries=self.arvrunner.num_retries) - except Exception as e: - logger.info("Error adding to components: %s", e) + if self.arvrunner.uuid: + try: + job = self.arvrunner.api.jobs().get(uuid=self.arvrunner.uuid).execute() + if job: + components = job["components"] + components[self.name] = record["uuid"] + self.arvrunner.api.jobs().update( + uuid=self.arvrunner.uuid, + body={ + "components": components + }).execute(num_retries=self.arvrunner.num_retries) + except Exception as e: + logger.info("Error adding to components: %s", e) def done(self, record): try: @@ -195,12 +242,13 @@ class ArvadosJob(object): keep_client=self.arvrunner.keep_client, num_retries=self.arvrunner.num_retries) log = logc.open(logc.keys()[0]) - dirs = {} - tmpdir = None - outdir = None - keepdir = None + dirs = { + "tmpdir": "/tmpdir", + "outdir": "/outdir", + "keep": "/keep" + } for l in log: - # Determine the tmpdir, outdir and keepdir paths from + # Determine the tmpdir, outdir and keep paths from # the job run. Unfortunately, we can't take the first # values we find (which are expected to be near the # top) and stop scanning because if the node fails and @@ -212,7 +260,7 @@ class ArvadosJob(object): dirs[g.group(1)] = g.group(2) if processStatus == "permanentFail": - done.logtail(logc, logger, "%s error log:" % self.arvrunner.label(self)) + done.logtail(logc, logger.error, "%s (%s) error log:" % (self.arvrunner.label(self), record["uuid"]), maxlen=40) with Perf(metrics, "output collection %s" % self.name): outputs = done.done(self, record, dirs["tmpdir"], @@ -233,13 +281,32 @@ class ArvadosJob(object): processStatus = "permanentFail" finally: self.output_callback(outputs, processStatus) - if record["uuid"] in self.arvrunner.processes: - del self.arvrunner.processes[record["uuid"]] + + def _get_intermediate_collection_info(self): + trash_time = None + if self.arvrunner.intermediate_output_ttl > 0: + trash_time = datetime.datetime.now() + datetime.timedelta(seconds=self.arvrunner.intermediate_output_ttl) + + current_container_uuid = None + try: + current_container = self.arvrunner.api.containers().current().execute(num_retries=self.arvrunner.num_retries) + current_container_uuid = current_container['uuid'] + except ApiError as e: + # Status code 404 just means we're not running in a container. + if e.resp.status != 404: + logger.info("Getting current container: %s", e) + props = {"type": "Intermediate", + "container": current_container_uuid} + + return {"name" : "Intermediate collection", + "trash_at" : trash_time, + "properties" : props} + class RunnerJob(Runner): """Submit and manage a Crunch job that runs crunch_scripts/cwl-runner.""" - def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs): + def arvados_job_spec(self, debug=False): """Create an Arvados job specification for this workflow. The returned dict can be used to create a job (i.e., passed as @@ -250,13 +317,13 @@ class RunnerJob(Runner): if self.tool.tool["id"].startswith("keep:"): self.job_order["cwl:tool"] = self.tool.tool["id"][5:] else: - packed = packed_workflow(self.arvrunner, self.tool) + packed = packed_workflow(self.arvrunner, self.tool, self.merged_map) wf_pdh = upload_workflow_collection(self.arvrunner, self.name, packed) self.job_order["cwl:tool"] = "%s/workflow.cwl#main" % wf_pdh 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) if self.output_name: self.job_order["arv:output_name"] = self.output_name @@ -269,6 +336,9 @@ class RunnerJob(Runner): if self.on_error: self.job_order["arv:on_error"] = self.on_error + if debug: + self.job_order["arv:debug"] = True + return { "script": "cwl-runner", "script_version": "master", @@ -281,8 +351,8 @@ class RunnerJob(Runner): } } - def run(self, *args, **kwargs): - job_spec = self.arvados_job_spec(*args, **kwargs) + def run(self, runtimeContext): + job_spec = self.arvados_job_spec(runtimeContext.debug) job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid) @@ -291,53 +361,34 @@ class RunnerJob(Runner): find_or_create=self.enable_reuse ).execute(num_retries=self.arvrunner.num_retries) - if self.enable_reuse: - # When reusing jobs, copy its output/log collection to the desired project - reused_collections = [('Output', job.get('output', None)), - ('Log', job.get('log', None))] - for col_type, pdh in [(n, p) for n, p in reused_collections if p]: - c = arvados.collection.Collection(pdh, - api_client=self.arvrunner.api, - keep_client=self.arvrunner.keep_client, - num_retries=self.arvrunner.num_retries) - c.save_new(name="{} of {}".format(col_type, self.name), - owner_uuid=self.arvrunner.project_uuid, - ensure_unique_name=True, - num_retries=self.arvrunner.num_retries) - logger.info("Copied reused job's %s to collection %s", - col_type.lower(), - c.manifest_locator()) - # Give read permission to the desired project on reused jobs - for job_name, job_uuid in job.get('components', {}).items(): - self.arvrunner.api.links().create(body={ - 'link_class': 'can_read', - 'tail_uuid': self.arvrunner.project_uuid, - 'head_uuid': job_uuid, - }).execute(num_retries=self.arvrunner.num_retries) - for k,v in job_spec["script_parameters"].items(): if v is False or v is None or isinstance(v, dict): job_spec["script_parameters"][k] = {"value": v} del job_spec["owner_uuid"] job_spec["job"] = job + + instance_spec = { + "owner_uuid": self.arvrunner.project_uuid, + "name": self.name, + "components": { + "cwl-runner": job_spec, + }, + "state": "RunningOnServer", + } + if not self.enable_reuse: + instance_spec["properties"] = {"run_options": {"enable_job_reuse": False}} + self.arvrunner.pipeline = self.arvrunner.api.pipeline_instances().create( - body={ - "owner_uuid": self.arvrunner.project_uuid, - "name": self.name, - "components": {"cwl-runner": job_spec }, - "state": "RunningOnServer"}).execute(num_retries=self.arvrunner.num_retries) + body=instance_spec).execute(num_retries=self.arvrunner.num_retries) logger.info("Created pipeline %s", self.arvrunner.pipeline["uuid"]) - if kwargs.get("wait") is False: + if runtimeContext.wait is False: self.uuid = self.arvrunner.pipeline["uuid"] return self.uuid = job["uuid"] - self.arvrunner.processes[self.uuid] = self - - if job["state"] in ("Complete", "Failed", "Cancelled"): - self.done(job) + self.arvrunner.process_submitted(self) class RunnerTemplate(object): @@ -353,7 +404,7 @@ class RunnerTemplate(object): } def __init__(self, runner, tool, job_order, enable_reuse, uuid, - submit_runner_ram=0, name=None): + submit_runner_ram=0, name=None, merged_map=None): self.runner = runner self.tool = tool self.job = RunnerJob( @@ -364,7 +415,8 @@ class RunnerTemplate(object): output_name=None, output_tags=None, submit_runner_ram=submit_runner_ram, - name=name) + name=name, + merged_map=merged_map) self.uuid = uuid def pipeline_component_spec(self):