5 from cwltool.errors import WorkflowException
6 from cwltool.process import get_feature, UnsupportedRequirement, shortname
7 from cwltool.pathmapper import adjustFiles
8 from cwltool.utils import aslist
10 import arvados.collection
12 from .arvdocker import arv_docker_get_image
14 from .runner import Runner
16 logger = logging.getLogger('arvados.cwl-runner')
18 class ArvadosContainer(object):
19 """Submit and manage a Crunch container request for executing a CWL CommandLineTool."""
21 def __init__(self, runner):
22 self.arvrunner = runner
26 def update_pipeline_component(self, r):
29 def run(self, dry_run=False, pull_image=True, **kwargs):
31 "command": self.command_line,
32 "owner_uuid": self.arvrunner.project_uuid,
34 "output_path": self.outdir,
39 runtime_constraints = {}
47 for f in self.pathmapper.files():
48 _, p, tp = self.pathmapper.mapper(f)
49 if tp == "Directory" and '/' not in p[6:]:
52 "portable_data_hash": p[6:]
55 for f in self.pathmapper.files():
56 _, p, tp = self.pathmapper.mapper(f)
57 if p[6:].split("/")[0] not in dirs:
60 "portable_data_hash": p[6:]
63 if self.generatefiles["listing"]:
64 raise UnsupportedRequirement("Generate files not supported")
66 container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
68 container_request["environment"].update(self.environment)
71 raise UnsupportedRequirement("Stdin redirection currently not suppported")
74 raise UnsupportedRequirement("Stderr redirection currently not suppported")
77 mounts["stdout"] = {"kind": "file",
78 "path": "%s/%s" % (self.outdir, self.stdout)}
80 (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
82 docker_req = {"dockerImageId": "arvados/jobs"}
84 container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
87 self.arvrunner.project_uuid)
89 resources = self.builder.resources
90 if resources is not None:
91 runtime_constraints["vcpus"] = resources.get("cores", 1)
92 runtime_constraints["ram"] = resources.get("ram") * 2**20
94 api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement")
96 runtime_constraints["API"] = True
98 runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
100 logger.warn("RuntimeConstraints not yet supported by container API")
102 partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
104 runtime_constraints["partition"] = aslist(partition_req["partition"])
106 container_request["mounts"] = mounts
107 container_request["runtime_constraints"] = runtime_constraints
110 response = self.arvrunner.api.container_requests().create(
111 body=container_request
112 ).execute(num_retries=self.arvrunner.num_retries)
114 self.arvrunner.processes[response["container_uuid"]] = self
116 logger.info("Container %s (%s) request state is %s", self.name, response["uuid"], response["state"])
118 if response["state"] == "Final":
120 except Exception as e:
121 logger.error("Got error %s" % str(e))
122 self.output_callback({}, "permanentFail")
124 def done(self, record):
126 if record["state"] == "Complete":
127 rcode = record["exit_code"]
128 if self.successCodes and rcode in self.successCodes:
129 processStatus = "success"
130 elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
131 processStatus = "temporaryFail"
132 elif self.permanentFailCodes and rcode in self.permanentFailCodes:
133 processStatus = "permanentFail"
135 processStatus = "success"
137 processStatus = "permanentFail"
139 processStatus = "permanentFail"
144 outputs = done.done(self, record, "/tmp", self.outdir, "/keep")
145 except WorkflowException as e:
146 logger.error("Error while collecting container outputs:\n%s", e, exc_info=(e if self.arvrunner.debug else False))
147 processStatus = "permanentFail"
148 except Exception as e:
149 logger.exception("Got unknown exception while collecting job outputs:")
150 processStatus = "permanentFail"
152 self.output_callback(outputs, processStatus)
154 del self.arvrunner.processes[record["uuid"]]
157 class RunnerContainer(Runner):
158 """Submit and manage a container that runs arvados-cwl-runner."""
160 def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
161 """Create an Arvados container request for this workflow.
163 The returned dict can be used to create a container passed as
164 the +body+ argument to container_requests().create().
167 workflowmapper = super(RunnerContainer, self).arvados_job_spec(dry_run=dry_run, pull_image=pull_image, **kwargs)
169 with arvados.collection.Collection(api_client=self.arvrunner.api,
170 keep_client=self.arvrunner.keep_client,
171 num_retries=self.arvrunner.num_retries) as jobobj:
172 with jobobj.open("cwl.input.json", "w") as f:
173 json.dump(self.job_order, f, sort_keys=True, indent=4)
174 jobobj.save_new(owner_uuid=self.arvrunner.project_uuid)
176 workflowname = os.path.basename(self.tool.tool["id"])
177 workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
178 workflowcollection = workflowmapper.mapper(self.tool.tool["id"])[1]
179 workflowcollection = workflowcollection[5:workflowcollection.index('/')]
180 jobpath = "/var/lib/cwl/job/cwl.input.json"
182 container_image = arv_docker_get_image(self.arvrunner.api,
183 {"dockerImageId": "arvados/jobs"},
185 self.arvrunner.project_uuid)
187 command = ["arvados-cwl-runner", "--local", "--api=containers"]
189 command.append("--output-name=" + self.output_name)
190 command.extend([workflowpath, jobpath])
194 "owner_uuid": self.arvrunner.project_uuid,
196 "output_path": "/var/spool/cwl",
197 "cwd": "/var/spool/cwl",
199 "state": "Committed",
200 "container_image": container_image,
202 "/var/lib/cwl/workflow": {
203 "kind": "collection",
204 "portable_data_hash": "%s" % workflowcollection
207 "kind": "collection",
208 "portable_data_hash": "%s/cwl.input.json" % jobobj.portable_data_hash()
212 "path": "/var/spool/cwl/cwl.output.json"
215 "kind": "collection",
219 "runtime_constraints": {
221 "ram": 1024*1024*256,
226 def run(self, *args, **kwargs):
227 kwargs["keepprefix"] = "keep:"
228 job_spec = self.arvados_job_spec(*args, **kwargs)
229 job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
231 response = self.arvrunner.api.container_requests().create(
233 ).execute(num_retries=self.arvrunner.num_retries)
235 self.uuid = response["uuid"]
236 self.arvrunner.processes[response["container_uuid"]] = self
238 logger.info("Submitted container %s", response["uuid"])
240 if response["state"] in ("Complete", "Failed", "Cancelled"):