from cwltool.errors import WorkflowException
from cwltool.process import get_feature, UnsupportedRequirement, shortname
from cwltool.pathmapper import adjustFiles
+from cwltool.utils import aslist
import arvados.collection
from .arvdocker import arv_docker_get_image
from . import done
-from .runner import Runner
+from .runner import Runner, arvados_jobs_image
logger = logging.getLogger('arvados.cwl-runner')
"kind": "tmp"
}
}
+ scheduling_parameters = {}
dirs = set()
for f in self.pathmapper.files():
(docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
if not docker_req:
- docker_req = {"dockerImageId": "arvados/jobs"}
+ docker_req = {"dockerImageId": arvados_jobs_image(self.arvrunner)}
container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
docker_req,
runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
if runtime_req:
- logger.warn("RuntimeConstraints not yet supported by container API")
+ if "keep_cache" in runtime_req:
+ runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"]
+
+ partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
+ if partition_req:
+ 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
try:
response = self.arvrunner.api.container_requests().create(
self.arvrunner.processes[response["container_uuid"]] = self
- logger.info("Container %s (%s) request state is %s", self.name, response["uuid"], response["state"])
+ 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 with container %s %s", self.name, response["uuid"], response["state"], container["uuid"], container["state"])
- if response["state"] == "Final":
- self.done(response)
+ if container["state"] in ("Complete", "Cancelled"):
+ self.done(container)
except Exception as e:
logger.error("Got error %s" % str(e))
self.output_callback({}, "permanentFail")
workflowcollection = workflowcollection[5:workflowcollection.index('/')]
jobpath = "/var/lib/cwl/job/cwl.input.json"
- container_image = arv_docker_get_image(self.arvrunner.api,
- {"dockerImageId": "arvados/jobs"},
- pull_image,
- self.arvrunner.project_uuid)
-
command = ["arvados-cwl-runner", "--local", "--api=containers"]
if self.output_name:
command.append("--output-name=" + self.output_name)
+
+ if self.enable_reuse:
+ command.append("--enable-reuse")
+ else:
+ command.append("--disable-reuse")
+
command.extend([workflowpath, jobpath])
return {
"cwd": "/var/spool/cwl",
"priority": 1,
"state": "Committed",
- "container_image": container_image,
+ "container_image": arvados_jobs_image(self.arvrunner),
"mounts": {
"/var/lib/cwl/workflow": {
"kind": "collection",