swf-monitor/scripts/augment-errors-entries.py

0001 """Add missing trailing entry fields to the errors entries of live snaps
0002 recorded before the current schema version.
0003
0004 The error-state component's entry rows grow at the end: the terminal
0005 status at schema version 3, the payload's transformation exit code at
0006 version 4, the core-seconds held at version 5 (docs/SNAPPER_ERRORS.md).
0007 Live snaps captured before a field joined carry rows without it. This
0008 augmentation adds every trailing field a row lacks, in place, looked up
0009 by PanDA job id from the job records with the publisher's own
0010 definitions, and recomputes the snap's component hash and state hash
0011 by the capture contract (snapper_ai capture: the sha256 of the
0012 canonical JSON of the component data, and of the components document).
0013 Every other field of the row and the snap stays as recorded; the
0014 component document carries an ``augmented`` note per field naming when
0015 it was added, how many rows received it, and how many had no job record
0016 left to read (those take the field's empty value). Overflow fold keys
0017 name no job and keep their recorded form.
0018
0019 Idempotent: rows already carrying every field are skipped, and a snap
0020 with nothing to add is not written. Dry-run default; ``--limit N``
0021 bounds an apply to the first N snaps needing the augmentation, for a
0022 trial.
0023
0024 Run under the venv with the swf-monitor project on the path:
0025
0026     cd <swf-monitor>/src && source <venv>/bin/activate && source ~/.env
0027     python <swf-monitor>/scripts/augment-errors-entries.py [--apply] [--limit N]
0028 """
0029
0030 import argparse
0031 import hashlib
0032 import json
0033 import os
0034 import sys
0035
0036 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0037
0038 import django  # noqa: E402
0039
0040 django.setup()
0041
0042 from django.db import connections, transaction  # noqa: E402
0043 from django.utils import timezone  # noqa: E402
0044
0045 from monitor_app.panda.constants import PANDA_SCHEMA  # noqa: E402
0046 from monitor_app.snapper_errors import (  # noqa: E402
0047     ENTRY_FIELDS,
0048     HELD_SQL,
0049     _exit_key,
0050 )
0051 from snapper_ai.models import SystemSnap  # noqa: E402
0052
0053 SCOPE = 'epicprod'
0054 COMPONENT = 'errors'
0055 BACKFILL_POLICY = 'backfill-errors-v1'
0056 WIDTH = len(ENTRY_FIELDS)
0057 # The value a field takes when the job has no record left to read.
0058 EMPTY = {'status': '', 'exitcode': '', 'held': 0}
0059
0060
0061 def _canonical_hash(value):
0062     """The capture contract's hash: sha256 of the canonical JSON."""
0063     encoded = json.dumps(
0064         value, allow_nan=False, ensure_ascii=False,
0065         separators=(',', ':'), sort_keys=True).encode('utf-8')
0066     return hashlib.sha256(encoded).hexdigest()
0067
0068
0069 def _job_fields(pandaids):
0070     """{pandaid: {field: value}} for the fields the job records supply
0071     — status, exitcode and held — over both tables. A job no longer in
0072     either table is absent from the result."""
0073     ids = sorted({int(p) for p in pandaids if p})
0074     out = {}
0075     if not ids:
0076         return out
0077     columns = (
0078         '"pandaid", "jobstatus", "transexitcode", "starttime", '
0079         '"endtime" AS "ended", "actualcorecount", "corecount"'
0080     )
0081     sql = (
0082         f'SELECT "pandaid", "jobstatus", "transexitcode", {HELD_SQL} '
0083         f'FROM (SELECT {columns} FROM "{PANDA_SCHEMA}"."jobsarchived4" '
0084         f'WHERE "pandaid" = ANY(%s) '
0085         f'UNION SELECT {columns} FROM "{PANDA_SCHEMA}"."jobsactive4" '
0086         f'WHERE "pandaid" = ANY(%s)) jobs'
0087     )
0088     with connections['panda'].cursor() as cursor:
0089         cursor.execute(sql, [ids, ids])
0090         for pandaid, status, exitcode, held in cursor.fetchall():
0091             out[int(pandaid)] = {
0092                 'status': str(status or ''),
0093                 'exitcode': _exit_key(exitcode),
0094                 'held': int(held or 0),
0095             }
0096     return out
0097
0098
0099 def _augment(rows, jobs):
0100     """Add every missing trailing field to each short row in place.
0101     Returns {field: (rows added to, rows whose job had no record)}."""
0102     stats = {}
0103     for row in rows:
0104         if len(row) >= WIDTH:
0105             continue
0106         job = jobs.get(int(row[0] or 0))
0107         for index in range(len(row), WIDTH):
0108             field = ENTRY_FIELDS[index]
0109             added, unresolved = stats.get(field) or (0, 0)
0110             if job is None:
0111                 row.append(EMPTY.get(field, ''))
0112                 unresolved += 1
0113             else:
0114                 row.append(job.get(field, EMPTY.get(field, '')))
0115             stats[field] = (added + 1, unresolved)
0116     return stats
0117
0118
0119 def _note(doc, stats, now_iso):
0120     """The augmented note: one entry per field added. An earlier note
0121     in the first form (a timestamp under 'exitcode' with 'rows' and
0122     'unresolved' beside it) is carried into the per-field form."""
0123     note = doc.get('augmented') or {}
0124     if isinstance(note.get('exitcode'), str):
0125         note = {'exitcode': {'at': note['exitcode'],
0126                              'rows': note.get('rows', 0),
0127                              'unresolved': note.get('unresolved', 0)}}
0128     for field, (added, unresolved) in stats.items():
0129         note[field] = {'at': now_iso, 'rows': added,
0130                        'unresolved': unresolved}
0131     doc['augmented'] = note
0132
0133
0134 def main():
0135     parser = argparse.ArgumentParser(
0136         description='Add missing trailing entry fields to pre-current '
0137                     'errors entries in live epicprod snaps.')
0138     parser.add_argument('--apply', action='store_true',
0139                         help='write the augmented snaps (default: dry run)')
0140     parser.add_argument('--limit', type=int, default=0,
0141                         help='with --apply, augment at most N snaps')
0142     args = parser.parse_args()
0143
0144     snaps = (SystemSnap.objects
0145              .filter(scope=SCOPE,
0146                      state__components__errors__data__has_key='entries')
0147              .exclude(capture_policy=BACKFILL_POLICY)
0148              .order_by('snap_time'))
0149     now_iso = timezone.now().isoformat()
0150     cache = {}
0151     seen = to_augment = written = 0
0152     totals = {}
0153     first = last = None
0154     for snap in snaps.iterator(chunk_size=100):
0155         seen += 1
0156         doc = snap.state['components'][COMPONENT]
0157         data = doc['data']
0158         rows = data.get('entries') or []
0159         if not any(len(row) < WIDTH for row in rows):
0160             continue
0161         interval = data.get('interval') or {}
0162         key = (interval.get('start'), interval.get('end'))
0163         if key not in cache:
0164             cache[key] = _job_fields(
0165                 row[0] for row in rows if len(row) < WIDTH)
0166         stats = _augment(rows, cache[key])
0167         to_augment += 1
0168         for field, (added, unresolved) in stats.items():
0169             a, u = totals.get(field) or (0, 0)
0170             totals[field] = (a + added, u + unresolved)
0171         first = first or snap.snap_time
0172         last = snap.snap_time
0173         if not args.apply or (args.limit and written >= args.limit):
0174             continue
0175         _note(doc, stats, now_iso)
0176         hashes = dict(snap.component_hashes or {})
0177         hashes[COMPONENT] = _canonical_hash(data)
0178         state_hash = _canonical_hash(snap.state['components'])
0179         with transaction.atomic():
0180             SystemSnap.objects.filter(pk=snap.pk).update(
0181                 state=snap.state, component_hashes=hashes,
0182                 state_hash=state_hash)
0183         written += 1
0184
0185     print(f'live snaps carrying entries: {seen}; needing augmentation: '
0186           f'{to_augment} ({first} to {last}); distinct intervals looked '
0187           f'up: {len(cache)}')
0188     for field, (added, unresolved) in totals.items():
0189         print(f'  {field}: added to {added} rows; {unresolved} rows '
0190               f'without a job record (empty value)')
0191     if not args.apply:
0192         print('\ndry run — nothing written; --apply augments the snaps')
0193         return 0
0194     print(f'\napplied: augmented {written} snaps'
0195           + (f' (limit {args.limit})' if args.limit else ''))
0196     return 0
0197
0198
0199 if __name__ == '__main__':
0200     sys.exit(main())