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
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
17 from .fsaccess import CollectionFetcher
19 logger = logging.getLogger('arvados.cwl-runner')
21 class ArvadosContainer(object):
22 """Submit and manage a Crunch container request for executing a CWL CommandLineTool."""
24 def __init__(self, runner):
25 self.arvrunner = runner
29 def update_pipeline_component(self, r):
32 def run(self, dry_run=False, pull_image=True, **kwargs):
34 "command": self.command_line,
35 "owner_uuid": self.arvrunner.project_uuid,
37 "output_path": self.outdir,
43 runtime_constraints = {}
49 scheduling_parameters = {}
52 for f in self.pathmapper.files():
53 _, p, tp = self.pathmapper.mapper(f)
54 if tp == "Directory" and '/' not in p[6:]:
57 "portable_data_hash": p[6:]
60 for f in self.pathmapper.files():
61 _, p, tp = self.pathmapper.mapper(f)
62 if p[6:].split("/")[0] not in dirs:
65 "portable_data_hash": p[6:]
68 if self.generatefiles["listing"]:
69 raise UnsupportedRequirement("InitialWorkDirRequirement not supported with --api=containers")
71 container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
73 container_request["environment"].update(self.environment)
76 raise UnsupportedRequirement("Stdin redirection currently not suppported")
79 raise UnsupportedRequirement("Stderr redirection currently not suppported")
82 mounts["stdout"] = {"kind": "file",
83 "path": "%s/%s" % (self.outdir, self.stdout)}
85 (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
87 docker_req = {"dockerImageId": arvados_jobs_image(self.arvrunner)}
89 container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
92 self.arvrunner.project_uuid)
94 resources = self.builder.resources
95 if resources is not None:
96 runtime_constraints["vcpus"] = resources.get("cores", 1)
97 runtime_constraints["ram"] = resources.get("ram") * 2**20
99 api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement")
101 runtime_constraints["API"] = True
103 runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
105 if "keep_cache" in runtime_req:
106 runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"]
108 partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
110 scheduling_parameters["partitions"] = aslist(partition_req["partition"])
112 container_request["mounts"] = mounts
113 container_request["runtime_constraints"] = runtime_constraints
114 container_request["use_existing"] = kwargs.get("enable_reuse", True)
115 container_request["scheduling_parameters"] = scheduling_parameters
117 if kwargs.get("runnerjob", "").startswith("arvwf:"):
118 wfuuid = kwargs["runnerjob"][6:kwargs["runnerjob"].index("#")]
119 wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
120 if container_request["name"] == "main":
121 container_request["name"] = wfrecord["name"]
122 container_request["properties"]["template_uuid"] = wfuuid
125 response = self.arvrunner.api.container_requests().create(
126 body=container_request
127 ).execute(num_retries=self.arvrunner.num_retries)
129 self.uuid = response["uuid"]
130 self.arvrunner.processes[self.uuid] = self
132 logger.info("%s %s state is %s", self.arvrunner.label(self), response["uuid"], response["state"])
134 if response["state"] == "Final":
136 except Exception as e:
137 logger.error("%s got error %s" % (self.arvrunner.label(self), str(e)))
138 self.output_callback({}, "permanentFail")
140 def done(self, record):
142 container = self.arvrunner.api.containers().get(
143 uuid=record["container_uuid"]
144 ).execute(num_retries=self.arvrunner.num_retries)
145 if container["state"] == "Complete":
146 rcode = container["exit_code"]
147 if self.successCodes and rcode in self.successCodes:
148 processStatus = "success"
149 elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
150 processStatus = "temporaryFail"
151 elif self.permanentFailCodes and rcode in self.permanentFailCodes:
152 processStatus = "permanentFail"
154 processStatus = "success"
156 processStatus = "permanentFail"
158 processStatus = "permanentFail"
160 if processStatus == "permanentFail":
161 logc = arvados.collection.CollectionReader(container["log"],
162 api_client=self.arvrunner.api,
163 keep_client=self.arvrunner.keep_client,
164 num_retries=self.arvrunner.num_retries)
165 done.logtail(logc, logger, self.arvrunner.label(self) + " ")
168 if container["output"]:
169 outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep")
170 except WorkflowException as e:
171 logger.error("%s unable to collect output from %s:\n%s",
172 self.arvrunner.label(self), container["output"], e, exc_info=(e if self.arvrunner.debug else False))
173 processStatus = "permanentFail"
174 except Exception as e:
175 logger.exception("%s while getting output object: %s", self.arvrunner.label(self), e)
176 self.output_callback({}, "permanentFail")
178 self.output_callback(outputs, processStatus)
180 if record["uuid"] in self.arvrunner.processes:
181 del self.arvrunner.processes[record["uuid"]]
184 class RunnerContainer(Runner):
185 """Submit and manage a container that runs arvados-cwl-runner."""
187 def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
188 """Create an Arvados container request for this workflow.
190 The returned dict can be used to create a container passed as
191 the +body+ argument to container_requests().create().
194 workflowmapper = super(RunnerContainer, self).arvados_job_spec(dry_run=dry_run, pull_image=pull_image, **kwargs)
197 "owner_uuid": self.arvrunner.project_uuid,
199 "output_path": "/var/spool/cwl",
200 "cwd": "/var/spool/cwl",
202 "state": "Committed",
203 "container_image": arvados_jobs_image(self.arvrunner),
205 "/var/lib/cwl/cwl.input.json": {
207 "content": self.job_order
211 "path": "/var/spool/cwl/cwl.output.json"
214 "kind": "collection",
218 "runtime_constraints": {
220 "ram": 1024*1024 * self.submit_runner_ram,
226 workflowcollection = workflowmapper.mapper(self.tool.tool["id"])[1]
227 if workflowcollection.startswith("keep:"):
228 workflowcollection = workflowcollection[5:workflowcollection.index('/')]
229 workflowname = os.path.basename(self.tool.tool["id"])
230 workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
231 container_req["mounts"]["/var/lib/cwl/workflow"] = {
232 "kind": "collection",
233 "portable_data_hash": "%s" % workflowcollection
235 elif workflowcollection.startswith("arvwf:"):
236 workflowpath = "/var/lib/cwl/workflow.json#main"
237 wfuuid = workflowcollection[6:workflowcollection.index("#")]
238 wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
239 wfobj = yaml.safe_load(wfrecord["definition"])
240 if container_req["name"].startswith("arvwf:"):
241 container_req["name"] = wfrecord["name"]
242 container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
246 container_req["properties"]["template_uuid"] = wfuuid
248 command = ["arvados-cwl-runner", "--local", "--api=containers", "--no-log-timestamps"]
250 command.append("--output-name=" + self.output_name)
253 command.append("--output-tags=" + self.output_tags)
255 if self.enable_reuse:
256 command.append("--enable-reuse")
258 command.append("--disable-reuse")
260 command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
262 container_req["command"] = command
267 def run(self, *args, **kwargs):
268 kwargs["keepprefix"] = "keep:"
269 job_spec = self.arvados_job_spec(*args, **kwargs)
270 job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
272 response = self.arvrunner.api.container_requests().create(
274 ).execute(num_retries=self.arvrunner.num_retries)
276 self.uuid = response["uuid"]
277 self.arvrunner.processes[self.uuid] = self
279 logger.info("%s submitted container %s", self.arvrunner.label(self), response["uuid"])
281 if response["state"] == "Final":
284 def done(self, record):
286 container = self.arvrunner.api.containers().get(
287 uuid=record["container_uuid"]
288 ).execute(num_retries=self.arvrunner.num_retries)
289 except Exception as e:
290 logger.exception("%s while getting runner container: %s", self.arvrunner.label(self), e)
291 self.arvrunner.output_callback({}, "permanentFail")
293 super(RunnerContainer, self).done(container)
295 if record["uuid"] in self.arvrunner.processes:
296 del self.arvrunner.processes[record["uuid"]]