Skip to main content

Testing Microservices with RabbitMQ Using Signadot Sandboxes

Two versions of a RabbitMQ consumer reading from one queue compete for its messages, so testing a consumer change on a shared cluster usually means a separate broker per developer or per pull request. This tutorial uses one broker instead: a topic exchange copies every message to a queue per consumer version, and a routing key in the message headers decides which version processes it. You deploy a Python publisher and consumer to Kubernetes, fork them into Signadot sandboxes, and watch each test message reach exactly one consumer.

Full source: signadot/examples/rabbitmq-signadot-demo

How it works​

The demo has a Flask publisher that publishes each POST /publish request as an order message to the orders_exchange topic exchange, and a consumer that processes the orders. Isolation rests on three pieces:

  1. The publisher propagates the routing key. A test request carries the sandbox's routing key in its baggage header, and OpenTelemetry copies it into the headers of the message it triggers.
  2. Every consumer version has its own queue. Each queue is bound to the exchange with the binding key #, so every version receives a copy of every message.
  3. Every consumer checks the key. A sandboxed consumer processes only messages carrying its own sandbox's key. The baseline consumer processes everything except messages whose key belongs to a sandbox of the consumer.

This diagram shows one message fanning out to the baseline queue and a sandbox's queue:

Which consumer processes a message depends only on its routing key:

Routing key in the messageBaseline consumerconsumer-v2 sandbox
NoneProcesses itSkips it
The consumer-v2 sandbox's keySkips itProcesses it
Any other key, for example a publisher sandbox'sProcesses itSkips it

Each of these messages is processed by exactly one consumer. Both consumers acknowledge every message they receive, processed or skipped, so skipped copies never pile up or get redelivered.

Two kinds of routing key

RabbitMQ has its own routing key: the string an exchange matches against queue bindings. Here it is always orders, and with the # binding every queue matches it. The Signadot routing key is a different value that identifies a sandbox, and it travels as sd-routing-key in the OpenTelemetry baggage. The exchange uses the first to deliver, the consumers use the second to select.

The publisher: OpenTelemetry carries the key​

The publisher never touches message headers itself. Its container starts under opentelemetry-instrument, whose Flask instrumentation extracts the baggage header from each incoming request and whose pika instrumentation injects that baggage into the AMQP headers of every message published while handling it. In publisher/app.py, the handler only reads the key for its response and its logs:

publisher/app.py
@app.route("/publish", methods=["POST"])
def publish_message():
    """
    OTel Flask instrumentation has already extracted baggage (if any) from the HTTP request.
    OTel pika instrumentation will inject baggage into AMQP automatically.
    """
    body_in = request.get_json(silent=True) or {}

    # Read-only view of the effective routing key for visibility
    rk = get_baggage("sd-routing-key")
    effective_rk = rk or "baseline"

    body = dict(body_in)
    body["routing_key"] = effective_rk

    conn, ch = get_rabbitmq_channel()
    try:
        ch.basic_publish(
            exchange=EXCHANGE_NAME,
            routing_key=AMQP_ROUTING_KEY,
            body=json.dumps(body),
            properties=pika.BasicProperties(
                delivery_mode=2,  # persistent
            ),
        )

The consumer: one queue per version, one check per message​

At startup, consumer/app.py declares a queue, binds it to the exchange with #, and starts consuming. The queue name, durability and auto-delete settings come from environment variables, with defaults that make the baseline's queue durable and a sandbox's queue temporary:

consumer/app.py
queue_name = os.environ.get("QUEUE_NAME", f"consumer-{self.sandbox_name}")
durable = (os.environ.get("QUEUE_DURABLE", "true" if self.is_baseline else "false").lower() == "true")
# ...
auto_delete = (os.environ.get("QUEUE_AUTO_DELETE", "false" if self.is_baseline else "true").lower() == "true")

res = self.channel.queue_declare(queue=queue_name, durable=durable, exclusive=exclusive, auto_delete=auto_delete)
# ...
self.channel.queue_bind(exchange=CUSTOMER_EXCHANGE, queue=self.queue_name, routing_key="#")

The baseline's manifest, k8s/consumer.yaml, sets all three variables explicitly, starting with QUEUE_NAME=consumer-baseline. That matters when you fork it in Step 3.

A consumer knows it is sandboxed because the Signadot Operator injects SIGNADOT_SANDBOX_ROUTING_KEY and SIGNADOT_SANDBOX_NAME into forked pods. The baseline also needs to know which keys belong to sandboxes of the consumer Deployment, so every 10 seconds it asks the Operator's route server, GET /api/v1/workloads/routing-rules on routeserver.signadot.svc:7778, and caches the answer. Then each message goes through one check:

consumer/app.py
def _should_process(self, properties: pika.BasicProperties) -> bool:
    msg_key = self._effective_routing_key()  # get_baggage("sd-routing-key")

    if not self.is_baseline:
        should = (msg_key == self.sandbox_routing_key)
        # ...
        return should

    if not msg_key:
        # ...
        return True

    with self._routes_lock:
        is_active = msg_key in self._active_routes

    if not is_active:
        # ...
        return True

    # ...
    return False

The message callback acknowledges every message in a finally block, whether it was processed or skipped. The RabbitMQ isolation guide describes how platform teams wrap this logic in a client library so services get it without writing it themselves.

Before you begin​

You need:

  • A Kubernetes cluster with the Signadot Operator installed. The consumer queries the Operator's route server.
  • The Signadot CLI, logged in with signadot auth login.
  • kubectl pointed at that cluster, git and jq.
  • Docker and a container registry, only if you want to build your own images. The manifests use prebuilt images.

Step 1: Deploy the baseline​

The baseline is the stable version of every service, shared by everyone. Clone the example and deploy RabbitMQ and Redis into the rabbitmq-demo namespace, then the publisher and consumer:

git clone https://github.com/signadot/examples.git
cd examples/rabbitmq-signadot-demo

kubectl apply -f k8s/namespace.yaml
kubectl apply -n rabbitmq-demo -f k8s/rabbitmq.yaml
kubectl apply -n rabbitmq-demo -f k8s/redis.yaml
kubectl wait --for=condition=ready pod -l app=rabbitmq -n rabbitmq-demo --timeout=120s
kubectl wait --for=condition=ready pod -l app=redis -n rabbitmq-demo --timeout=60s

kubectl apply -n rabbitmq-demo -f k8s/publisher.yaml
kubectl apply -n rabbitmq-demo -f k8s/consumer.yaml
kubectl get pods -n rabbitmq-demo

All four pods should reach Running. The publisher and consumer show 2/2 ready containers because their pod templates carry the sidecar.signadot.com/inject: "true" annotation, which adds the Signadot sidecar that routes requests by routing key.

Build your own images instead

The manifests run ghcr.io/emminex23/rabbitmq-publisher and ghcr.io/emminex23/rabbitmq-consumer. To run your own code, build and push both images, then change the image: fields in k8s/publisher.yaml and k8s/consumer.yaml before you apply them:

docker build -t YOUR_REGISTRY/rabbitmq-publisher:latest ./publisher
docker push YOUR_REGISTRY/rabbitmq-publisher:latest

docker build -t YOUR_REGISTRY/rabbitmq-consumer:latest ./consumer
docker push YOUR_REGISTRY/rabbitmq-consumer:latest

The publisher sandbox in Step 3 runs a :dev tag of the publisher image, so push one of those too and update signadot/sandboxes/publisher.yaml to match.

YOUR_REGISTRY is a placeholder. For images of your own services, use a private repository and give the cluster pull access to it, for example with an image pull secret.

Step 2: Send a baseline message​

Before creating any sandbox, confirm that a message without a routing key flows from the publisher to the baseline consumer. signadot local connect makes cluster service names such as publisher.rabbitmq-demo.svc reachable from your machine:

export CLUSTER_NAME=<your-signadot-cluster-name>
signadot local connect --cluster=$CLUSTER_NAME

Publish an order:

curl -X POST http://publisher.rabbitmq-demo.svc:8080/publish \
  -H "Content-Type: application/json" \
  -d '{"order_id":"001","amount":100,"message":"Baseline order"}'

The response shows that the message carried no routing key:

{"message":{"amount":100,"message":"Baseline order","order_id":"001","routing_key":"baseline"},"routing_key":"baseline","status":"published"}

The baseline consumer processes it:

kubectl logs -l app=consumer -n rabbitmq-demo --tail=50 | grep PROCESSING
Defaulted container "consumer" out of: consumer, sd-sidecar, sd-init-networking (init)
2025-08-22 15:13:28,935 - INFO - [Baseline] --> PROCESSING ORDER 001 amount=100

Step 3: Fork the consumer and the publisher​

A sandbox forks a baseline workload and runs your changed version next to it. Create two: consumer-v2, a fork of the consumer, and publisher-v2, a fork of the publisher that runs the :dev image.

A fork starts as a copy of its baseline Deployment, environment variables included. The baseline consumer's manifest sets QUEUE_NAME=consumer-baseline, so a fork that kept it would read from the baseline's queue and compete with it. Before you apply signadot/sandboxes/consumer.yaml, add a customizations block that gives the fork its own temporary queue:

signadot/sandboxes/consumer.yaml
name: consumer-v2
spec:
  labels:
    demo: rabbitmq
    signadot.com/sandbox: consumer-v2
  cluster: "@{cluster}"
  forks:
  - forkOf:
      kind: Deployment
      namespace: rabbitmq-demo
      name: consumer
    customizations:
      env:
      - name: QUEUE_NAME
        value: consumer-v2
      - name: QUEUE_DURABLE
        value: "false"
      - name: QUEUE_AUTO_DELETE
        value: "true"
  defaultRouteGroup:
    endpoints:
    - name: publisher
      target: http://publisher.rabbitmq-demo.svc:8080

QUEUE_AUTO_DELETE makes RabbitMQ delete the consumer-v2 queue when the sandbox's consumer disconnects, so deleting the sandbox leaves no queue collecting copies of every message. Now create both sandboxes:

signadot sandbox apply -f signadot/sandboxes/consumer.yaml --set cluster=$CLUSTER_NAME
signadot sandbox apply -f signadot/sandboxes/publisher.yaml --set cluster=$CLUSTER_NAME

The baseline consumer refreshes its list of sandbox keys every 10 seconds, so give it a moment before the next step.

Step 4: Send messages with and without a routing key​

Send one message for each row of the table in How it works and check which consumer processes it.

No routing key. Send order 002 exactly as before:

curl -X POST http://publisher.rabbitmq-demo.svc:8080/publish \
  -H "Content-Type: application/json" \
  -d '{"order_id":"002","amount":200,"message":"Baseline order"}'
{"message":{"amount":200,"message":"Baseline order","order_id":"002","routing_key":"baseline"},"routing_key":"baseline","status":"published"}

The consumer sandbox's key. Read consumer-v2's routing key and send order 003 with it in the baggage header, the way a test or the Signadot Chrome extension would:

ROUTING_KEY=$(signadot sandbox get consumer-v2 -o json | jq -r '.routingKey')
echo "Sandbox routing key: $ROUTING_KEY"

curl -X POST http://publisher.rabbitmq-demo.svc:8080/publish \
  -H "Content-Type: application/json" \
  -H "baggage: sd-routing-key=$ROUTING_KEY" \
  -d '{"order_id":"003","amount":300,"message":"Sandbox order"}'
Sandbox routing key: th44pmjmhc7sq
{"message":{"amount":300,"message":"Sandbox order","order_id":"003","routing_key":"th44pmjmhc7sq"},"routing_key":"th44pmjmhc7sq","status":"published"}

The baseline publisher handled this request, since consumer-v2 does not fork the publisher, and the message it published carries the sandbox's key. Watch the baseline consumer:

kubectl logs -f -n rabbitmq-demo -l app=consumer | grep PROCESSING
2025-08-22 15:10:08,492 - INFO - [Baseline] --> PROCESSING ORDER 002 amount=200

The baseline consumer processed order 002 and skipped 003. To read the sandboxed consumer's logs, find its Deployment, which is named after the sandbox and the baseline workload:

kubectl get deployments -n rabbitmq-demo
kubectl logs -f -n rabbitmq-demo deploy/consumer-v2-dep-consumer-<id> | grep PROCESSING
2025-08-22 15:10:21,367 - INFO - [consumer-v2] --> PROCESSING ORDER 003 amount=300

The sandbox processed only order 003, the one carrying its key.

Another sandbox's key. Repeat the request with publisher-v2's key and a new order ID:

PUBLISHER_KEY=$(signadot sandbox get publisher-v2 -o json | jq -r '.routingKey')

curl -X POST http://publisher.rabbitmq-demo.svc:8080/publish \
  -H "Content-Type: application/json" \
  -H "baggage: sd-routing-key=$PUBLISHER_KEY" \
  -d '{"order_id":"004","amount":400,"message":"Publisher sandbox order"}'

This time the request reaches the forked publisher, and the response shows publisher-v2's key. No consumer sandbox claims that key, so the baseline consumer processes order 004 and consumer-v2 skips it. A key that no consumer fork claims falls through to the baseline, which is how a change to the publisher alone gets tested against the stable consumer.

Clean up​

Delete the sandboxes, then the demo. RabbitMQ removes the consumer-v2 queue when the sandbox's consumer goes away.

signadot sandbox delete consumer-v2
signadot sandbox delete publisher-v2
kubectl delete namespace rabbitmq-demo
signadot local disconnect

Going further​

Test the publisher and consumer together​

When a change spans both services, a route group combines several sandboxes under one routing key. The demo includes one, signadot/routegroups/demo.yaml, which matches every sandbox labeled demo: rabbitmq:

signadot routegroup apply -f signadot/routegroups/demo.yaml --set cluster=$CLUSTER_NAME

The route group's key reaches the forked publisher, but the demo's sandboxed consumer accepts only its own sandbox's key, and the baseline consumer skips the route group's key because the group contains a fork of the consumer. To process route group messages, have the sandboxed consumer fetch its keys from the route server as well, by adding destinationSandboxName to the routing-rules query, as the consumer in the Kafka tutorial does. Delete the route group with signadot routegroup delete rabbitmq-demo.

Run it for every pull request​

The same flow runs in CI: build the image for the changed service, create a sandbox that runs it, and send your end-to-end tests through the publisher with the sandbox's routing key in the baggage header. Every pull request gets its own consumer queue on the shared broker, and deleting the sandbox removes it. See Set up PR sandboxes.

Frequently asked questions​

How do you test microservices that use RabbitMQ?​

Run the consumer version under test next to the stable one on the same broker, each with its own queue bound to the same exchange, so both receive every message. A routing key travels from the test request into the message headers, and each consumer processes only the messages meant for it: the sandboxed consumer takes messages carrying its sandbox key, and the baseline consumer takes everything else. Other developers and the baseline keep using the same broker undisturbed.

Do I need a separate RabbitMQ broker for each test environment?​

No. A broker per developer or per pull request gives strong isolation, but you also have to copy every publisher and consumer that uses it, and the copies are slow to create and drift out of date. With message-level isolation, one shared broker serves the baseline and every sandbox, and each test environment adds only a queue for the consumer it changes.

Why does each consumer version need its own queue?​

RabbitMQ delivers each message in a queue to only one of the queue's consumers. If a sandboxed consumer reads from the baseline's queue, the two compete, and each one receives some of the messages meant for the other. Giving every consumer version its own queue, bound to the same exchange, means every version sees every message and can decide for itself whether to process it.

How does the routing key get from an HTTP request into a RabbitMQ message?​

Through OpenTelemetry context propagation. The test request carries the key in its baggage header, the publisher's Flask instrumentation extracts it, and the pika instrumentation writes the baggage into the AMQP headers of every message published while handling that request. On the consuming side, the pika instrumentation restores the baggage before the message callback runs, so the consumer can read the key.

Is Signadot's routing key the same as a RabbitMQ routing key?​

No. A RabbitMQ routing key is the string an exchange matches against queue bindings to decide where a message goes; in this example it is orders. Signadot's routing key identifies a sandbox or route group and travels in the message headers as sd-routing-key in the OpenTelemetry baggage. The exchange uses the first to copy the message to every queue, and the consumers use the second to decide which of them processes it.

Does this approach work with Kafka, Pub/Sub or SQS?​

Yes. The pattern is the same everywhere: propagate the routing key in message metadata, let every consumer version see every message, and have each consumer process only the messages meant for it. What changes is the broker mechanism that gives each version its own copy: a consumer group in Kafka, a subscription in Google Pub/Sub, and a queue per version or released messages in Amazon SQS.

See also​