# Middleware & Observability (/integrations/watermill/how-to/middleware-and-observability)
Because `watermill-kubemq` implements Watermill's `message.Publisher` and `message.Subscriber` interfaces faithfully, the standard Watermill middleware stack, metrics component, and tracing conventions all work unchanged on top of KubeMQ. This guide covers middleware compatibility per pattern, OpenTelemetry trace propagation, Prometheus metrics, KEDA autoscaling, and logging — everything you need to operate a Watermill-over-KubeMQ deployment in production.
## Prerequisites [#prerequisites]
* `github.com/kubemq-io/watermill-kubemq` installed, with a Watermill `Router` and at least one Publisher/Subscriber already wired up (see [Getting Started with Watermill](/integrations/watermill/tutorials/getting-started))
* A running KubeMQ broker reachable on `localhost:50000`
* KEDA v2.x and the `kubemq-keda-scaler` deployed in the cluster if you use [KEDA Autoscaling](#keda-autoscaling) below
## Middleware Compatibility [#middleware-compatibility]
All standard Watermill middleware works with the plugin. A few middlewares are only meaningful for specific KubeMQ patterns — for example, `Retry` relies on Nack-triggered redelivery, which only the Queues pattern provides, while `InstantAck` makes sense only for the fire-and-forget Events and EventsStore patterns.
| Middleware | Events | EventsStore | Queues | Notes |
| -------------- | ------ | ----------- | ------ | -------------------------------------------------------- |
| Retry | N/A | N/A | Yes | Nack triggers redelivery; only meaningful for Queues |
| Throttle | Yes | Yes | Yes | Rate-limits handler invocations |
| CorrelationID | Yes | Yes | Yes | Stored in Tags via Metadata |
| Timeout | Yes | Yes | Yes | Context deadline propagated |
| Poison Queue | N/A | N/A | Yes | Application-level DLQ (alternative to KubeMQ native DLQ) |
| CircuitBreaker | Yes | Yes | Yes | Handler-level |
| Deduplication | Yes | Yes | Yes | Uses UUID from Tags |
| InstantAck | Yes | Yes | N/A | For fire-and-forget patterns |
| Recoverer | Yes | Yes | Yes | Panic recovery |
| Prometheus | Yes | Yes | Yes | Built-in metrics component |
| OpenTelemetry | Yes | Yes | Yes | Community middleware |
`Retry` and `Poison Queue` depend on Nack-driven redelivery, so they only apply to the **Queues** pattern. For application-level dead-lettering on Queues — both the Poison Queue middleware and the native `QueueMessagePolicy` route — see the [Queues](/integrations/watermill/how-to/queues) capability page.
## Adding Middleware to the Router [#adding-middleware-to-the-router]
Middleware is registered on the Watermill `Router` with `AddMiddleware`. Each middleware wraps every handler attached to the router, in the order it was added. The `Recoverer` middleware — which converts a panic in a handler into an error instead of crashing the process — is the one you almost always want:
```go title="middleware.go"
import "github.com/ThreeDotsLabs/watermill/message/router/middleware"
router, err := message.NewRouter(message.RouterConfig{}, logger)
if err != nil {
log.Fatal(err)
}
// Catch panics in handlers and convert them to errors (Nack on Queues).
router.AddMiddleware(middleware.Recoverer)
```
You can stack several middlewares. On the **Queues** pattern, a common production layering is throttle → retry → recoverer, so that a transient handler failure is retried with backoff before the message is ultimately nacked:
```go title="middleware_stack.go"
import "time"
router.AddMiddleware(
middleware.Throttle{
Count: 100,
Duration: time.Second,
}.Middleware,
middleware.Retry{
MaxRetries: 3,
InitialInterval: time.Millisecond * 100,
Logger: logger,
}.Middleware,
middleware.Recoverer,
)
```
The repository ships runnable middleware examples for each of the common cases. With a broker running on `localhost:50000`, run them from the project root:
```bash
go run ./examples/middleware/recoverer/main.go
go run ./examples/middleware/retry/main.go
go run ./examples/middleware/throttle/main.go
go run ./examples/middleware/correlation-id/main.go
go run ./examples/middleware/poison-queue/main.go
go run ./examples/middleware/custom-middleware/main.go
```
The `recoverer`, `retry`, and `poison-queue` examples use `Pattern: kubemq.PatternQueues`, because their behavior depends on Nack-driven redelivery. The `throttle` and `correlation-id` examples work across all patterns.
## OpenTelemetry Trace Propagation [#opentelemetry-trace-propagation]
The plugin propagates trace context automatically using the [W3C Trace Context](https://www.w3.org/TR/trace-context/) format carried in KubeMQ Tags. Internally it injects the trace context into the message metadata before publishing and extracts it again after receiving, both via `otel.GetTextMapPropagator()`:
```go title="pkg/kubemq/helpers.go"
// injectTraceContext injects OTel trace context into the message metadata
// before marshaling, so it flows through to KubeMQ Tags.
func injectTraceContext(msg *message.Message) {
otel.GetTextMapPropagator().Inject(
msg.Context(),
propagation.MapCarrier(msg.Metadata),
)
}
// extractTraceContext extracts OTel trace context from message metadata.
func extractTraceContext(msg *message.Message) {
ctx := otel.GetTextMapPropagator().Extract(
msg.Context(),
propagation.MapCarrier(msg.Metadata),
)
msg.SetContext(ctx)
}
```
Because these run inside the plugin, you only have to do three things in your application:
### Register a global propagator [#register-a-global-propagator]
The plugin reads `otel.GetTextMapPropagator()`, so register a `TraceContext{}` propagator globally once at startup. **This is required** — without it, the injected and extracted carriers are no-ops and no trace flows through.
```go title="otel_setup.go"
import (
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/propagation"
)
otel.SetTextMapPropagator(propagation.TraceContext{})
```
### Attach a span before publishing [#attach-a-span-before-publishing]
Start a span and set its context on the message with `msg.SetContext` **before** calling `Publish`. The plugin's `injectTraceContext` then writes the `traceparent` (and `tracestate`) headers into the message metadata.
```go title="publish_with_span.go"
tracer := otel.Tracer("watermill-kubemq-example")
spanCtx, span := tracer.Start(ctx, "publish-order")
defer span.End()
msg := message.NewMessage(watermill.NewUUID(), []byte("Order created"))
msg.SetContext(spanCtx) // propagated into KubeMQ tags
if err := pub.Publish("watermill-obs.otel-trace", msg); err != nil {
log.Fatal(err)
}
```
### Build a child span after receiving [#build-a-child-span-after-receiving]
On the receiving side, `extractTraceContext` has already restored the propagated context. Read it back with `received.Context()` and start a child span from it:
```go title="receive_with_span.go"
received := <-msgs
// Trace context was restored by the plugin; use it as the parent.
_, childSpan := tracer.Start(received.Context(), "process-order")
defer childSpan.End()
// ... process the message ...
received.Ack()
```
A complete, runnable program lives in the repository. It wires a stdout span exporter, publishes a message inside a span, and starts a child span from the propagated context after receiving:
```bash
go run ./examples/observability/otel-trace-propagation/main.go
```
The OTel trace-propagation example pulls in OpenTelemetry SDK dependencies that are not part of the plugin's default module. Add them before running the example:
```bash
go get go.opentelemetry.io/otel/sdk@v1.42.0
go get go.opentelemetry.io/otel/exporters/stdout/stdouttrace@v1.42.0
```
### Span Attributes Set by the Plugin [#span-attributes-set-by-the-plugin]
When tracing is active, the plugin annotates its spans with the following [OpenTelemetry messaging semantic-convention](https://opentelemetry.io/docs/specs/semconv/messaging/) attributes:
| Attribute | Value |
| ------------------------------- | ------------------------------------- |
| `messaging.system` | `"kubemq"` |
| `messaging.operation.name` | `"publish"`, `"receive"`, `"process"` |
| `messaging.destination.name` | Topic/channel name |
| `messaging.message.id` | Watermill UUID |
| `messaging.client.id` | KubeMQ ClientID |
| `messaging.consumer.group.name` | ConsumerGroup (if set) |
| `server.address` | KubeMQ host |
| `server.port` | KubeMQ port |
### Reading Propagated Trace Headers [#reading-propagated-trace-headers]
The W3C Trace Context headers arrive as ordinary message metadata, so you can inspect them directly on a received message — useful for logging the trace ID or correlating with an external system. The `traceparent` key carries the trace and span IDs; `tracestate` carries vendor-specific state when present:
```go title="read_traceparent.go"
received := <-msgs
// traceparent format: 00---
traceparent := received.Metadata.Get("traceparent")
tracestate := received.Metadata.Get("tracestate")
log.Printf("traceparent=%s tracestate=%s", traceparent, tracestate)
// e.g. traceparent=00---01
```
## Prometheus Metrics [#prometheus-metrics]
Watermill's built-in Prometheus metrics component works out of the box. Build a metrics builder with the `watermill` namespace and `kubemq` subsystem, attach it to the router with `AddPrometheusRouterMetrics`, and expose a `/metrics` endpoint on a port of your choosing:
```go title="prometheus.go"
import (
"net/http"
"github.com/ThreeDotsLabs/watermill/components/metrics"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
)
reg := prometheus.NewRegistry()
metricsBuilder := metrics.NewPrometheusMetricsBuilder(reg, "watermill", "kubemq")
metricsBuilder.AddPrometheusRouterMetrics(router)
// Expose the metrics endpoint on a dedicated port.
go func() {
http.ListenAndServe(":8081", promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
}()
```
This registers three histograms, named from the `watermill` namespace and `kubemq` subsystem:
| Metric | Measures |
| ----------------------------------------------------------- | ----------------------------------- |
| `watermill_kubemq_publish_time_seconds` | Time spent publishing a message |
| `watermill_kubemq_subscriber_received_message_time_seconds` | Time a message spent being received |
| `watermill_kubemq_handler_execution_time_seconds` | Time spent executing a handler |
This `:8081/metrics` endpoint is your **application's** Prometheus surface for Watermill handler metrics — it is separate from the KubeMQ broker's own metrics on port `9090`. Scrape both: the broker's `9090` for queue depth and broker health, and your app's port for handler-level latency.
## KEDA Autoscaling [#keda-autoscaling]
KubeMQ has an existing [KEDA](https://keda.sh/) scaler for Kubernetes-native autoscaling. Pair it with the **Queues** pattern to scale your Watermill consumer pods by queue depth: as backlog grows past a threshold, KEDA adds replicas; as it drains, replicas scale back down.
Install the scaler from the [kubemq-keda](https://github.com/kubemq-io/kubemq-keda) repository, then create a `ScaledObject` with a `kubemq` trigger. The trigger's `address` points at the broker's gRPC port, `channel` is the queue your consumers poll, and `queueLength` is the target backlog per replica:
```yaml title="scaledobject.yaml"
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
name: watermill-worker
spec:
scaleTargetRef:
name: watermill-worker-deployment
minReplicaCount: 1
maxReplicaCount: 20
triggers:
- type: kubemq
metadata:
address: "kubemq.kubemq.svc.cluster.local:50000"
channel: "ml.requests"
queueLength: "10"
```
### When KEDA Helps [#when-keda-helps]
KEDA autoscaling is meaningful only for the **Queues** pattern, because that is where backlog is an observable, drainable quantity:
* **Queues** — messages persist until acked, so a growing queue is a clear signal that consumers are falling behind. KEDA reacts by adding replicas, and each new pod joins the same `ConsumerGroup` to compete for messages. This is the autoscaling sweet spot.
* **Events** — fire-and-forget with no backlog to measure; there is nothing for the scaler to react to.
* **EventsStore** — consumption advances an offset rather than draining a queue, so queue-depth scaling does not apply.
For KEDA to distribute work across the scaled replicas, your Watermill subscribers must share a `ConsumerGroup` on the Queues pattern, so KubeMQ load-balances messages across competing consumers rather than fanning out to all of them.
## Logging [#logging]
The plugin emits structured logs through the Watermill `LoggerAdapter` you pass in `PublisherConfig.Logger` and `SubscriberConfig.Logger`. Publish and subscribe lifecycle events are logged at `Trace` level, and failures at `Error` level. The quickest way to see this output while troubleshooting is the built-in `StdLogger` with debug and trace enabled:
```go title="logging.go"
// debug=true, trace=true — verbose lifecycle logging for troubleshooting
logger := watermill.NewStdLogger(true, true)
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
Address: "localhost:50000",
Pattern: kubemq.PatternEvents,
Logger: logger,
})
```
To route logs into an external framework (zap, zerolog, logrus), implement `watermill.LoggerAdapter` and pass your implementation as the `Logger`. Each method receives structured `watermill.LogFields`:
```go title="custom_logger.go"
type PrefixLogger struct{ prefix string }
func (l *PrefixLogger) Error(msg string, err error, fields watermill.LogFields) {
log.Printf("[%s][ERROR] %s: %v %v", l.prefix, msg, err, fields)
}
func (l *PrefixLogger) Info(msg string, fields watermill.LogFields) { /* ... */ }
func (l *PrefixLogger) Debug(msg string, fields watermill.LogFields) { /* ... */ }
func (l *PrefixLogger) Trace(msg string, fields watermill.LogFields) { /* ... */ }
func (l *PrefixLogger) With(fields watermill.LogFields) watermill.LoggerAdapter {
return l
}
```
The repository ships a logging example that demonstrates both the built-in `StdLogger` and a custom adapter:
```bash
go run ./examples/observability/watermill-logging/main.go
```
## Related [#related]
* [Connection & Configuration](/integrations/watermill/how-to/configuration) — address, auth, TLS, client reuse, and health-check endpoints to expose alongside your `/metrics` endpoint
* [Queues](/integrations/watermill/how-to/queues) — native `QueueMessagePolicy` DLQ and the Poison Queue middleware
* [Concepts](/integrations/watermill/concepts/concepts) — how the Router, middleware, and Ack/Nack map onto KubeMQ patterns