Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/integrations/faststream.md
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ bootstrapper_config = FastStreamConfig(
KafkaBroker: KafkaTelemetryMiddleware,
RedisBroker: RedisTelemetryMiddleware,
},
prometheus_metrics_path="/metrics",
prometheus_middleware_cls={
KafkaBroker: KafkaPrometheusMiddleware,
RedisBroker: RedisPrometheusMiddleware,
Expand Down
17 changes: 8 additions & 9 deletions docs/introduction/configuration.md
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,8 @@ every removed option.
Prometheus is the cheapest of the three non-logging instruments; see [Performance](performance.md).

Prometheus is on by default when its extra is installed, serving metrics at `prometheus_metrics_path`
(default: `/metrics`). Set `prometheus_metrics_path` to an empty string to disable it.
(default: `/metrics`). Set `prometheus_metrics_path` to an empty string to disable it. FastStream is
the exception: its default is empty; see [Prometheus FastStream](#prometheus-faststream).

Additional parameters:

Expand Down Expand Up @@ -164,20 +165,18 @@ Additional parameters for Litestar integration:

Prometheus's integration for FastStream requires `prometheus_client` package.

To bootstrap Prometheus for FastStream, you must provide at least one of:
On FastStream `prometheus_metrics_path` defaults to an empty string since 2.0, so the metrics endpoint
is off until you set one. Once set, it serves `prometheus_client.REGISTRY`, the registry FastAPI,
Litestar and FastMCP serve too. That includes the default process and platform collectors and any
collector you register without naming a registry.

- `prometheus_middleware_cls` - the broker metrics middleware, e.g.
`faststream.redis.prometheus.RedisPrometheusMiddleware`. It is constructed with the instrument's
registry and added to every broker of the first broker's type. For an application with brokers of
different types, pass a mapping from broker class to middleware class instead; see
[multiple brokers](../integrations/faststream.md#multiple-brokers).
- `prometheus_collector_registry` - a `prometheus_client.CollectorRegistry` of your own, used in
place of the fresh one the instrument would otherwise build.

With neither, nothing would populate the registry, so the metrics endpoint is not mounted and the
instrument is skipped. Unlike the other frameworks, FastStream serves a private registry rather than
`prometheus_client.REGISTRY`, so an endpoint with no middleware and no injected registry would have
nothing to report.
- `prometheus_collector_registry` - a `prometheus_client.CollectorRegistry` of your own, served and
passed to the middleware in place of `prometheus_client.REGISTRY`.

### Prometheus FastAPI

Expand Down
18 changes: 2 additions & 16 deletions lite_bootstrap/bootstrappers/faststream_bootstrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,7 @@ class FastStreamConfig(
| Mapping[type, type[FastStreamPrometheusMiddlewareProtocol]]
| None
) = None
prometheus_metrics_path: str = ""
prometheus_collector_registry: "prometheus_client.CollectorRegistry | None" = None
faststream_log_level: int = logging.WARNING
faststream_health_check_broker_timeout: float = 5.0
Expand Down Expand Up @@ -196,29 +197,14 @@ def bootstrap(self) -> None:
)


def _make_collector_registry() -> "prometheus_client.CollectorRegistry":
return prometheus_client.CollectorRegistry()


@dataclasses.dataclass(kw_only=True)
class FastStreamPrometheusInstrument(PrometheusInstrument[FastStreamConfig]):
collector_registry: "prometheus_client.CollectorRegistry" = dataclasses.field(init=False)
not_configured_reason = (
PrometheusInstrument.not_configured_reason
+ ", or neither prometheus_middleware_cls nor prometheus_collector_registry is set"
)
missing_dependency_message = "prometheus_client is not installed"

def __post_init__(self) -> None:
injected = self.bootstrap_config.prometheus_collector_registry
self.collector_registry = injected if injected is not None else _make_collector_registry()

@classmethod
def is_configured(cls, bootstrap_config: "FastStreamConfig") -> bool: # ty: ignore[invalid-method-override]
return super().is_configured(bootstrap_config) and (
bootstrap_config.prometheus_middleware_cls is not None
or bootstrap_config.prometheus_collector_registry is not None
)
self.collector_registry = injected if injected is not None else prometheus_client.REGISTRY

@staticmethod
def dependencies_installed() -> bool:
Expand Down
42 changes: 34 additions & 8 deletions tests/test_faststream_bootstrap.py
Original file line number Diff line number Diff line change
Expand Up @@ -392,25 +392,51 @@ async def test_faststream_prometheus_uses_injected_registry(broker: RedisBroker,
bootstrapper.teardown()


async def test_faststream_prometheus_is_skipped_without_middleware_or_registry(broker: RedisBroker) -> None:
bootstrap_config = dataclasses.replace(
build_faststream_config(broker=broker),
prometheus_middleware_cls=None,
prometheus_collector_registry=None,
def test_faststream_prometheus_is_skipped_without_metrics_path(broker: RedisBroker) -> None:
bootstrap_config = FastStreamConfig(
application=faststream.asgi.AsgiFastStream(broker),
prometheus_middleware_cls=RedisPrometheusMiddleware,
logging_buffer_capacity=0,
)
bootstrapper = FastStreamBootstrapper(bootstrap_config=bootstrap_config)
skipped = dict(bootstrapper.skipped_instruments)
assert FastStreamPrometheusInstrument in skipped
assert "prometheus_collector_registry" in skipped[FastStreamPrometheusInstrument]
assert skipped[FastStreamPrometheusInstrument] == "prometheus_metrics_path is empty or not valid"


async def test_faststream_prometheus_serves_the_default_registry(broker: RedisBroker) -> None:
counter_name = f"default_registry_counter_{uuid.uuid4().hex}_total"
counter = prometheus_client.Counter(counter_name, "Default registry counter")
counter.inc()
bootstrap_config = dataclasses.replace(
build_faststream_config(broker=broker), prometheus_middleware_cls=None, prometheus_collector_registry=None
)
bootstrapper = FastStreamBootstrapper(bootstrap_config=bootstrap_config)
application = bootstrapper.bootstrap()
try:
instruments = [one for one in bootstrapper.instruments if isinstance(one, FastStreamPrometheusInstrument)]
assert [one.collector_registry for one in instruments] == [prometheus_client.REGISTRY]
with TestClient(app=application) as test_client:
async with TestRedisBroker(broker):
response = test_client.get(bootstrap_config.prometheus_metrics_path)
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.status_code == status.HTTP_200_OK
assert counter_name.encode() in response.content
assert b"python_info" in response.content
finally:
bootstrapper.teardown()
prometheus_client.REGISTRY.unregister(counter)


def test_faststream_default_registry_takes_two_brokers_and_a_second_bootstrap() -> None:
for _ in range(2):
brokers: list[BrokerUsecase[typing.Any, typing.Any, typing.Any]] = [RedisBroker(), RedisBroker()]
bootstrapper = _bootstrap_with_brokers(
brokers, opentelemetry_log_traces=False, opentelemetry_middleware_cls=None
)
try:
for one in brokers:
assert _middleware_types(one) == [RedisPrometheusMiddleware]
finally:
bootstrapper.teardown()


def test_faststream_logging_teardown_runs_super_when_broker_write_raises(
Expand Down
Loading