swf-monitor/scripts/segfault-diagnosis-trigger.py

0001 #!/usr/bin/env python3
0002 """segfault-diagnosis-trigger.py — submit a crash signature's LLM study.
0003
0004 Stage 5 of swf-epicprod docs/SEGFAULT_DIAGNOSIS.md, on the assessment
0005 harness pattern (EPICPROD_ASSESSMENTS_V1.md): the bundle is assembled
0006 deterministically from the signature record (the signature, its trace,
0007 the representative and reproduction job records, the configuration with
0008 the image the task ran, the reproduction package README when one is
0009 built), stored as a hidden corun bundle Page in the segfault bundle
0010 section, and submitted as the run's prompt content with the
0011 ``segfault_diagnosis`` definition (POST /prompts/, then POST /jobs/).
0012 The run's job id is recorded on the signature (``data['diagnosis']``)
0013 and a ``segfault_diagnosis_triggered`` action is logged either way.
0014
0015 Django-bootstrap standalone script, run by the ops agent for the
0016 Diagnose action and for the automatic trigger, or by hand::
0017
0018     cd /data/wenauseic/github/swf-monitor/src
0019     ../../swf-testbed/.venv/bin/python ../scripts/segfault-diagnosis-trigger.py --key exit139:task39623 [--dry-run]
0020
0021 Environment: SWF_MONITOR_URL, CORUN_API_URL (or CORUN_BASE_URL),
0022 CORUN_API_TOKEN, CORUN_SEGFAULT_SECTION, CORUN_SEGFAULT_BUNDLE_SECTION,
0023 CORUN_SEGFAULT_DEFINITION (printed by swf_epicprod.segfault.bootstrap).
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 django.utils import timezone  # noqa: E402
0041
0042 from monitor_app.epicprod_logging import log_epicprod_action  # noqa: E402
0043 from monitor_app.models import CrashSignature, EpicProdJob  # noqa: E402
0044 from monitor_app.segfaults import signature_detail  # noqa: E402
0045 from swf_epicprod.assessment.trigger import (CORUN_API_TOKEN, CORUN_API_URL,  # noqa: E402
0046                                              CORUN_WEB_URL, _request)
0047 from swf_epicprod.segfault import spec  # noqa: E402
0048
0049 SECTION = os.environ.get('CORUN_SEGFAULT_SECTION', spec.DEFAULT_SECTION)
0050 BUNDLE_SECTION = os.environ.get('CORUN_SEGFAULT_BUNDLE_SECTION', spec.DEFAULT_BUNDLE_SECTION)
0051 DEFINITION = os.environ.get('CORUN_SEGFAULT_DEFINITION', '')
0052 MONITOR_URL = os.environ.get('SWF_MONITOR_URL', '').rstrip('/')
0053 log = logging.getLogger('segfault-diagnosis-trigger')
0054
0055
0056 def _job_record(pandaid):
0057     job = EpicProdJob.objects.filter(pandaid=int(pandaid)).only('pandaid', 'jeditaskid', 'seq_number', 'data').first()
0058     if job is None:
0059         return None
0060     return {'pandaid': job.pandaid, 'jeditaskid': job.jeditaskid, 'seq_number': job.seq_number,
0061             'crash': (job.data or {}).get('crash') or {}}
0062
0063
0064 def assemble(key):
0065     """The evidence bundle for one signature, deterministic."""
0066     detail = signature_detail(key, jobs_limit=20)
0067     if detail is None:
0068         raise ValueError(f'no crash signature {key}')
0069     sig = CrashSignature.objects.get(key=key)
0070     trace = detail.get('trace') or {}
0071     representative = _job_record(trace['source_pandaid']) if trace.get('source_pandaid') else None
0072     reproductions = []
0073     for e in detail.get('reproduction') or []:
0074         entry = dict(e)
0075         if e.get('canary_pandaid'):
0076             entry['canary_job'] = _job_record(e['canary_pandaid'])
0077         reproductions.append(entry)
0078     readme = ''
0079     path = (sig.package or {}).get('path')
0080     if path and os.path.isfile(os.path.join(path, 'README.md')):
0081         with open(os.path.join(path, 'README.md')) as fh:
0082             readme = fh.read()[:20000]
0083     manifest = [{'source': 'signature record', 'ok': True},
0084                 {'source': 'trace', 'ok': trace.get('trace_status') == 'found'},
0085                 {'source': 'representative job', 'ok': representative is not None},
0086                 {'source': 'reproductions', 'ok': bool(reproductions)},
0087                 {'source': 'package README', 'ok': bool(readme)}]
0088     return {
0089         'schema': 'epicprod-segfault-bundle/1',
0090         'generated_at': datetime.now(dt_timezone.utc).isoformat(timespec='seconds'),
0091         'params': {'key': key},
0092         'signature': {k: v for k, v in detail.items()
0093                       if k not in ('jobs', 'trace', 'reproduction', 'data')},
0094         'trace': trace,
0095         'representative_job': representative,
0096         'crashed_jobs': detail.get('jobs') or [],
0097         'reproductions': reproductions,
0098         'package_readme': readme,
0099         'monitor_urls': {
0100             'signature': f'{MONITOR_URL}/panda/segfaults/{key}/',
0101             'catalog': f'{MONITOR_URL}/panda/segfaults/',
0102         },
0103         'manifest': manifest,
0104         'degraded': not all(m['ok'] for m in manifest[:2]),
0105     }
0106
0107
0108 def persist_bundle(bundle):
0109     key = bundle['params']['key']
0110     page = _request(
0111         f'{CORUN_API_URL}/pages/',
0112         payload={
0113             'section': BUNDLE_SECTION,
0114             'title': f"ePIC segfault diagnosis evidence {key} — {bundle['generated_at']}",
0115             'content': '```json\n' + json.dumps(bundle, indent=1, default=str) + '\n```\n',
0116             'data': {'ui_visible': False, 'artifact_type': 'segfault_diagnosis_evidence_bundle',
0117                      'source_system': 'epicprod', 'subject_type': 'crash_signature',
0118                      'subject_key': key, 'generated_at': bundle['generated_at'],
0119                      'evidence_bundle': bundle},
0120             'tags': ['evidence-bundle', 'epicprod', 'segfault'],
0121         },
0122         token=CORUN_API_TOKEN)
0123     bundle_id = str(page.get('group_id') or '')
0124     if not bundle_id:
0125         raise RuntimeError(f'corun bundle Page response contained no group id: {page!r}')
0126     bundle['artifact'] = {'type': 'corun_page', 'id': bundle_id,
0127                           'url': f'{CORUN_WEB_URL}/page/{bundle_id}/', 'section': BUNDLE_SECTION}
0128     return bundle_id
0129
0130
0131 def submit(key, requested_by='', dry_run=False):
0132     bundle = assemble(key)
0133     if dry_run:
0134         print(json.dumps({'key': key, 'dry_run': True, 'degraded': bundle['degraded'],
0135                           'manifest': bundle['manifest'],
0136                           'bytes': len(json.dumps(bundle, default=str))}))
0137         return 0
0138     if not (CORUN_API_URL and CORUN_API_TOKEN and DEFINITION):
0139         raise RuntimeError('CORUN_API_URL, CORUN_API_TOKEN and CORUN_SEGFAULT_DEFINITION are required')
0140     bundle_id = persist_bundle(bundle)
0141     content = json.dumps({
0142         'submitted_at': datetime.now(dt_timezone.utc).isoformat(timespec='seconds'),
0143         'bundle': bundle,
0144     }, default=str)
0145     prompt = _request(f'{CORUN_API_URL}/prompts/',
0146                       payload={'section': SECTION, 'content': content, 'definition_id': DEFINITION},
0147                       token=CORUN_API_TOKEN)
0148     job = _request(f'{CORUN_API_URL}/jobs/',
0149                    payload={'prompt_group_id': str(prompt.get('group_id') or ''),
0150                             'definition_id': DEFINITION},
0151                    token=CORUN_API_TOKEN)
0152     job_id = str(job.get('id') or job.get('job_id') or '')
0153     if not job_id:
0154         raise RuntimeError(f'corun job response contained no job id: {job!r}')
0155     sig = CrashSignature.objects.get(key=key)
0156     data = dict(sig.data or {})
0157     data['diagnosis'] = {'job_id': job_id, 'prompt_group_id': str(prompt.get('group_id') or ''),
0158                          'bundle_id': bundle_id,
0159                          'submitted_at': timezone.now().isoformat(timespec='seconds'),
0160                          'requested_by': requested_by, 'state': 'submitted'}
0161     sig.data = data
0162     sig.save(update_fields=['data', 'updated_at'])
0163     return {'key': key, 'job_id': job_id, 'prompt_group_id': data['diagnosis']['prompt_group_id'],
0164             'bundle_id': bundle_id, 'degraded': bundle['degraded']}
0165
0166
0167 def main():
0168     ap = argparse.ArgumentParser(description=__doc__.split('\n\n')[0])
0169     ap.add_argument('--key', required=True)
0170     ap.add_argument('--requested-by', default='')
0171     ap.add_argument('--dry-run', action='store_true', help='assemble and report, submit nothing')
0172     args = ap.parse_args()
0173     logging.basicConfig(level=logging.INFO, stream=sys.stderr,
0174                         format='%(asctime)s %(name)s %(levelname)s %(message)s')
0175     t0 = time.monotonic()
0176     try:
0177         result = submit(args.key, args.requested_by, args.dry_run)
0178     except Exception as e:                                    # noqa: BLE001
0179         log.exception('trigger failed')
0180         log_epicprod_action('segfault-diagnosis', 'segfault_diagnosis_triggered', outcome='error',
0181                             subject_type='crash_signature', subject_key=args.key,
0182                             username=args.requested_by, sublevel='normal', live_default=True,
0183                             level=logging.ERROR, duration_ms=int((time.monotonic() - t0) * 1000),
0184                             message=f'segfault diagnosis trigger {args.key} failed: {str(e)[:200]}',
0185                             reason=str(e)[:300])
0186         print(json.dumps({'key': args.key, 'error': str(e)[:300]}))
0187         return 1
0188     if result == 0:
0189         return 0
0190     log_epicprod_action('segfault-diagnosis', 'segfault_diagnosis_triggered', outcome='ok',
0191                         subject_type='crash_signature', subject_key=args.key,
0192                         username=args.requested_by, sublevel='normal', live_default=True,
0193                         duration_ms=int((time.monotonic() - t0) * 1000),
0194                         message=f"segfault diagnosis {args.key} submitted: corun job {result['job_id']}",
0195                         job_id=result['job_id'], degraded=result['degraded'])
0196     print(json.dumps(result))
0197     return 0
0198
0199
0200 if __name__ == '__main__':
0201     sys.exit(main())