swf-monitor/scripts/live-watch-trigger.py

0001 #!/usr/bin/env python3
0002 """live-watch-trigger.py — submit one live watch run.
0003
0004 The front end of the live watch (swf-epicprod docs/EPICPROD_ASSESSMENTS.md,
0005 The live watch), on the segfault diagnosis's harness pattern: the bundle is
0006 assembled deterministically (``swf_epicprod.livewatch.evidence``: the
0007 channel's posts, the action record's failure groups and flapping counts,
0008 the publication policy, the mechanical floor), stored as a hidden corun
0009 bundle Page, and submitted as the run's prompt content with the
0010 ``live_watch`` definition (POST /prompts/, then POST /jobs/). The run is
0011 recorded in PersistentState under ``live_watch`` and a quiet
0012 ``live_watch_triggered`` action is logged either way.
0013
0014 Run by the ops agent's ``live_watch`` handler (every four hours by cron
0015 enqueue), or by hand::
0016
0017     cd /data/wenauseic/github/swf-monitor/src
0018     ../../swf-testbed/.venv/bin/python ../scripts/live-watch-trigger.py [--dry-run]
0019
0020 Environment: CORUN_API_URL (or CORUN_BASE_URL), CORUN_API_TOKEN,
0021 CORUN_LIVEWATCH_SECTION, CORUN_LIVEWATCH_BUNDLE_SECTION,
0022 CORUN_LIVEWATCH_DEFINITION (printed by swf_epicprod.livewatch.bootstrap),
0023 MATTERMOST_TOKEN for the channel read.
0024 """
0025 import argparse
0026 import json
0027 import logging
0028 import os
0029 import sys
0030 import time
0031 from datetime import datetime, timezone as dt_timezone
0032
0033 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
0034 sys.path.insert(0, os.path.join(THIS_DIR, '..', 'src'))
0035 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0036
0037 import django  # noqa: E402
0038 django.setup()
0039
0040 from monitor_app.epicprod_logging import log_epicprod_action  # noqa: E402
0041 from monitor_app.models import PersistentState  # noqa: E402
0042 from swf_epicprod.assessment.trigger import (CORUN_API_TOKEN, CORUN_API_URL,  # noqa: E402
0043                                              CORUN_WEB_URL, _request)
0044 from swf_epicprod.livewatch import evidence, spec  # noqa: E402
0045
0046 SECTION = os.environ.get('CORUN_LIVEWATCH_SECTION', spec.DEFAULT_SECTION)
0047 BUNDLE_SECTION = os.environ.get('CORUN_LIVEWATCH_BUNDLE_SECTION', spec.DEFAULT_BUNDLE_SECTION)
0048 DEFINITION = os.environ.get('CORUN_LIVEWATCH_DEFINITION', '')
0049 log = logging.getLogger('live-watch-trigger')
0050
0051
0052 def persist_bundle(bundle):
0053     page = _request(
0054         f'{CORUN_API_URL}/pages/',
0055         payload={
0056             'section': BUNDLE_SECTION,
0057             'title': f"ePIC live watch evidence — {bundle['generated_at']}",
0058             'content': '```json\n' + json.dumps(bundle, indent=1, default=str) + '\n```\n',
0059             'data': {'ui_visible': False, 'artifact_type': 'live_watch_evidence_bundle',
0060                      'source_system': 'epicprod', 'subject_type': 'live_channel',
0061                      'subject_key': 'epicprod-live', 'generated_at': bundle['generated_at'],
0062                      'evidence_bundle': bundle},
0063             'tags': ['evidence-bundle', 'epicprod', 'live-watch'],
0064         },
0065         token=CORUN_API_TOKEN)
0066     bundle_id = str(page.get('group_id') or '')
0067     if not bundle_id:
0068         raise RuntimeError(f'corun bundle Page response contained no group id: {page!r}')
0069     bundle['artifact'] = {'type': 'corun_page', 'id': bundle_id,
0070                           'url': f'{CORUN_WEB_URL}/page/{bundle_id}/', 'section': BUNDLE_SECTION}
0071     return bundle_id
0072
0073
0074 def submit(requested_by='', dry_run=False):
0075     bundle = evidence.assemble()
0076     if dry_run:
0077         print(json.dumps({'dry_run': True, 'floor': bundle['floor'], 'degraded': bundle['degraded'],
0078                           'posts': len(bundle['channel'].get('posts') or []),
0079                           'groups': len(bundle['failure_groups']),
0080                           'bytes': len(json.dumps(bundle, default=str))}))
0081         return 0
0082     if not (CORUN_API_URL and CORUN_API_TOKEN and DEFINITION):
0083         raise RuntimeError('CORUN_API_URL, CORUN_API_TOKEN and CORUN_LIVEWATCH_DEFINITION are required')
0084     bundle_id = persist_bundle(bundle)
0085     content = json.dumps({
0086         'submitted_at': datetime.now(dt_timezone.utc).isoformat(timespec='seconds'),
0087         'bundle': bundle,
0088     }, default=str)
0089     prompt = _request(f'{CORUN_API_URL}/prompts/',
0090                       payload={'section': SECTION, 'content': content, 'definition_id': DEFINITION},
0091                       token=CORUN_API_TOKEN)
0092     job = _request(f'{CORUN_API_URL}/jobs/',
0093                    payload={'prompt_group_id': str(prompt.get('group_id') or ''),
0094                             'definition_id': DEFINITION},
0095                    token=CORUN_API_TOKEN)
0096     job_id = str(job.get('id') or job.get('job_id') or '')
0097     if not job_id:
0098         raise RuntimeError(f'corun job response contained no job id: {job!r}')
0099     state = dict(PersistentState.get_state().get('live_watch') or {})
0100     state['current'] = {'job_id': job_id, 'prompt_group_id': str(prompt.get('group_id') or ''),
0101                         'bundle_id': bundle_id, 'submitted_at': bundle['generated_at'],
0102                         'requested_by': requested_by, 'state': 'submitted',
0103                         'floor': bundle['floor']['verdict']}
0104     PersistentState.update_state({'live_watch': state})
0105     return {'job_id': job_id, 'bundle_id': bundle_id, 'floor': bundle['floor']['verdict'],
0106             'degraded': bundle['degraded']}
0107
0108
0109 def main():
0110     ap = argparse.ArgumentParser(description=__doc__.split('\n\n')[0])
0111     ap.add_argument('--requested-by', default='')
0112     ap.add_argument('--dry-run', action='store_true', help='assemble and report, submit nothing')
0113     args = ap.parse_args()
0114     logging.basicConfig(level=logging.INFO, stream=sys.stderr,
0115                         format='%(asctime)s %(name)s %(levelname)s %(message)s')
0116     t0 = time.monotonic()
0117     try:
0118         result = submit(args.requested_by, args.dry_run)
0119     except Exception as e:                                    # noqa: BLE001
0120         log.exception('trigger failed')
0121         # A failed trigger is a watch that did not run: visible in the channel.
0122         log_epicprod_action('live-watch', 'live_watch_triggered', outcome='error',
0123                             username=args.requested_by, sublevel='normal', live_default=True,
0124                             level=logging.ERROR, duration_ms=int((time.monotonic() - t0) * 1000),
0125                             message=f'live watch trigger failed: {str(e)[:200]}',
0126                             reason=str(e)[:300])
0127         print(json.dumps({'error': str(e)[:300]}))
0128         return 1
0129     if result == 0:
0130         return 0
0131     log_epicprod_action('live-watch', 'live_watch_triggered', outcome='ok',
0132                         username=args.requested_by, sublevel='low', live_default=False,
0133                         duration_ms=int((time.monotonic() - t0) * 1000),
0134                         message=f"live watch submitted: corun job {result['job_id']}, floor {result['floor']}",
0135                         job_id=result['job_id'], floor=result['floor'], degraded=result['degraded'])
0136     print(json.dumps(result))
0137     return 0
0138
0139
0140 if __name__ == '__main__':
0141     sys.exit(main())