X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e96a1b2bef7762c2896273af5d15625f4a845c2e..d6a05be901e501bd10c00702fc8540dc1efba68a:/sdk/cwl/arvados_cwl/arvworkflow.py diff --git a/sdk/cwl/arvados_cwl/arvworkflow.py b/sdk/cwl/arvados_cwl/arvworkflow.py index 4fe82a6fe1..c0b479c37f 100644 --- a/sdk/cwl/arvados_cwl/arvworkflow.py +++ b/sdk/cwl/arvados_cwl/arvworkflow.py @@ -13,6 +13,8 @@ import logging from schema_salad.sourceline import SourceLine, cmap import schema_salad.ref_resolver +import arvados.collection + from cwltool.pack import pack from cwltool.load_tool import fetch_document, resolve_and_validate_document from cwltool.process import shortname @@ -37,11 +39,64 @@ metrics = logging.getLogger('arvados.cwl-runner.metrics') max_res_pars = ("coresMin", "coresMax", "ramMin", "ramMax", "tmpdirMin", "tmpdirMax") sum_res_pars = ("outdirMin", "outdirMax") -def upload_workflow(arvRunner, tool, job_order, project_uuid, uuid=None, +def make_wrapper_workflow(arvRunner, main, packed, project_uuid, name): + col = arvados.collection.Collection(api_client=arvRunner.api, + keep_client=arvRunner.keep_client) + + with col.open("workflow.json", "wt") as f: + json.dump(packed, f, sort_keys=True, indent=4, separators=(',',': ')) + + pdh = col.portable_data_hash() + + existing = arvRunner.api.collections().list(filters=[["portable_data_hash", "=", pdh], ["owner_uuid", "=", project_uuid]]).execute(num_retries=arvRunner.num_retries) + if len(existing["items"]) == 0: + col.save_new(name=name, owner_uuid=project_uuid, ensure_unique_name=True) + + # now construct the wrapper + + step = { + "id": "#main/step", + "in": [], + "out": [], + "run": "keep:%s/workflow.json#main" % pdh + } + + wrapper = { + "class": "Workflow", + "id": "#main", + "inputs": main["inputs"], + "outputs": [], + "steps": [step] + } + + for i in main["inputs"]: + step["in"].append({ + "id": "#main/step/%s" % shortname(i["id"]), + "source": i["id"] + }) + + for i in main["outputs"]: + step["out"].append({"id": "#main/step/%s" % shortname(i["id"])}) + wrapper["outputs"].append({"outputSource": "#main/step/%s" % shortname(i["id"]), + "type": i["type"], + "id": i["id"]}) + + wrapper["requirements"] = [{"class": "SubworkflowFeatureRequirement"}] + + if main.get("requirements"): + wrapper["requirements"].extend(main["requirements"]) + if main.get("hints"): + wrapper["hints"] = main["hints"] + + return json.dumps({"cwlVersion": "v1.2", "$graph": [wrapper]}, sort_keys=True, indent=4, separators=(',',': ')) + +def upload_workflow(arvRunner, tool, job_order, project_uuid, + runtimeContext, uuid=None, submit_runner_ram=0, name=None, merged_map=None, - submit_runner_image=None): + submit_runner_image=None, + git_info=None): - packed = packed_workflow(arvRunner, tool, merged_map) + packed = packed_workflow(arvRunner, tool, merged_map, runtimeContext, git_info) adjustDirObjs(job_order, trim_listing) adjustFileObjs(job_order, trim_anonymous_location) @@ -57,7 +112,8 @@ def upload_workflow(arvRunner, tool, job_order, project_uuid, uuid=None, name = tool.tool.get("label", os.path.basename(tool.tool["id"])) upload_dependencies(arvRunner, name, tool.doc_loader, - packed, tool.tool["id"], False) + packed, tool.tool["id"], False, + runtimeContext) wf_runner_resources = None @@ -72,18 +128,22 @@ def upload_workflow(arvRunner, tool, job_order, project_uuid, uuid=None, wf_runner_resources = {"class": "http://arvados.org/cwl#WorkflowRunnerResources"} hints.append(wf_runner_resources) - wf_runner_resources["acrContainerImage"] = arvados_jobs_image(arvRunner, submit_runner_image or "arvados/jobs:"+__version__) + wf_runner_resources["acrContainerImage"] = arvados_jobs_image(arvRunner, + submit_runner_image or "arvados/jobs:"+__version__, + runtimeContext) if submit_runner_ram: wf_runner_resources["ramMin"] = submit_runner_ram main["hints"] = hints + wrapper = make_wrapper_workflow(arvRunner, main, packed, project_uuid, name) + body = { "workflow": { "name": name, "description": tool.tool.get("doc", ""), - "definition":json.dumps(packed, sort_keys=True, indent=4, separators=(',',': ')) + "definition": wrapper }} if project_uuid: body["workflow"]["owner_uuid"] = project_uuid @@ -142,8 +202,13 @@ class ArvadosWorkflowStep(WorkflowStep): **argv ): # type: (...) -> None - super(ArvadosWorkflowStep, self).__init__(toolpath_object, pos, loadingContext, *argc, **argv) - self.tool["class"] = "WorkflowStep" + if arvrunner.fast_submit: + self.tool = toolpath_object + self.tool["inputs"] = [] + self.tool["outputs"] = [] + else: + super(ArvadosWorkflowStep, self).__init__(toolpath_object, pos, loadingContext, *argc, **argv) + self.tool["class"] = "WorkflowStep" self.arvrunner = arvrunner def job(self, joborder, output_callback, runtimeContext): @@ -194,7 +259,8 @@ class ArvadosWorkflow(Workflow): self.doc_loader, joborder, joborder.get("id", "#"), - False) + False, + runtimeContext) if self.wf_pdh is None: packed = pack(self.loadingContext, self.tool["id"], loader=self.doc_loader) @@ -237,7 +303,8 @@ class ArvadosWorkflow(Workflow): self.doc_loader, packed, self.tool["id"], - False) + False, + runtimeContext) # Discover files/directories referenced by the # workflow (mainly "default" values) @@ -301,7 +368,7 @@ class ArvadosWorkflow(Workflow): if self.wf_pdh is None: adjustFileObjs(packed, keepmount) adjustDirObjs(packed, keepmount) - self.wf_pdh = upload_workflow_collection(self.arvrunner, shortname(self.tool["id"]), packed) + self.wf_pdh = upload_workflow_collection(self.arvrunner, shortname(self.tool["id"]), packed, runtimeContext) self.loadingContext = self.loadingContext.copy() self.loadingContext.metadata = self.loadingContext.metadata.copy()