X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/2f953026bc4baeccb78ca82acc4d07cad37625b8..6ab526bb7fef3d7d42ff728fa30444e75de7be38:/sdk/cwl/arvados_cwl/arvjob.py diff --git a/sdk/cwl/arvados_cwl/arvjob.py b/sdk/cwl/arvados_cwl/arvjob.py index 30056c7c61..a620a20749 100644 --- a/sdk/cwl/arvados_cwl/arvjob.py +++ b/sdk/cwl/arvados_cwl/arvjob.py @@ -4,20 +4,22 @@ import copy import json import time -from cwltool.process import get_feature, shortname +from cwltool.process import get_feature, shortname, UnsupportedRequirement from cwltool.errors import WorkflowException from cwltool.draft2tool import revmap_file, CommandLineTool from cwltool.load_tool import fetch_document from cwltool.builder import Builder -from cwltool.pathmapper import adjustDirObjs +from cwltool.pathmapper import adjustFileObjs, adjustDirObjs + +from schema_salad.sourceline import SourceLine import ruamel.yaml as yaml import arvados.collection from .arvdocker import arv_docker_get_image -from .runner import Runner, arvados_jobs_image, packed_workflow, trim_listing -from .pathmapper import InitialWorkDirPathMapper +from .runner import Runner, arvados_jobs_image, packed_workflow, upload_workflow_collection, trim_anonymous_location +from .pathmapper import VwdPathMapper, trim_listing from .perf import Perf from . import done from ._version import __version__ @@ -49,8 +51,8 @@ class ArvadosJob(object): keep_client=self.arvrunner.keep_client, num_retries=self.arvrunner.num_retries) script_parameters["task.vwd"] = {} - generatemapper = InitialWorkDirPathMapper([self.generatefiles], "", "", - separateDirs=False) + generatemapper = VwdPathMapper([self.generatefiles], "", "", + separateDirs=False) with Perf(metrics, "createfiles %s" % self.name): for f, p in generatemapper.items(): @@ -58,8 +60,9 @@ class ArvadosJob(object): with vwd.open(p.target, "w") as n: n.write(p.resolved.encode("utf-8")) - with Perf(metrics, "generatefiles.save_new %s" % self.name): - vwd.save_new() + if vwd: + with Perf(metrics, "generatefiles.save_new %s" % self.name): + vwd.save_new() for f, p in generatemapper.items(): if p.type == "File": @@ -107,6 +110,7 @@ class ArvadosJob(object): if runtime_req: if "keep_cache" in runtime_req: runtime_constraints["keep_cache_mb_per_task"] = runtime_req["keep_cache"] + runtime_constraints["min_ram_mb_per_node"] += runtime_req["keep_cache"] if "outputDirType" in runtime_req: if runtime_req["outputDirType"] == "local_output_dir": script_parameters["task.keepTmpOutput"] = False @@ -139,11 +143,12 @@ class ArvadosJob(object): self.update_pipeline_component(response) - logger.info("%s %s is %s", self.arvrunner.label(self), response["uuid"], response["state"]) - - if response["state"] in ("Complete", "Failed", "Cancelled"): + 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) + else: + logger.info("%s %s is %s", self.arvrunner.label(self), response["uuid"], response["state"]) except Exception as e: logger.exception("%s error" % (self.arvrunner.label(self))) self.output_callback({}, "permanentFail") @@ -234,28 +239,6 @@ class ArvadosJob(object): class RunnerJob(Runner): """Submit and manage a Crunch job that runs crunch_scripts/cwl-runner.""" - def upload_workflow_collection(self, packed): - collection = arvados.collection.Collection(api_client=self.arvrunner.api, - keep_client=self.arvrunner.keep_client, - num_retries=self.arvrunner.num_retries) - with collection.open("workflow.cwl", "w") as f: - f.write(yaml.round_trip_dump(packed)) - - exists = self.arvrunner.api.collections().list(filters=[["owner_uuid", "=", self.arvrunner.project_uuid], - ["portable_data_hash", "=", collection.portable_data_hash()], - ["name", "like", self.name+"%"]]).execute(num_retries=self.arvrunner.num_retries) - - if exists["items"]: - logger.info("Using collection %s", exists["items"][0]["uuid"]) - else: - collection.save_new(name=self.name, - owner_uuid=self.arvrunner.project_uuid, - ensure_unique_name=True, - num_retries=self.arvrunner.num_retries) - logger.info("Uploaded to %s", collection.manifest_locator()) - - return collection.portable_data_hash() - def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs): """Create an Arvados job specification for this workflow. @@ -268,10 +251,12 @@ class RunnerJob(Runner): self.job_order["cwl:tool"] = self.tool.tool["id"][5:] else: packed = packed_workflow(self.arvrunner, self.tool) - wf_pdh = self.upload_workflow_collection(packed) - self.job_order["cwl:tool"] = "%s/workflow.cwl" % wf_pdh + 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) if self.output_name: self.job_order["arv:output_name"] = self.output_name @@ -306,6 +291,30 @@ 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} @@ -385,10 +394,12 @@ class RunnerTemplate(object): if not isinstance(types, list): types = [types] param['required'] = 'null' not in types - non_null_types = set(types) - set(['null']) + non_null_types = [t for t in types if t != "null"] if len(non_null_types) == 1: the_type = [c for c in non_null_types][0] - dataclass = self.type_to_dataclass.get(the_type) + dataclass = None + if isinstance(the_type, basestring): + dataclass = self.type_to_dataclass.get(the_type) if dataclass: param['dataclass'] = dataclass # Note: If we didn't figure out a single appropriate