symfonic.core.observability.metrics_store

metrics_store

Pluggable persistence for per-LLM-call metrics (v8.8.0).

The in-memory :class:ConversationMetricsCollector resets on every worker restart. This module adds a storage-agnostic durable sink so the admin Usage dashboard and Conversation Detail view survive restarts, without coupling the framework to any one database.

Design mirrors two existing precedents in this repo:

• Budget (budget.py / postgres_budget_store.py): a pluggable store interface injected via a process-wide setter, with a Postgres reference impl; core issues statements against a table it does not own.
• Embeddings (memory/embeddings/factory.py): a string-identifier factory that resolves postgres / mongo / a dotted path to a concrete backend shipped behind an extra, raising a ValueError with the exact pip install hint on a missing dependency.

Public surface:

• :class:MetricsSink / :class:MetricsReader / :class:MetricsStore — the storage-agnostic Protocols a backend implements.
• :class:BufferedMetricsSink — backend-agnostic hot-path buffer + background batch flusher; wraps any store so on_llm_end never awaits DB I/O. This is where batching lives, once, for all backends.
• :func:make_metrics_store — config-string factory.
• :func:set_metrics_store / :func:get_metrics_store — process-wide hook the collector fan-out and admin reads share.

BufferedMetricsSink

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

Hot-path buffer around any :class:MetricsStore.

record() is a synchronous O(1) enqueue so the collector fan-out in on_llm_end never awaits DB I/O. A background task flushes the queue every flush_interval_s or once batch_size rows are pending, calling the wrapped store's record_calls once per batch. All read / lifecycle methods delegate to the wrapped store, so get_metrics_store() returns a single object serving both write and read.

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._dropped = 0
    self._task: asyncio.Task[None] | None = None
    self._stopped = False
    self._wake = asyncio.Event()
    # Serialise the two drain paths (background loop + explicit flush())
    # so they never pop the same row twice or interleave record_calls.
    self._drain_lock = asyncio.Lock()

dropped property

dropped: int

Rows discarded due to a full queue (backpressure signal for ops).

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:
        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."""
    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)  # deque(maxlen) evicts the oldest when full
    if len(self._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 snapshot-shaped rows for one conversation, oldest-first.

The shape MUST match ConversationMetricsCollector.snapshots() so the scaffold's build_conversation_detail() is reused unchanged.

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 snapshot-shaped rows for one conversation, oldest-first.

    The shape MUST match ``ConversationMetricsCollector.snapshots()`` so
    the scaffold's ``build_conversation_detail()`` is reused unchanged.
    """
    ...

list_conversations async

list_conversations(
    tenant_id: str, limit: int, offset: int
) -> tuple[list[dict[str, Any]], int]

Return (rows, total) — per-conversation token rollup.

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

async def list_conversations(  # pragma: no cover - protocol
    self, tenant_id: str, limit: int, offset: int,
) -> tuple[list[dict[str, Any]], int]:
    """Return ``(rows, total)`` — per-conversation token rollup."""
    ...

summary async

summary(
    tenant_id: str, start: date, end: date
) -> dict[str, Any]

Return {conversations, total_llm_calls, total_cost_usd, total_tokens}.

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

async def summary(  # pragma: no cover - protocol
    self, tenant_id: str, start: date, end: date,
) -> dict[str, Any]:
    """Return ``{conversations, total_llm_calls, total_cost_usd, total_tokens}``."""
    ...

token_usage async

token_usage(
    tenant_id: str, start: date, end: date, group_by: str
) -> list[dict[str, Any]]

Return DailyUsageRow / ModelUsageRow shaped dicts (group_by in {day, model}).

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

async def token_usage(  # pragma: no cover - protocol
    self, tenant_id: str, start: date, end: date, group_by: str,
) -> list[dict[str, Any]]:
    """Return DailyUsageRow / ModelUsageRow shaped dicts (``group_by`` in {day, model})."""
    ...

MetricsSink

Bases: Protocol

Durable write side of a metrics backend.

aclose async

aclose() -> None

Release driver resources.

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

async def aclose(self) -> None:  # pragma: no cover - protocol
    """Release driver resources."""
    ...

flush async

flush() -> None

Deterministically drain any internal buffer (tests / shutdown).

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

async def flush(self) -> None:  # pragma: no cover - protocol
    """Deterministically drain any internal buffer (tests / shutdown)."""
    ...

prune async

prune(cutoff: datetime) -> int

Delete rows older than cutoff; return the number removed.

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

async def prune(self, cutoff: datetime) -> int:  # pragma: no cover
    """Delete rows older than ``cutoff``; return the number removed."""
    ...

record_calls async

record_calls(rows: list[dict[str, Any]]) -> None

Persist a batch of snapshot-shaped per-call dicts.

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

async def record_calls(self, rows: list[dict[str, Any]]) -> None:
    """Persist a batch of snapshot-shaped per-call dicts."""
    ...

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.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