From 96bdd9baed221163c3f71379b9e717e0007a5dc6 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sun, 14 Jun 2026 17:38:23 +0300 Subject: [PATCH] feat: surface drain_timeout on the bundled Prometheus/OTel adapters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #92 added the drain_timeout recorder event but only the raw recorder seam saw it — the bundled adapters ignored it (forward-compatible) and the docs didn't mention it, so operators got no out-of-the-box metric. Complete the feature: - Prometheus: faststream_outbox_drain_timeout_total counter (consume labels). - OTel: messaging.outbox.drain_timeout meter counter (operation=process). - Docs: observability.md recorder-event table + event set; CLAUDE.md and architecture/metrics.md emission lists. Added PLR0911 to the two adapters' __call__ noqa (one more event branch in the flat event-dispatch dispatch, same shape as the existing C901/PLR0912 suppressions). Tests: test_{prometheus,otel}_drain_timeout_* assert the counter increments. just test -> 518 passed, 100% coverage; just lint clean; mkdocs --strict builds. Co-Authored-By: Claude Opus 4.8 (1M context) --- CLAUDE.md | 2 +- architecture/metrics.md | 2 +- docs/usage/observability.md | 3 ++- faststream_outbox/metrics/opentelemetry.py | 11 ++++++++++- faststream_outbox/metrics/prometheus.py | 12 +++++++++++- tests/test_metrics_opentelemetry.py | 9 +++++++++ tests/test_metrics_prometheus.py | 7 +++++++ 7 files changed, 41 insertions(+), 5 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index ee2e45a..02fac73 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -174,7 +174,7 @@ Caller owns the `AsyncEngine` — the broker never disposes it. The engine lives Two complementary seams — **don't collapse them.** -- **Recorder seam** (`OutboxBroker(..., metrics_recorder=...)`): `Callable[[str, Mapping[str, Any]], None]`. Subscriber emits `fetched`, `dispatched`, `acked`, `nacked_retried`, `nacked_terminal`, `lease_lost`, plus `dlq_written` when `dlq_table` is set. Producer emits `published`. Default `_noop_recorder` lets sites fire unconditionally. Every call site is wrapped in `try/except` + DEBUG log. **Recorder must not block** (sync `Counter.inc()` fine; HTTP/StatsD not). `dlq_written` vs `nacked_terminal` divergence detects DLQ misconfiguration. +- **Recorder seam** (`OutboxBroker(..., metrics_recorder=...)`): `Callable[[str, Mapping[str, Any]], None]`. Subscriber emits `fetched`, `dispatched`, `acked`, `nacked_retried`, `nacked_terminal`, `lease_lost`, `drain_timeout` (on a timed-out `stop()` drain), plus `dlq_written` when `dlq_table` is set. Producer emits `published`. The bundled Prometheus/OTel adapters translate every one of these (`drain_timeout` → `_outbox_drain_timeout_total` / `messaging.outbox.drain_timeout`). Default `_noop_recorder` lets sites fire unconditionally. Every call site is wrapped in `try/except` + DEBUG log. **Recorder must not block** (sync `Counter.inc()` fine; HTTP/StatsD not). `dlq_written` vs `nacked_terminal` divergence detects DLQ misconfiguration. - **Native middleware** (`opentelemetry/`, `prometheus/`): thin subclasses of upstream's `TelemetryMiddleware[OutboxPublishCommand]` and `PrometheusMiddleware[OutboxInnerMessage, OutboxPublishCommand]`. Register via the public `OutboxBroker(..., middlewares=[...])` constructor kwarg (forwarded internally as `broker_middlewares`). Fire on `consume_scope` (via `dispatch_one → self.consume(row)`) and `publish_scope` (via `_basic_publish`). Why two: middleware owns `consume_scope` / `publish_scope` (spans, durations, status, size). Recorder owns events **outside** the bus — `fetched` (no `StreamMessage` at fetch time), `lease_lost` (after `consume_scope` exits), `nacked_terminal(reason="max_deliveries")` (before consume opens). Each fires for events the other physically cannot observe. diff --git a/architecture/metrics.md b/architecture/metrics.md index 45fe97b..3c5d8ed 100644 --- a/architecture/metrics.md +++ b/architecture/metrics.md @@ -6,7 +6,7 @@ User-facing: `docs/usage/observability.md`. Invariant summary: `CLAUDE.md` § Me `OutboxBroker(..., metrics_recorder=...)` accepts a `MetricsRecorder = Callable[[str, Mapping[str, Any]], None]`. The default (`_noop_recorder`) lets instrumentation sites call unconditionally. The recorder threads through `OutboxBrokerConfig.metrics_recorder` to two places: -- **Subscriber emission points** (`OutboxSubscriber._emit_metric`): `fetched`, `dispatched`, `acked`, `nacked_retried`, `nacked_terminal`, `lease_lost`, plus `dlq_written` when `dlq_table` is configured. +- **Subscriber emission points** (`OutboxSubscriber._emit_metric`): `fetched`, `dispatched`, `acked`, `nacked_retried`, `nacked_terminal`, `lease_lost`, `drain_timeout` (a `stop()` drain that exceeded `graceful_timeout`), plus `dlq_written` when `dlq_table` is configured. - **Producer emission point** (`OutboxProducer._emit_metric`): `published`. The producer reads the recorder from its own constructor kwarg (passed in alongside the config field) so the canonical insert path doesn't have to reach through the broker config at call time. diff --git a/docs/usage/observability.md b/docs/usage/observability.md index 75d9d7f..ab2b90b 100644 --- a/docs/usage/observability.md +++ b/docs/usage/observability.md @@ -28,7 +28,7 @@ from faststream_outbox import MetricsRecorder, OutboxBroker def recorder(event: str, tags: dict) -> None: # event ∈ {fetched, dispatched, acked, nacked_retried, nacked_terminal, - # lease_lost, dlq_written, published} + # lease_lost, dlq_written, drain_timeout, published} # tags always include "queue"; subscriber-side events also include "subscriber" print(event, tags) @@ -82,6 +82,7 @@ broken recorder never poisons the dispatch loop. | `lease_lost` | `queue`, `subscriber`, `phase`, `row_id`, `deliveries_count` | | Terminal or retry write found `rowcount == 0` (`phase` = `terminal` \| `retry`) | | `published` | `queue`, `status`, `count`, `size_bytes`, `duration_seconds` | `exception_type` | Producer, after the INSERT executes (pre-commit; also fires on error with `status="error"`) | | `dlq_written` | `queue`, `subscriber`, `deliveries_count`, `failure_reason` | `exception_type` | DLQ CTE wrote an audit row. `exception_type` is **omitted** — not set to `None` — when the terminal had no exception (`max_deliveries`, or a manual `reject()` without one) | +| `drain_timeout` | `queue`, `subscriber`, `drain_timeout_seconds` | | A `stop()` drain exceeded `graceful_timeout`; in-flight rows were abandoned to lease-expiry retry. `queue` is the subscriber's **first** queue | `reason` on `nacked_terminal` is one of `max_deliveries`, `retry_terminal`, `rejected`. The same value lands in the DLQ diff --git a/faststream_outbox/metrics/opentelemetry.py b/faststream_outbox/metrics/opentelemetry.py index 70e5de9..2e6b96a 100644 --- a/faststream_outbox/metrics/opentelemetry.py +++ b/faststream_outbox/metrics/opentelemetry.py @@ -149,6 +149,11 @@ def __init__( unit="event", description="DLQ audit rows written by terminal flush, broken down by reason", ) + self._drain_timeout = self._meter.create_counter( + name="messaging.outbox.drain_timeout", + unit="event", + description="Drains that exceeded graceful_timeout, abandoning in-flight rows to lease-expiry retry", + ) def _attrs(self, tags: Mapping[str, typing.Any], *, operation: str) -> dict[str, typing.Any]: attrs: dict[str, typing.Any] = { @@ -161,7 +166,7 @@ def _attrs(self, tags: Mapping[str, typing.Any], *, operation: str) -> dict[str, attrs[_ATTR_HANDLER] = handler return attrs - def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None: # noqa: C901, PLR0912 + def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None: # noqa: C901, PLR0911, PLR0912 if event == "fetched": self._fetch_batches.add(1, self._attrs(tags, operation="receive")) return @@ -204,6 +209,10 @@ def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None: # noqa: self._dlq_written.add(1, attrs) return + if event == "drain_timeout": + self._drain_timeout.add(1, self._attrs(tags, operation="process")) + return + if event == "published": attrs = self._attrs(tags, operation="publish") attrs[_ATTR_STATUS] = tags.get("status", "success") diff --git a/faststream_outbox/metrics/prometheus.py b/faststream_outbox/metrics/prometheus.py index add2495..2707858 100644 --- a/faststream_outbox/metrics/prometheus.py +++ b/faststream_outbox/metrics/prometheus.py @@ -212,6 +212,12 @@ def __init__( [*consume_labels, "reason"], registry=registry, ) + self._drain_timeout = Counter( + f"{p}_outbox_drain_timeout_total", + "Drains that exceeded graceful_timeout, abandoning in-flight rows to lease-expiry retry.", + consume_labels, + registry=registry, + ) def _resolve_custom_values(self, tags: Mapping[str, typing.Any]) -> tuple[str, ...]: return tuple( @@ -234,7 +240,7 @@ 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 __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None: # noqa: C901, PLR0912 + def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None: # noqa: C901, PLR0911, PLR0912 consume_base = self._consume_values(tags) if event == "fetched": @@ -282,6 +288,10 @@ def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None: # noqa: self._dlq_written.labels(*consume_base, tags["failure_reason"]).inc() return + if event == "drain_timeout": + self._drain_timeout.labels(*consume_base).inc() + return + if event == "published": publish_base = self._publish_values(tags) status = tags.get("status", "success") diff --git a/tests/test_metrics_opentelemetry.py b/tests/test_metrics_opentelemetry.py index f6552b5..63d6f2b 100644 --- a/tests/test_metrics_opentelemetry.py +++ b/tests/test_metrics_opentelemetry.py @@ -123,6 +123,15 @@ def test_otel_unknown_event_is_silently_ignored() -> None: assert _collect_metrics(reader) == {} +def test_otel_drain_timeout_emits_counter() -> None: + """The subscriber-emitted drain_timeout event increments its dedicated meter counter.""" + reader, rec = _reader_and_recorder() + rec("drain_timeout", {"queue": "q", "subscriber": "h", "drain_timeout_seconds": 0.2}) + metrics = _collect_metrics(reader) + assert "messaging.outbox.drain_timeout" in metrics + assert sum(p.value for p in metrics["messaging.outbox.drain_timeout"].data_points) == 1 + + def test_otel_dlq_written_emits_counter_with_reason_attr() -> None: reader, rec = _reader_and_recorder() rec( diff --git a/tests/test_metrics_prometheus.py b/tests/test_metrics_prometheus.py index 959930f..9c9712d 100644 --- a/tests/test_metrics_prometheus.py +++ b/tests/test_metrics_prometheus.py @@ -219,6 +219,13 @@ def test_prometheus_unknown_event_is_silently_ignored() -> None: rec("future_event_not_yet_added", {"queue": "q", "subscriber": "h"}) # forward-compat +def test_prometheus_drain_timeout_increments() -> None: + """The subscriber-emitted drain_timeout event increments its dedicated counter.""" + reg, rec = _make_recorder() + rec("drain_timeout", {"queue": "q", "subscriber": "h", "drain_timeout_seconds": 0.2}) + assert _sample(reg, "faststream_outbox_drain_timeout_total", _base_labels()) == 1.0 + + def test_prometheus_dlq_written_records_reason_label() -> None: reg, rec = _make_recorder() rec(