src/monitor_app/live_notices.py

0001 """Human-channel selection, separate from the complete action record.
0002
0003 Only the Mattermost delivery uses this policy (docs/NOTICE_ROUTING.md).
0004 The caller persists ``failures`` alongside its stream position; it contains
0005 only outstanding automated failures, not a second incident archive.
0006 """
0007 import json
0008 import re
0009
0010
0011 STATE_KEY = 'epicprod_live_failures'
0012 ROUTINE_SUCCESSES = frozenset({
0013     'catalog_sync', 'catalog_import', 'past_import', 'questionnaire_import',
0014     'association_sweep', 'rucio_sweep', 'evgen_sweep', 'segfault_inventory',
0015     'segfault_dig', 'segfault_study', 'segfault_notice', 'storage_door_cycle',
0016 })
0017 # Maintenance passes: their automated runs never reach the channel, failures
0018 # and recoveries included (Torre, 2026-10-08); the action log and the alarms
0019 # keep them.
0020 MAINTENANCE = frozenset({
0021     'stash_drain', 'storage_sweep', 'storage_door_cycle', 'log_rescue', 'log_grant',
0022     'registrar', 'report_sweep', 'node_measure_ingest', 'harvester_stdout_capture',
0023     'batch_log_capture', 'batch_log_learn', 'file_events_measure',
0024     'panda_sandbox_keepalive', 'es_closeout_cycle', 'system_status_refresh',
0025     'snapper_capture', 'rucio_snapshot_update', 'rucio_arrivals_sweep',
0026     'delivery_daily_rebuild', 'campaign_progress_refresh',
0027 })
0028 FAILURE_OUTCOMES = frozenset({'error', 'timeout', 'partial', 'unrecorded'})
0029 PROGRESS_PATTERNS = {
0030     'catalog_import': r'\b(\d+) new\b',
0031     'past_import': r'\bcreated=(\d+)',
0032     'questionnaire_import': r'\b(\d+) (?:new|updated)\b',
0033     'association_sweep': r'\b(?:new|intaken)=(\d+)',
0034     'segfault_inventory': r'\b(\d+) (?:crashed jobs|rows added|new)\b',
0035     'segfault_dig': r'\b(\d+) (?:dug|traced)\b',
0036     'segfault_study': r'\b(\d+) (?:studies queued|traced signatures marked)\b',
0037     'segfault_notice': r'\b(\d+) (?:need a reproduction|traced unread)\b',
0038 }
0039
0040
0041 def select_notice(row, extra, failures, live_policy):
0042     """Return delivery attributes, or None for a quiet channel event.
0043
0044     Observe quiet successes too, so recovery is not hidden by a source's
0045     routine-success default. Explicit live overrides remain authoritative;
0046     subscription filters are still applied to the returned attributes.
0047     Never mutate the original record or another subscriber's attributes.
0048     """
0049     if row.app_name != 'epicprod':
0050         return extra
0051     action = str(extra.get('action') or '')
0052     outcome = str(extra.get('outcome') or '')
0053     username = str(extra.get('username') or '')
0054     automated = username == 'cron' or username.endswith('_cron')
0055     notice = dict(extra)
0056     failure = outcome in FAILURE_OUTCOMES
0057     if failure and not notice.get('reason'):
0058         notice['reason'] = str(notice.get('summary') or row.message or '')[:300]
0059
0060     quiet = False
0061     if automated and action in MAINTENANCE:
0062         quiet = True
0063     elif automated:
0064         key = json.dumps([action, row.instance_name,
0065                           extra.get('subject_type'), extra.get('subject_key'),
0066                           extra.get('source'), extra.get('operation')])
0067         if failure:
0068             current = {'outcome': outcome, 'reason': notice.get('reason', '')}
0069             quiet = failures.get(key) == current
0070             failures[key] = current
0071             # A source's routine importance must not hide its first failure.
0072             notice.update(live_default=True, sublevel='high')
0073         elif outcome == 'ok':
0074             previous = failures.pop(key, None)
0075             if previous:
0076                 notice.update(
0077                     live_default=True, sublevel='normal',
0078                     operation=f'{action}_recovered',
0079                     summary=('Recovered; previous failure: '
0080                              f"{previous['reason'] or previous['outcome']}"))
0081                 return notice
0082             quiet = action in ROUTINE_SUCCESSES
0083             pattern = PROGRESS_PATTERNS.get(action)
0084             if pattern and any(int(v) for v in re.findall(
0085                     pattern, str(extra.get('summary') or ''))):
0086                 quiet = False
0087         elif action == 'node_guard_decision' and outcome == 'would_exclude':
0088             quiet = True
0089
0090     if quiet and live_policy.get(action) is not True:
0091         return None
0092     return notice