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, arvados_jobs_image
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 = {}
45 scheduling_parameters = {}
48 for f in self.pathmapper.files():
49 _, p, tp = self.pathmapper.mapper(f)
50 if tp == "Directory" and '/' not in p[6:]:
53 "portable_data_hash": p[6:]
56 for f in self.pathmapper.files():
57 _, p, tp = self.pathmapper.mapper(f)
58 if p[6:].split("/")[0] not in dirs:
61 "portable_data_hash": p[6:]
64 if self.generatefiles["listing"]:
65 raise UnsupportedRequirement("InitialWorkDirRequirement not supported with --api=containers")
67 container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
69 container_request["environment"].update(self.environment)
72 raise UnsupportedRequirement("Stdin redirection currently not suppported")
75 raise UnsupportedRequirement("Stderr redirection currently not suppported")
78 mounts["stdout"] = {"kind": "file",
79 "path": "%s/%s" % (self.outdir, self.stdout)}
81 (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
83 docker_req = {"dockerImageId": arvados_jobs_image(self.arvrunner)}
85 container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
88 self.arvrunner.project_uuid)
90 resources = self.builder.resources
91 if resources is not None:
92 runtime_constraints["vcpus"] = resources.get("cores", 1)
93 runtime_constraints["ram"] = resources.get("ram") * 2**20
95 api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement")
97 runtime_constraints["API"] = True
99 runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
101 if "keep_cache" in runtime_req:
102 runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"]
104 partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
106 scheduling_parameters["partitions"] = aslist(partition_req["partition"])
108 container_request["mounts"] = mounts
109 container_request["runtime_constraints"] = runtime_constraints
110 container_request["use_existing"] = kwargs.get("enable_reuse", True)
111 container_request["scheduling_parameters"] = scheduling_parameters
114 response = self.arvrunner.api.container_requests().create(
115 body=container_request
116 ).execute(num_retries=self.arvrunner.num_retries)
118 self.arvrunner.processes[response["uuid"]] = self
120 logger.info("Container request %s (%s) state is %s", self.name, response["uuid"], response["state"])
122 if response["state"] == "Final":
124 except Exception as e:
125 logger.error("Got error %s" % str(e))
126 self.output_callback({}, "permanentFail")
128 def done(self, record):
130 container = self.arvrunner.api.containers().get(
131 uuid=record["container_uuid"]
132 ).execute(num_retries=self.arvrunner.num_retries)
133 if container["state"] == "Complete":
134 rcode = container["exit_code"]
135 if self.successCodes and rcode in self.successCodes:
136 processStatus = "success"
137 elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
138 processStatus = "temporaryFail"
139 elif self.permanentFailCodes and rcode in self.permanentFailCodes:
140 processStatus = "permanentFail"
142 processStatus = "success"
144 processStatus = "permanentFail"
146 processStatus = "permanentFail"
150 if container["output"]:
152 outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep")
153 except Exception as e:
154 logger.error("Got error %s" % str(e))
155 self.output_callback({}, "permanentFail")
156 self.output_callback(outputs, processStatus)
158 del self.arvrunner.processes[record["uuid"]]
161 class RunnerContainer(Runner):
162 """Submit and manage a container that runs arvados-cwl-runner."""
164 def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
165 """Create an Arvados container request for this workflow.
167 The returned dict can be used to create a container passed as
168 the +body+ argument to container_requests().create().
171 workflowmapper = super(RunnerContainer, self).arvados_job_spec(dry_run=dry_run, pull_image=pull_image, **kwargs)
173 with arvados.collection.Collection(api_client=self.arvrunner.api,
174 keep_client=self.arvrunner.keep_client,
175 num_retries=self.arvrunner.num_retries) as jobobj:
176 with jobobj.open("cwl.input.json", "w") as f:
177 json.dump(self.job_order, f, sort_keys=True, indent=4)
178 jobobj.save_new(owner_uuid=self.arvrunner.project_uuid)
180 workflowname = os.path.basename(self.tool.tool["id"])
181 workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
182 workflowcollection = workflowmapper.mapper(self.tool.tool["id"])[1]
183 workflowcollection = workflowcollection[5:workflowcollection.index('/')]
184 jobpath = "/var/lib/cwl/job/cwl.input.json"
186 command = ["arvados-cwl-runner", "--local", "--api=containers"]
188 command.append("--output-name=" + self.output_name)
191 command.append("--output-tags=" + self.output_tags)
193 if self.enable_reuse:
194 command.append("--enable-reuse")
196 command.append("--disable-reuse")
198 command.extend([workflowpath, jobpath])
202 "owner_uuid": self.arvrunner.project_uuid,
204 "output_path": "/var/spool/cwl",
205 "cwd": "/var/spool/cwl",
207 "state": "Committed",
208 "container_image": arvados_jobs_image(self.arvrunner),
210 "/var/lib/cwl/workflow": {
211 "kind": "collection",
212 "portable_data_hash": "%s" % workflowcollection
215 "kind": "collection",
216 "portable_data_hash": "%s/cwl.input.json" % jobobj.portable_data_hash()
220 "path": "/var/spool/cwl/cwl.output.json"
223 "kind": "collection",
227 "runtime_constraints": {
229 "ram": 1024*1024 * self.submit_runner_ram,
234 def run(self, *args, **kwargs):
235 kwargs["keepprefix"] = "keep:"
236 job_spec = self.arvados_job_spec(*args, **kwargs)
237 job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
239 response = self.arvrunner.api.container_requests().create(
241 ).execute(num_retries=self.arvrunner.num_retries)
243 self.uuid = response["uuid"]
244 self.arvrunner.processes[response["uuid"]] = self
246 logger.info("Submitted container %s", response["uuid"])
248 if response["state"] == "Final":
251 def done(self, record):
253 container = self.arvrunner.api.containers().get(
254 uuid=record["container_uuid"]
255 ).execute(num_retries=self.arvrunner.num_retries)
256 except Exception as e:
257 logger.exception("While getting runner container: %s", e)
258 self.arvrunner.output_callback({}, "permanentFail")
259 del self.arvrunner.processes[record["uuid"]]
261 super(RunnerContainer, self).done(container)
263 del self.arvrunner.processes[record["uuid"]]