import json
import os
+import ruamel.yaml as yaml
+
+from schema_salad.sourceline import SourceLine
+
from cwltool.errors import WorkflowException
from cwltool.process import get_feature, UnsupportedRequirement, shortname
from cwltool.pathmapper import adjustFiles
from .arvdocker import arv_docker_get_image
from . import done
from .runner import Runner, arvados_jobs_image
+from .fsaccess import CollectionFetcher
logger = logging.getLogger('arvados.cwl-runner')
"output_path": self.outdir,
"cwd": self.outdir,
"priority": 1,
- "state": "Committed"
+ "state": "Committed",
+ "properties": {}
}
runtime_constraints = {}
mounts = {
"kind": "tmp"
}
}
+ scheduling_parameters = {}
dirs = set()
for f in self.pathmapper.files():
}
if self.generatefiles["listing"]:
- raise UnsupportedRequirement("Generate files not supported")
+ raise SourceLine(self.tool.get_requirement("InitialWorkDirRequirement")[0], None, UnsupportedRequirement).makeError("InitialWorkDirRequirement not supported with --api=containers")
container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
if self.environment:
container_request["environment"].update(self.environment)
if self.stdin:
- raise UnsupportedRequirement("Stdin redirection currently not suppported")
+ raise SourceLine(self.tool.tool, "stdin", UnsupportedRequirement).makeError("Stdin redirection currently not suppported")
if self.stderr:
- raise UnsupportedRequirement("Stderr redirection currently not suppported")
+ raise SourceLine(self.tool.tool, "stderr", UnsupportedRequirement).makeError("Stderr redirection currently not suppported")
if self.stdout:
mounts["stdout"] = {"kind": "file",
partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
if partition_req:
- runtime_constraints["partition"] = aslist(partition_req["partition"])
+ scheduling_parameters["partitions"] = aslist(partition_req["partition"])
container_request["mounts"] = mounts
container_request["runtime_constraints"] = runtime_constraints
container_request["use_existing"] = kwargs.get("enable_reuse", True)
+ container_request["scheduling_parameters"] = scheduling_parameters
+
+ if kwargs.get("runnerjob", "").startswith("arvwf:"):
+ wfuuid = kwargs["runnerjob"][6:kwargs["runnerjob"].index("#")]
+ wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
+ if container_request["name"] == "main":
+ container_request["name"] = wfrecord["name"]
+ container_request["properties"]["template_uuid"] = wfuuid
try:
response = self.arvrunner.api.container_requests().create(
body=container_request
).execute(num_retries=self.arvrunner.num_retries)
- self.arvrunner.processes[response["container_uuid"]] = self
+ self.uuid = response["uuid"]
+ self.arvrunner.processes[self.uuid] = self
- container = self.arvrunner.api.containers().get(
- uuid=response["container_uuid"]
- ).execute(num_retries=self.arvrunner.num_retries)
+ logger.info("Container request %s (%s) state is %s", self.name, response["uuid"], response["state"])
- logger.info("Container request %s (%s) state is %s with container %s %s", self.name, response["uuid"], response["state"], container["uuid"], container["state"])
-
- if container["state"] in ("Complete", "Cancelled"):
- self.done(container)
+ if response["state"] == "Final":
+ self.done(response)
except Exception as e:
logger.error("Got error %s" % str(e))
self.output_callback({}, "permanentFail")
def done(self, record):
try:
- if record["state"] == "Complete":
- rcode = record["exit_code"]
+ container = self.arvrunner.api.containers().get(
+ uuid=record["container_uuid"]
+ ).execute(num_retries=self.arvrunner.num_retries)
+ if container["state"] == "Complete":
+ rcode = container["exit_code"]
if self.successCodes and rcode in self.successCodes:
processStatus = "success"
elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
else:
processStatus = "permanentFail"
- try:
- outputs = {}
- if record["output"]:
- outputs = done.done(self, record, "/tmp", self.outdir, "/keep")
- except WorkflowException as e:
- logger.error("Error while collecting container outputs:\n%s", e, exc_info=(e if self.arvrunner.debug else False))
- processStatus = "permanentFail"
- except Exception as e:
- logger.exception("Got unknown exception while collecting job outputs:")
- processStatus = "permanentFail"
+ outputs = {}
+ if container["output"]:
+ try:
+ outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep")
+ except Exception as e:
+ logger.error("Got error %s" % str(e))
+ self.output_callback({}, "permanentFail")
self.output_callback(outputs, processStatus)
finally:
del self.arvrunner.processes[record["uuid"]]
workflowmapper = super(RunnerContainer, self).arvados_job_spec(dry_run=dry_run, pull_image=pull_image, **kwargs)
- with arvados.collection.Collection(api_client=self.arvrunner.api,
- keep_client=self.arvrunner.keep_client,
- num_retries=self.arvrunner.num_retries) as jobobj:
- with jobobj.open("cwl.input.json", "w") as f:
- json.dump(self.job_order, f, sort_keys=True, indent=4)
- jobobj.save_new(owner_uuid=self.arvrunner.project_uuid)
-
- workflowname = os.path.basename(self.tool.tool["id"])
- workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
- workflowcollection = workflowmapper.mapper(self.tool.tool["id"])[1]
- workflowcollection = workflowcollection[5:workflowcollection.index('/')]
- jobpath = "/var/lib/cwl/job/cwl.input.json"
-
- command = ["arvados-cwl-runner", "--local", "--api=containers"]
- if self.output_name:
- command.append("--output-name=" + self.output_name)
-
- if self.output_tags:
- command.append("--output-tags=" + self.output_tags)
-
- if self.enable_reuse:
- command.append("--enable-reuse")
- else:
- command.append("--disable-reuse")
-
- command.extend([workflowpath, jobpath])
-
- return {
- "command": command,
+ container_req = {
"owner_uuid": self.arvrunner.project_uuid,
"name": self.name,
"output_path": "/var/spool/cwl",
"state": "Committed",
"container_image": arvados_jobs_image(self.arvrunner),
"mounts": {
- "/var/lib/cwl/workflow": {
- "kind": "collection",
- "portable_data_hash": "%s" % workflowcollection
- },
- jobpath: {
- "kind": "collection",
- "portable_data_hash": "%s/cwl.input.json" % jobobj.portable_data_hash()
+ "/var/lib/cwl/cwl.input.json": {
+ "kind": "json",
+ "content": self.job_order
},
"stdout": {
"kind": "file",
},
"runtime_constraints": {
"vcpus": 1,
- "ram": 1024*1024*256,
+ "ram": 1024*1024 * self.submit_runner_ram,
"API": True
- }
+ },
+ "properties": {}
}
+ workflowcollection = workflowmapper.mapper(self.tool.tool["id"])[1]
+ if workflowcollection.startswith("keep:"):
+ workflowcollection = workflowcollection[5:workflowcollection.index('/')]
+ workflowname = os.path.basename(self.tool.tool["id"])
+ workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
+ container_req["mounts"]["/var/lib/cwl/workflow"] = {
+ "kind": "collection",
+ "portable_data_hash": "%s" % workflowcollection
+ }
+ elif workflowcollection.startswith("arvwf:"):
+ workflowpath = "/var/lib/cwl/workflow.json#main"
+ wfuuid = workflowcollection[6:workflowcollection.index("#")]
+ wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
+ wfobj = yaml.safe_load(wfrecord["definition"])
+ if container_req["name"].startswith("arvwf:"):
+ container_req["name"] = wfrecord["name"]
+ container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
+ "kind": "json",
+ "json": wfobj
+ }
+ container_req["properties"]["template_uuid"] = wfuuid
+
+ command = ["arvados-cwl-runner", "--local", "--api=containers"]
+ if self.output_name:
+ command.append("--output-name=" + self.output_name)
+
+ if self.output_tags:
+ command.append("--output-tags=" + self.output_tags)
+
+ if self.enable_reuse:
+ command.append("--enable-reuse")
+ else:
+ command.append("--disable-reuse")
+
+ command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
+
+ container_req["command"] = command
+
+ return container_req
+
+
def run(self, *args, **kwargs):
kwargs["keepprefix"] = "keep:"
job_spec = self.arvados_job_spec(*args, **kwargs)
).execute(num_retries=self.arvrunner.num_retries)
self.uuid = response["uuid"]
- self.arvrunner.processes[response["container_uuid"]] = self
+ self.arvrunner.processes[self.uuid] = self
logger.info("Submitted container %s", response["uuid"])
- if response["state"] in ("Complete", "Failed", "Cancelled"):
+ if response["state"] == "Final":
self.done(response)
+
+ def done(self, record):
+ try:
+ container = self.arvrunner.api.containers().get(
+ uuid=record["container_uuid"]
+ ).execute(num_retries=self.arvrunner.num_retries)
+ except Exception as e:
+ logger.exception("While getting runner container: %s", e)
+ self.arvrunner.output_callback({}, "permanentFail")
+ del self.arvrunner.processes[record["uuid"]]
+ else:
+ super(RunnerContainer, self).done(container)
+ finally:
+ del self.arvrunner.processes[record["uuid"]]
projects / arvados.git / blobdiff