diff --git a/docs/usage/observability.md b/docs/usage/observability.md index c09b179..fdd3d3e 100644 --- a/docs/usage/observability.md +++ b/docs/usage/observability.md @@ -158,3 +158,55 @@ middleware stack, so publish-scope middleware metrics and the recorder Consume-scope metrics fire too. In sync mode the handler runs inside `publish`, so a publish-scope duration includes the handler's run time; the recorder's `duration_seconds` covers only the in-memory insert. + +## Queue lag + +Set `queue_lag_interval_seconds` on your existing `PrometheusRecorder` to +enable sampling. Omit it or pass `None` to disable sampling. + +```python +broker = OutboxBroker( + engine, + outbox_table=outbox_table, + metrics_recorder=PrometheusRecorder( + app_name="checkout", + registry=registry, + queue_lag_interval_seconds=60, + ), +) +``` + +The broker starts and stops the sampler. It samples immediately, then waits +the configured interval between rounds. At startup it reads the lease TTL +for each registered subscriber queue. If subscribers overlap on a queue, +the sampler uses the shortest TTL. + +Both gauges have `table` and `queue` labels: + +- `faststream_outbox_oldest_due_message_age_seconds` is the time since the + oldest claimable row became due. Rows under active leases and future timers + and retries are excluded. A successful sample of an empty claimable queue is + zero. +- `faststream_outbox_queue_lag_last_success_timestamp_seconds` is the database + time of the last successful sample. It starts at zero and stays unchanged + on errors, as does the age gauge, which starts at NaN. Alert separately on + stale samples and scrape target availability. + +Each queue adds one read per round with a +five-second maximum timeout (or the sampling interval if shorter). The +query uses the existing fetch indexes and can scan past active leases to +find an expired lease. If several replicas share an outbox, enabling the +sampler on one replica avoids redundant reads. + +For a queue with many active leases, consider this optional index on the +actual outbox table. Run the statement outside a transaction. + +```sql +CREATE INDEX CONCURRENTLY outbox_lag_lease_idx +ON outbox (queue, acquired_at) INCLUDE (next_attempt_at) +WHERE acquired_token IS NOT NULL; +``` + +The index lets Postgres narrow the leased branch by the expiry cutoff. It +adds maintenance work on claims and terminal writes. If many leases have +already expired, the query can still scan and sort the qualifying rows. diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index 052969e..31d16f4 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -11,6 +11,7 @@ import typing import warnings from collections.abc import Iterable, Sequence +from contextlib import AbstractAsyncContextManager from types import TracebackType import anyio @@ -34,9 +35,11 @@ from faststream_outbox.configs import LastExceptionRenderer, OutboxBrokerConfig from faststream_outbox.message import OutboxInnerMessage, TimerConflict from faststream_outbox.metrics import MetricsRecorder, _noop_recorder +from faststream_outbox.metrics.prometheus import PrometheusRecorder from faststream_outbox.publisher.producer import OutboxProducer from faststream_outbox.registrator import OutboxRegistrator from faststream_outbox.response import _REQUEST_UNSUPPORTED_MSG, OutboxPublishCommand, _validate_publish_args +from faststream_outbox.subscriber.usecase import OutboxSubscriber _logger = logging.getLogger(__name__) @@ -52,8 +55,6 @@ from sqlalchemy import Table from sqlalchemy.ext.asyncio import AsyncEngine - from faststream_outbox.subscriber.usecase import OutboxSubscriber - def _spec_url(engine: "AsyncEngine | None", outbox_table: "Table") -> list[str]: """AsyncAPI server URL(s) for the broker spec. @@ -155,6 +156,7 @@ def __init__( # noqa: PLR0913 ) -> None: self._outbox_table = outbox_table self._dlq_table = dlq_table + self._queue_lag: AbstractAsyncContextManager[None] | None = None client = OutboxClient(engine, outbox_table, dlq_table=dlq_table) if engine is not None else None fd_config = FastDependsConfig(use_fastdepends=apply_types, serializer=serializer) recorder: MetricsRecorder = metrics_recorder or _noop_recorder @@ -235,6 +237,27 @@ async def start(self) -> None: await super().start() self._warn_on_unstarted_foreign_publishers() self._warn_on_duplicate_queues() + await self._start_queue_lag() + + async def _start_queue_lag(self) -> None: + """Start the recorder's sampler for the queues registered on this broker.""" + config = self.config.broker_config + recorder = config.metrics_recorder + if self._queue_lag is not None or not isinstance(recorder, PrometheusRecorder) or config.engine is None: + return + if recorder._queue_lag_interval_seconds is None: # noqa: SLF001 + return + lease_ttl_by_queue: dict[str, float] = {} + for sub in self.subscribers: + if not isinstance(sub, OutboxSubscriber): + continue + ttl = sub._config.lease_ttl_seconds # noqa: SLF001 + for queue in sub._queues: # noqa: SLF001 + lease_ttl_by_queue[queue] = min(ttl, lease_ttl_by_queue.get(queue, ttl)) + context = recorder._queue_lag_context(config.engine, self._outbox_table, lease_ttl_by_queue) # noqa: SLF001 + if context is not None: + await context.__aenter__() + self._queue_lag = context def _check_pool_capacity(self) -> None: """Raise when a fixed-size pool can't hold every subscriber's connections at once. @@ -347,6 +370,9 @@ async def stop(self, *_args: object, **_kwargs: object) -> None: # otherwise an external cancellation of stop() during the gather leaves # running=True over already-stopped subscribers (ping() would lie). self.running = False + queue_lag, self._queue_lag = self._queue_lag, None + if queue_lag is not None: + await queue_lag.__aexit__(None, None, None) results = await asyncio.gather( *(sub.stop() for sub in subs), return_exceptions=True, diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index 5b57df6..37a0b05 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -197,6 +197,16 @@ def _seconds(name: str) -> "ColumnElement[typing.Any]": return func.make_interval(0, 0, 0, 0, 0, 0, bindparam(name, type_=Float)) +def _claimable_branches( + t: "Table", lease_cutoff: "ColumnElement[typing.Any]" +) -> tuple[tuple["ColumnElement[bool]", ...], tuple["ColumnElement[bool]", ...]]: + """Return separate lease predicates for unleased rows and expired leases.""" + return ( + (t.c.acquired_token.is_(None),), + (t.c.acquired_token.is_not(None), t.c.acquired_at < lease_cutoff), + ) + + def _build_fetch_stmt(outbox: "Table") -> "Update": """Build the claim statement once; queues, limit, lease TTL and the new token are bind parameters. @@ -222,8 +232,9 @@ def branch(name: str, *lease_state: "ColumnElement[bool]") -> "CTE": ) return select(per_queue.c.id, per_queue.c.next_attempt_at).select_from(queues.join(per_queue, true())).cte(name) - pending = branch("pending", t.c.acquired_token.is_(None)) - expired = branch("expired", t.c.acquired_token.is_not(None), t.c.acquired_at < lease_cutoff) + unleased, expired_lease = _claimable_branches(t, lease_cutoff) + pending = branch("pending", *unleased) + expired = branch("expired", *expired_lease) candidates = union_all( select(pending.c.id, pending.c.next_attempt_at), select(expired.c.id, expired.c.next_attempt_at) ).subquery("candidates") diff --git a/faststream_outbox/metrics/lag.py b/faststream_outbox/metrics/lag.py new file mode 100644 index 0000000..3715663 --- /dev/null +++ b/faststream_outbox/metrics/lag.py @@ -0,0 +1,77 @@ +"""Opt-in Prometheus sampling of outbox queue lag.""" + +import asyncio +import logging +import typing +from collections.abc import AsyncGenerator, Mapping +from contextlib import asynccontextmanager + +from sqlalchemy import bindparam, extract, func, select + +from faststream_outbox.client import _claimable_branches, _seconds + + +if typing.TYPE_CHECKING: + from prometheus_client import Gauge + from sqlalchemy import Select, Table + from sqlalchemy.ext.asyncio import AsyncEngine + +_logger = logging.getLogger(__name__) + + +def _lag_statement(table: "Table") -> "Select[int, int]": + """Select the oldest claimable row age and sample time from the database clock.""" + now = func.now() + unleased, expired = _claimable_branches(table, now - _seconds("lease_ttl")) + due = ( + select(table.c.next_attempt_at) + .where(table.c.queue == bindparam("queue"), table.c.next_attempt_at <= now) + .order_by(table.c.next_attempt_at) + .limit(1) + ) + pending = due.where(*unleased).scalar_subquery() + overdue = due.where(*expired, table.c.next_attempt_at <= func.coalesce(pending, now)).scalar_subquery() + return select(extract("epoch", now - func.least(pending, overdue)), extract("epoch", now)) + + +@asynccontextmanager +async def _sample_queue_lag( + engine: "AsyncEngine", + table: "Table", + *, + age: "Gauge", + freshness: "Gauge", + lease_ttl_by_queue: Mapping[str, float], + interval_seconds: float, +) -> AsyncGenerator[None, None]: + """Sample the oldest claimable row for each queue until context exit.""" + statement = _lag_statement(table) + for queue in lease_ttl_by_queue: + age.labels(table.fullname, queue).set(float("nan")) + freshness.labels(table.fullname, queue).set(0) + + async def sample() -> None: + while True: + for queue, ttl in lease_ttl_by_queue.items(): + try: + async with asyncio.timeout(min(interval_seconds, 5.0)), engine.connect() as connection: + delay, sampled_at = ( + await connection.execute(statement, {"queue": queue, "lease_ttl": ttl}) + ).one() + age.labels(table.fullname, queue).set(float(delay or 0)) + freshness.labels(table.fullname, queue).set(float(sampled_at)) + except Exception: + _logger.warning("Failed to sample outbox lag for queue %r", queue, exc_info=True) + await asyncio.sleep(interval_seconds) + + task = asyncio.create_task(sample()) + try: + yield + finally: + task.cancel() + try: + await asyncio.gather(task, return_exceptions=True) + finally: + for queue in lease_ttl_by_queue: + age.remove(table.fullname, queue) + freshness.remove(table.fullname, queue) diff --git a/faststream_outbox/metrics/prometheus.py b/faststream_outbox/metrics/prometheus.py index 129a683..095e512 100644 --- a/faststream_outbox/metrics/prometheus.py +++ b/faststream_outbox/metrics/prometheus.py @@ -29,6 +29,7 @@ ) """ +import math import typing from collections.abc import Callable, Mapping, Sequence @@ -36,10 +37,13 @@ from faststream_outbox._import_checker import is_prometheus_client_installed from faststream_outbox.metrics import BROKER_SYSTEM +from faststream_outbox.metrics.lag import _sample_queue_lag if typing.TYPE_CHECKING: from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram + from sqlalchemy import Table + from sqlalchemy.ext.asyncio import AsyncEngine # ``faststream.prometheus`` imports ``prometheus_client`` at its module top, so its # import must be guarded behind the same probe as the third-party import. Importing @@ -89,13 +93,15 @@ class PrometheusRecorder: upstream so series fall onto the existing exposition. received_messages_size_buckets: bucket boundaries for the message-size histogram. Defaults to FastStream's ``DEFAULT_SIZE_BUCKETS``. - custom_labels: extra labels appended to every metric. Same shape as + custom_labels: extra labels appended to event metrics. Same shape as upstream: ``dict[str, str | Callable[[Mapping], str]]``. Callable values receive the event's ``tags`` mapping and return the label value. ``published`` tags carry no ``subscriber``. include_bus_metrics: if False, skip the consume/publish families that ``OutboxPrometheusMiddleware`` also records and keep only the ``{metrics_prefix}_outbox_*`` series. Set it when registering both. + queue_lag_interval_seconds: seconds between lag samples. Defaults to + ``None``, which disables sampling. """ @@ -108,10 +114,30 @@ def __init__( received_messages_size_buckets: Sequence[float] | None = None, custom_labels: dict[str, str | Callable[[Mapping[str, typing.Any]], str]] | None = None, include_bus_metrics: bool = True, + queue_lag_interval_seconds: float | None = None, ) -> None: if not is_prometheus_client_installed: msg = "PrometheusRecorder requires the 'prometheus' extra: pip install 'faststream-outbox[prometheus]'" raise ImportError(msg) + if queue_lag_interval_seconds is not None and ( + not math.isfinite(queue_lag_interval_seconds) or queue_lag_interval_seconds <= 0 + ): + msg = "queue_lag_interval_seconds must be finite and positive." + raise ValueError(msg) + self._queue_lag_interval_seconds = queue_lag_interval_seconds + if queue_lag_interval_seconds is not None: + self._queue_lag_age = Gauge( + "faststream_outbox_oldest_due_message_age_seconds", + "Age of the oldest claimable row, or zero for an empty claimable queue.", + ("table", "queue"), + registry=registry, + ) + self._queue_lag_freshness = Gauge( + "faststream_outbox_queue_lag_last_success_timestamp_seconds", + "Database timestamp of the last successful queue lag sample.", + ("table", "queue"), + registry=registry, + ) self._app_name = "" if app_name is EMPTY else app_name self._custom_label_keys = list((custom_labels or {}).keys()) self._custom_label_resolvers = list((custom_labels or {}).values()) @@ -242,6 +268,19 @@ def _publish_values(self, tags: Mapping[str, typing.Any]) -> tuple[str, ...]: destination = tags.get("queue", "") return (self._app_name, BROKER_SYSTEM, destination, *self._resolve_custom_values(tags)) + def _queue_lag_context(self, engine: "AsyncEngine", table: "Table", lease_ttl_by_queue: Mapping[str, float]): + interval = self._queue_lag_interval_seconds + if interval is None or not lease_ttl_by_queue: + return None + return _sample_queue_lag( + engine, + table, + age=self._queue_lag_age, + freshness=self._queue_lag_freshness, + lease_ttl_by_queue=lease_ttl_by_queue, + interval_seconds=interval, + ) + def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None: if event == "published": if self._include_bus_metrics: diff --git a/tests/test_queue_lag.py b/tests/test_queue_lag.py new file mode 100644 index 0000000..75c1b79 --- /dev/null +++ b/tests/test_queue_lag.py @@ -0,0 +1,103 @@ +import asyncio +from unittest.mock import patch +from uuid import uuid4 + +import pytest +from faststream.kafka import KafkaRouter +from prometheus_client import CollectorRegistry +from sqlalchemy import MetaData, Table, func, insert, text +from sqlalchemy.ext.asyncio import AsyncEngine + +from faststream_outbox import OutboxBroker, make_outbox_table +from faststream_outbox.metrics.prometheus import PrometheusRecorder + + +@pytest.mark.asyncio +async def test_queue_lag_samples_only_claimable_rows(pg_engine: AsyncEngine, outbox_table: Table) -> None: + async with pg_engine.begin() as conn: + for queue, due, lease in [ + ("work", "-30 seconds", None), + ("work", "-60 seconds", "-40 seconds"), + ("work", "-90 seconds", "-1 second"), + ("future", "1 hour", None), + ]: + await conn.execute( + insert(outbox_table).values( + queue=queue, + payload=b"", + next_attempt_at=func.now() + text(f"interval '{due}'"), + acquired_at=func.now() + text(f"interval '{lease}'") if lease else None, + acquired_token=uuid4() if lease else None, + ) + ) + registry = CollectorRegistry() + age = "faststream_outbox_oldest_due_message_age_seconds" + freshness = "faststream_outbox_queue_lag_last_success_timestamp_seconds" + work = {"table": outbox_table.fullname, "queue": "work"} + future = {"table": outbox_table.fullname, "queue": "future"} + + recorder = PrometheusRecorder(registry=registry, queue_lag_interval_seconds=60) + broker = OutboxBroker(pg_engine, outbox_table=outbox_table, metrics_recorder=recorder) + subscriber = broker.subscriber(["work", "future"], lease_ttl_seconds=20) + + with patch.object(subscriber, "start"): + async with broker: + await broker.start() + async with asyncio.timeout(5): + while not registry.get_sample_value(freshness, future): # noqa: ASYNC110 + await asyncio.sleep(0.01) + sample = registry.get_sample_value(age, work) + assert sample is not None + assert 60 <= sample < 90 + assert registry.get_sample_value(age, future) == 0 + + assert registry.get_sample_value(age, work) is None + + +@pytest.mark.asyncio +async def test_queue_lag_is_disabled_by_default(pg_engine: AsyncEngine, outbox_table: Table) -> None: + registry = CollectorRegistry() + recorder = PrometheusRecorder(registry=registry) + async with OutboxBroker(pg_engine, outbox_table=outbox_table, metrics_recorder=recorder): + assert all("queue_lag" not in metric.name for metric in registry.collect()) + assert all("oldest_due_message_age" not in metric.name for metric in registry.collect()) + + +@pytest.mark.parametrize("interval", [0.0, -1.0, float("inf"), float("nan")]) +def test_queue_lag_rejects_invalid_interval(interval: float) -> None: + with pytest.raises(ValueError, match="queue_lag_interval_seconds must be finite and positive"): + PrometheusRecorder(registry=CollectorRegistry(), queue_lag_interval_seconds=interval) + + +@pytest.mark.asyncio +async def test_queue_lag_ignores_foreign_subscribers(pg_engine: AsyncEngine, outbox_table: Table) -> None: + registry = CollectorRegistry() + recorder = PrometheusRecorder(registry=registry, queue_lag_interval_seconds=60) + router = KafkaRouter() + subscriber = router.subscriber("foreign") + broker = OutboxBroker(pg_engine, outbox_table=outbox_table, metrics_recorder=recorder) + broker.include_router(router) # ty: ignore[invalid-argument-type] + + with patch.object(subscriber, "start"): + async with broker: + labels = {"table": outbox_table.fullname, "queue": "foreign"} + assert ( + registry.get_sample_value("faststream_outbox_queue_lag_last_success_timestamp_seconds", labels) is None + ) + + +@pytest.mark.asyncio +async def test_queue_lag_reports_failed_sample(pg_engine: AsyncEngine, caplog: pytest.LogCaptureFixture) -> None: + registry = CollectorRegistry() + recorder = PrometheusRecorder(registry=registry, queue_lag_interval_seconds=60) + missing_table = make_outbox_table(MetaData(), "missing_lag_outbox") + broker = OutboxBroker(pg_engine, outbox_table=missing_table, metrics_recorder=recorder) + subscriber = broker.subscriber("work") + + with patch.object(subscriber, "start"): + async with broker: + async with asyncio.timeout(5): + while "Failed to sample outbox lag" not in caplog.text: # noqa: ASYNC110 + await asyncio.sleep(0.01) + labels = {"table": missing_table.fullname, "queue": "work"} + assert registry.get_sample_value("faststream_outbox_queue_lag_last_success_timestamp_seconds", labels) == 0