src/monitor_app/notice_router.py

0001 """Notice routing: match new action-stream events to consumer-registered
0002 subscriptions and deliver them (docs/NOTICE_ROUTING.md).
0003
0004 ``route_new_events()`` runs once per cycle of the stream-tailer process
0005 (the epicprod-live publisher). It keeps its own high-water mark, so
0006 routing and Mattermost publication advance independently; a routing
0007 failure never blocks publication and vice versa. Buffered-pull delivery
0008 writes one CapcomNotice row per (matched event, subscriber); push
0009 plugins are added as named delivery modes when they come.
0010 """
0011 import logging
0012
0013 logger = logging.getLogger(__name__)
0014
0015 STATE_KEY = 'notice_router_last_id'
0016 # Per-cycle row cap: a quiet-period backlog drains over a few cycles
0017 # rather than one giant pass; nothing is dropped, the mark just advances
0018 # as far as the cap allows.
0019 BATCH_MAX = 500
0020
0021
0022 def _init_high_water():
0023     """Start at the current stream head — never replay history."""
0024     from monitor_app.models import AppLog, PersistentState
0025     state = PersistentState.get_state()
0026     if STATE_KEY not in state:
0027         head = (AppLog.objects.order_by('-id')
0028                 .values_list('id', flat=True).first()) or 0
0029         PersistentState.update_state({STATE_KEY: head})
0030         logger.info("notice router: initialized high-water mark at %s", head)
0031
0032
0033 def _matches(sub, row, action, extra, live_policy):
0034     """One subscription against one event: name (exact or trailing-*
0035     prefix), then equality over the structured fields. A list-valued
0036     filter means membership, so one subscription covers a value set
0037     ({'operation': ['pause', 'resume']}). Two reserved keys reach beyond
0038     the event's own attributes: ``app_name`` matches the record's logging
0039     namespace, and ``live`` matches the event's effective live state —
0040     the runtime live-policy override where one exists, else the record's
0041     ``live_default`` — so a subscription can select the live stream as
0042     data (NOTICE_ROUTING.md § Subscriptions)."""
0043     if sub.event.endswith('*'):
0044         if not action.startswith(sub.event[:-1]):
0045             return False
0046     elif action != sub.event:
0047         return False
0048     for key, want in (sub.filters or {}).items():
0049         if key == 'app_name':
0050             have = row.app_name
0051         elif key == 'live':
0052             override = live_policy.get(action)
0053             have = bool(override if override is not None
0054                         else extra.get('live_default'))
0055         else:
0056             have = extra.get(key)
0057         if isinstance(want, list):
0058             if have not in want:
0059                 return False
0060         elif have != want:
0061             return False
0062     return True
0063
0064
0065 SEVERITIES = ('info', 'warning', 'alarm', 'error')
0066
0067
0068 def _compose(row, extra):
0069     """Deterministic notice content from the event (NOTICE_ROUTING.md §
0070     Delivery): title from the action and subject, detail from the
0071     record's explanatory fields, URL and severity from the event's own
0072     ``url``/``severity`` attributes when it carries them, else the log
0073     record link and outcome-derived severity."""
0074     from monitor_app.models import external_face_base_url
0075
0076     action = str(extra.get('action') or row.funcname or 'event')
0077     title_action = str(extra.get('operation') or action)
0078     subject = str(extra.get('subject_label') or extra.get('subject_key') or '')
0079     title = title_action.replace('_', ' ')
0080     if subject:
0081         title = f'{title}: {subject}'
0082     outcome = str(extra.get('outcome') or '')
0083     detail = str(extra.get('narration') or extra.get('reason')
0084                  or extra.get('summary') or '')
0085     if outcome and outcome != 'ok':
0086         detail = f'{outcome.upper()} — {detail}' if detail else outcome.upper()
0087     severity = str(extra.get('severity') or '')
0088     if severity not in SEVERITIES:
0089         severity = 'info' if outcome in ('', 'ok') else 'warning'
0090     url = str(extra.get('url') or '')
0091     if url.startswith('/'):
0092         url = f'{external_face_base_url()}/prod{url}'
0093     elif not url:
0094         url = f'{external_face_base_url()}/prod/logs/{row.id}/'
0095     return {
0096         'source': 'swf-notices',
0097         'severity': severity,
0098         'title': title[:300],
0099         'detail': detail[:2000],
0100         'url': url[:500],
0101     }
0102
0103
0104 def route_new_events():
0105     """One routing pass: deliver new matching events, advance the mark.
0106
0107     Called from the tailer cycle; exceptions propagate to the caller's
0108     cycle guard and are logged there — a failed pass retries from the
0109     same mark next cycle.
0110     """
0111     from monitor_app.epicprod_logging import get_live_policy
0112     from monitor_app.models import (AppLog, CapcomNotice,
0113                                     NoticeSubscription, PersistentState)
0114     from monitor_app.notice_plugins import PLUGINS
0115
0116     _init_high_water()
0117     last_id = int(PersistentState.get_state().get(STATE_KEY) or 0)
0118     subs = list(NoticeSubscription.objects.filter(enabled=True))
0119     rows = list(AppLog.objects.filter(id__gt=last_id,
0120                                       extra_data__has_key='action')
0121                 .order_by('id')[:BATCH_MAX])
0122     if not rows:
0123         return 0
0124     live_policy = get_live_policy()
0125     delivered = 0
0126     active_plugins = []
0127     failed_plugins = set()
0128     for row in rows:
0129         extra = row.extra_data if isinstance(row.extra_data, dict) else {}
0130         action = str(extra.get('action') or '')
0131         for sub in subs:
0132             if not action or not _matches(sub, row, action, extra,
0133                                           live_policy):
0134                 continue
0135             if sub.delivery == 'buffer':
0136                 content = _compose(row, extra)
0137                 _, created = CapcomNotice.objects.get_or_create(
0138                     subscriber=sub.subscriber,
0139                     dedup_key=f'event:{row.id}:{sub.subscriber}',
0140                     defaults=content)
0141                 delivered += int(created)
0142                 continue
0143             plugin = PLUGINS.get(sub.delivery)
0144             if plugin is None:
0145                 logger.error(
0146                     "notice router: unknown delivery %r on subscription "
0147                     "%s ← %s; row %s not delivered",
0148                     sub.delivery, sub.subscriber, sub.event, row.id)
0149                 continue
0150             # Push is at-most-once: a failure is logged and the pass
0151             # continues, so a push outage never stalls buffered delivery.
0152             # A plugin whose start_pass fails sits out the rest of the
0153             # pass rather than re-failing on every matched row.
0154             if sub.delivery in failed_plugins:
0155                 continue
0156             try:
0157                 if plugin not in active_plugins:
0158                     plugin.start_pass()
0159                     active_plugins.append(plugin)
0160                 plugin.deliver(row, extra)
0161                 delivered += 1
0162             except Exception:
0163                 if plugin not in active_plugins:
0164                     failed_plugins.add(sub.delivery)
0165                 logger.exception(
0166                     "notice router: push delivery %r failed for row %s "
0167                     "(subscription %s ← %s)",
0168                     sub.delivery, row.id, sub.subscriber, sub.event)
0169         PersistentState.update_state({STATE_KEY: int(row.id)})
0170     for plugin in active_plugins:
0171         try:
0172             plugin.end_pass()
0173         except Exception:
0174             logger.exception("notice router: end_pass failed for a plugin")
0175     if delivered:
0176         logger.info("notice router: delivered %d notices (through row %s)",
0177                     delivered, rows[-1].id)
0178     return delivered