swf-monitor/scripts/backfill-action-stream.py

0001 #!/usr/bin/env python
0002 """One-off backfill of early epicprod action-stream records.
0003
0004 Records written before the sublevel/live axes (2026-07-05) carry no
0005 sublevel/live_default and are therefore invisible to the live view and its
0006 channels; records written before the reason plumbing carry non-ok outcomes
0007 with no failure reason. This stamps both, from recoverable sources only:
0008
0009 - axes: sublevel/live_default from the ACTION_DEFAULTS catalog by action id
0010   (an action absent from the catalog gets low/False, the conservative
0011   default).
0012 - reason, payload_log_fetch errors: the cache .error marker's last_error,
0013   written by the same failed fetch ($SWF_TMP_DIR/panda-logs/<task>/<job>/).
0014 - reason, association_sweep timeout: deterministic from the outcome and the
0015   record's own measured duration.
0016 - reason, questionnaire_import skipped: the documented skip cause —
0017   SysConfig questionnaire_csv_url unset.
0018
0019 Anything else non-ok with no recoverable cause is reported and left alone —
0020 a backfill must not fabricate. Backfilled records are marked
0021 extra_data['backfilled'] with what was stamped. Dry run by default; --apply
0022 writes.
0023 """
0024 import argparse
0025 import json
0026 import os
0027
0028
0029 def main():
0030     parser = argparse.ArgumentParser()
0031     parser.add_argument("--apply", action="store_true",
0032                         help="write changes (default: dry-run report)")
0033     args = parser.parse_args()
0034
0035     os.environ.setdefault("DJANGO_SETTINGS_MODULE", "swf_monitor_project.settings")
0036     import django
0037     django.setup()
0038     from monitor_app.epicprod_logging import (ACTION_DEFAULTS,
0039                                               EPICPROD_APP_NAME,
0040                                               SUBLEVEL_VALUES)
0041     from monitor_app.models import AppLog
0042
0043     swf_tmp = os.environ.get("SWF_TMP_DIR", "/data/swf-tmp")
0044     stamped_axes = stamped_reason = unrecovered = 0
0045
0046     rows = (AppLog.objects.filter(app_name=EPICPROD_APP_NAME)
0047             .exclude(extra_data__isnull=True).order_by('id'))
0048     for row in rows.iterator():
0049         extra = row.extra_data if isinstance(row.extra_data, dict) else {}
0050         action = str(extra.get('action') or '')
0051         if not action:
0052             continue
0053         changed = []
0054
0055         if extra.get('sublevel') not in SUBLEVEL_VALUES:
0056             decl = ACTION_DEFAULTS.get(action) or {}
0057             extra['sublevel'] = decl.get('sublevel', 'low')
0058             extra['live_default'] = bool(decl.get('live', False))
0059             changed.append('axes')
0060
0061         outcome = str(extra.get('outcome') or '')
0062         if outcome not in ('', 'ok') and not extra.get('reason'):
0063             reason = ''
0064             if action == 'payload_log_fetch':
0065                 marker = os.path.join(swf_tmp, 'panda-logs',
0066                                       str(extra.get('jeditaskid') or ''),
0067                                       str(extra.get('subject_key') or ''),
0068                                       '.error')
0069                 try:
0070                     with open(marker) as f:
0071                         reason = str(json.load(f).get('last_error') or '')
0072                 except (OSError, ValueError):
0073                     reason = ''
0074             elif action == 'association_sweep' and outcome == 'timeout':
0075                 dur = extra.get('duration_ms')
0076                 reason = (f"timed out after {int(dur) // 1000}s"
0077                           if dur else "timed out")
0078             elif action == 'questionnaire_import' and outcome == 'skipped':
0079                 reason = 'questionnaire_csv_url unset'
0080             if reason:
0081                 extra['reason'] = reason[:300]
0082                 row.message = f"{row.message} — {reason[:300]}"
0083                 changed.append('reason')
0084             else:
0085                 unrecovered += 1
0086                 print(f"UNRECOVERED: #{row.id} {row.timestamp} {action} "
0087                       f"{outcome} — no recoverable cause, left alone")
0088
0089         if changed:
0090             stamped_axes += 'axes' in changed
0091             stamped_reason += 'reason' in changed
0092             extra['backfilled'] = '+'.join(changed)
0093             print(f"{'APPLY' if args.apply else 'DRY'}: #{row.id} "
0094                   f"{row.timestamp} {action} <- {extra['backfilled']}"
0095                   + (f" ({extra.get('reason', '')})" if 'reason' in changed else ''))
0096             if args.apply:
0097                 row.extra_data = extra
0098                 row.save(update_fields=['extra_data', 'message'])
0099
0100     print(f"\nsummary: axes={stamped_axes} reason={stamped_reason} "
0101           f"unrecovered={unrecovered} applied={bool(args.apply)}")
0102     return 0
0103
0104
0105 if __name__ == "__main__":
0106     raise SystemExit(main())