sdk/cwl/arvados_cwl/arvcontainer.py

   1 import logging
   2 import json
   3 import os
   4
   5 import ruamel.yaml as yaml
   6
   7 from schema_salad.sourceline import SourceLine
   8
   9 from cwltool.errors import WorkflowException
  10 from cwltool.process import get_feature, UnsupportedRequirement, shortname
  11 from cwltool.pathmapper import adjustFiles
  12 from cwltool.utils import aslist
  13
  14 import arvados.collection
  15
  16 from .arvdocker import arv_docker_get_image
  17 from . import done
  18 from .runner import Runner, arvados_jobs_image
  19 from .fsaccess import CollectionFetcher
  20
  21 logger = logging.getLogger('arvados.cwl-runner')
  22
  23 class ArvadosContainer(object):
  24     """Submit and manage a Crunch container request for executing a CWL CommandLineTool."""
  25
  26     def __init__(self, runner):
  27         self.arvrunner = runner
  28         self.running = False
  29         self.uuid = None
  30
  31     def update_pipeline_component(self, r):
  32         pass
  33
  34     def run(self, dry_run=False, pull_image=True, **kwargs):
  35         container_request = {
  36             "command": self.command_line,
  37             "owner_uuid": self.arvrunner.project_uuid,
  38             "name": self.name,
  39             "output_path": self.outdir,
  40             "cwd": self.outdir,
  41             "priority": 1,
  42             "state": "Committed",
  43             "properties": {}
  44         }
  45         runtime_constraints = {}
  46         mounts = {
  47             self.outdir: {
  48                 "kind": "tmp"
  49             }
  50         }
  51         scheduling_parameters = {}
  52
  53         dirs = set()
  54         for f in self.pathmapper.files():
  55             _, p, tp = self.pathmapper.mapper(f)
  56             if tp == "Directory" and '/' not in p[6:]:
  57                 mounts[p] = {
  58                     "kind": "collection",
  59                     "portable_data_hash": p[6:]
  60                 }
  61                 dirs.add(p[6:])
  62         for f in self.pathmapper.files():
  63             _, p, tp = self.pathmapper.mapper(f)
  64             if p[6:].split("/")[0] not in dirs:
  65                 mounts[p] = {
  66                     "kind": "collection",
  67                     "portable_data_hash": p[6:]
  68                 }
  69
  70         if self.generatefiles["listing"]:
  71             raise SourceLine(self.tool.get_requirement("InitialWorkDirRequirement")[0], None, UnsupportedRequirement).makeError("InitialWorkDirRequirement not supported with --api=containers")
  72
  73         container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
  74         if self.environment:
  75             container_request["environment"].update(self.environment)
  76
  77         if self.stdin:
  78             raise SourceLine(self.tool.tool, "stdin", UnsupportedRequirement).makeError("Stdin redirection currently not suppported")
  79
  80         if self.stderr:
  81             raise SourceLine(self.tool.tool, "stderr", UnsupportedRequirement).makeError("Stderr redirection currently not suppported")
  82
  83         if self.stdout:
  84             mounts["stdout"] = {"kind": "file",
  85                                 "path": "%s/%s" % (self.outdir, self.stdout)}
  86
  87         (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
  88         if not docker_req:
  89             docker_req = {"dockerImageId": arvados_jobs_image(self.arvrunner)}
  90
  91         container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
  92                                                                      docker_req,
  93                                                                      pull_image,
  94                                                                      self.arvrunner.project_uuid)
  95
  96         resources = self.builder.resources
  97         if resources is not None:
  98             runtime_constraints["vcpus"] = resources.get("cores", 1)
  99             runtime_constraints["ram"] = resources.get("ram") * 2**20
 100
 101         api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement")
 102         if api_req:
 103             runtime_constraints["API"] = True
 104
 105         runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
 106         if runtime_req:
 107             if "keep_cache" in runtime_req:
 108                 runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"]
 109
 110         partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
 111         if partition_req:
 112             scheduling_parameters["partitions"] = aslist(partition_req["partition"])
 113
 114         container_request["mounts"] = mounts
 115         container_request["runtime_constraints"] = runtime_constraints
 116         container_request["use_existing"] = kwargs.get("enable_reuse", True)
 117         container_request["scheduling_parameters"] = scheduling_parameters
 118
 119         if kwargs.get("runnerjob", "").startswith("arvwf:"):
 120             wfuuid = kwargs["runnerjob"][6:kwargs["runnerjob"].index("#")]
 121             wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
 122             if container_request["name"] == "main":
 123                 container_request["name"] = wfrecord["name"]
 124             container_request["properties"]["template_uuid"] = wfuuid
 125
 126         try:
 127             response = self.arvrunner.api.container_requests().create(
 128                 body=container_request
 129             ).execute(num_retries=self.arvrunner.num_retries)
 130
 131             self.uuid = response["uuid"]
 132             self.arvrunner.processes[self.uuid] = self
 133
 134             logger.info("Container request %s (%s) state is %s", self.name, response["uuid"], response["state"])
 135
 136             if response["state"] == "Final":
 137                 self.done(response)
 138         except Exception as e:
 139             logger.error("Got error %s" % str(e))
 140             self.output_callback({}, "permanentFail")
 141
 142     def done(self, record):
 143         try:
 144             container = self.arvrunner.api.containers().get(
 145                 uuid=record["container_uuid"]
 146             ).execute(num_retries=self.arvrunner.num_retries)
 147             if container["state"] == "Complete":
 148                 rcode = container["exit_code"]
 149                 if self.successCodes and rcode in self.successCodes:
 150                     processStatus = "success"
 151                 elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
 152                     processStatus = "temporaryFail"
 153                 elif self.permanentFailCodes and rcode in self.permanentFailCodes:
 154                     processStatus = "permanentFail"
 155                 elif rcode == 0:
 156                     processStatus = "success"
 157                 else:
 158                     processStatus = "permanentFail"
 159             else:
 160                 processStatus = "permanentFail"
 161
 162             outputs = {}
 163
 164             if container["output"]:
 165                 try:
 166                     outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep")
 167                 except Exception as e:
 168                     logger.error("Got error %s" % str(e))
 169                     self.output_callback({}, "permanentFail")
 170             self.output_callback(outputs, processStatus)
 171         finally:
 172             del self.arvrunner.processes[record["uuid"]]
 173
 174
 175 class RunnerContainer(Runner):
 176     """Submit and manage a container that runs arvados-cwl-runner."""
 177
 178     def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
 179         """Create an Arvados container request for this workflow.
 180
 181         The returned dict can be used to create a container passed as
 182         the +body+ argument to container_requests().create().
 183         """
 184
 185         workflowmapper = super(RunnerContainer, self).arvados_job_spec(dry_run=dry_run, pull_image=pull_image, **kwargs)
 186
 187         container_req = {
 188             "owner_uuid": self.arvrunner.project_uuid,
 189             "name": self.name,
 190             "output_path": "/var/spool/cwl",
 191             "cwd": "/var/spool/cwl",
 192             "priority": 1,
 193             "state": "Committed",
 194             "container_image": arvados_jobs_image(self.arvrunner),
 195             "mounts": {
 196                 "/var/lib/cwl/cwl.input.json": {
 197                     "kind": "json",
 198                     "content": self.job_order
 199                 },
 200                 "stdout": {
 201                     "kind": "file",
 202                     "path": "/var/spool/cwl/cwl.output.json"
 203                 },
 204                 "/var/spool/cwl": {
 205                     "kind": "collection",
 206                     "writable": True
 207                 }
 208             },
 209             "runtime_constraints": {
 210                 "vcpus": 1,
 211                 "ram": 1024*1024 * self.submit_runner_ram,
 212                 "API": True
 213             },
 214             "properties": {}
 215         }
 216
 217         workflowcollection = workflowmapper.mapper(self.tool.tool["id"])[1]
 218         if workflowcollection.startswith("keep:"):
 219             workflowcollection = workflowcollection[5:workflowcollection.index('/')]
 220             workflowname = os.path.basename(self.tool.tool["id"])
 221             workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
 222             container_req["mounts"]["/var/lib/cwl/workflow"] = {
 223                 "kind": "collection",
 224                 "portable_data_hash": "%s" % workflowcollection
 225                 }
 226         elif workflowcollection.startswith("arvwf:"):
 227             workflowpath = "/var/lib/cwl/workflow.json#main"
 228             wfuuid = workflowcollection[6:workflowcollection.index("#")]
 229             wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
 230             wfobj = yaml.safe_load(wfrecord["definition"])
 231             if container_req["name"].startswith("arvwf:"):
 232                 container_req["name"] = wfrecord["name"]
 233             container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
 234                 "kind": "json",
 235                 "json": wfobj
 236             }
 237             container_req["properties"]["template_uuid"] = wfuuid
 238
 239         command = ["arvados-cwl-runner", "--local", "--api=containers"]
 240         if self.output_name:
 241             command.append("--output-name=" + self.output_name)
 242
 243         if self.output_tags:
 244             command.append("--output-tags=" + self.output_tags)
 245
 246         if self.enable_reuse:
 247             command.append("--enable-reuse")
 248         else:
 249             command.append("--disable-reuse")
 250
 251         command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
 252
 253         container_req["command"] = command
 254
 255         return container_req
 256
 257
 258     def run(self, *args, **kwargs):
 259         kwargs["keepprefix"] = "keep:"
 260         job_spec = self.arvados_job_spec(*args, **kwargs)
 261         job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
 262
 263         response = self.arvrunner.api.container_requests().create(
 264             body=job_spec
 265         ).execute(num_retries=self.arvrunner.num_retries)
 266
 267         self.uuid = response["uuid"]
 268         self.arvrunner.processes[self.uuid] = self
 269
 270         logger.info("Submitted container %s", response["uuid"])
 271
 272         if response["state"] == "Final":
 273             self.done(response)
 274
 275     def done(self, record):
 276         try:
 277             container = self.arvrunner.api.containers().get(
 278                 uuid=record["container_uuid"]
 279             ).execute(num_retries=self.arvrunner.num_retries)
 280         except Exception as e:
 281             logger.exception("While getting runner container: %s", e)
 282             self.arvrunner.output_callback({}, "permanentFail")
 283             del self.arvrunner.processes[record["uuid"]]
 284         else:
 285             super(RunnerContainer, self).done(container)
 286         finally:
 287             del self.arvrunner.processes[record["uuid"]]