swf-monitor/scripts/backfill-catalog-intervals.py

0001 #!/usr/bin/env python3
0002 """Backfill per-interval catalog snaps into snap history
0003 (docs/SNAPPER_CATALOG.md, Backfill).
0004
0005 Reconstructs the catalog component's registrations group for every
0006 five-minute interval from the digest every production job carries in
0007 the PanDA record (payloadRegistration, payloadExit, payloadVersion in
0008 jobmetrics), and the latency group from the payload reports at hand
0009 (the metatable for finished jobs, the swept reports for failed ones).
0010 One synthetic snap per non-empty interval is written under capture
0011 policy ``backfill-catalog-v1``, reconstructed evidence distinguishable
0012 from observed snaps, carrying only the catalog component in the live
0013 publisher's envelope shape. The probe and registrar groups are absent
0014 in a backfilled snap: the probe was not made at the time, and the
0015 backlog is a gauge of the moment; the card and the curves treat an
0016 absent group as not recorded, never as zero.
0017
0018 Idempotent: --apply first removes prior backfill-catalog-v1 snaps for
0019 the scope, and writes only intervals ending strictly before the first
0020 live catalog snap's interval start (or up to now when none exists).
0021 Dry run by default.
0022
0023 Run under the venv with the swf-monitor project on the path:
0024     cd <swf-monitor>/src && source <venv>/bin/activate && source ~/.env
0025     python <swf-monitor>/scripts/backfill-catalog-intervals.py \\
0026         [--since 2026-09-15] [--step-minutes 5] [--apply]
0027 """
0028 import argparse
0029 import datetime as dt
0030 import os
0031 import sys
0032
0033 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0034 import django  # noqa: E402
0035 django.setup()
0036
0037 from django.utils import timezone  # noqa: E402
0038
0039 from monitor_app.snapper_catalog import (  # noqa: E402
0040     ASSESSMENT_POLICY_VERSION, CATALOG_REGISTRATION, COMPONENT_NAME,
0041     COMPONENT_SCHEMA_VERSION, PUBLISHER_IDENTITY, SCOPE, _ended_jobs,
0042     _iso_utc, _latency_rows, latency_from_rows, registrations_from_rows,
0043 )
0044 from snapper_ai.models import SystemSnap  # noqa: E402
0045
0046 CAPTURE_POLICY = 'backfill-catalog-v1'
0047
0048
0049 def _interval_snaps(start, end, step):
0050     out = []
0051     lead = start
0052     while lead < end:
0053         upto = min(lead + step, end)
0054         rows = _ended_jobs(lead, upto)
0055         if rows:
0056             registrations = registrations_from_rows(rows)
0057             latency = latency_from_rows(_latency_rows(lead, upto))
0058             out.append((lead, upto, registrations, latency))
0059         lead = upto
0060     return out
0061
0062
0063 def main():
0064     parser = argparse.ArgumentParser(
0065         description='Backfill per-interval catalog snaps into epicprod snap history.')
0066     parser.add_argument('--since', default='2026-09-15',
0067                         help='first day to reconstruct (ET midnight), default 2026-09-15')
0068     parser.add_argument('--step-minutes', type=int, default=5)
0069     parser.add_argument('--apply', action='store_true', help='write the snaps (dry run without)')
0070     args = parser.parse_args()
0071
0072     now = timezone.now()
0073     step = dt.timedelta(minutes=args.step_minutes)
0074     live_first = (SystemSnap.objects
0075                   .filter(scope=SCOPE, state__components__has_key=COMPONENT_NAME)
0076                   .exclude(capture_policy=CAPTURE_POLICY)
0077                   .order_by('snap_time').values('state').first())
0078     if live_first:
0079         seam_iso = live_first['state']['components'][COMPONENT_NAME]['data']['interval']['start']
0080         seam = dt.datetime.fromisoformat(seam_iso.replace('Z', '+00:00'))
0081     else:
0082         seam = now
0083     from zoneinfo import ZoneInfo
0084     start = dt.datetime.strptime(args.since, '%Y-%m-%d').replace(tzinfo=ZoneInfo('America/New_York'))
0085     start = start.astimezone(dt.timezone.utc)
0086     grid_end = seam.replace(second=0, microsecond=0)
0087     grid_end -= dt.timedelta(minutes=grid_end.minute % args.step_minutes)
0088     print(f'grid: {args.step_minutes}m step, {start.isoformat()} -> {grid_end.isoformat()}')
0089     print('seam: ' + (f'first live catalog interval starts {seam.isoformat()}' if live_first
0090                       else 'now (no live catalog snap yet)'))
0091
0092     snaps = []
0093     day = start
0094     while day < grid_end:
0095         day_end = min(day + dt.timedelta(days=1), grid_end)
0096         snaps.extend(_interval_snaps(day, day_end, step))
0097         day = day_end
0098     if live_first and seam > grid_end:
0099         snaps.extend(_interval_snaps(grid_end, seam, step))
0100
0101     total = sum(r['jobs'] for _, _, r, _ in snaps)
0102     lost = sum((r['by_outcome'] or {}).get('failed', 0) for _, _, r, _ in snaps)
0103     print(f'non-empty intervals: {len(snaps)}, jobs {total}, registrations lost {lost}')
0104     for lead, upto, r, lat in sorted(snaps, key=lambda s: -(s[2]['by_outcome'].get('failed', 0)))[:3]:
0105         print(f'  {upto.isoformat()}: {r["jobs"]} jobs, {r["by_outcome"]}, exits {r["failed_by_exit"]}, '
0106               f'registered median {(lat.get("registered") or {}).get("median_s")} s')
0107     if not args.apply:
0108         print('\ndry run; nothing written; --apply writes the snaps')
0109         return 0
0110
0111     removed = SystemSnap.objects.filter(scope=SCOPE, capture_policy=CAPTURE_POLICY).delete()
0112     written = 0
0113     for lead, upto, registrations, latency in snaps:
0114         SystemSnap.objects.create(
0115             scope=SCOPE,
0116             # Three seconds past the grid instant: the errors backfill
0117             # holds +2 s and the live captures the aligned boundaries.
0118             snap_time=upto + dt.timedelta(seconds=3),
0119             observed_at=now, completed_at=now, snap_schema_version=1,
0120             capture_policy=CAPTURE_POLICY, encoding='full', reasons=['backfill'],
0121             changed_components=[COMPONENT_NAME],
0122             component_revisions={COMPONENT_NAME: 0},
0123             registration_versions={COMPONENT_NAME: COMPONENT_SCHEMA_VERSION},
0124             component_hashes={}, state_hash='',
0125             state={'components': {COMPONENT_NAME: {
0126                 'v': COMPONENT_SCHEMA_VERSION,
0127                 'data': {'interval': {'start': _iso_utc(lead), 'end': _iso_utc(upto)},
0128                          'registrations': registrations, 'latency': latency,
0129                          'backfilled': True},
0130                 'registration': CATALOG_REGISTRATION,
0131                 'revision': 0,
0132                 'registration_version': COMPONENT_SCHEMA_VERSION,
0133                 'assessed_at': upto.isoformat(), 'source_as_of': upto.isoformat(),
0134                 'accepted_at': now.isoformat(),
0135                 'assessment_policy': ASSESSMENT_POLICY_VERSION,
0136                 'publisher_identity': PUBLISHER_IDENTITY,
0137             }}},
0138         )
0139         written += 1
0140     print(f'\napplied: removed prior backfill {removed[0]}, wrote {written} snaps')
0141     return 0
0142
0143
0144 if __name__ == '__main__':
0145     sys.exit(main())