- def arvExecutor(self, tool, job_order, input_basedir, args, **kwargs):
- events = arvados.events.subscribe(arvados.api('v1'), [["object_uuid", "is_a", "arvados#job"]], self.on_message)
-
- try:
- self.api.collections().get(uuid=crunchrunner_pdh).execute()
- except arvados.errors.ApiError as e:
- import httplib2
- h = httplib2.Http(ca_certs=arvados.util.ca_certs_path())
- resp, content = h.request(crunchrunner_download, "GET")
- resp2, content2 = h.request(certs_download, "GET")
- with arvados.collection.Collection() as col:
- with col.open("crunchrunner", "w") as f:
- f.write(content)
- with col.open("ca-certificates.crt", "w") as f:
- f.write(content2)
-
- col.save_new("crunchrunner binary", ensure_unique_name=True)
-
- self.fs_access = CollectionFsAccess(input_basedir)
-
- kwargs["fs_access"] = self.fs_access
- kwargs["enable_reuse"] = args.enable_reuse
-
- kwargs["outdir"] = "$(task.outdir)"
- kwargs["tmpdir"] = "$(task.tmpdir)"
-
- if kwargs.get("conformance_test"):
- return cwltool.main.single_job_executor(tool, job_order, input_basedir, args, **kwargs)
+ def check_writable(self, obj):
+ if isinstance(obj, dict):
+ if obj.get("writable"):
+ raise UnsupportedRequirement("InitialWorkDir feature 'writable: true' not supported")
+ for v in obj.itervalues():
+ self.check_writable(v)
+ if isinstance(obj, list):
+ for v in obj:
+ self.check_writable(v)
+
+ def make_output_collection(self, name, outputObj):
+ outputObj = copy.deepcopy(outputObj)
+
+ files = []
+ def capture(fileobj):
+ files.append(fileobj)
+
+ adjustDirObjs(outputObj, capture)
+ adjustFileObjs(outputObj, capture)
+
+ generatemapper = FinalOutputPathMapper(files, "", "", separateDirs=False)
+
+ final = arvados.collection.Collection(api_client=self.api,
+ keep_client=self.keep_client,
+ num_retries=self.num_retries)
+
+ srccollections = {}
+ for k,v in generatemapper.items():
+ sp = k.split("/")
+ srccollection = sp[0][5:]
+ if srccollection not in srccollections:
+ srccollections[srccollection] = arvados.collection.CollectionReader(
+ srccollection,
+ api_client=self.api,
+ keep_client=self.keep_client,
+ num_retries=self.num_retries)
+ reader = srccollections[srccollection]
+ try:
+ srcpath = "/".join(sp[1:]) if len(sp) > 1 else "."
+ final.copy(srcpath, v.target, source_collection=reader, overwrite=False)
+ except IOError as e:
+ logger.warn("While preparing output collection: %s", e)
+
+ def rewrite(fileobj):
+ fileobj["location"] = generatemapper.mapper(fileobj["location"]).target
+ for k in ("basename", "size", "listing"):
+ if k in fileobj:
+ del fileobj[k]
+
+ adjustDirObjs(outputObj, rewrite)
+ adjustFileObjs(outputObj, rewrite)
+
+ with final.open("cwl.output.json", "w") as f:
+ json.dump(outputObj, f, sort_keys=True, indent=4, separators=(',',': '))
+
+ final.save_new(name=name, owner_uuid=self.project_uuid, ensure_unique_name=True)
+
+ logger.info("Final output collection %s \"%s\" (%s)", final.portable_data_hash(),
+ final.api_response()["name"],
+ final.manifest_locator())
+
+ self.final_output_collection = final
+
+ def arv_executor(self, tool, job_order, **kwargs):
+ self.debug = kwargs.get("debug")
+
+ tool.visit(self.check_writable)
+
+ useruuid = self.api.users().current().execute()["uuid"]
+ self.project_uuid = kwargs.get("project_uuid") if kwargs.get("project_uuid") else useruuid
+ self.pipeline = None
+ make_fs_access = kwargs.get("make_fs_access") or partial(CollectionFsAccess,
+ api_client=self.api,
+ keep_client=self.keep_client)
+ self.fs_access = make_fs_access(kwargs["basedir"])
+
+ if kwargs.get("create_template"):
+ tmpl = RunnerTemplate(self, tool, job_order, kwargs.get("enable_reuse"))
+ tmpl.save()
+ # cwltool.main will write our return value to stdout.
+ return tmpl.uuid
+
+ if kwargs.get("create_workflow") or kwargs.get("update_workflow"):
+ return upload_workflow(self, tool, job_order, self.project_uuid, kwargs.get("update_workflow"))
+
+ self.ignore_docker_for_reuse = kwargs.get("ignore_docker_for_reuse")
+
+ kwargs["make_fs_access"] = make_fs_access
+ kwargs["enable_reuse"] = kwargs.get("enable_reuse")
+ kwargs["use_container"] = True
+ kwargs["tmpdir_prefix"] = "tmp"
+ kwargs["on_error"] = "continue"
+ kwargs["compute_checksum"] = kwargs.get("compute_checksum")
+
+ if self.work_api == "containers":
+ kwargs["outdir"] = "/var/spool/cwl"
+ kwargs["docker_outdir"] = "/var/spool/cwl"
+ kwargs["tmpdir"] = "/tmp"
+ kwargs["docker_tmpdir"] = "/tmp"
+ elif self.work_api == "jobs":
+ kwargs["outdir"] = "$(task.outdir)"
+ kwargs["docker_outdir"] = "$(task.outdir)"
+ kwargs["tmpdir"] = "$(task.tmpdir)"
+
+ runnerjob = None
+ if kwargs.get("submit"):
+ if self.work_api == "containers":
+ if tool.tool["class"] == "CommandLineTool":
+ runnerjob = tool.job(job_order,
+ self.output_callback,
+ **kwargs).next()
+ else:
+ runnerjob = RunnerContainer(self, tool, job_order, kwargs.get("enable_reuse"), self.output_name)
+ else:
+ runnerjob = RunnerJob(self, tool, job_order, kwargs.get("enable_reuse"), self.output_name)
+
+ if not kwargs.get("submit") and "cwl_runner_job" not in kwargs and not self.work_api == "containers":
+ # Create pipeline for local run
+ self.pipeline = self.api.pipeline_instances().create(
+ body={
+ "owner_uuid": self.project_uuid,
+ "name": shortname(tool.tool["id"]),
+ "components": {},
+ "state": "RunningOnClient"}).execute(num_retries=self.num_retries)
+ logger.info("Pipeline instance %s", self.pipeline["uuid"])
+
+ if runnerjob and not kwargs.get("wait"):
+ runnerjob.run()
+ return runnerjob.uuid
+
+ self.poll_api = arvados.api('v1')
+ self.polling_thread = threading.Thread(target=self.poll_states)
+ self.polling_thread.start()
+
+ if runnerjob:
+ jobiter = iter((runnerjob,))