swf-monitor/scripts/backfill-panda-counters.py

0001 """Backfill cumulative terminal-outcome counters into snap history.
0002
0003 Reconstructs the panda component's version-5 cumulative terminal
0004 counters (jobs.cum, per-site cum and cum_failed_by_class) at a regular
0005 grid of historical instants, from the complete recorded job history in
0006 the PanDA database. One synthetic snap per grid instant is written
0007 with capture policy ``backfill-panda-v1`` — reconstructed evidence,
0008 explicitly distinguishable from observed snaps — carrying only the
0009 counter fields in the live publisher's envelope shape. The counters
0010 are absolute counts of terminal events with end times at or before
0011 each instant, the same origin the live publisher seeds from, so the
0012 backfilled grid and the live counter chain form one consistent record.
0013
0014 Idempotent: --apply first removes prior backfill-panda-v1 snaps for
0015 the scope, and writes only instants strictly before the earliest live
0016 snap carrying jobs.cum (or up to now when none exists yet). Dry-run
0017 default.
0018
0019 Run under the venv with the swf-monitor project on the path:
0020
0021     cd <swf-monitor>/src && source <venv>/bin/activate && source ~/.env
0022     python <swf-monitor>/scripts/backfill-panda-counters.py \\
0023         [--start 2026-07-01] [--step-hours 1] [--apply]
0024 """
0025
0026 import argparse
0027 import datetime as dt
0028 import os
0029 import sys
0030
0031 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0032
0033 import django  # noqa: E402
0034
0035 django.setup()
0036
0037 from django.db import connections  # noqa: E402
0038 from django.utils import timezone  # noqa: E402
0039
0040 from monitor_app.panda.constants import (  # noqa: E402
0041     ERROR_COMPONENTS,
0042     PANDA_SCHEMA,
0043 )
0044 from monitor_app.snapper_panda import (  # noqa: E402
0045     ASSESSMENT_POLICY_VERSION,
0046     MAX_SITES,
0047     PANDA_REGISTRATION,
0048     PUBLISHER_IDENTITY,
0049     TERMINAL_JOB_STATUSES,
0050 )
0051 from snapper_ai.models import SystemSnap  # noqa: E402
0052
0053 CAPTURE_POLICY = 'backfill-panda-v1'
0054
0055
0056 def _hourly_outcome_buckets(until):
0057     """Terminal-outcome counts per (hour bucket, site, status, failure
0058     class) over the full recorded history up to ``until``, deduplicated
0059     across the active and archived tables."""
0060     class_case = ' '.join(
0061         f'WHEN "{c["code"]}" > 0 THEN \'{c["name"]}\''
0062         for c in ERROR_COMPONENTS
0063     )
0064     err_fields = ', '.join(f'"{c["code"]}"' for c in ERROR_COMPONENTS)
0065     placeholders = ', '.join(['%s'] * len(TERMINAL_JOB_STATUSES))
0066     bounds = f'"endtime" <= %s AND "jobstatus" IN ({placeholders})'
0067     params = [until, *TERMINAL_JOB_STATUSES]
0068     sql = f"""
0069         SELECT date_trunc('hour', "endtime") AS bucket,
0070                COALESCE("computingsite", 'unknown'), "jobstatus",
0071                CASE WHEN "jobstatus" = 'failed'
0072                     THEN CASE {class_case} ELSE 'other' END
0073                     ELSE '' END,
0074                COUNT(*)
0075         FROM (
0076             SELECT "pandaid", "endtime", "jobstatus",
0077                    "computingsite", {err_fields}
0078             FROM "{PANDA_SCHEMA}"."jobsactive4" WHERE {bounds}
0079             UNION
0080             SELECT "pandaid", "endtime", "jobstatus",
0081                    "computingsite", {err_fields}
0082             FROM "{PANDA_SCHEMA}"."jobsarchived4" WHERE {bounds}
0083         ) completed
0084         GROUP BY 1, 2, 3, 4
0085         ORDER BY 1
0086     """
0087     with connections['panda'].cursor() as cursor:
0088         cursor.execute(sql, params + params)
0089         return cursor.fetchall()
0090
0091
0092 def _bump(counter, key, count):
0093     counter[key] = int(counter.get(key) or 0) + count
0094
0095
0096 def _counters_at_instants(instants, buckets):
0097     """Walk the hour buckets once, emitting deep-copied counter maps at
0098     each grid instant. Bucket times are naive UTC from the database;
0099     a bucket belongs to instant t when the whole hour ends at or
0100     before t."""
0101     scope_cum = {}
0102     site_cums = {}
0103     results = []
0104     index = 0
0105     for instant in instants:
0106         cutoff = instant.replace(tzinfo=None)
0107         while index < len(buckets):
0108             bucket, site, status, fclass, count = buckets[index]
0109             if bucket + dt.timedelta(hours=1) > cutoff:
0110                 break
0111             count = int(count or 0)
0112             _bump(scope_cum, status, count)
0113             entry = site_cums.setdefault(
0114                 str(site or 'unknown'), {'cum': {}, 'classes': {}})
0115             _bump(entry['cum'], status, count)
0116             if status == 'failed' and fclass:
0117                 _bump(entry['classes'], fclass, count)
0118             index += 1
0119         ranked = sorted(
0120             site_cums,
0121             key=lambda name: (-sum(site_cums[name]['cum'].values()), name),
0122         )[:MAX_SITES]
0123         sites = {}
0124         for name in ranked:
0125             entry = site_cums[name]
0126             block = {'cum': dict(entry['cum'])}
0127             if entry['classes']:
0128                 block['cum_failed_by_class'] = dict(entry['classes'])
0129             sites[name] = block
0130         results.append((instant, dict(scope_cum), sites))
0131     return results
0132
0133
0134 def main():
0135     parser = argparse.ArgumentParser(
0136         description='Backfill cumulative terminal-outcome counters '
0137                     'into epicprod snap history.')
0138     parser.add_argument('--start', default='2026-07-01',
0139                         help='first grid instant, UTC date or ISO '
0140                              '(default 2026-07-01)')
0141     parser.add_argument('--step-hours', type=float, default=1.0,
0142                         help='grid spacing in hours (default 1)')
0143     parser.add_argument('--apply', action='store_true',
0144                         help='write the snaps (dry run without)')
0145     args = parser.parse_args()
0146
0147     start = dt.datetime.fromisoformat(args.start)
0148     if start.tzinfo is None:
0149         start = start.replace(tzinfo=dt.timezone.utc)
0150     now = timezone.now()
0151
0152     live_first = (
0153         SystemSnap.objects
0154         .filter(scope='epicprod',
0155                 state__components__panda__data__jobs__has_key='cum')
0156         .exclude(capture_policy=CAPTURE_POLICY)
0157         .order_by('snap_time')
0158         .values_list('snap_time', flat=True)
0159         .first())
0160     end = live_first or now
0161     if start >= end:
0162         print(f'nothing to do: start {start} is not before end {end}')
0163         return 0
0164
0165     instants = []
0166     step = dt.timedelta(hours=args.step_hours)
0167     instant = start
0168     while instant < end:
0169         instants.append(instant)
0170         instant = instant + step
0171
0172     buckets = _hourly_outcome_buckets(end)
0173     results = _counters_at_instants(instants, buckets)
0174
0175     print(f'grid: {len(instants)} instants, {args.step_hours}h step, '
0176           f'{start.isoformat()} -> {instants[-1].isoformat()}')
0177     print(f'end boundary: '
0178           f'{"earliest live counter snap " + end.isoformat() if live_first else "now"}')
0179     print(f'hour buckets: {len(buckets)}')
0180     for instant, scope_cum, sites in results[-3:]:
0181         google = sites.get('BNL_ePIC_GOOGLE') or {}
0182         print(f'  {instant.isoformat()}: scope {scope_cum} | '
0183               f'BNL_ePIC_GOOGLE {google.get("cum")} '
0184               f'{google.get("cum_failed_by_class")}')
0185
0186     if not args.apply:
0187         print('\ndry run — nothing written; --apply writes the snaps')
0188         return 0
0189
0190     removed = SystemSnap.objects.filter(
0191         scope='epicprod', capture_policy=CAPTURE_POLICY).delete()
0192     written = 0
0193     for instant, scope_cum, sites in results:
0194         # One second past the grid instant: live captures land on
0195         # aligned 30-second boundaries, so the stamp never collides
0196         # with a real snap under the (scope, snap_time) uniqueness.
0197         SystemSnap.objects.create(
0198             scope='epicprod',
0199             snap_time=instant + dt.timedelta(seconds=1),
0200             observed_at=now,
0201             completed_at=now,
0202             snap_schema_version=1,
0203             capture_policy=CAPTURE_POLICY,
0204             encoding='full',
0205             reasons=['backfill'],
0206             changed_components=['panda'],
0207             component_revisions={'panda': 0},
0208             registration_versions={'panda': 5},
0209             component_hashes={},
0210             state_hash='',
0211             state={'components': {'panda': {
0212                 'v': 1,
0213                 'data': {'jobs': {'cum': scope_cum, 'sites': sites}},
0214                 'registration': PANDA_REGISTRATION,
0215                 'revision': 0,
0216                 'registration_version': 5,
0217                 'assessed_at': instant.isoformat(),
0218                 'source_as_of': instant.isoformat(),
0219                 'accepted_at': now.isoformat(),
0220                 'assessment_policy': ASSESSMENT_POLICY_VERSION,
0221                 'publisher_identity': PUBLISHER_IDENTITY,
0222             }}},
0223         )
0224         written += 1
0225     print(f'\napplied: removed prior backfill {removed[0]}, '
0226           f'wrote {written} snaps')
0227     return 0
0228
0229
0230 if __name__ == '__main__':
0231     sys.exit(main())