5 import ruamel.yaml as yaml
7 from cwltool.errors import WorkflowException
8 from cwltool.process import get_feature, UnsupportedRequirement, shortname
9 from cwltool.pathmapper import adjustFiles, adjustDirObjs
10 from cwltool.utils import aslist
12 import arvados.collection
14 from .arvdocker import arv_docker_get_image
16 from .runner import Runner, arvados_jobs_image, packed_workflow, trim_listing
17 from .fsaccess import CollectionFetcher
18 from .pathmapper import NoFollowPathMapper
19 from .perf import Perf
21 logger = logging.getLogger('arvados.cwl-runner')
22 metrics = logging.getLogger('arvados.cwl-runner.metrics')
24 class ArvadosContainer(object):
25 """Submit and manage a Crunch container request for executing a CWL CommandLineTool."""
27 def __init__(self, runner):
28 self.arvrunner = runner
32 def update_pipeline_component(self, r):
35 def run(self, dry_run=False, pull_image=True, **kwargs):
37 "command": self.command_line,
38 "owner_uuid": self.arvrunner.project_uuid,
40 "output_path": self.outdir,
46 runtime_constraints = {}
52 scheduling_parameters = {}
55 for f in self.pathmapper.files():
56 pdh, p, tp = self.pathmapper.mapper(f)
57 if tp == "Directory" and '/' not in pdh:
60 "portable_data_hash": pdh[5:]
64 for f in self.pathmapper.files():
65 res, p, tp = self.pathmapper.mapper(f)
66 pdh, path = res[5:].split("/", 1)
70 "portable_data_hash": pdh
73 mounts[p]["path"] = path
75 with Perf(metrics, "generatefiles %s" % self.name):
76 if self.generatefiles["listing"]:
77 vwd = arvados.collection.Collection(api_client=self.arvrunner.api,
78 keep_client=self.arvrunner.keep_client,
79 num_retries=self.arvrunner.num_retries)
80 generatemapper = NoFollowPathMapper([self.generatefiles], "", "",
83 with Perf(metrics, "createfiles %s" % self.name):
84 for f, p in generatemapper.items():
87 elif p.type in ("File", "Directory"):
88 source, path = self.arvrunner.fs_access.get_collection(p.resolved)
89 vwd.copy(path, p.target, source_collection=source)
90 elif p.type == "CreateFile":
91 with vwd.open(p.target, "w") as n:
92 n.write(p.resolved.encode("utf-8"))
94 with Perf(metrics, "generatefiles.save_new %s" % self.name):
97 for f, p in generatemapper.items():
100 mountpoint = "%s/%s" % (self.outdir, p.target)
101 mounts[mountpoint] = {"kind": "collection",
102 "portable_data_hash": vwd.portable_data_hash(),
105 container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
107 container_request["environment"].update(self.environment)
110 raise UnsupportedRequirement("Stdin redirection currently not suppported")
113 raise UnsupportedRequirement("Stderr redirection currently not suppported")
116 mounts["stdout"] = {"kind": "file",
117 "path": "%s/%s" % (self.outdir, self.stdout)}
119 (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
121 docker_req = {"dockerImageId": "arvados/jobs"}
123 container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
126 self.arvrunner.project_uuid)
128 resources = self.builder.resources
129 if resources is not None:
130 runtime_constraints["vcpus"] = resources.get("cores", 1)
131 runtime_constraints["ram"] = resources.get("ram") * 2**20
133 api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement")
135 runtime_constraints["API"] = True
137 runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
139 if "keep_cache" in runtime_req:
140 runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"] * 2**20
142 partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
144 scheduling_parameters["partitions"] = aslist(partition_req["partition"])
146 container_request["mounts"] = mounts
147 container_request["runtime_constraints"] = runtime_constraints
148 container_request["use_existing"] = kwargs.get("enable_reuse", True)
149 container_request["scheduling_parameters"] = scheduling_parameters
151 if kwargs.get("runnerjob", "").startswith("arvwf:"):
152 wfuuid = kwargs["runnerjob"][6:kwargs["runnerjob"].index("#")]
153 wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
154 if container_request["name"] == "main":
155 container_request["name"] = wfrecord["name"]
156 container_request["properties"]["template_uuid"] = wfuuid
159 response = self.arvrunner.api.container_requests().create(
160 body=container_request
161 ).execute(num_retries=self.arvrunner.num_retries)
163 self.uuid = response["uuid"]
164 self.arvrunner.processes[self.uuid] = self
166 if response["state"] == "Final":
167 logger.info("%s reused container %s", self.arvrunner.label(self), response["container_uuid"])
170 logger.info("%s %s state is %s", self.arvrunner.label(self), response["uuid"], response["state"])
171 except Exception as e:
172 logger.error("%s got error %s" % (self.arvrunner.label(self), str(e)))
173 self.output_callback({}, "permanentFail")
175 def done(self, record):
177 container = self.arvrunner.api.containers().get(
178 uuid=record["container_uuid"]
179 ).execute(num_retries=self.arvrunner.num_retries)
180 if container["state"] == "Complete":
181 rcode = container["exit_code"]
182 if self.successCodes and rcode in self.successCodes:
183 processStatus = "success"
184 elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
185 processStatus = "temporaryFail"
186 elif self.permanentFailCodes and rcode in self.permanentFailCodes:
187 processStatus = "permanentFail"
189 processStatus = "success"
191 processStatus = "permanentFail"
193 processStatus = "permanentFail"
195 if processStatus == "permanentFail":
196 logc = arvados.collection.CollectionReader(container["log"],
197 api_client=self.arvrunner.api,
198 keep_client=self.arvrunner.keep_client,
199 num_retries=self.arvrunner.num_retries)
200 done.logtail(logc, logger, "%s error log:" % self.arvrunner.label(self))
203 if container["output"]:
204 outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep")
205 except WorkflowException as e:
206 logger.error("%s unable to collect output from %s:\n%s",
207 self.arvrunner.label(self), container["output"], e, exc_info=(e if self.arvrunner.debug else False))
208 processStatus = "permanentFail"
209 except Exception as e:
210 logger.exception("%s while getting output object: %s", self.arvrunner.label(self), e)
211 processStatus = "permanentFail"
213 self.output_callback(outputs, processStatus)
214 if record["uuid"] in self.arvrunner.processes:
215 del self.arvrunner.processes[record["uuid"]]
218 class RunnerContainer(Runner):
219 """Submit and manage a container that runs arvados-cwl-runner."""
221 def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
222 """Create an Arvados container request for this workflow.
224 The returned dict can be used to create a container passed as
225 the +body+ argument to container_requests().create().
228 adjustDirObjs(self.job_order, trim_listing)
231 "owner_uuid": self.arvrunner.project_uuid,
233 "output_path": "/var/spool/cwl",
234 "cwd": "/var/spool/cwl",
236 "state": "Committed",
237 "container_image": arvados_jobs_image(self.arvrunner, self.jobs_image),
239 "/var/lib/cwl/cwl.input.json": {
241 "content": self.job_order
245 "path": "/var/spool/cwl/cwl.output.json"
248 "kind": "collection",
252 "runtime_constraints": {
254 "ram": 1024*1024 * self.submit_runner_ram,
260 if self.tool.tool.get("id", "").startswith("keep:"):
261 sp = self.tool.tool["id"].split('/')
262 workflowcollection = sp[0][5:]
263 workflowname = "/".join(sp[1:])
264 workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
265 container_req["mounts"]["/var/lib/cwl/workflow"] = {
266 "kind": "collection",
267 "portable_data_hash": "%s" % workflowcollection
270 packed = packed_workflow(self.arvrunner, self.tool)
271 workflowpath = "/var/lib/cwl/workflow.json#main"
272 container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
276 if self.tool.tool.get("id", "").startswith("arvwf:"):
277 container_req["properties"]["template_uuid"] = self.tool.tool["id"][6:33]
279 command = ["arvados-cwl-runner", "--local", "--api=containers", "--no-log-timestamps"]
281 command.append("--output-name=" + self.output_name)
282 container_req["output_name"] = self.output_name
285 command.append("--output-tags=" + self.output_tags)
287 if kwargs.get("debug"):
288 command.append("--debug")
290 if self.enable_reuse:
291 command.append("--enable-reuse")
293 command.append("--disable-reuse")
296 command.append("--on-error=" + self.on_error)
298 command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
300 container_req["command"] = command
305 def run(self, *args, **kwargs):
306 kwargs["keepprefix"] = "keep:"
307 job_spec = self.arvados_job_spec(*args, **kwargs)
308 job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
310 response = self.arvrunner.api.container_requests().create(
312 ).execute(num_retries=self.arvrunner.num_retries)
314 self.uuid = response["uuid"]
315 self.arvrunner.processes[self.uuid] = self
317 logger.info("%s submitted container %s", self.arvrunner.label(self), response["uuid"])
319 if response["state"] == "Final":
322 def done(self, record):
324 container = self.arvrunner.api.containers().get(
325 uuid=record["container_uuid"]
326 ).execute(num_retries=self.arvrunner.num_retries)
327 except Exception as e:
328 logger.exception("%s while getting runner container: %s", self.arvrunner.label(self), e)
329 self.arvrunner.output_callback({}, "permanentFail")
331 super(RunnerContainer, self).done(container)
333 if record["uuid"] in self.arvrunner.processes:
334 del self.arvrunner.processes[record["uuid"]]