# Observability & Middleware (/integrations/faststream/how-to/observability)
`kubemq-faststream` is a first-class FastStream broker, not a thin transport shim. Every message — published or consumed, across all five patterns — flows through the full FastStream pipeline: parser, decoder, and the complete middleware chain. Because the broker runs that chain unchanged, **standard FastStream middleware composes directly** with KubeMQ messages. You attach middleware exactly as you would on a Kafka or Redis broker, and the same `TestKubeMQBroker` runs the pipeline end-to-end in tests.
This is the seam for observability and cross-cutting concerns: Prometheus metrics, OpenTelemetry tracing, request logging, timing, validation — all of it slots in without touching your handlers.
## Prerequisites [#prerequisites]
* A `KubeMQBroker` app already wired up (see [Configuration & Security](/integrations/faststream/how-to/configuration))
* The optional backend package for whichever middleware you enable — `prometheus-client` for `PrometheusMiddleware`, `opentelemetry-sdk` / `opentelemetry-api` for `TelemetryMiddleware`
## How Middleware Attaches [#how-middleware-attaches]
Middleware is registered on the broker constructor through the `middlewares` parameter. It accepts a sequence of FastStream middleware classes or instances:
```python title="broker.py constructor"
broker = KubeMQBroker(
"kubemq://localhost:50000",
middlewares=(my_middleware,),
)
```
Each middleware wraps the handler call. The **first** middleware in the sequence is the **outermost** wrapper — it sees the message before any other middleware and returns after them. Both the live broker and `TestKubeMQBroker` execute the full chain, so the behavior you verify in tests matches production.
Install the optional dependency for each middleware **before** enabling it. The built-in Prometheus and OpenTelemetry middlewares live in FastStream but pull their backends from separate packages: `prometheus-client` for metrics and `opentelemetry-sdk` / `opentelemetry-api` for tracing. The examples below raise a clear `SystemExit` if the package is missing.
## Prometheus Metrics [#prometheus-metrics]
FastStream ships a `PrometheusMiddleware` that records message counts, processing latencies, and error counters automatically — no handler changes required. You expose the collected metrics over an HTTP endpoint that Prometheus scrapes.
Install the backend first:
```bash
pip install prometheus-client
```
Construct the middleware with a `CollectorRegistry`, attach it to the broker, and start an HTTP metrics server once the app is up:
```python title="prometheus_metrics.py"
import asyncio
from faststream import FastStream
from faststream.prometheus import PrometheusMiddleware
from prometheus_client import CollectorRegistry, start_http_server
from kubemq_faststream import KubeMQBroker
registry = CollectorRegistry()
prometheus_middleware = PrometheusMiddleware(registry=registry)
broker = KubeMQBroker(
"kubemq://localhost:50000",
middlewares=(prometheus_middleware,),
)
app = FastStream(broker)
CHANNEL = "example.middleware.prometheus"
@broker.subscriber(events=CHANNEL)
async def handle_message(msg: dict) -> None:
# Metrics are recorded automatically — no instrumentation here.
print(f"[Prometheus] Processed: {msg}")
@app.after_startup
async def run_demo() -> None:
start_http_server(port=9090, registry=registry)
print("Prometheus metrics available at http://localhost:9090/metrics")
for i in range(5):
await broker.publish({"event_id": i, "type": "metric-demo"}, events=CHANNEL)
if __name__ == "__main__":
asyncio.run(app.run())
```
The middleware increments its counters as messages pass through the consume scope. Point a Prometheus scrape config at `http://localhost:9090/metrics` to collect them.
The metrics server in this example binds port `9090` on the local host. KubeMQ's own shared HTTP server also defaults to `9090`. When you run both the broker and the metrics endpoint on the same machine, pick a different port for `start_http_server` (for example `8000`) to avoid a collision — the broker itself connects only over gRPC on `50000`.
## OpenTelemetry Tracing [#opentelemetry-tracing]
FastStream's `TelemetryMiddleware` creates a span for each KubeMQ message and propagates trace context through the message, so a trace can span multiple services connected via KubeMQ. You wire it to a standard OpenTelemetry `TracerProvider`.
Install the backend first:
```bash
pip install opentelemetry-sdk opentelemetry-api
```
Set up a provider (here exporting spans to the console for demonstration), build the middleware from it, and attach it to the broker:
```python title="opentelemetry_tracing.py"
import asyncio
from faststream import FastStream
from faststream.opentelemetry import TelemetryMiddleware
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor
from kubemq_faststream import KubeMQBroker
resource = Resource.create({"service.name": "kubemq-faststream-demo"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)
telemetry_middleware = TelemetryMiddleware(tracer_provider=provider)
broker = KubeMQBroker(
"kubemq://localhost:50000",
middlewares=(telemetry_middleware,),
)
app = FastStream(broker)
CHANNEL = "example.middleware.otel"
@broker.subscriber(events=CHANNEL)
async def handle_message(msg: dict) -> None:
# Spans are created and linked automatically.
print(f"[OTel] Traced message: {msg}")
@app.after_startup
async def run_demo() -> None:
for i in range(3):
await broker.publish({"trace_id": i, "action": "otel-demo"}, events=CHANNEL)
if __name__ == "__main__":
asyncio.run(app.run())
```
In production, swap `ConsoleSpanExporter` for an OTLP exporter pointed at your collector. Trace context rides along with each published message, so a consumer in another service continues the same trace.
## Custom Middleware [#custom-middleware]
For logging, timing, validation, or any application-specific concern, subclass `BaseMiddleware` and override `consume_scope`. The method receives `call_next` (the next stage in the chain) and the message; you wrap the call however you need. No external packages are required.
```python title="custom_middleware.py"
import asyncio
import logging
import time
from collections.abc import Awaitable, Callable
from typing import Any
from faststream import BaseMiddleware, FastStream
from kubemq_faststream import KubeMQBroker
logging.basicConfig(level=logging.INFO)
mw_logger = logging.getLogger("custom_middleware")
class TimingMiddleware(BaseMiddleware):
"""Logs wall-clock time spent in each handler."""
async def consume_scope(
self,
call_next: Callable[[Any], Awaitable[Any]],
msg: Any,
) -> Any:
start = time.monotonic()
try:
result = await call_next(msg)
finally:
elapsed_ms = (time.monotonic() - start) * 1000
mw_logger.info("Handler took %.1f ms", elapsed_ms)
return result
broker = KubeMQBroker(
"kubemq://localhost:50000",
middlewares=(TimingMiddleware,),
)
app = FastStream(broker)
CHANNEL = "example.middleware.custom"
@broker.subscriber(events=CHANNEL)
async def handle_message(msg: dict) -> None:
await asyncio.sleep(0.05)
print(f"[Custom] Processed: {msg}")
if __name__ == "__main__":
asyncio.run(app.run())
```
The `try/finally` ensures the timing log fires even when the handler raises. A few common variations:
* **Logging** — log the message before and after `call_next`, tagging each with a request ID.
* **Validation** — inspect the message and raise before calling `call_next` to reject malformed input.
* **Error capture** — wrap `call_next` in `try/except` to record or swallow handler exceptions.
## Composing Multiple Middlewares [#composing-multiple-middlewares]
Pass several middlewares as one sequence and they stack in order — the first is the outermost wrapper. This lets you run a custom logging layer, an error-catch layer, Prometheus, and OpenTelemetry together in a single chain. A robust pattern is to add the optional middlewares only when their backend package is importable, so the app degrades gracefully when a dependency is absent.
```python title="middleware_chain.py"
import asyncio
import logging
from collections.abc import Awaitable, Callable
from typing import Any
from faststream import BaseMiddleware, FastStream
from kubemq_faststream import KubeMQBroker
logging.basicConfig(level=logging.INFO)
chain_logger = logging.getLogger("middleware_chain")
class RequestIdMiddleware(BaseMiddleware):
"""Assigns a sequential request ID to each processed message."""
_counter: int = 0
async def consume_scope(
self,
call_next: Callable[[Any], Awaitable[Any]],
msg: Any,
) -> Any:
RequestIdMiddleware._counter += 1
req_id = RequestIdMiddleware._counter
chain_logger.info("[ReqID=%d] Before handler", req_id)
result = await call_next(msg)
chain_logger.info("[ReqID=%d] After handler", req_id)
return result
class ErrorCatchMiddleware(BaseMiddleware):
"""Catches handler exceptions and logs them without crashing."""
async def consume_scope(
self,
call_next: Callable[[Any], Awaitable[Any]],
msg: Any,
) -> Any:
try:
return await call_next(msg)
except Exception:
chain_logger.exception("Handler error caught by middleware")
return None
middlewares: list[Any] = [RequestIdMiddleware, ErrorCatchMiddleware]
try:
from faststream.prometheus import PrometheusMiddleware
from prometheus_client import CollectorRegistry
middlewares.append(PrometheusMiddleware(registry=CollectorRegistry()))
chain_logger.info("Prometheus middleware added to chain")
except ImportError:
chain_logger.info("prometheus-client not installed — skipping Prometheus middleware")
try:
from faststream.opentelemetry import TelemetryMiddleware
from opentelemetry.sdk.trace import TracerProvider
middlewares.append(TelemetryMiddleware(tracer_provider=TracerProvider()))
chain_logger.info("OpenTelemetry middleware added to chain")
except ImportError:
chain_logger.info("opentelemetry-sdk not installed — skipping OTel middleware")
broker = KubeMQBroker(
"kubemq://localhost:50000",
middlewares=middlewares,
)
app = FastStream(broker)
@broker.subscriber(events="example.middleware.chain")
async def handle_message(msg: dict) -> None:
print(f"[Chain] Handled: {msg}")
if __name__ == "__main__":
asyncio.run(app.run())
```
With both optional backends installed, a message passes through `RequestIdMiddleware` → `ErrorCatchMiddleware` → Prometheus → OpenTelemetry → your handler, then unwinds in reverse. Note that `ErrorCatchMiddleware` sits *outside* the metrics and tracing layers, so the inner middlewares still observe and record the failing message before the error is caught.
## Error Handling & Graceful Shutdown [#error-handling--graceful-shutdown]
Observability is most valuable around failures, so it helps to understand how the broker behaves when things go wrong.
### Handler Exceptions [#handler-exceptions]
When a subscriber raises, FastStream catches the exception, logs it, and the application keeps processing subsequent messages — one bad message does not crash the app. For **events** (fire-and-forget), the failing message is not re-delivered. For **queues**, settlement follows the subscriber's `AckPolicy` (the default `AckPolicy.ACK` nacks and requeues on error). An `ErrorCatchMiddleware` like the one above can also intercept exceptions before they reach FastStream's default logging.
```python title="handler_exception.py"
@broker.subscriber(events="example.error.exception")
async def handle_message(msg: dict) -> None:
global counter # noqa: PLW0603
counter += 1
if counter == 2:
raise ValueError(f"Simulated failure on message #{counter}")
print(f"Message #{counter} processed OK: {msg}")
```
The app continues running after the `ValueError` and processes message `#3` normally.
### Automatic Reconnection [#automatic-reconnection]
The underlying KubeMQ SDK handles gRPC reconnection transparently. If the broker restarts mid-run, the client re-establishes its channels and delivery resumes — your code does not need retry wiring for the connection itself. gRPC keepalive (configured via `keepalive_time_ms` / `keepalive_timeout_ms`) keeps the channel healthy between messages.
```python title="reconnection.py"
@app.after_startup
async def run_demo() -> None:
for i in range(1, 11):
try:
await broker.publish({"seq": i}, events="example.error.reconnect")
print(f"Published #{i}")
except Exception as exc:
print(f"Publish #{i} failed (will retry): {exc}")
await asyncio.sleep(1)
```
### Graceful Shutdown & Draining [#graceful-shutdown--draining]
On `SIGTERM` or `SIGINT`, the broker stops accepting new work and waits for in-flight handlers to finish before closing its connections. The `graceful_timeout` constructor option (default `15.0` seconds) caps how long it waits — handlers still running when the timeout elapses are cancelled. Set it to match your slowest handler so in-flight messages drain cleanly:
```python title="graceful_shutdown.py"
import signal
broker = KubeMQBroker("kubemq://localhost:50000", graceful_timeout=10.0)
app = FastStream(broker)
@broker.subscriber(events="example.error.shutdown")
async def handle_message(msg: dict) -> None:
print(f"Processing: {msg}")
await asyncio.sleep(0.5)
print(f"Done: {msg}")
@app.after_startup
async def run_demo() -> None:
def request_stop(signum: int, _frame: object) -> None:
sig_name = signal.Signals(signum).name
print(f"\n{sig_name} received — requesting graceful stop...")
asyncio.get_event_loop().call_soon_threadsafe(asyncio.ensure_future, app.stop())
signal.signal(signal.SIGINT, request_stop)
signal.signal(signal.SIGTERM, request_stop)
```
Running under the FastStream CLI (`faststream run app:app`) wires signal handling and graceful shutdown for you, honoring the same `graceful_timeout`. The manual signal handlers above are for scripts run directly with `python`.
## Serialization Through the Pipeline [#serialization-through-the-pipeline]
The parser and decoder are part of the same pipeline that runs the middleware chain, so your choice of serialization travels with every message — published and consumed alike. By default, messages are encoded and decoded as **JSON**, which transparently handles dicts, dataclasses, and Pydantic models.
To use a different format, supply a custom `decoder` (and, for outbound encoding, a matching `parser`) on the broker constructor. The decoder receives the raw FastStream message and returns the object handed to your handler.
No configuration needed — JSON is the default codec.
```python title="json_default.py"
broker = KubeMQBroker("kubemq://localhost:50000")
@broker.subscriber(events="example.serial.json")
async def handle_message(msg: dict) -> None:
print(f"[JSON] {msg}")
```
Compact binary encoding. Requires `pip install msgpack`.
```python title="msgpack_serializer.py"
from typing import Any
import msgpack
async def msgpack_decoder(msg: Any) -> Any:
raw = msg.body if hasattr(msg, "body") else msg
if isinstance(raw, bytes):
return msgpack.unpackb(raw, raw=False)
return raw
broker = KubeMQBroker(
"kubemq://localhost:50000",
decoder=msgpack_decoder,
)
```
Strongly-typed contracts. Requires `pip install protobuf`.
```python title="protobuf_messages.py"
from typing import Any
from google.protobuf import json_format, struct_pb2
def protobuf_decode(raw: bytes) -> dict:
struct = struct_pb2.Struct()
struct.ParseFromString(raw)
return json_format.MessageToDict(struct)
async def protobuf_decoder(msg: Any) -> Any:
raw = msg.body if hasattr(msg, "body") else msg
if isinstance(raw, bytes):
return protobuf_decode(raw)
return raw
broker = KubeMQBroker(
"kubemq://localhost:50000",
decoder=protobuf_decoder,
)
```
Any application-specific format — here a simple `key=value|key=value` codec.
```python title="custom_serializer.py"
from typing import Any
SEPARATOR = "|"
def custom_decode(raw: bytes) -> dict:
text = raw.decode() if isinstance(raw, bytes) else str(raw)
result = {}
for pair in text.split(SEPARATOR):
if "=" in pair:
k, v = pair.split("=", 1)
result[k] = v
return result
async def custom_decoder(msg: Any) -> Any:
raw = msg.body if hasattr(msg, "body") else msg
if isinstance(raw, bytes):
return custom_decode(raw)
return raw
broker = KubeMQBroker(
"kubemq://localhost:50000",
decoder=custom_decoder,
)
```
Because the decoder runs inside the pipeline, it composes with middleware: a metrics or tracing middleware sees every message regardless of which codec decoded it, and `TestKubeMQBroker` exercises the same decoder you ship.
## Running the Examples [#running-the-examples]
All of the snippets above run against a local KubeMQ broker. Start one in Docker — `kubemq-faststream` needs only the gRPC port:
There is no connector to enable and no HTTP flag to set: FastStream connects directly over gRPC on `50000`. Install the relevant optional backend (`prometheus-client`, `opentelemetry-sdk`, `msgpack`, or `protobuf`) before running an example that requires it.
## Related [#related]