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

0001 """Backfill per-interval error-event snaps into snap history.
0002
0003 Reconstructs the errors component's version-3 interval entries — every
0004 job that ended faulty within each grid interval, as rows of
0005 (pandaid, jeditaskid, category, endtime, status) — from the recorded
0006 job history in the PanDA database, on a regular grid at the live
0007 capture cadence. One synthetic snap per non-empty interval is written with
0008 capture policy ``backfill-errors-v1`` — reconstructed evidence,
0009 explicitly distinguishable from observed snaps — carrying only the
0010 errors component in the live publisher's envelope shape. A job reports
0011 errors once, upon completion, so each failed job lands in exactly one
0012 interval; the backfilled intervals and the live interval chain form
0013 one consistent record. Empty intervals write nothing, matching the
0014 live quiet behavior.
0015
0016 Idempotent: --apply first removes prior backfill-errors-v1 snaps for
0017 the scope, and writes only intervals ending strictly before the
0018 earliest live snap carrying version-2 entries (or up to now when none
0019 exists yet). Earlier version-1 counter snaps are ignored as a
0020 boundary: the entry record supersedes them over the same span.
0021 Dry-run default.
0022
0023 Run under the venv with the swf-monitor project on the path:
0024
0025     cd <swf-monitor>/src && source <venv>/bin/activate && source ~/.env
0026     python <swf-monitor>/scripts/backfill-errors-entries.py \\
0027         [--days 30] [--step-minutes 5] [--apply]
0028 """
0029
0030 import argparse
0031 import datetime as dt
0032 import os
0033 import sys
0034
0035 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0036
0037 import django  # noqa: E402
0038
0039 django.setup()
0040
0041 from django.utils import timezone  # noqa: E402
0042
0043 from monitor_app.snapper_errors import (  # noqa: E402
0044     ASSESSMENT_POLICY_VERSION,
0045     ERRORS_REGISTRATION,
0046     MAX_ENTRIES,
0047     PUBLISHER_IDENTITY,
0048     _category_key,
0049     _entry_rows,
0050     _iso_utc,
0051 )
0052 from snapper_ai.models import SystemSnap  # noqa: E402
0053
0054 CAPTURE_POLICY = 'backfill-errors-v1'
0055 SCOPE = 'epicprod'
0056 COMPONENT = 'errors'
0057
0058
0059 def _interval_snaps(day_start, day_end, step):
0060     """(interval_start, interval_end, entries, overflow) per non-empty
0061     grid interval in one slice, from one query over the job records.
0062     Interval ends are grid edges capped at the slice end, so a partial
0063     final interval closes exactly at day_end."""
0064     rows = _entry_rows(day_start, day_end)
0065     intervals = []
0066     lead = day_start
0067     edge = min(day_start + step, day_end)
0068     entries = []
0069     overflow_total = 0
0070     overflow_categories = {}
0071
0072     def close():
0073         nonlocal entries, overflow_total, overflow_categories
0074         if entries:
0075             overflow = (
0076                 {'total': overflow_total,
0077                  'by_category': overflow_categories}
0078                 if overflow_total else None
0079             )
0080             intervals.append((lead, edge, list(entries), overflow))
0081         entries = []
0082         overflow_total = 0
0083         overflow_categories = {}
0084
0085     for pandaid, taskid, comp, code, endtime, status in rows:
0086         stamp = endtime if endtime.tzinfo else endtime.replace(
0087             tzinfo=dt.timezone.utc)
0088         while stamp > edge:
0089             close()
0090             lead = edge
0091             edge = min(edge + step, day_end)
0092         category = _category_key(comp, code)
0093         if len(entries) < MAX_ENTRIES:
0094             entries.append([
0095                 int(pandaid or 0),
0096                 int(taskid or 0),
0097                 category,
0098                 _iso_utc(endtime),
0099                 str(status or ''),
0100             ])
0101         else:
0102             overflow_total += 1
0103             fold_key = f'{category}@{status or ""}'
0104             overflow_categories[fold_key] = (
0105                 overflow_categories.get(fold_key) or 0) + 1
0106     close()
0107     return intervals
0108
0109
0110 def main():
0111     parser = argparse.ArgumentParser(
0112         description='Backfill per-interval error-event snaps into '
0113                     'epicprod snap history.')
0114     parser.add_argument('--days', type=int, default=30,
0115                         help='trailing horizon in days (default 30)')
0116     parser.add_argument('--step-minutes', type=int, default=5,
0117                         help='grid spacing in minutes (default 5, the '
0118                              'live capture cadence)')
0119     parser.add_argument('--apply', action='store_true',
0120                         help='write the snaps (dry run without)')
0121     args = parser.parse_args()
0122
0123     now = timezone.now()
0124     step = dt.timedelta(minutes=args.step_minutes)
0125
0126     # The seam is the START of the first live entries interval: the
0127     # backfilled record must tile exactly against the live chain, every
0128     # job in exactly one interval. Until the version-2 maintainer has
0129     # published (deploy first, then backfill), the seam falls back to
0130     # now — fine for a dry run, re-run after deploy for the real seam.
0131     live_first = (
0132         SystemSnap.objects
0133         .filter(scope=SCOPE,
0134                 state__components__errors__data__has_key='entries')
0135         .exclude(capture_policy=CAPTURE_POLICY)
0136         .order_by('snap_time')
0137         .values('state')
0138         .first())
0139     if live_first:
0140         seam_iso = (live_first['state']['components']['errors']['data']
0141                     ['interval']['start'])
0142         seam = dt.datetime.fromisoformat(seam_iso.replace('Z', '+00:00'))
0143     else:
0144         seam = now
0145     start = seam - dt.timedelta(days=args.days)
0146     # Align the grid to round step boundaries.
0147     start = start.replace(second=0, microsecond=0)
0148     start -= dt.timedelta(minutes=start.minute % args.step_minutes)
0149     grid_end = seam.replace(second=0, microsecond=0)
0150     grid_end -= dt.timedelta(minutes=grid_end.minute % args.step_minutes)
0151
0152     print(f'grid: {args.step_minutes}m step, '
0153           f'{start.isoformat()} -> {grid_end.isoformat()}')
0154     print(f'seam: '
0155           f'{"first live entries interval starts " + seam.isoformat() if live_first else "now (no live entries snap yet — deploy first, then re-run)"}')
0156
0157     snaps = []
0158     day = start
0159     while day < grid_end:
0160         day_end = min(day + dt.timedelta(days=1), grid_end)
0161         snaps.extend(_interval_snaps(day, day_end, step))
0162         day = day_end
0163     if live_first and seam > grid_end:
0164         # One final partial interval closes the record exactly at the
0165         # live chain's first interval start.
0166         snaps.extend(_interval_snaps(grid_end, seam, step))
0167
0168     total_entries = sum(len(entries) for _, _, entries, _ in snaps)
0169     total_overflow = sum(
0170         (overflow or {}).get('total') or 0 for _, _, _, overflow in snaps)
0171     busiest = max(snaps, key=lambda s: len(s[2]) + ((s[3] or {}).get('total') or 0),
0172                   default=None)
0173     print(f'non-empty intervals: {len(snaps)}, entries {total_entries}, '
0174           f'overflow {total_overflow}')
0175     if busiest:
0176         _, end_stamp, entries, overflow = busiest
0177         print(f'busiest interval ends {end_stamp.isoformat()}: '
0178               f'{len(entries)} entries'
0179               + (f' + {overflow["total"]} overflow' if overflow else ''))
0180     for _, end_stamp, entries, overflow in snaps[-3:]:
0181         categories = {}
0182         for _, _, category, _, _ in entries:
0183             categories[category] = (categories.get(category) or 0) + 1
0184         top = sorted(categories.items(), key=lambda kv: -kv[1])[:3]
0185         print(f'  {end_stamp.isoformat()}: {len(entries)} entries, top {top}')
0186
0187     if not args.apply:
0188         print('\ndry run — nothing written; --apply writes the snaps')
0189         return 0
0190
0191     removed = SystemSnap.objects.filter(
0192         scope=SCOPE, capture_policy=CAPTURE_POLICY).delete()
0193     written = 0
0194     for lead, end_stamp, entries, overflow in snaps:
0195         # Two seconds past the grid instant: live captures land on
0196         # aligned 30-second boundaries and the hourly counter backfill
0197         # (backfill-panda-counters.py) owns the one-second offset, so
0198         # this stamp never collides under the (scope, snap_time)
0199         # uniqueness.
0200         SystemSnap.objects.create(
0201             scope=SCOPE,
0202             snap_time=end_stamp + dt.timedelta(seconds=2),
0203             observed_at=now,
0204             completed_at=now,
0205             snap_schema_version=1,
0206             capture_policy=CAPTURE_POLICY,
0207             encoding='full',
0208             reasons=['backfill'],
0209             changed_components=[COMPONENT],
0210             component_revisions={COMPONENT: 0},
0211             registration_versions={COMPONENT: 3},
0212             component_hashes={},
0213             state_hash='',
0214             state={'components': {COMPONENT: {
0215                 'v': 3,
0216                 'data': dict(
0217                     {'interval': {'start': _iso_utc(lead),
0218                                   'end': _iso_utc(end_stamp)},
0219                      'entries': entries},
0220                     **({'overflow': overflow} if overflow else {})),
0221                 'registration': ERRORS_REGISTRATION,
0222                 'revision': 0,
0223                 'registration_version': 3,
0224                 'assessed_at': end_stamp.isoformat(),
0225                 'source_as_of': end_stamp.isoformat(),
0226                 'accepted_at': now.isoformat(),
0227                 'assessment_policy': ASSESSMENT_POLICY_VERSION,
0228                 'publisher_identity': PUBLISHER_IDENTITY,
0229             }}},
0230         )
0231         written += 1
0232     print(f'\napplied: removed prior backfill {removed[0]}, '
0233           f'wrote {written} snaps')
0234     return 0
0235
0236
0237 if __name__ == '__main__':
0238     sys.exit(main())