src/monitor_app/harvester_stdout.py

0001 """Harvester stdout records: capture, store, read.
0002
0003 The Kubernetes harvester queues (BNL_ePIC_GOOGLE today) upload a
0004 worker's stdout alone, to the PanDA server's cache, and the server
0005 purges that cache on a flat seven-day mtime cutoff (panda-server
0006 copyArchive.py); nothing else of those jobs' logs exists. This module
0007 copies the stdout of the jobs worth keeping while it is there, so the
0008 job page and the dig read our copy for as long as the record needs it
0009 (docs/EPICPROD_OPS.md, Harvester stdout records).
0010
0011 Which jobs: every job whose pilot id names a cache stdout, failed
0012 always, finished behind the ``harvester_stdout.finished`` switch (their
0013 account is the payload report). The capture runs hourly, well inside
0014 the week, and a fetch that fails leaves a marker with its reason, so a
0015 copy that could not be made is visible rather than absent; a 404 is
0016 the cache having let the file go.
0017
0018 The store is a filesystem tree outside the reclaimable scratch cache,
0019 so the thirty-day prune of ``panda-logs`` never touches it. Layout::
0020
0021     <root>/<jeditaskid>/<pandaid>.stdout.gz   the text, gzip (220 KB against 4 MB)
0022     <root>/<jeditaskid>/<pandaid>.json     source, fetched, bytes, status
0023     <root>/<jeditaskid>/<pandaid>.error    why the fetch failed
0024
0025 Lifetime: ``harvester_stdout.keep_days``, 30 by default (0 keeps them);
0026 the hourly pass prunes copies older than that by their capture time.
0027 """
0028 import gzip
0029 import json
0030 import logging
0031 import os
0032 import ssl
0033 import urllib.request
0034 from datetime import datetime, timedelta, timezone as dt_timezone
0035
0036 from django.conf import settings
0037 from django.db import connections
0038
0039 from .panda.constants import PANDA_SCHEMA
0040
0041 logger = logging.getLogger(__name__)
0042
0043 # A worker's stdout is 120 to 220 KB gzip on the wire, about 4 MB as text
0044 # (the whole pilot log); the cap is far above that and a body that
0045 # reaches it is recorded as capped rather than silently shortened.
0046 FETCH_MAX_BYTES = 64 * 1024 * 1024
0047 FETCH_TIMEOUT_S = 60
0048 DEFAULTS = {'finished': False, 'keep_days': 30, 'backfill_days': 7}
0049
0050
0051 def store_root():
0052     return getattr(settings, 'HARVESTER_STDOUT_ROOT',
0053                    '/data/wenauseic/swf-monitor/harvester-stdout')
0054
0055
0056 def setting(key):
0057     from .models import SysConfig
0058     return SysConfig.get_setting(f'harvester_stdout.{key}', DEFAULTS[key])
0059
0060
0061 def _paths(root, jeditaskid, pandaid):
0062     d = os.path.join(root, str(int(jeditaskid or 0)))
0063     base = os.path.join(d, str(int(pandaid)))
0064     return d, f'{base}.stdout.gz', f'{base}.json', f'{base}.error'
0065
0066
0067 def stored(pandaid, jeditaskid=None, root=None):
0068     """The copy of one job, or None: ``{pandaid, status, path, body,
0069     bytes, captured_at, source}``, status 'captured' or 'failed' (the
0070     failed form's body is the reason)."""
0071     root = root or store_root()
0072     if jeditaskid is None:
0073         jeditaskid = _task_of(pandaid)
0074         if jeditaskid is None:
0075             return None
0076     _, out, meta, err = _paths(root, jeditaskid, pandaid)
0077     info = {}
0078     try:
0079         with open(meta, encoding='utf-8') as f:
0080             info = json.load(f) or {}
0081     except (OSError, ValueError):
0082         info = {}
0083     for path, status in ((out, 'captured'), (err, 'failed')):
0084         if os.path.exists(path):
0085             try:
0086                 if path.endswith('.gz'):
0087                     with gzip.open(path, 'rt', encoding='utf-8', errors='replace') as f:
0088                         body = f.read()
0089                 else:
0090                     with open(path, encoding='utf-8', errors='replace') as f:
0091                         body = f.read()
0092             except (OSError, EOFError) as e:                  # noqa: BLE001
0093                 logger.error('harvester stdout unreadable at %s: %s', path, e)
0094                 return None
0095             return {'pandaid': int(pandaid), 'status': status, 'path': path,
0096                     'body': body, 'bytes': len(body),
0097                     'captured_at': info.get('fetched'), 'source': info.get('source')}
0098     return None
0099
0100
0101 def _task_of(pandaid):
0102     from .models import EpicProdJob
0103     row = EpicProdJob.objects.filter(pandaid=int(pandaid)).only('jeditaskid').first()
0104     if row is not None and row.jeditaskid:
0105         return int(row.jeditaskid)
0106     try:
0107         with connections['panda'].cursor() as cursor:
0108             cursor.execute(f'SELECT "jeditaskid" FROM "{PANDA_SCHEMA}"."jobsarchived4" '
0109                            f'WHERE "pandaid" = %s', [int(pandaid)])
0110             r = cursor.fetchone()
0111             return int(r[0]) if r and r[0] else None
0112     except Exception as e:                                    # noqa: BLE001
0113         logger.error('harvester stdout: task of %s unknown: %s', pandaid, e)
0114         return None
0115
0116
0117 def cache_url(pilotid):
0118     """The PanDA-cache stdout URL a pilot id names, or ''."""
0119     url = str(pilotid or '').split('|')[0].strip()
0120     return url if ('/cache/' in url and 'pandaserver' in url) else ''
0121
0122
0123 def candidates(since, finished=False, limit=None):
0124     """Jobs ended since ``since`` whose pilot id names a cache stdout:
0125     failed, and finished when asked. ``[(pandaid, jeditaskid, url,
0126     jobstatus, endtime)]``."""
0127     statuses = ['failed'] + (['finished'] if finished else [])
0128     sql = f"""
0129         SELECT "pandaid", "jeditaskid", "pilotid", "jobstatus", "endtime"
0130         FROM "{PANDA_SCHEMA}"."jobsarchived4"
0131         WHERE "processingtype" = 'epicproduction'
0132           AND "jobstatus" = ANY(%s)
0133           AND "pilotid" LIKE '%%/cache/%%'
0134           AND "endtime" >= %s
0135         ORDER BY "endtime" DESC
0136     """
0137     if limit:
0138         sql += f' LIMIT {int(limit)}'
0139     try:
0140         with connections['panda'].cursor() as cursor:
0141             cursor.execute(sql, [statuses, since])
0142             rows = cursor.fetchall()
0143     except Exception as e:                                    # noqa: BLE001
0144         logger.error('harvester stdout candidate query failed: %s', e)
0145         return []
0146     out = []
0147     for pandaid, jeditaskid, pilotid, status, endtime in rows:
0148         url = cache_url(pilotid)
0149         if url and jeditaskid:
0150             out.append((int(pandaid), int(jeditaskid), url, status, endtime))
0151     return out
0152
0153
0154 def _fetch(url):
0155     """The stdout as gzip bytes and its text length. The server serves
0156     it gzip-encoded under a .out name and the bytes are kept as they
0157     came; the chain of its TLS certificate is not in this host's trust
0158     store, and no credential travels either way."""
0159     context = ssl._create_unverified_context()
0160     request = urllib.request.Request(
0161         url, headers={'User-Agent': 'swf-monitor/harvester-stdout',
0162                       'Accept-Encoding': 'gzip'})
0163     with urllib.request.urlopen(request, timeout=FETCH_TIMEOUT_S, context=context) as response:
0164         raw = response.read(FETCH_MAX_BYTES)
0165         encoding = (response.headers.get('Content-Encoding') or '').lower()
0166     if encoding == 'gzip' or raw[:2] == b'\x1f\x8b':
0167         text_len = len(gzip.decompress(raw))
0168         return raw, text_len
0169     return gzip.compress(raw), len(raw)
0170
0171
0172 def _write(path, text):
0173     tmp = f'{path}.part'
0174     with open(tmp, 'w', encoding='utf-8') as f:
0175         f.write(text)
0176     os.replace(tmp, path)
0177
0178
0179 def _write_bytes(path, data):
0180     tmp = f'{path}.part'
0181     with open(tmp, 'wb') as f:
0182         f.write(data)
0183     os.replace(tmp, path)
0184
0185
0186 def capture(since=None, finished=None, limit=None, root=None):
0187     """Copy every candidate's stdout not yet stored. Idempotent: a job
0188     with a body or a failure marker is skipped. Returns the counts and
0189     the jobs captured and gone."""
0190     root = root or store_root()
0191     if finished is None:
0192         finished = bool(setting('finished'))
0193     since = since or (datetime.now(dt_timezone.utc) - timedelta(days=int(setting('backfill_days') or 7)))
0194     rows = candidates(since, finished=finished, limit=limit)
0195     counts = {'candidates': len(rows), 'skipped': 0, 'captured': 0, 'gone': 0, 'failed': 0,
0196               'bytes': 0}
0197     captured, gone = [], []
0198     now = datetime.now(dt_timezone.utc).isoformat()
0199     for pandaid, jeditaskid, url, status, endtime in rows:
0200         d, out, meta, err = _paths(root, jeditaskid, pandaid)
0201         if os.path.exists(out) or os.path.exists(err):
0202             counts['skipped'] += 1
0203             continue
0204         os.makedirs(d, exist_ok=True)
0205         try:
0206             body, text_len = _fetch(url)
0207         except Exception as e:                                # noqa: BLE001
0208             is_gone = getattr(e, 'code', None) == 404
0209             reason = (f'{"gone from the PanDA cache" if is_gone else e.__class__.__name__}: {e}\n'
0210                       f'source: {url}\nchecked: {now}\n')
0211             _write(err, reason)
0212             _write(meta, json.dumps({'source': url, 'fetched': now, 'status': 'gone' if is_gone else 'failed',
0213                                      'jobstatus': status, 'endtime': endtime.isoformat() if endtime else None}))
0214             counts['gone' if is_gone else 'failed'] += 1
0215             if is_gone:
0216                 gone.append(pandaid)
0217             continue
0218         _write_bytes(out, body)
0219         _write(meta, json.dumps({'source': url, 'fetched': now, 'status': 'captured',
0220                                  'bytes': text_len, 'stored_bytes': len(body),
0221                                  'capped': len(body) >= FETCH_MAX_BYTES,
0222                                  'jobstatus': status,
0223                                  'endtime': endtime.isoformat() if endtime else None}))
0224         counts['captured'] += 1
0225         counts['bytes'] += len(body)
0226         captured.append(pandaid)
0227     return counts, captured, gone
0228
0229
0230 def prune(keep_days=None, root=None, dry_run=False):
0231     """Remove copies older than ``keep_days`` (their fetched time); 0 or
0232     None keeps everything. Returns the paths removed."""
0233     root = root or store_root()
0234     keep_days = int(keep_days if keep_days is not None else (setting('keep_days') or 0))
0235     if keep_days <= 0:
0236         return []
0237     cutoff = datetime.now(dt_timezone.utc) - timedelta(days=keep_days)
0238     removed = []
0239     try:
0240         tasks = os.listdir(root)
0241     except OSError:
0242         return removed
0243     for task in tasks:
0244         tdir = os.path.join(root, task)
0245         if not os.path.isdir(tdir):
0246             continue
0247         for name in os.listdir(tdir):
0248             if not name.endswith('.json'):
0249                 continue
0250             meta = os.path.join(tdir, name)
0251             try:
0252                 with open(meta, encoding='utf-8') as f:
0253                     fetched = datetime.fromisoformat((json.load(f) or {}).get('fetched'))
0254             except (OSError, ValueError, TypeError):
0255                 continue
0256             if fetched >= cutoff:
0257                 continue
0258             base = meta[:-5]
0259             for path in (f'{base}.stdout.gz', f'{base}.stdout', f'{base}.error', meta):
0260                 if os.path.exists(path):
0261                     removed.append(path)
0262                     if not dry_run:
0263                         try:
0264                             os.unlink(path)
0265                         except OSError as e:                  # noqa: BLE001
0266                             logger.error('harvester stdout prune failed for %s: %s', path, e)
0267     return removed