6 import ruamel.yaml as yaml
8 from cwltool.errors import WorkflowException
9 from cwltool.process import get_feature, UnsupportedRequirement, shortname
10 from cwltool.pathmapper import adjustFileObjs, adjustDirObjs
11 from cwltool.utils import aslist
13 import arvados.collection
15 from .arvdocker import arv_docker_get_image
17 from .runner import Runner, arvados_jobs_image, packed_workflow, trim_anonymous_location
18 from .fsaccess import CollectionFetcher
19 from .pathmapper import NoFollowPathMapper, trim_listing
20 from .perf import Perf
22 logger = logging.getLogger('arvados.cwl-runner')
23 metrics = logging.getLogger('arvados.cwl-runner.metrics')
25 class ArvadosContainer(object):
26 """Submit and manage a Crunch container request for executing a CWL CommandLineTool."""
28 def __init__(self, runner):
29 self.arvrunner = runner
33 def update_pipeline_component(self, r):
36 def run(self, dry_run=False, pull_image=True, **kwargs):
38 "command": self.command_line,
39 "owner_uuid": self.arvrunner.project_uuid,
41 "output_path": self.outdir,
47 runtime_constraints = {}
49 resources = self.builder.resources
50 if resources is not None:
51 runtime_constraints["vcpus"] = resources.get("cores", 1)
52 runtime_constraints["ram"] = resources.get("ram") * 2**20
57 "capacity": resources.get("outdirSize", 0) * 2**20
61 "capacity": resources.get("tmpdirSize", 0) * 2**20
64 scheduling_parameters = {}
66 rf = [self.pathmapper.mapper(f) for f in self.pathmapper.referenced_files]
67 rf.sort(key=lambda k: k.resolved)
69 for resolved, target, tp, stg in rf:
72 if prevdir and target.startswith(prevdir):
77 targetdir = os.path.dirname(target)
78 sp = resolved.split("/", 1)
79 pdh = sp[0][5:] # remove "keep:"
82 "portable_data_hash": pdh
88 path = os.path.dirname(sp[1])
89 if path and path != "/":
90 mounts[targetdir]["path"] = path
91 prevdir = targetdir + "/"
93 with Perf(metrics, "generatefiles %s" % self.name):
94 if self.generatefiles["listing"]:
95 vwd = arvados.collection.Collection(api_client=self.arvrunner.api,
96 keep_client=self.arvrunner.keep_client,
97 num_retries=self.arvrunner.num_retries)
98 generatemapper = NoFollowPathMapper([self.generatefiles], "", "",
101 with Perf(metrics, "createfiles %s" % self.name):
102 for f, p in generatemapper.items():
105 elif p.type in ("File", "Directory"):
106 source, path = self.arvrunner.fs_access.get_collection(p.resolved)
107 vwd.copy(path, p.target, source_collection=source)
108 elif p.type == "CreateFile":
109 with vwd.open(p.target, "w") as n:
110 n.write(p.resolved.encode("utf-8"))
112 with Perf(metrics, "generatefiles.save_new %s" % self.name):
115 for f, p in generatemapper.items():
118 mountpoint = "%s/%s" % (self.outdir, p.target)
119 mounts[mountpoint] = {"kind": "collection",
120 "portable_data_hash": vwd.portable_data_hash(),
123 container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
125 container_request["environment"].update(self.environment)
128 sp = self.stdin[6:].split("/", 1)
129 mounts["stdin"] = {"kind": "collection",
130 "portable_data_hash": sp[0],
134 mounts["stderr"] = {"kind": "file",
135 "path": "%s/%s" % (self.outdir, self.stderr)}
138 mounts["stdout"] = {"kind": "file",
139 "path": "%s/%s" % (self.outdir, self.stdout)}
141 (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
143 docker_req = {"dockerImageId": "arvados/jobs"}
145 container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
148 self.arvrunner.project_uuid)
150 api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement")
152 runtime_constraints["API"] = True
154 runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
156 if "keep_cache" in runtime_req:
157 runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"] * 2**20
158 if "outputDirType" in runtime_req:
159 if runtime_req["outputDirType"] == "local_output_dir":
160 # Currently the default behavior.
162 elif runtime_req["outputDirType"] == "keep_output_dir":
163 mounts[self.outdir]= {
164 "kind": "collection",
168 partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
170 scheduling_parameters["partitions"] = aslist(partition_req["partition"])
172 intermediate_output_req, _ = get_feature(self, "http://arvados.org/cwl#IntermediateOutput")
173 if intermediate_output_req:
174 self.output_ttl = intermediate_output_req["outputTTL"]
176 self.output_ttl = self.arvrunner.intermediate_output_ttl
178 if self.output_ttl < 0:
179 raise WorkflowError("Invalid value %d for output_ttl, cannot be less than zero" % container_request["output_ttl"])
181 container_request["output_ttl"] = self.output_ttl
182 container_request["mounts"] = mounts
183 container_request["runtime_constraints"] = runtime_constraints
184 container_request["use_existing"] = kwargs.get("enable_reuse", True)
185 container_request["scheduling_parameters"] = scheduling_parameters
187 if kwargs.get("runnerjob", "").startswith("arvwf:"):
188 wfuuid = kwargs["runnerjob"][6:kwargs["runnerjob"].index("#")]
189 wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
190 if container_request["name"] == "main":
191 container_request["name"] = wfrecord["name"]
192 container_request["properties"]["template_uuid"] = wfuuid
195 response = self.arvrunner.api.container_requests().create(
196 body=container_request
197 ).execute(num_retries=self.arvrunner.num_retries)
199 self.uuid = response["uuid"]
200 self.arvrunner.processes[self.uuid] = self
202 if response["state"] == "Final":
203 logger.info("%s reused container %s", self.arvrunner.label(self), response["container_uuid"])
206 logger.info("%s %s state is %s", self.arvrunner.label(self), response["uuid"], response["state"])
207 except Exception as e:
208 logger.error("%s got error %s" % (self.arvrunner.label(self), str(e)))
209 self.output_callback({}, "permanentFail")
211 def done(self, record):
214 self.arvrunner.add_intermediate_output(record["output_uuid"])
216 container = self.arvrunner.api.containers().get(
217 uuid=record["container_uuid"]
218 ).execute(num_retries=self.arvrunner.num_retries)
219 if container["state"] == "Complete":
220 rcode = container["exit_code"]
221 if self.successCodes and rcode in self.successCodes:
222 processStatus = "success"
223 elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
224 processStatus = "temporaryFail"
225 elif self.permanentFailCodes and rcode in self.permanentFailCodes:
226 processStatus = "permanentFail"
228 processStatus = "success"
230 processStatus = "permanentFail"
232 processStatus = "permanentFail"
234 if processStatus == "permanentFail":
235 logc = arvados.collection.CollectionReader(container["log"],
236 api_client=self.arvrunner.api,
237 keep_client=self.arvrunner.keep_client,
238 num_retries=self.arvrunner.num_retries)
239 done.logtail(logc, logger, "%s error log:" % self.arvrunner.label(self))
241 if container["output"]:
242 outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep")
243 except WorkflowException as e:
244 logger.error("%s unable to collect output from %s:\n%s",
245 self.arvrunner.label(self), container["output"], e, exc_info=(e if self.arvrunner.debug else False))
246 processStatus = "permanentFail"
247 except Exception as e:
248 logger.exception("%s while getting output object: %s", self.arvrunner.label(self), e)
249 processStatus = "permanentFail"
251 self.output_callback(outputs, processStatus)
252 if record["uuid"] in self.arvrunner.processes:
253 del self.arvrunner.processes[record["uuid"]]
256 class RunnerContainer(Runner):
257 """Submit and manage a container that runs arvados-cwl-runner."""
259 def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
260 """Create an Arvados container request for this workflow.
262 The returned dict can be used to create a container passed as
263 the +body+ argument to container_requests().create().
266 adjustDirObjs(self.job_order, trim_listing)
267 adjustFileObjs(self.job_order, trim_anonymous_location)
268 adjustDirObjs(self.job_order, trim_anonymous_location)
271 "owner_uuid": self.arvrunner.project_uuid,
273 "output_path": "/var/spool/cwl",
274 "cwd": "/var/spool/cwl",
276 "state": "Committed",
277 "container_image": arvados_jobs_image(self.arvrunner, self.jobs_image),
279 "/var/lib/cwl/cwl.input.json": {
281 "content": self.job_order
285 "path": "/var/spool/cwl/cwl.output.json"
288 "kind": "collection",
292 "runtime_constraints": {
294 "ram": 1024*1024 * self.submit_runner_ram,
300 if self.tool.tool.get("id", "").startswith("keep:"):
301 sp = self.tool.tool["id"].split('/')
302 workflowcollection = sp[0][5:]
303 workflowname = "/".join(sp[1:])
304 workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
305 container_req["mounts"]["/var/lib/cwl/workflow"] = {
306 "kind": "collection",
307 "portable_data_hash": "%s" % workflowcollection
310 packed = packed_workflow(self.arvrunner, self.tool)
311 workflowpath = "/var/lib/cwl/workflow.json#main"
312 container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
316 if self.tool.tool.get("id", "").startswith("arvwf:"):
317 container_req["properties"]["template_uuid"] = self.tool.tool["id"][6:33]
319 command = ["arvados-cwl-runner", "--local", "--api=containers", "--no-log-timestamps"]
321 command.append("--output-name=" + self.output_name)
322 container_req["output_name"] = self.output_name
325 command.append("--output-tags=" + self.output_tags)
327 if kwargs.get("debug"):
328 command.append("--debug")
330 if self.enable_reuse:
331 command.append("--enable-reuse")
333 command.append("--disable-reuse")
336 command.append("--on-error=" + self.on_error)
338 if self.intermediate_output_ttl:
339 command.append("--intermediate-output-ttl=%d" % self.intermediate_output_ttl)
341 command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
343 container_req["command"] = command
348 def run(self, *args, **kwargs):
349 kwargs["keepprefix"] = "keep:"
350 job_spec = self.arvados_job_spec(*args, **kwargs)
351 job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
353 response = self.arvrunner.api.container_requests().create(
355 ).execute(num_retries=self.arvrunner.num_retries)
357 self.uuid = response["uuid"]
358 self.arvrunner.processes[self.uuid] = self
360 logger.info("%s submitted container %s", self.arvrunner.label(self), response["uuid"])
362 if response["state"] == "Final":
365 def done(self, record):
367 container = self.arvrunner.api.containers().get(
368 uuid=record["container_uuid"]
369 ).execute(num_retries=self.arvrunner.num_retries)
370 except Exception as e:
371 logger.exception("%s while getting runner container: %s", self.arvrunner.label(self), e)
372 self.arvrunner.output_callback({}, "permanentFail")
374 super(RunnerContainer, self).done(container)
376 if record["uuid"] in self.arvrunner.processes:
377 del self.arvrunner.processes[record["uuid"]]