sdk/cwl/arvados_cwl/done.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: Apache-2.0
   4
   5 import re
   6 from cwltool.errors import WorkflowException
   7 from collections import deque
   8
   9 def done(self, record, tmpdir, outdir, keepdir):
  10     cols = [
  11         ("output", "Output %s of %s" % (record["output"][0:7], self.name), record["output"]),
  12         ("log", "Log of %s" % (record["uuid"]), record["log"])
  13     ]
  14
  15     for coltype, colname, colpdh in cols:
  16         # check if collection already exists with same owner, name and content
  17         collection_exists = self.arvrunner.api.collections().list(
  18             filters=[["owner_uuid", "=", self.arvrunner.project_uuid],
  19                      ['portable_data_hash', '=', colpdh],
  20                      ["name", "=", colname]]
  21         ).execute(num_retries=self.arvrunner.num_retries)
  22
  23         if not collection_exists["items"]:
  24             # Create a collection located in the same project as the
  25             # pipeline with the contents of the output/log.
  26             # First, get output/log record.
  27             collections = self.arvrunner.api.collections().list(
  28                 limit=1,
  29                 filters=[['portable_data_hash', '=', colpdh]],
  30                 select=["manifest_text"]
  31             ).execute(num_retries=self.arvrunner.num_retries)
  32
  33             if not collections["items"]:
  34                 raise WorkflowException(
  35                     "[job %s] %s '%s' cannot be found on API server" % (
  36                         self.name, coltype, colpdh))
  37
  38             # Create new collection in the parent project
  39             # with the output/log contents.
  40             self.arvrunner.api.collections().create(body={
  41                 "owner_uuid": self.arvrunner.project_uuid,
  42                 "name": colname,
  43                 "portable_data_hash": colpdh,
  44                 "manifest_text": collections["items"][0]["manifest_text"]
  45             }, ensure_unique_name=True).execute(
  46                 num_retries=self.arvrunner.num_retries)
  47
  48     return done_outputs(self, record, tmpdir, outdir, keepdir)
  49
  50 def done_outputs(self, record, tmpdir, outdir, keepdir):
  51     self.builder.outdir = outdir
  52     self.builder.pathmapper.keepdir = keepdir
  53     return self.collect_outputs("keep:" + record["output"])
  54
  55 crunchstat_re = re.compile(r"^\d{4}-\d\d-\d\d_\d\d:\d\d:\d\d [a-z0-9]{5}-8i9sb-[a-z0-9]{15} \d+ \d+ stderr crunchstat:")
  56 timestamp_re = re.compile(r"^(\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d+Z) (.*)")
  57
  58 def logtail(logcollection, logfunc, header, maxlen=25):
  59     if len(logcollection) == 0:
  60         logfunc("%s\n%s", header, "  ** log is empty **")
  61         return
  62
  63     containersapi = ("crunch-run.txt" in logcollection)
  64     mergelogs = {}
  65
  66     for log in list(logcollection.keys()):
  67         if not containersapi or log in ("crunch-run.txt", "stdout.txt", "stderr.txt"):
  68             logname = log[:-4]
  69             logt = deque([], maxlen)
  70             mergelogs[logname] = logt
  71             with logcollection.open(log) as f:
  72                 for l in f:
  73                     if containersapi:
  74                         g = timestamp_re.match(l)
  75                         logt.append((g.group(1), g.group(2)))
  76                     elif not crunchstat_re.match(l):
  77                         logt.append(l)
  78
  79     if containersapi:
  80         keys = list(mergelogs.keys())
  81         loglines = []
  82         while True:
  83             earliest = None
  84             for k in keys:
  85                 if mergelogs[k]:
  86                     if earliest is None or mergelogs[k][0][0] < mergelogs[earliest][0][0]:
  87                         earliest = k
  88             if earliest is None:
  89                 break
  90             ts, msg = mergelogs[earliest].popleft()
  91             loglines.append("%s %s %s" % (ts, earliest, msg))
  92         loglines = loglines[-maxlen:]
  93     else:
  94         loglines = list(mergelogs.values())[0]
  95
  96     logtxt = "\n  ".join(l.strip() for l in loglines)
  97     logfunc("%s\n\n  %s", header, logtxt)