X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/d4bb1f8a3c68288f45d2c1fa618c333825fcc8d0..3af6db5dc4e2f08b2ebb49a82109c4325ad7fcc4:/sdk/cwl/arvados_cwl/arvcontainer.py diff --git a/sdk/cwl/arvados_cwl/arvcontainer.py b/sdk/cwl/arvados_cwl/arvcontainer.py index 413435db23..aa088c5e8a 100644 --- a/sdk/cwl/arvados_cwl/arvcontainer.py +++ b/sdk/cwl/arvados_cwl/arvcontainer.py @@ -5,12 +5,13 @@ import os 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') @@ -41,6 +42,7 @@ class ArvadosContainer(object): "kind": "tmp" } } + scheduling_parameters = {} dirs = set() for f in self.pathmapper.files(): @@ -78,7 +80,7 @@ class ArvadosContainer(object): (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, @@ -90,8 +92,23 @@ class ArvadosContainer(object): runtime_constraints["vcpus"] = resources.get("cores", 1) runtime_constraints["ram"] = resources.get("ram") * 2**20 + api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement") + if api_req: + runtime_constraints["API"] = True + + runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints") + if runtime_req: + 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( @@ -100,10 +117,14 @@ class ArvadosContainer(object): 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) - if response["state"] == "Final": - self.done(response) + 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) except Exception as e: logger.error("Got error %s" % str(e)) self.output_callback({}, "permanentFail") @@ -153,7 +174,9 @@ class RunnerContainer(Runner): 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) as jobobj: + 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) @@ -164,20 +187,26 @@ class RunnerContainer(Runner): 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 { - "command": ["arvados-cwl-runner", "--local", "--api=containers", workflowpath, jobpath], + "command": command, "owner_uuid": self.arvrunner.project_uuid, "name": self.name, "output_path": "/var/spool/cwl", "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",