sdk/cwl/arvados_cwl/arvcontainer.py

   1 import logging
   2 import json
   3 import os
   4
   5 from cwltool.errors import WorkflowException
   6 from cwltool.process import get_feature, adjustFiles, UnsupportedRequirement, shortname
   7
   8 import arvados.collection
   9
  10 from .arvdocker import arv_docker_get_image
  11 from . import done
  12 from .runner import Runner
  13
  14 logger = logging.getLogger('arvados.cwl-runner')
  15
  16 class ArvadosContainer(object):
  17     """Submit and manage a Crunch container request for executing a CWL CommandLineTool."""
  18
  19     def __init__(self, runner):
  20         self.arvrunner = runner
  21         self.running = False
  22         self.uuid = None
  23
  24     def update_pipeline_component(self, r):
  25         pass
  26
  27     def run(self, dry_run=False, pull_image=True, **kwargs):
  28         container_request = {
  29             "command": self.command_line,
  30             "owner_uuid": self.arvrunner.project_uuid,
  31             "name": self.name,
  32             "output_path": self.outdir,
  33             "cwd": self.outdir,
  34             "priority": 1,
  35             "state": "Committed"
  36         }
  37         runtime_constraints = {}
  38         mounts = {
  39             self.outdir: {
  40                 "kind": "tmp"
  41             }
  42         }
  43
  44         for f in self.pathmapper.files():
  45             _, p = self.pathmapper.mapper(f)
  46             mounts[p] = {
  47                 "kind": "collection",
  48                 "portable_data_hash": p[6:]
  49             }
  50
  51         if self.generatefiles:
  52             raise UnsupportedRequirement("Generate files not supported")
  53
  54         container_request["environment"] = {"TMPDIR": "/tmp"}
  55         if self.environment:
  56             container_request["environment"].update(self.environment)
  57
  58         if self.stdin:
  59             raise UnsupportedRequirement("Stdin redirection currently not suppported")
  60
  61         if self.stdout:
  62             mounts["stdout"] = {"kind": "file",
  63                                 "path": "%s/%s" % (self.outdir, self.stdout)}
  64
  65         (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
  66         if not docker_req:
  67             docker_req = {"dockerImageId": "arvados/jobs"}
  68
  69         container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
  70                                                                      docker_req,
  71                                                                      pull_image,
  72                                                                      self.arvrunner.project_uuid)
  73
  74         resources = self.builder.resources
  75         if resources is not None:
  76             runtime_constraints["vcpus"] = resources.get("cores", 1)
  77             runtime_constraints["ram"] = resources.get("ram") * 2**20
  78
  79         container_request["mounts"] = mounts
  80         container_request["runtime_constraints"] = runtime_constraints
  81
  82         try:
  83             response = self.arvrunner.api.container_requests().create(
  84                 body=container_request
  85             ).execute(num_retries=self.arvrunner.num_retries)
  86
  87             self.arvrunner.processes[response["container_uuid"]] = self
  88
  89             logger.info("Container %s (%s) request state is %s", self.name, response["container_uuid"], response["state"])
  90
  91             if response["state"] == "Final":
  92                 self.done(response)
  93         except Exception as e:
  94             logger.error("Got error %s" % str(e))
  95             self.output_callback({}, "permanentFail")
  96
  97     def done(self, record):
  98         try:
  99             if record["state"] == "Complete":
 100                 rcode = record["exit_code"]
 101                 if self.successCodes and rcode in self.successCodes:
 102                     processStatus = "success"
 103                 elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
 104                     processStatus = "temporaryFail"
 105                 elif self.permanentFailCodes and rcode in self.permanentFailCodes:
 106                     processStatus = "permanentFail"
 107                 elif rcode == 0:
 108                     processStatus = "success"
 109                 else:
 110                     processStatus = "permanentFail"
 111             else:
 112                 processStatus = "permanentFail"
 113
 114             try:
 115                 outputs = {}
 116                 if record["output"]:
 117                     outputs = done.done(self, record, "/tmp", self.outdir, "/keep")
 118             except WorkflowException as e:
 119                 logger.error("Error while collecting container outputs:\n%s", e, exc_info=(e if self.arvrunner.debug else False))
 120                 processStatus = "permanentFail"
 121             except Exception as e:
 122                 logger.exception("Got unknown exception while collecting job outputs:")
 123                 processStatus = "permanentFail"
 124
 125             self.output_callback(outputs, processStatus)
 126         finally:
 127             del self.arvrunner.processes[record["uuid"]]
 128
 129
 130 class RunnerContainer(Runner):
 131     """Submit and manage a container that runs arvados-cwl-runner."""
 132
 133     def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
 134         """Create an Arvados container request for this workflow.
 135
 136         The returned dict can be used to create a container passed as
 137         the +body+ argument to container_requests().create().
 138         """
 139
 140         workflowmapper = super(RunnerContainer, self).arvados_job_spec(dry_run=dry_run, pull_image=pull_image, **kwargs)
 141
 142         with arvados.collection.Collection(api_client=self.arvrunner.api) as jobobj:
 143             with jobobj.open("cwl.input.json", "w") as f:
 144                 json.dump(self.job_order, f, sort_keys=True, indent=4)
 145             jobobj.save_new(owner_uuid=self.arvrunner.project_uuid)
 146
 147         workflowname = os.path.basename(self.tool.tool["id"])
 148         workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
 149         workflowcollection = workflowmapper.mapper(self.tool.tool["id"])[1]
 150         workflowcollection = workflowcollection[5:workflowcollection.index('/')]
 151         jobpath = "/var/lib/cwl/job/cwl.input.json"
 152
 153         container_image = arv_docker_get_image(self.arvrunner.api,
 154                                                {"dockerImageId": "arvados/jobs"},
 155                                                pull_image,
 156                                                self.arvrunner.project_uuid)
 157
 158         return {
 159             "command": ["arvados-cwl-runner", "--local", "--api=containers", workflowpath, jobpath],
 160             "owner_uuid": self.arvrunner.project_uuid,
 161             "name": self.name,
 162             "output_path": "/var/spool/cwl",
 163             "cwd": "/var/spool/cwl",
 164             "priority": 1,
 165             "state": "Committed",
 166             "container_image": container_image,
 167             "mounts": {
 168                 "/var/lib/cwl/workflow": {
 169                     "kind": "collection",
 170                     "portable_data_hash": "%s" % workflowcollection
 171                 },
 172                 jobpath: {
 173                     "kind": "collection",
 174                     "portable_data_hash": "%s/cwl.input.json" % jobobj.portable_data_hash()
 175                 },
 176                 "stdout": {
 177                     "kind": "file",
 178                     "path": "/var/spool/cwl/cwl.output.json"
 179                 },
 180                 "/var/spool/cwl": {
 181                     "kind": "collection",
 182                     "writable": True
 183                 }
 184             },
 185             "runtime_constraints": {
 186                 "vcpus": 1,
 187                 "ram": 1024*1024*256,
 188                 "API": True
 189             }
 190         }
 191
 192     def run(self, *args, **kwargs):
 193         kwargs["keepprefix"] = "keep:"
 194         job_spec = self.arvados_job_spec(*args, **kwargs)
 195         job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
 196
 197         response = self.arvrunner.api.container_requests().create(
 198             body=job_spec
 199         ).execute(num_retries=self.arvrunner.num_retries)
 200
 201         self.uuid = response["uuid"]
 202         self.arvrunner.processes[response["container_uuid"]] = self
 203
 204         logger.info("Submitted container %s", response["uuid"])
 205
 206         if response["state"] in ("Complete", "Failed", "Cancelled"):
 207             self.done(response)