alerts

azure_bootstrap.alerts

Tier 2 tiered-alert dispatcher.

Apps register a sender (any callable matching AlertSender) at startup, then code anywhere in the app can fire alert_dev_team(...) — the dispatcher handles dedup, rate-limit, escalation, and HTML rendering.

Modules:

Name	Description
dispatcher	Tiered alert dispatcher: WARN (log-only), ERROR (digest + escalation), CRITICAL (email).
escalation	Sliding-window error history for the ERROR→CRITICAL escalation ladder.
render	HTML rendering for alert emails and digest fragments.

Functions:

Name	Description
alert_dev_team	Emit a tiered alert. Best-effort — never raises.
drain_pending_alerts	Return and clear the pending-digest list. Used by daily digest builders.
install_global_exception_hooks	Wire sys.excepthook and the asyncio loop exception handler to fire
register_dispatcher	Wire the alerts module to a live sender. Idempotent.
reset_state	Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1.
render_pending_alerts_html	Render the pending-alerts digest fragment for inclusion in a daily email.
bump_counter	Thread-safe increment. Never raises.
counter_snapshot	Return a copy of the counter map.

alert_dev_team

alert_dev_team(severity: AlertSeverity | str, subject: str, context: dict[str, Any] | None = None, dedup_key: str | None = None) -> None

Emit a tiered alert. Best-effort — never raises.

Source code in azure_bootstrap/alerts/dispatcher.py

def alert_dev_team(
    severity: AlertSeverity | str,
    subject: str,
    context: dict[str, Any] | None = None,
    dedup_key: str | None = None,
) -> None:
    """Emit a tiered alert. Best-effort — never raises."""
    try:
        if isinstance(severity, str):
            try:
                sev = AlertSeverity(severity)
            except ValueError:
                sev = AlertSeverity.ERROR
        else:
            sev = severity
        ctx = dict(context) if context else {}
        key = dedup_key if dedup_key else subject
        now = time.monotonic()

        with _state.lock:
            existing = _state.dedup.get(key)
            if existing is not None and (now - existing.first_seen) <= _dedup_window():
                existing.count += 1
                existing.last_seen = now
                for k, v in ctx.items():
                    existing.context.setdefault(k, v)
                _logger.debug(
                    "alerts: deduped",
                    extra={"operation": "alerts.alert_dev_team", "dedup_key": key},
                )
                return
            rec = AlertRecord(
                severity=sev,
                subject=subject,
                context=ctx,
                dedup_key=key,
                first_seen=now,
                last_seen=now,
            )
            _state.dedup[key] = rec
            if len(_state.dedup) > _DEDUP_MAX_ENTRIES:
                cutoff = now - _dedup_window()
                _state.dedup = {k: v for k, v in _state.dedup.items() if v.last_seen >= cutoff}

        level_map = {
            AlertSeverity.WARN: logging.WARNING,
            AlertSeverity.ERROR: logging.ERROR,
            AlertSeverity.CRITICAL: logging.CRITICAL,
        }
        _logger.log(
            level_map[sev],
            "%s",
            subject,
            extra={
                "operation": "alerts.alert_dev_team",
                "severity": sev.value,
                "dedup_key": key,
                "alert_context": _redact(ctx),
            },
        )

        if sev is AlertSeverity.WARN:
            bump_counter("alerts.warn")
            return

        if sev is AlertSeverity.ERROR:
            bump_counter("alerts.error")
            with _state.lock:
                _state.pending_digest.append(rec)
                history = _state.error_history.setdefault(key, deque(maxlen=_ERROR_HISTORY_MAXLEN))
                escalate = should_escalate(
                    history,
                    threshold=_escalation_threshold(),
                    window_seconds=_escalation_window(),
                )
            if escalate:
                bump_counter("alerts.escalated")
                escalated = AlertRecord(
                    severity=AlertSeverity.CRITICAL,
                    subject=f"[ESCALATED] {subject}",
                    context={
                        **ctx,
                        "_escalation_count": len(history),
                        "_escalation_window_seconds": _escalation_window(),
                    },
                    dedup_key=f"escalated:{key}",
                    first_seen=now,
                    last_seen=now,
                )
                _send_critical(escalated)
            return

        # CRITICAL
        bump_counter("alerts.critical")
        _send_critical(rec)
    except Exception:  # never propagate from alerts
        try:
            _logger.exception("alerts: dispatch failed")
        except Exception:
            pass

drain_pending_alerts

drain_pending_alerts() -> list[AlertRecord]

Return and clear the pending-digest list. Used by daily digest builders.

Source code in azure_bootstrap/alerts/dispatcher.py

def drain_pending_alerts() -> list[AlertRecord]:
    """Return and clear the pending-digest list. Used by daily digest builders."""
    with _state.lock:
        out = list(_state.pending_digest)
        _state.pending_digest.clear()
    return out

install_global_exception_hooks

install_global_exception_hooks() -> None

Wire sys.excepthook and the asyncio loop exception handler to fire CRITICAL alerts. Always chains to the previous handlers.

Source code in azure_bootstrap/alerts/dispatcher.py

def install_global_exception_hooks() -> None:
    """Wire ``sys.excepthook`` and the asyncio loop exception handler to fire
    CRITICAL alerts. Always chains to the previous handlers.
    """
    previous_excepthook = sys.excepthook

    def _hook(exc_type: type, exc: BaseException, tb: Any) -> None:
        try:
            alert_dev_team(
                AlertSeverity.CRITICAL,
                subject=f"Uncaught {exc_type.__name__}: {str(exc)[:120]}",
                context={
                    "exception_type": exc_type.__name__,
                    "error": str(exc)[:500],
                    "traceback": "".join(traceback.format_exception(exc_type, exc, tb))[-2000:],
                },
                dedup_key=f"uncaught:{exc_type.__name__}",
            )
        except Exception:
            pass
        try:
            previous_excepthook(exc_type, exc, tb)
        except Exception:
            pass

    sys.excepthook = _hook  # type: ignore[assignment]

    def _async_handler(loop: asyncio.AbstractEventLoop, ctx: dict[str, Any]) -> None:
        try:
            exc = ctx.get("exception")
            exc_type_name = type(exc).__name__ if exc else "AsyncioError"
            msg = ctx.get("message") or (str(exc) if exc else "")
            alert_dev_team(
                AlertSeverity.CRITICAL,
                subject=f"Asyncio uncaught {exc_type_name}: {msg[:120]}",
                context={
                    "exception_type": exc_type_name,
                    "error": str(exc)[:500] if exc else "",
                    "message": str(msg)[:500],
                },
                dedup_key=f"uncaught_async:{exc_type_name}",
            )
        except Exception:
            pass
        try:
            loop.default_exception_handler(ctx)
        except Exception:
            pass

    try:
        loop = asyncio.get_event_loop()
        loop.set_exception_handler(_async_handler)
    except RuntimeError:
        # No event loop yet — future loops will use the default handler.
        pass

register_dispatcher

register_dispatcher(sender: AlertSender, recipients: list[str] | None = None) -> None

Wire the alerts module to a live sender. Idempotent.

Source code in azure_bootstrap/alerts/dispatcher.py

def register_dispatcher(
    sender: AlertSender,
    recipients: list[str] | None = None,
) -> None:
    """Wire the alerts module to a live sender. Idempotent."""
    with _state.lock:
        _state.sender = sender
        _state.recipients = list(recipients) if recipients is not None else _parse_recipients()

reset_state

reset_state() -> None

Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1.

Source code in azure_bootstrap/alerts/dispatcher.py

def reset_state() -> None:
    """Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1."""
    if os.environ.get("AZURE_BOOTSTRAP_ALLOW_RESET") != "1":
        raise RuntimeError("reset_state is test-only — set AZURE_BOOTSTRAP_ALLOW_RESET=1")
    with _state.lock:
        _state.sender = None
        _state.recipients = []
        _state.dedup.clear()
        _state.pending_digest.clear()
        _state.sent_timestamps.clear()
        _state.error_history.clear()

render_pending_alerts_html

render_pending_alerts_html(records: list[AlertRecord]) -> str

Render the pending-alerts digest fragment for inclusion in a daily email.

Source code in azure_bootstrap/alerts/render.py

def render_pending_alerts_html(records: list[AlertRecord]) -> str:
    """Render the pending-alerts digest fragment for inclusion in a daily email."""
    if not records:
        return ""
    rows = "".join(_render_alert_row(r) for r in records)
    return (
        "<h3>Batched alerts since last digest</h3>"
        '<table style="border-collapse:collapse;border:1px solid #ddd;width:100%;">'
        "<thead><tr>"
        '<th style="padding:4px 8px;border:1px solid #ddd;">Severity</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Count</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Subject</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Dedup key</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Context</th>'
        "</tr></thead><tbody>" + rows + "</tbody></table>"
    )

bump_counter

bump_counter(name: str, n: int = 1) -> None

Thread-safe increment. Never raises.

Source code in azure_bootstrap/counters/__init__.py

def bump_counter(name: str, n: int = 1) -> None:
    """Thread-safe increment. Never raises."""
    if not isinstance(name, str) or not name:
        return
    try:
        with _lock:
            _counters[name] = _counters.get(name, 0) + int(n)
    except Exception:
        pass

counter_snapshot

counter_snapshot() -> dict[str, int]

Return a copy of the counter map.

Source code in azure_bootstrap/counters/__init__.py

def counter_snapshot() -> dict[str, int]:
    """Return a copy of the counter map."""
    with _lock:
        return dict(_counters)