X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/691a08fd486e53ce656ae20fea84c2b3ec1a1088..12e22cbf3bbf047c001624be54ec802999bb5c1e:/sdk/cwl/arvados_cwl/arvcontainer.py?ds=sidebyside diff --git a/sdk/cwl/arvados_cwl/arvcontainer.py b/sdk/cwl/arvados_cwl/arvcontainer.py index 94a7579f20..fa32456c3f 100644 --- a/sdk/cwl/arvados_cwl/arvcontainer.py +++ b/sdk/cwl/arvados_cwl/arvcontainer.py @@ -3,7 +3,8 @@ import json import os from cwltool.errors import WorkflowException -from cwltool.process import get_feature, adjustFiles, UnsupportedRequirement, shortname +from cwltool.process import get_feature, UnsupportedRequirement, shortname +from cwltool.pathmapper import adjustFiles import arvados.collection @@ -29,38 +30,51 @@ class ArvadosContainer(object): "command": self.command_line, "owner_uuid": self.arvrunner.project_uuid, "name": self.name, - "output_path": "/var/spool/cwl", - "cwd": "/var/spool/cwl", + "output_path": self.outdir, + "cwd": self.outdir, "priority": 1, "state": "Committed" } runtime_constraints = {} mounts = { - "/var/spool/cwl": { + self.outdir: { "kind": "tmp" } } + dirs = set() for f in self.pathmapper.files(): - _, p = self.pathmapper.mapper(f) - mounts[p] = { - "kind": "collection", - "portable_data_hash": p[6:] - } + _, p, tp = self.pathmapper.mapper(f) + if tp == "Directory" and '/' not in p[6:]: + mounts[p] = { + "kind": "collection", + "portable_data_hash": p[6:] + } + dirs.add(p[6:]) + for f in self.pathmapper.files(): + _, p, tp = self.pathmapper.mapper(f) + if p[6:].split("/")[0] not in dirs: + mounts[p] = { + "kind": "collection", + "portable_data_hash": p[6:] + } - if self.generatefiles: + if self.generatefiles["listing"]: raise UnsupportedRequirement("Generate files not supported") - container_request["environment"] = {"TMPDIR": "/tmp"} + container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir} if self.environment: container_request["environment"].update(self.environment) if self.stdin: raise UnsupportedRequirement("Stdin redirection currently not suppported") + if self.stderr: + raise UnsupportedRequirement("Stderr redirection currently not suppported") + if self.stdout: mounts["stdout"] = {"kind": "file", - "path": "/var/spool/cwl/%s" % (self.stdout)} + "path": "%s/%s" % (self.outdir, self.stdout)} (docker_req, docker_is_req) = get_feature(self, "DockerRequirement") if not docker_req: @@ -76,6 +90,14 @@ class ArvadosContainer(object): runtime_constraints["vcpus"] = resources.get("cores", 1) runtime_constraints["ram"] = resources.get("ram") * 2**20 + api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement") + if api_req: + runtime_constraints["API"] = True + + runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints") + if runtime_req: + logger.warn("RuntimeConstraints not yet supported by container API") + container_request["mounts"] = mounts container_request["runtime_constraints"] = runtime_constraints @@ -86,7 +108,7 @@ class ArvadosContainer(object): self.arvrunner.processes[response["container_uuid"]] = self - logger.info("Container %s (%s) request state is %s", self.name, response["container_uuid"], response["state"]) + logger.info("Container %s (%s) request state is %s", self.name, response["uuid"], response["state"]) if response["state"] == "Final": self.done(response) @@ -114,7 +136,7 @@ class ArvadosContainer(object): try: outputs = {} if record["output"]: - outputs = done.done(self, record, "/tmp", "/var/spool/cwl", "/keep") + outputs = done.done(self, record, "/tmp", self.outdir, "/keep") except WorkflowException as e: logger.error("Error while collecting container outputs:\n%s", e, exc_info=(e if self.arvrunner.debug else False)) processStatus = "permanentFail" @@ -139,7 +161,9 @@ class RunnerContainer(Runner): workflowmapper = super(RunnerContainer, self).arvados_job_spec(dry_run=dry_run, pull_image=pull_image, **kwargs) - with arvados.collection.Collection(api_client=self.arvrunner.api) as jobobj: + with arvados.collection.Collection(api_client=self.arvrunner.api, + keep_client=self.arvrunner.keep_client, + num_retries=self.arvrunner.num_retries) as jobobj: with jobobj.open("cwl.input.json", "w") as f: json.dump(self.job_order, f, sort_keys=True, indent=4) jobobj.save_new(owner_uuid=self.arvrunner.project_uuid) @@ -155,8 +179,13 @@ class RunnerContainer(Runner): pull_image, self.arvrunner.project_uuid) + command = ["arvados-cwl-runner", "--local", "--api=containers"] + if self.output_name: + command.append("--output-name=" + self.output_name) + command.extend([workflowpath, jobpath]) + return { - "command": ["arvados-cwl-runner", "--local", "--api=containers", workflowpath, jobpath], + "command": command, "owner_uuid": self.arvrunner.project_uuid, "name": self.name, "output_path": "/var/spool/cwl",