symfonic.core.observability.metrics_store

metrics_store

Storage-agnostic durable metrics contracts, buffering, and factory.

BufferedMetricsSink

BufferedMetricsSink(store: MetricsStore, *, batch_size: int = 25, flush_interval_s: float = 2.0, max_queue: int = 10000)

Source code in src/symfonic/core/observability/metrics_store.py

def __init__(
    self,
    store: MetricsStore,
    *,
    batch_size: int = 25,
    flush_interval_s: float = 2.0,
    max_queue: int = 10_000,
) -> None:
    self._store = store
    self._batch_size = max(1, batch_size)
    self._flush_interval_s = flush_interval_s
    self._max_queue = max(1, max_queue)
    self._queue: deque[dict[str, Any]] = deque(maxlen=self._max_queue)
    self._event_queue: deque[dict[str, Any]] = deque(maxlen=self._max_queue)
    self._dropped = 0
    self._failed_call_records = 0
    self._failed_event_records = 0
    self._lost_fallback_events = 0
    self._task: asyncio.Task[None] | None = None
    self._stopped = False
    self._wake = asyncio.Event()
    self._drain_lock = asyncio.Lock()
    self._admission_lock = Lock()
    self._blocked_tenants: set[str] = set()

dropped property

dropped: int

All rows irrevocably lost through overflow or failed persistence.

loss_counts property

loss_counts: dict[str, int]

Losses by durable record family, safe for health reporting.

lost_fallback_events property

lost_fallback_events: int

Fallback rows evicted, failed, or unconfirmed after cancellation.

block_tenant async

block_tenant(tenant_id: str) -> None

Reject future rows, await active writes, and discard queued rows.

Source code in src/symfonic/core/observability/metrics_store.py

async def block_tenant(self, tenant_id: str) -> None:
    """Reject future rows, await active writes, and discard queued rows."""
    if not tenant_id:
        raise ValueError("tenant_id must be nonempty")
    with self._admission_lock:
        self._blocked_tenants.add(tenant_id)
    async with self._drain_lock:
        with self._admission_lock:
            self._queue = deque(
                (row for row in self._queue if row.get("tenant_id") != tenant_id),
                maxlen=self._max_queue,
            )
            self._event_queue = deque(
                (row for row in self._event_queue if row.get("tenant_id") != tenant_id),
                maxlen=self._max_queue,
            )

ensure_indexes

ensure_indexes() -> None

Delegate to the wrapped store's index bootstrap, if it has one.

Source code in src/symfonic/core/observability/metrics_store.py

def ensure_indexes(self) -> None:
    """Delegate to the wrapped store's index bootstrap, if it has one."""
    ensure = getattr(self._store, "ensure_indexes", None)
    if ensure is not None:
        ensure()

flush async

flush() -> None

Drain the whole queue now (deterministic for tests / shutdown).

Source code in src/symfonic/core/observability/metrics_store.py

async def flush(self) -> None:
    """Drain the whole queue now (deterministic for tests / shutdown)."""
    while self._queue or self._event_queue:
        await self._drain_once()
    flush = getattr(self._store, "flush", None)
    if flush is not None:
        with contextlib.suppress(Exception):
            await flush()

record

record(row: dict[str, Any]) -> None

Synchronous, non-blocking enqueue. Drops oldest on overflow.

Source code in src/symfonic/core/observability/metrics_store.py

def record(self, row: dict[str, Any]) -> None:
    """Synchronous, non-blocking enqueue. Drops oldest on overflow."""
    with self._admission_lock:
        if row.get("tenant_id") in self._blocked_tenants:
            return
        if len(self._queue) >= self._max_queue:
            self._dropped += 1
            if self._dropped == 1 or self._dropped % 1000 == 0:
                logger.warning(
                    "metrics buffer full (max_queue=%d); dropped %d rows so far",
                    self._max_queue, self._dropped,
                )
        self._queue.append(row)
        if len(self._queue) >= self._batch_size:
            self._wake.set()
    self._ensure_task()

record_event

record_event(row: dict[str, Any]) -> None

Enqueue one metadata-only execution event.

Source code in src/symfonic/core/observability/metrics_store.py

def record_event(self, row: dict[str, Any]) -> None:
    """Enqueue one metadata-only execution event."""
    with self._admission_lock:
        if row.get("tenant_id") in self._blocked_tenants:
            return
        if len(self._event_queue) >= self._max_queue:
            self._dropped += 1
            if self._event_queue[0].get("kind") == "kernel_fallback.v1":
                self._lost_fallback_events += 1
        self._event_queue.append(row)
        if len(self._event_queue) >= self._batch_size:
            self._wake.set()
    self._ensure_task()

start

start() -> None

Explicitly start the background flusher (lifespan startup).

Source code in src/symfonic/core/observability/metrics_store.py

def start(self) -> None:
    """Explicitly start the background flusher (lifespan startup)."""
    self._ensure_task()

MetricsReader

Bases: Protocol

Rich, tenant-scoped read side backing the admin dashboards.

conversation_detail async

conversation_detail(tenant_id: str, conversation_id: str) -> list[dict[str, Any]]

Return rows shaped like ConversationMetricsCollector.snapshots().

Source code in src/symfonic/core/observability/metrics_store.py

async def conversation_detail(  # pragma: no cover - protocol
    self, tenant_id: str, conversation_id: str,
) -> list[dict[str, Any]]:
    """Return rows shaped like ``ConversationMetricsCollector.snapshots()``."""
    ...

MetricsSink

Bases: Protocol

Durable write side of a metrics backend.

MetricsStore

Bases: MetricsSink, MetricsReader, Protocol

A full metrics backend: durable writes + rich tenant-scoped reads.

get_metrics_store

get_metrics_store() -> BufferedMetricsSink | MetricsStore | None

Return the registered metrics store, or None (in-memory only).

Source code in src/symfonic/core/observability/metrics_store.py

def get_metrics_store() -> BufferedMetricsSink | MetricsStore | None:
    """Return the registered metrics store, or ``None`` (in-memory only)."""
    return _metrics_store

make_metrics_store

make_metrics_store(identifier: str | None, *, postgres_session_factory: Any = None, mongo_database_factory: Any = None, mongo_retention_days: int = 90) -> MetricsStore | None

Resolve a METRICS_STORE identifier to a concrete backend.

none / "" → None (in-memory collector only, today's default). postgres → :class:PostgresMetricsStore (needs postgres_session_factory + symfonic-core[postgres]). mongo → :class:MongoMetricsStore (needs mongo_database_factory + symfonic-core[mongodb]). A dotted path (my_pkg.mod:MyStore or my_pkg.mod.MyStore) → the adopter's zero-arg-constructible MetricsStore.

Raises ValueError with a pip install hint on a missing extra and on an unknown identifier, so misconfiguration fails loudly at startup.

Source code in src/symfonic/core/observability/metrics_store_factory.py

def make_metrics_store(
    identifier: str | None,
    *,
    postgres_session_factory: Any = None,
    mongo_database_factory: Any = None,
    mongo_retention_days: int = 90,
) -> MetricsStore | None:
    """Resolve a ``METRICS_STORE`` identifier to a concrete backend.

    ``none`` / ``""`` → ``None`` (in-memory collector only, today's default).
    ``postgres`` → :class:`PostgresMetricsStore` (needs ``postgres_session_factory``
    + ``symfonic-core[postgres]``). ``mongo`` → :class:`MongoMetricsStore`
    (needs ``mongo_database_factory`` + ``symfonic-core[mongodb]``). A dotted
    path (``my_pkg.mod:MyStore`` or ``my_pkg.mod.MyStore``) → the adopter's
    zero-arg-constructible ``MetricsStore``.

    Raises ``ValueError`` with a ``pip install`` hint on a missing extra and
    on an unknown identifier, so misconfiguration fails loudly at startup.
    """
    raw = (identifier or "").strip()
    if raw == "" or raw.lower() == "none":
        return None
    low = raw.lower()

    if low == "postgres":
        if postgres_session_factory is None:
            raise ValueError(
                "METRICS_STORE=postgres requires a postgres_session_factory",
            )
        try:
            from symfonic.core.observability.postgres_metrics_store import (  # noqa: PLC0415
                PostgresMetricsStore,
            )
        except ImportError as exc:  # pragma: no cover - exercised via extra
            raise ValueError(
                "METRICS_STORE=postgres requires SQLAlchemy: "
                "pip install 'symfonic-core[postgres]'",
            ) from exc
        return PostgresMetricsStore(postgres_session_factory)

    if low in ("mongo", "mongodb"):
        if mongo_database_factory is None:
            raise ValueError(
                "METRICS_STORE=mongo requires a mongo_database_factory",
            )
        try:
            from symfonic.core.observability.mongo_metrics_store import (  # noqa: PLC0415
                MongoMetricsStore,
            )
        except ImportError as exc:  # pragma: no cover - exercised via extra
            raise ValueError(
                "METRICS_STORE=mongo requires motor: "
                "pip install 'symfonic-core[mongodb]'",
            ) from exc
        return MongoMetricsStore(
            mongo_database_factory, retention_days=mongo_retention_days,
        )

    if ":" in raw or "." in raw:
        cls = _import_dotted(raw)
        return cls()

    raise ValueError(
        f"unknown METRICS_STORE {raw!r}; expected 'none', 'postgres', "
        "'mongo', or a dotted path 'pkg.module:Class'",
    )

set_metrics_store

set_metrics_store(store: BufferedMetricsSink | MetricsStore | None) -> None

Register the process-wide metrics store (or None to disable).

Source code in src/symfonic/core/observability/metrics_store.py

def set_metrics_store(store: BufferedMetricsSink | MetricsStore | None) -> None:
    """Register the process-wide metrics store (or ``None`` to disable)."""
    global _metrics_store
    _metrics_store = store