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 cwltool.errors import WorkflowException
   8 from cwltool.process import get_feature, UnsupportedRequirement, shortname
   9 from cwltool.pathmapper import adjustFiles, adjustDirObjs
  10 from cwltool.utils import aslist
  11
  12 import arvados.collection
  13
  14 from .arvdocker import arv_docker_get_image
  15 from . import done
  16 from .runner import Runner, arvados_jobs_image, packed_workflow, trim_listing
  17 from .fsaccess import CollectionFetcher
  18 from .pathmapper import NoFollowPathMapper
  19 from .perf import Perf
  20
  21 logger = logging.getLogger('arvados.cwl-runner')
  22 metrics = logging.getLogger('arvados.cwl-runner.metrics')
  23
  24 class ArvadosContainer(object):
  25     """Submit and manage a Crunch container request for executing a CWL CommandLineTool."""
  26
  27     def __init__(self, runner):
  28         self.arvrunner = runner
  29         self.running = False
  30         self.uuid = None
  31
  32     def update_pipeline_component(self, r):
  33         pass
  34
  35     def run(self, dry_run=False, pull_image=True, **kwargs):
  36         container_request = {
  37             "command": self.command_line,
  38             "owner_uuid": self.arvrunner.project_uuid,
  39             "name": self.name,
  40             "output_path": self.outdir,
  41             "cwd": self.outdir,
  42             "priority": 1,
  43             "state": "Committed",
  44             "properties": {}
  45         }
  46         runtime_constraints = {}
  47         mounts = {
  48             self.outdir: {
  49                 "kind": "tmp"
  50             }
  51         }
  52         scheduling_parameters = {}
  53
  54         dirs = set()
  55         for f in self.pathmapper.files():
  56             pdh, p, tp = self.pathmapper.mapper(f)
  57             if tp == "Directory" and '/' not in pdh:
  58                 mounts[p] = {
  59                     "kind": "collection",
  60                     "portable_data_hash": pdh[5:]
  61                 }
  62                 dirs.add(pdh)
  63
  64         for f in self.pathmapper.files():
  65             res, p, tp = self.pathmapper.mapper(f)
  66             if res.startswith("keep:"):
  67                 res = res[5:]
  68             elif res.startswith("/keep/"):
  69                 res = res[6:]
  70             else:
  71                 continue
  72             sp = res.split("/", 1)
  73             pdh = sp[0]
  74             if pdh not in dirs:
  75                 mounts[p] = {
  76                     "kind": "collection",
  77                     "portable_data_hash": pdh
  78                 }
  79                 if len(sp) == 2:
  80                     mounts[p]["path"] = sp[1]
  81
  82         with Perf(metrics, "generatefiles %s" % self.name):
  83             if self.generatefiles["listing"]:
  84                 vwd = arvados.collection.Collection(api_client=self.arvrunner.api,
  85                                                     keep_client=self.arvrunner.keep_client,
  86                                                     num_retries=self.arvrunner.num_retries)
  87                 generatemapper = NoFollowPathMapper([self.generatefiles], "", "",
  88                                                     separateDirs=False)
  89
  90                 with Perf(metrics, "createfiles %s" % self.name):
  91                     for f, p in generatemapper.items():
  92                         if not p.target:
  93                             pass
  94                         elif p.type in ("File", "Directory"):
  95                             source, path = self.arvrunner.fs_access.get_collection(p.resolved)
  96                             vwd.copy(path, p.target, source_collection=source)
  97                         elif p.type == "CreateFile":
  98                             with vwd.open(p.target, "w") as n:
  99                                 n.write(p.resolved.encode("utf-8"))
 100
 101                 with Perf(metrics, "generatefiles.save_new %s" % self.name):
 102                     vwd.save_new()
 103
 104                 for f, p in generatemapper.items():
 105                     if not p.target:
 106                         continue
 107                     mountpoint = "%s/%s" % (self.outdir, p.target)
 108                     mounts[mountpoint] = {"kind": "collection",
 109                                           "portable_data_hash": vwd.portable_data_hash(),
 110                                           "path": p.target}
 111
 112         container_request["environment"] = {"TMPDIR": self.tmpdir, "HOME": self.outdir}
 113         if self.environment:
 114             container_request["environment"].update(self.environment)
 115
 116         if self.stdin:
 117             sp = self.stdin[6:].split("/", 1)
 118             mounts["stdin"] = {"kind": "collection",
 119                                 "portable_data_hash": sp[0],
 120                                 "path": sp[1]}
 121
 122         if self.stderr:
 123             mounts["stderr"] = {"kind": "file",
 124                                 "path": "%s/%s" % (self.outdir, self.stderr)}
 125
 126         if self.stdout:
 127             mounts["stdout"] = {"kind": "file",
 128                                 "path": "%s/%s" % (self.outdir, self.stdout)}
 129
 130         (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
 131         if not docker_req:
 132             docker_req = {"dockerImageId": "arvados/jobs"}
 133
 134         container_request["container_image"] = arv_docker_get_image(self.arvrunner.api,
 135                                                                      docker_req,
 136                                                                      pull_image,
 137                                                                      self.arvrunner.project_uuid)
 138
 139         resources = self.builder.resources
 140         if resources is not None:
 141             runtime_constraints["vcpus"] = resources.get("cores", 1)
 142             runtime_constraints["ram"] = resources.get("ram") * 2**20
 143
 144         api_req, _ = get_feature(self, "http://arvados.org/cwl#APIRequirement")
 145         if api_req:
 146             runtime_constraints["API"] = True
 147
 148         runtime_req, _ = get_feature(self, "http://arvados.org/cwl#RuntimeConstraints")
 149         if runtime_req:
 150             if "keep_cache" in runtime_req:
 151                 runtime_constraints["keep_cache_ram"] = runtime_req["keep_cache"] * 2**20
 152
 153         partition_req, _ = get_feature(self, "http://arvados.org/cwl#PartitionRequirement")
 154         if partition_req:
 155             scheduling_parameters["partitions"] = aslist(partition_req["partition"])
 156
 157         container_request["mounts"] = mounts
 158         container_request["runtime_constraints"] = runtime_constraints
 159         container_request["use_existing"] = kwargs.get("enable_reuse", True)
 160         container_request["scheduling_parameters"] = scheduling_parameters
 161
 162         if kwargs.get("runnerjob", "").startswith("arvwf:"):
 163             wfuuid = kwargs["runnerjob"][6:kwargs["runnerjob"].index("#")]
 164             wfrecord = self.arvrunner.api.workflows().get(uuid=wfuuid).execute(num_retries=self.arvrunner.num_retries)
 165             if container_request["name"] == "main":
 166                 container_request["name"] = wfrecord["name"]
 167             container_request["properties"]["template_uuid"] = wfuuid
 168
 169         try:
 170             response = self.arvrunner.api.container_requests().create(
 171                 body=container_request
 172             ).execute(num_retries=self.arvrunner.num_retries)
 173
 174             self.uuid = response["uuid"]
 175             self.arvrunner.processes[self.uuid] = self
 176
 177             if response["state"] == "Final":
 178                 logger.info("%s reused container %s", self.arvrunner.label(self), response["container_uuid"])
 179                 self.done(response)
 180             else:
 181                 logger.info("%s %s state is %s", self.arvrunner.label(self), response["uuid"], response["state"])
 182         except Exception as e:
 183             logger.error("%s got error %s" % (self.arvrunner.label(self), str(e)))
 184             self.output_callback({}, "permanentFail")
 185
 186     def done(self, record):
 187         try:
 188             container = self.arvrunner.api.containers().get(
 189                 uuid=record["container_uuid"]
 190             ).execute(num_retries=self.arvrunner.num_retries)
 191             if container["state"] == "Complete":
 192                 rcode = container["exit_code"]
 193                 if self.successCodes and rcode in self.successCodes:
 194                     processStatus = "success"
 195                 elif self.temporaryFailCodes and rcode in self.temporaryFailCodes:
 196                     processStatus = "temporaryFail"
 197                 elif self.permanentFailCodes and rcode in self.permanentFailCodes:
 198                     processStatus = "permanentFail"
 199                 elif rcode == 0:
 200                     processStatus = "success"
 201                 else:
 202                     processStatus = "permanentFail"
 203             else:
 204                 processStatus = "permanentFail"
 205
 206             if processStatus == "permanentFail":
 207                 logc = arvados.collection.CollectionReader(container["log"],
 208                                                            api_client=self.arvrunner.api,
 209                                                            keep_client=self.arvrunner.keep_client,
 210                                                            num_retries=self.arvrunner.num_retries)
 211                 done.logtail(logc, logger, "%s error log:" % self.arvrunner.label(self))
 212
 213             outputs = {}
 214             if container["output"]:
 215                 outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep")
 216         except WorkflowException as e:
 217             logger.error("%s unable to collect output from %s:\n%s",
 218                          self.arvrunner.label(self), container["output"], e, exc_info=(e if self.arvrunner.debug else False))
 219             processStatus = "permanentFail"
 220         except Exception as e:
 221             logger.exception("%s while getting output object: %s", self.arvrunner.label(self), e)
 222             processStatus = "permanentFail"
 223         finally:
 224             self.output_callback(outputs, processStatus)
 225             if record["uuid"] in self.arvrunner.processes:
 226                 del self.arvrunner.processes[record["uuid"]]
 227
 228
 229 class RunnerContainer(Runner):
 230     """Submit and manage a container that runs arvados-cwl-runner."""
 231
 232     def arvados_job_spec(self, dry_run=False, pull_image=True, **kwargs):
 233         """Create an Arvados container request for this workflow.
 234
 235         The returned dict can be used to create a container passed as
 236         the +body+ argument to container_requests().create().
 237         """
 238
 239         adjustDirObjs(self.job_order, trim_listing)
 240
 241         container_req = {
 242             "owner_uuid": self.arvrunner.project_uuid,
 243             "name": self.name,
 244             "output_path": "/var/spool/cwl",
 245             "cwd": "/var/spool/cwl",
 246             "priority": 1,
 247             "state": "Committed",
 248             "container_image": arvados_jobs_image(self.arvrunner, self.jobs_image),
 249             "mounts": {
 250                 "/var/lib/cwl/cwl.input.json": {
 251                     "kind": "json",
 252                     "content": self.job_order
 253                 },
 254                 "stdout": {
 255                     "kind": "file",
 256                     "path": "/var/spool/cwl/cwl.output.json"
 257                 },
 258                 "/var/spool/cwl": {
 259                     "kind": "collection",
 260                     "writable": True
 261                 }
 262             },
 263             "runtime_constraints": {
 264                 "vcpus": 1,
 265                 "ram": 1024*1024 * self.submit_runner_ram,
 266                 "API": True
 267             },
 268             "properties": {}
 269         }
 270
 271         if self.tool.tool.get("id", "").startswith("keep:"):
 272             sp = self.tool.tool["id"].split('/')
 273             workflowcollection = sp[0][5:]
 274             workflowname = "/".join(sp[1:])
 275             workflowpath = "/var/lib/cwl/workflow/%s" % workflowname
 276             container_req["mounts"]["/var/lib/cwl/workflow"] = {
 277                 "kind": "collection",
 278                 "portable_data_hash": "%s" % workflowcollection
 279             }
 280         else:
 281             packed = packed_workflow(self.arvrunner, self.tool)
 282             workflowpath = "/var/lib/cwl/workflow.json#main"
 283             container_req["mounts"]["/var/lib/cwl/workflow.json"] = {
 284                 "kind": "json",
 285                 "content": packed
 286             }
 287             if self.tool.tool.get("id", "").startswith("arvwf:"):
 288                 container_req["properties"]["template_uuid"] = self.tool.tool["id"][6:33]
 289
 290         command = ["arvados-cwl-runner", "--local", "--api=containers", "--no-log-timestamps"]
 291         if self.output_name:
 292             command.append("--output-name=" + self.output_name)
 293             container_req["output_name"] = self.output_name
 294
 295         if self.output_tags:
 296             command.append("--output-tags=" + self.output_tags)
 297
 298         if kwargs.get("debug"):
 299             command.append("--debug")
 300
 301         if self.enable_reuse:
 302             command.append("--enable-reuse")
 303         else:
 304             command.append("--disable-reuse")
 305
 306         if self.on_error:
 307             command.append("--on-error=" + self.on_error)
 308
 309         command.extend([workflowpath, "/var/lib/cwl/cwl.input.json"])
 310
 311         container_req["command"] = command
 312
 313         return container_req
 314
 315
 316     def run(self, *args, **kwargs):
 317         kwargs["keepprefix"] = "keep:"
 318         job_spec = self.arvados_job_spec(*args, **kwargs)
 319         job_spec.setdefault("owner_uuid", self.arvrunner.project_uuid)
 320
 321         response = self.arvrunner.api.container_requests().create(
 322             body=job_spec
 323         ).execute(num_retries=self.arvrunner.num_retries)
 324
 325         self.uuid = response["uuid"]
 326         self.arvrunner.processes[self.uuid] = self
 327
 328         logger.info("%s submitted container %s", self.arvrunner.label(self), response["uuid"])
 329
 330         if response["state"] == "Final":
 331             self.done(response)
 332
 333     def done(self, record):
 334         try:
 335             container = self.arvrunner.api.containers().get(
 336                 uuid=record["container_uuid"]
 337             ).execute(num_retries=self.arvrunner.num_retries)
 338         except Exception as e:
 339             logger.exception("%s while getting runner container: %s", self.arvrunner.label(self), e)
 340             self.arvrunner.output_callback({}, "permanentFail")
 341         else:
 342             super(RunnerContainer, self).done(container)
 343         finally:
 344             if record["uuid"] in self.arvrunner.processes:
 345                 del self.arvrunner.processes[record["uuid"]]