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,
42 runtime_constraints = {}
48 scheduling_parameters = {}
51 for f in self.pathmapper.files():
52 _, p, tp = self.pathmapper.mapper(f)
53 if tp == "Directory" and '/' not in p[6:]:
56 "portable_data_hash": p[6:]
59 for f in self.pathmapper.files():
60 _, p, tp = self.pathmapper.mapper(f)
61 if p[6:].split("/")[0] not in dirs:
64 "portable_data_hash": p[6:]
67 if self.generatefiles["listing"]:
68 raise UnsupportedRequirement("InitialWorkDirRequirement not supported with --api=containers")
70 container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
72 container_request["environment"].update(self.environment)
75 raise UnsupportedRequirement("Stdin redirection currently not suppported")
78 raise UnsupportedRequirement("Stderr redirection currently not suppported")
81 mounts["stdout"] = {"kind": "file",
82 "path": "%s/%s" % (self.outdir, self.stdout)}
84 (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
86 docker_req = {"dockerImageId": arvados_jobs_image(self.arvrunner)}
88 container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
91 self.arvrunner.project_uuid)
93 resources = self.builder.resources
94 if resources is not None:
95 runtime_constraints["vcpus"] = resources.get("cores", 1)
96 runtime_constraints["ram"] = resources.get("ram") * 2**20
98 api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement")
100 runtime_constraints["API"] = True
102 runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
104 if "keep_cache" in runtime_req:
105 runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"]
107 partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
109 scheduling_parameters["partitions"] = aslist(partition_req["partition"])
111 container_request["mounts"] = mounts
112 container_request["runtime_constraints"] = runtime_constraints
113 container_request["use_existing"] = kwargs.get("enable_reuse", True)
114 container_request["scheduling_parameters"] = scheduling_parameters
117 response = self.arvrunner.api.container_requests().create(
118 body=container_request
119 ).execute(num_retries=self.arvrunner.num_retries)
121 self.arvrunner.processes[response["uuid"]] = self
123 logger.info("Container request %s (%s) state is %s", self.name, response["uuid"], response["state"])
125 if response["state"] == "Final":
127 except Exception as e:
128 logger.error("Got error %s" % str(e))
129 self.output_callback({}, "permanentFail")
131 def done(self, record):
133 container = self.arvrunner.api.containers().get(
134 uuid=record["container_uuid"]
135 ).execute(num_retries=self.arvrunner.num_retries)
136 if container["state"] == "Complete":
137 rcode = container["exit_code"]
138 if self.successCodes and rcode in self.successCodes:
139 processStatus = "success"
140 elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
141 processStatus = "temporaryFail"
142 elif self.permanentFailCodes and rcode in self.permanentFailCodes:
143 processStatus = "permanentFail"
145 processStatus = "success"
147 processStatus = "permanentFail"
149 processStatus = "permanentFail"
153 if container["output"]:
155 outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep")
156 except Exception as e:
157 logger.error("Got error %s" % str(e))
158 self.output_callback({}, "permanentFail")
159 self.output_callback(outputs, processStatus)
161 del self.arvrunner.processes[record["uuid"]]
164 class RunnerContainer(Runner):
165 """Submit and manage a container that runs arvados-cwl-runner."""
167 def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
168 """Create an Arvados container request for this workflow.
170 The returned dict can be used to create a container passed as
171 the +body+ argument to container_requests().create().
174 workflowmapper = super(RunnerContainer, self).arvados_job_spec(dry_run=dry_run, pull_image=pull_image, **kwargs)
177 "owner_uuid": self.arvrunner.project_uuid,
179 "output_path": "/var/spool/cwl",
180 "cwd": "/var/spool/cwl",
182 "state": "Committed",
183 "container_image": arvados_jobs_image(self.arvrunner),
185 "/var/lib/cwl/cwl.input.json": {
187 "content": self.job_order
191 "path": "/var/spool/cwl/cwl.output.json"
194 "kind": "collection",
198 "runtime_constraints": {
200 "ram": 1024*1024 * self.submit_runner_ram,
205 workflowcollection = workflowmapper.mapper(self.tool.tool["id"])[1]
206 if workflowcollection.startswith("keep:"):
207 workflowcollection = workflowcollection[5:workflowcollection.index('/')]
208 workflowname = os.path.basename(self.tool.tool["id"])
209 workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
210 container_req["mounts"]["/var/lib/cwl/workflow"] = {
211 "kind": "collection",
212 "portable_data_hash": "%s" % workflowcollection
214 elif workflowcollection.startswith("arvwf:"):
215 workflowpath = "/var/lib/cwl/workflow.json#main"
216 fetcher = CollectionFetcher({}, None,
217 api_client=self.arvrunner.api,
218 keep_client=self.arvrunner.keep_client)
219 wfobj = yaml.safe_load(fetcher.fetch_text(workflowcollection))
220 container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
225 command = ["arvados-cwl-runner", "--local", "--api=containers"]
227 command.append("--output-name=" + self.output_name)
230 command.append("--output-tags=" + self.output_tags)
232 if self.enable_reuse:
233 command.append("--enable-reuse")
235 command.append("--disable-reuse")
237 command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
239 container_req["command"] = command
244 def run(self, *args, **kwargs):
245 kwargs["keepprefix"] = "keep:"
246 job_spec = self.arvados_job_spec(*args, **kwargs)
247 job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
249 response = self.arvrunner.api.container_requests().create(
251 ).execute(num_retries=self.arvrunner.num_retries)
253 self.uuid = response["uuid"]
254 self.arvrunner.processes[response["uuid"]] = self
256 logger.info("Submitted container %s", response["uuid"])
258 if response["state"] == "Final":
261 def done(self, record):
263 container = self.arvrunner.api.containers().get(
264 uuid=record["container_uuid"]
265 ).execute(num_retries=self.arvrunner.num_retries)
266 except Exception as e:
267 logger.exception("While getting runner container: %s", e)
268 self.arvrunner.output_callback({}, "permanentFail")
269 del self.arvrunner.processes[record["uuid"]]
271 super(RunnerContainer, self).done(container)
273 del self.arvrunner.processes[record["uuid"]]