feat: add prometheus state metrics

docs/docs/sqlbroker/tutorial.md

@@ -228,3 +228,57 @@ And relay the messages from the database to another broker.
 ```python linenums="1"
 {!> docs_src/sqlbroker/transactional_outbox.py [ln:30-51]!}
 ```
+
+## Observability
+
+FastStream already supplies Prometheus metrics for message publishing and processing rates and latencies through its
+[Prometheus middleware](../getting-started/observability/prometheus.md){.external-link target="_blank"}.
+
+<figure markdown="span">
+  ![Grafana panels from the FastStream Prometheus middleware: publish and process rates, publish and process duration percentiles, messages in process, and received message size](../assets/img/faststream-processing-metrics.png){ width="100%" .on-glb }
+</figure>
+
+SQLBroker additionally provides metrics derived from the messages persisted in the database:
+
+- `sqlbroker_messages` — messages in the primary table, labeled by `queue` and
+  `state`.
+- `sqlbroker_most_overdue_message_age_seconds` — how long the most overdue message has
+  been eligible for processing, labeled by `queue` and `state`.
+- `sqlbroker_archived_messages` — messages in the archive table, labeled by
+  `queue` and `state`.
+- `sqlbroker_state_collection_last_success_timestamp_seconds` — Unix timestamp
+  of the last successful database sample.
+
+<figure markdown="span">
+  ![Grafana panels from the SQLBroker state sampler: messages by queue and state, oldest message age, and archived messages by queue and state](../assets/img/sqlbroker-state-metrics.png){ width="100%" .on-glb }
+</figure>
+
+### Standalone sampler
+
+If the sampler runs in every broker node, each node queries the shared database and reports database-wide values and exports a duplicate copy of the same series. Prefer one standalone sampler per database, using the packaged
+`sqlbroker-state-metrics` command:
+
+```console
+pip install "faststream-sqlbroker[cli]"
+sqlbroker-state-metrics \
+    --host 0.0.0.0 \
+    --port 8000 \
+    --message-table message \
+    --archive-table message_archive \
+    --interval 30 \
+    --database-url postgresql+asyncpg://user:pass@localhost/mydb # pragma: allowlist secret
+```
+
+### In-broker sampler
+
+The sampler can also run as part of the broker. Install the Prometheus dependency and pass the registry exposed by your metrics endpoint to `SqlBrokerStateMetricsConfig`:
+
+```console
+pip install "faststream-sqlbroker[prometheus]"
+```
+
+```python linenums="1"
+{!> docs_src/sqlbroker/observability_in_broker.py !}
+```
+
+Mount `metrics_app` at `/metrics` in your ASGI application. These database-wide gauges must not be summed across instances. This applies even to nominally single-node deployments because rolling restarts can briefly run the old and new broker nodes at the same time.

docs/docs_src/sqlbroker/observability_in_broker.py

@@ -0,0 +1,19 @@
+from prometheus_client import CollectorRegistry, make_asgi_app
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from faststream_sqlbroker import SqlBroker
+from faststream_sqlbroker.sqlbroker.observability import (
+    SqlBrokerStateMetricsConfig,
+)
+
+engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
+registry = CollectorRegistry()
+
+broker = SqlBroker(
+    engine=engine,
+    state_metrics_config=SqlBrokerStateMetricsConfig(
+        registry=registry,
+        interval=30,
+    ),
+)
+metrics_app = make_asgi_app(registry=registry)

faststream_sqlbroker/sqlbroker/broker/broker.py

@@ -32,6 +32,10 @@
     from faststream.specification.schema.extra.tag import Tag, TagDict
 
     from faststream_sqlbroker.sqlbroker.client import SqlBrokerBaseClient
+    from faststream_sqlbroker.sqlbroker.observability import (
+        SqlBrokerStateMetricsConfig,
+        SqlBrokerStateSampler as SqlBrokerStateSamplerType,
+    )
 
 
 class SqlBroker(
@@ -49,6 +53,7 @@ def __init__(
         engine: AsyncEngine,
         schema: SqlBrokerSchemaConfig | None = None,
         validate_schema_on_start: bool = True,
+        state_metrics_config: "SqlBrokerStateMetricsConfig | None" = None,
         # broker base args
         graceful_timeout: float | None = 15.0,
         decoder: Optional["CustomCallable"] = None,
@@ -123,16 +128,34 @@ def __init__(
             ),
         )
 
+        self._state_metrics_sampler: SqlBrokerStateSamplerType | None
+        if state_metrics_config is not None:
+            from faststream_sqlbroker.sqlbroker.observability.sampler import (
+                SqlBrokerStateSampler,
+            )
+
+            self._state_metrics_sampler = SqlBrokerStateSampler(
+                engine=engine,
+                schema=config.schema,
+                config=state_metrics_config,
+            )
+        else:
+            self._state_metrics_sampler = None
+
     async def start(self) -> None:
         await self.connect()
         await super().start()
+        if self._state_metrics_sampler is not None:
+            self._state_metrics_sampler.start()
 
     async def stop(
         self,
         exc_type: type[BaseException] | None = None,
         exc_val: BaseException | None = None,
         exc_tb: Optional["TracebackType"] = None,
     ) -> None:
+        if self._state_metrics_sampler is not None:
+            await self._state_metrics_sampler.stop()
         await super().stop(exc_type, exc_val, exc_tb)
         if self.config.broker_config.engine:
             await self.config.broker_config.engine.dispose(close=True)

faststream_sqlbroker/sqlbroker/observability/__init__.py

@@ -0,0 +1,23 @@
+from .metrics import SqlBrokerStateMetrics
+from .queries import archived_message_state_summary_query, message_state_summary_query
+from .sampler import SqlBrokerStateMetricsConfig, SqlBrokerStateSampler
+from .snapshot import (
+    ArchivedMessageStateSummary,
+    MessageStateSummary,
+    SqlBrokerStateSnapshot,
+    load_state_snapshot,
+    state_snapshot_from_rows,
+)
+
+__all__ = (
+    "ArchivedMessageStateSummary",
+    "MessageStateSummary",
+    "SqlBrokerStateMetrics",
+    "SqlBrokerStateMetricsConfig",
+    "SqlBrokerStateSampler",
+    "SqlBrokerStateSnapshot",
+    "archived_message_state_summary_query",
+    "load_state_snapshot",
+    "message_state_summary_query",
+    "state_snapshot_from_rows",
+)

faststream_sqlbroker/sqlbroker/observability/cli.py

@@ -0,0 +1,124 @@
+import asyncio
+import logging
+import signal
+from enum import Enum
+from typing import Annotated
+
+import typer
+from prometheus_client import CollectorRegistry, start_http_server
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from faststream_sqlbroker.sqlbroker.observability.sampler import (
+    SqlBrokerStateMetricsConfig,
+    SqlBrokerStateSampler,
+)
+from faststream_sqlbroker.sqlbroker.schema import SqlBrokerSchemaConfig
+
+cli = typer.Typer(add_completion=False, pretty_exceptions_short=True)
+
+
+class LogLevel(str, Enum):
+    DEBUG = "DEBUG"
+    INFO = "INFO"
+    WARNING = "WARNING"
+    ERROR = "ERROR"
+
+
+async def serve(
+    *,
+    database_url: str,
+    host: str,
+    port: int,
+    interval: float,
+    namespace: str,
+    message_table: str,
+    archive_table: str | None,
+    queues: list[str] | None,
+) -> None:
+    logger = logging.getLogger(__name__)
+    engine = create_async_engine(database_url)
+    registry = CollectorRegistry()
+    sampler = SqlBrokerStateSampler(
+        engine=engine,
+        schema=SqlBrokerSchemaConfig(
+            message_table_name=message_table,
+            message_archive_table_name=archive_table,
+        ),
+        config=SqlBrokerStateMetricsConfig(
+            registry=registry,
+            interval=interval,
+            namespace=namespace,
+            queues=queues,
+        ),
+        logger=logger,
+    )
+    server, thread = start_http_server(port, addr=host, registry=registry)
+    stopped = asyncio.Event()
+    loop = asyncio.get_running_loop()
+    for signum in (signal.SIGINT, signal.SIGTERM):
+        loop.add_signal_handler(signum, stopped.set)
+
+    logger.info("Serving SQLBroker state metrics on http://%s:%d/metrics", host, port)
+    sampler.start()
+    try:
+        await stopped.wait()
+    finally:
+        await sampler.stop()
+        server.shutdown()
+        thread.join()
+        await engine.dispose()
+
+
+@cli.command()
+def state_metrics(
+    database_url: Annotated[
+        str,
+        typer.Option(
+            help="SQLAlchemy async URL, for example postgresql+asyncpg://user:pass@host/db.",  # pragma: allowlist secret
+        ),
+    ],
+    host: Annotated[str, typer.Option(help="Metrics listen address.")] = "127.0.0.1",
+    port: Annotated[int, typer.Option(help="Metrics listen port.")] = 8000,
+    interval: Annotated[
+        float,
+        typer.Option(help="Database sampling interval in seconds."),
+    ] = 30.0,
+    namespace: Annotated[
+        str, typer.Option(help="Prometheus metric namespace.")
+    ] = "sqlbroker",
+    message_table: Annotated[
+        str,
+        typer.Option(help="SQLBroker primary message table name."),
+    ] = "message",
+    archive_table: Annotated[
+        str,
+        typer.Option(
+            help="SQLBroker archive table name; use an empty value when disabled.",
+        ),
+    ] = "message_archive",
+    queue: Annotated[
+        list[str] | None,
+        typer.Option(
+            help="Queue to collect; repeat to select multiple queues. All queues by default.",
+        ),
+    ] = None,
+    log_level: Annotated[LogLevel, typer.Option(case_sensitive=False)] = LogLevel.INFO,
+) -> None:
+    """Expose persisted FastStream SQLBroker state as Prometheus metrics."""
+    logging.basicConfig(level=log_level.value)
+    asyncio.run(
+        serve(
+            database_url=database_url,
+            host=host,
+            port=port,
+            interval=interval,
+            namespace=namespace,
+            message_table=message_table,
+            archive_table=archive_table or None,
+            queues=queue or None,
+        ),
+    )
+
+
+def main() -> None:
+    cli(prog_name="sqlbroker-state-metrics")

faststream_sqlbroker/sqlbroker/observability/metrics.py

@@ -0,0 +1,79 @@
+from datetime import datetime, timezone
+
+from prometheus_client import CollectorRegistry, Gauge
+
+from faststream_sqlbroker.sqlbroker.observability.snapshot import (
+    SqlBrokerStateSnapshot,
+)
+
+
+class SqlBrokerStateMetrics:
+    """Cached Prometheus metrics populated from a persisted-state snapshot."""
+
+    def __init__(
+        self,
+        *,
+        registry: CollectorRegistry,
+        namespace: str = "sqlbroker",
+    ) -> None:
+        self.messages = Gauge(
+            "messages",
+            "Current messages persisted in the SQLBroker primary table.",
+            ("queue", "state"),
+            namespace=namespace,
+            registry=registry,
+        )
+        self.most_overdue_message_age_seconds = Gauge(
+            "most_overdue_message_age_seconds",
+            "Lag of the most overdue SQLBroker message by queue and state.",
+            ("queue", "state"),
+            namespace=namespace,
+            registry=registry,
+        )
+        self.archived_messages = Gauge(
+            "archived_messages",
+            "Current messages persisted in the SQLBroker archive table.",
+            ("queue", "state"),
+            namespace=namespace,
+            registry=registry,
+        )
+        self.last_success_timestamp_seconds = Gauge(
+            "state_collection_last_success_timestamp_seconds",
+            "Unix timestamp of the last successful SQLBroker state collection.",
+            namespace=namespace,
+            registry=registry,
+        )
+        self._labels: set[tuple[str, str]] = set()
+        self._archive_labels: set[tuple[str, str]] = set()
+
+    def apply(self, snapshot: SqlBrokerStateSnapshot) -> None:
+        current: set[tuple[str, str]] = set()
+        for item in snapshot.messages:
+            labels = (item.queue, item.state.value)
+            current.add(labels)
+            self.messages.labels(*labels).set(item.message_count)
+            age = max(
+                0.0,
+                (snapshot.collected_at - item.oldest_next_attempt_at).total_seconds(),
+            )
+            self.most_overdue_message_age_seconds.labels(*labels).set(age)
+
+        for labels in self._labels - current:
+            self.messages.remove(*labels)
+            self.most_overdue_message_age_seconds.remove(*labels)
+        self._labels = current
+
+        archive_current: set[tuple[str, str]] = set()
+        for archived_item in snapshot.archived_messages:
+            labels = (archived_item.queue, archived_item.state.value)
+            archive_current.add(labels)
+            self.archived_messages.labels(*labels).set(archived_item.message_count)
+
+        for labels in self._archive_labels - archive_current:
+            self.archived_messages.remove(*labels)
+        self._archive_labels = archive_current
+
+    def record_success(self, *, collected_at: datetime) -> None:
+        self.last_success_timestamp_seconds.set(
+            collected_at.replace(tzinfo=timezone.utc).timestamp()
+        )