"""Monitoring emitter: fire-and-forget queue + background dispatcher. The emitter is the single seam between producers (gateway status hooks, the diagnostic log handler) and consumers (the OTLP streamers). Its contract is the hot-path invariant: ``emit()`` MUST return in O(microseconds), MUST NOT block on disk/network, and MUST NEVER raise into the caller. A monitoring failure is logged locally and dropped — it can never affect the gateway or a session. Mechanism: * ``emit(event)`` does a non-blocking ``queue.put_nowait`` wrapped in a bare except. On a full queue it drops the *oldest* event and counts the drop. * A daemon thread drains the queue and fans each batch out to subscribers (the OTLP metric/span/log streamers). Each subscriber is fail-isolated — a slow or raising subscriber never affects the hot path or its peers. Nothing is persisted here. Monitoring is an egress path, not a local store; if no subscriber is attached, events simply age out of the ring buffer. """ from __future__ import annotations import logging import queue import threading import time from typing import Any, Dict, Optional logger = logging.getLogger(__name__) _MAX_QUEUE = 10_000 # ring-buffer depth; oldest dropped when full _DRAIN_BATCH = 256 class MonitoringEmitter: """Owns the queue, the dispatcher thread, and the subscriber list.""" def __init__(self, *, enabled: bool = True) -> None: self._enabled = enabled self._q: "queue.Queue[Dict[str, Any]]" = queue.Queue(maxsize=_MAX_QUEUE) self._dropped = 0 self._dispatched = 0 self._stop = threading.Event() self._started = False self._lock = threading.Lock() self._thread: Optional[threading.Thread] = None # Live subscribers (the OTLP streamers). Called from the dispatcher # thread, fully fail-isolated. Each subscriber is callable(batch: list[dict]). self._subscribers: list = [] # ── public API (hot path) ─────────────────────────────────────────────── def emit(self, event: Any) -> None: """Enqueue an event. Never blocks, never raises. ``event`` may be a dataclass with ``to_dict()`` or a plain dict. """ if not self._enabled: return try: payload = event.to_dict() if hasattr(event, "to_dict") else dict(event) payload.setdefault("ts_ns", time.time_ns()) self._ensure_started() try: self._q.put_nowait(payload) except queue.Full: # Drop oldest to make room — bounded memory, newest-wins. try: self._q.get_nowait() self._q.task_done() self._dropped += 1 self._q.put_nowait(payload) except Exception: self._dropped += 1 except Exception: # the hot-path invariant: never propagate logger.debug("monitoring emit failed", exc_info=True) # ── lifecycle ─────────────────────────────────────────────────────────── def _ensure_started(self) -> None: if self._started: return with self._lock: if self._started: return self._thread = threading.Thread( target=self._run, name="hermes-monitoring-dispatch", daemon=True ) self._thread.start() self._started = True def _run(self) -> None: while not self._stop.is_set(): try: first = self._q.get(timeout=0.5) except queue.Empty: continue batch = [first] while len(batch) < _DRAIN_BATCH: try: batch.append(self._q.get_nowait()) except queue.Empty: break try: self._dispatch(batch) finally: for _ in batch: self._q.task_done() def _dispatch(self, batch) -> None: # Fan-out to subscribers (OTLP streamers) — fully fail-isolated. for sub in list(self._subscribers): try: sub(batch) except Exception: logger.debug("monitoring subscriber failed", exc_info=True) self._dispatched += len(batch) def subscribe(self, callback) -> None: """Register a live batch subscriber (callable(batch: list[dict])).""" if callback not in self._subscribers: self._subscribers.append(callback) self._enabled = True def unsubscribe(self, callback) -> None: try: self._subscribers.remove(callback) except ValueError: pass if not self._subscribers: self._enabled = False # ── introspection / shutdown (tests, CLI) ─────────────────────────────── def flush(self, timeout: float = 2.0) -> None: """Wait boundedly for queued and in-flight batches to finish dispatch.""" if timeout <= 0: return finished = threading.Event() def _wait_for_completion() -> None: self._q.join() finished.set() waiter = threading.Thread( target=_wait_for_completion, name="hermes-monitoring-flush", daemon=True, ) waiter.start() finished.wait(timeout=timeout) def stats(self) -> Dict[str, int]: return { "queued": self._q.qsize(), "dispatched": self._dispatched, "dropped": self._dropped, "subscribers": len(self._subscribers), } def close(self) -> None: self._stop.set() if self._thread is not None: self._thread.join(timeout=2.0) self._started = False # ── process-wide singleton ────────────────────────────────────────────────── _EMITTER: Optional[MonitoringEmitter] = None _EMITTER_LOCK = threading.Lock() def get_emitter() -> MonitoringEmitter: """Return the process-wide monitoring emitter.""" global _EMITTER if _EMITTER is not None: return _EMITTER with _EMITTER_LOCK: if _EMITTER is None: # Collection is opt-in. A plane exporter enables the singleton by # attaching its first subscriber; until then producers are no-ops. _EMITTER = MonitoringEmitter(enabled=False) return _EMITTER def emit(event: Any) -> None: """Module-level convenience: emit via the singleton.""" get_emitter().emit(event) def reset_emitter_for_tests(emitter: Optional[MonitoringEmitter] = None) -> None: """Swap the singleton (tests only).""" global _EMITTER with _EMITTER_LOCK: if _EMITTER is not None and emitter is not _EMITTER: try: _EMITTER.close() except Exception: pass _EMITTER = emitter # Back-compat alias for the salvaged class name used in emozilla's tests. TelemetryEmitter = MonitoringEmitter __all__ = [ "MonitoringEmitter", "TelemetryEmitter", "get_emitter", "emit", "reset_emitter_for_tests", ]