Middleware & 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

• 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)
• 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 below

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.

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:

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:

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:

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

OpenTelemetry Trace Propagation

The plugin propagates trace context automatically using the W3C 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():

// 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:

import (
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/propagation"
)

otel.SetTextMapPropagator(propagation.TraceContext{})

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)
}

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:

go run ./examples/observability/otel-trace-propagation/main.go

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

When tracing is active, the plugin annotates its spans with the following OpenTelemetry messaging semantic-convention attributes:

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:

received := <-msgs

// traceparent format: 00-<trace-id>-<span-id>-<flags>
traceparent := received.Metadata.Get("traceparent")
tracestate := received.Metadata.Get("tracestate")

log.Printf("traceparent=%s tracestate=%s", traceparent, tracestate)
// e.g. traceparent=00-<trace-id>-<span-id>-01

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:

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:

KEDA Autoscaling

KubeMQ has an existing KEDA 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 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:

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

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.

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:

// 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:

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:

go run ./examples/observability/watermill-logging/main.go

Related

• Connection & Configuration — address, auth, TLS, client reuse, and health-check endpoints to expose alongside your /metrics endpoint
• Queues — native QueueMessagePolicy DLQ and the Poison Queue middleware
• Concepts — how the Router, middleware, and Ack/Nack map onto KubeMQ patterns