cki_tools.amqp_bridge

Single service to act as a bridge between external AMQP 0.91, AMQP 1.0, SQS queues and Kafka topics and a single RabbitMQ exchange.

                          RabbitMQ
                        +----------------------------------------+
+--------------+        |  +----------+          +------------+  |
| AMQP091      +-----+---->+ Exchange +----+---->+ Consumer_1 |  |
+--------------+     |  |  +----------+    |     +------------+  |
+--------------+     |  |                  |     +------------+  |
| AMQP10       +-----+  |                  +---->+ Consumer_2 |  |
+--------------+     |  |                        +------------+  |
+--------------+     |  |                        :            :  |
| SQS          +-----+  |                        +------------+  |
+--------------+     |  |                        | Consumer_n |  |
+--------------+     |  +----------------------------------------+
| Kafka(rx,tx) +<----+
+--------------+

A different process needs to be spawned for each server to mirror. Each one will subscribe to a given list of topics and forward them to the configured RabbitMQ exchange.

In case of Kafka, the bridge can be bidirectional, with one process for each direction. Kafka TX consumes from RabbitMQ and produces to Kafka, while Kafka RX consumes from Kafka and produces to RabbitMQ.

Usage

usage: python3 -m cki_tools.amqp_bridge [-h]
           [--from-datagrepper] [--start START] [--end END]

options:
  -h, --help          show this help message and exit

Retrieve messages from datagrepper:
  --from-datagrepper  Retrieve messages from datagrepper instead of listening to AMQP
  --start START       Start time for datagrepper messages in ISO format
  --end END           End time for datagrepper messages in ISO format

When called without any arguments, the service will start listening to the configured AMQP servers and forward messages to the RabbitMQ exchange.

When called with --from-datagrepper, the service will retrieve messages from datagrepper instead of listening to AMQP. The --start and --end options specify the time range for the retrieved messages.

In case of Kafka, the service can bridge messages from RabbitMQ to Kafka but also from Kafka to RabbitMQ. This is controlled by the protocol configuration parameter that, in case of Kafka, can be kafka_rx or kafka_tx.

Configuration

CKI_DEPLOYMENT_ENVIRONMENT

On staging developments (CKI_DEPLOYMENT_ENVIRONMENT != production), the behaviour changes the following way:

• On AMQP 0.91, queue_name is not used and a volatile queue with an uuid name is created instead.
• On AMQP 1.0, queues (queue://) in receiver_urls are translated into topics (topic://).
• Messages are not forwarded to the destination but printed instead.

AMQP_BRIDGE_CONFIG

AMQP 1.0 Configuration

name: foobar
protocol: amqp10
message_topics:
  - queue://Consumer.consumer-name.foo.VirtualTopic.foo.>
  - queue://Consumer.consumer-name.foo.VirtualTopic.bar.>
receiver_urls:
  - amqps://messaging-broker01.foobar.com:5671
  - amqps://messaging-broker02.foobar.com:5671
cert_path: /path/to/cert.pem

AMQP 0.91 Configuration

name: foobar
protocol: amqp091
routing_keys:
  - some.routing.key
  - other.key.#
host: host.org
port: 5671
cafile: /path/to/ca.pem
certfile: /path/to/cert.pem
virtual_host: /virtual_host_name
exchange: exchange.name
queue_name: queue.name

SQS Configuration

The SQS receiver operates in mirror mode, passing through original message attributes without adding bridge headers.

On message processing failures, messages will be re-queued after the visibility_timeout period. The default timeout of 10 minutes matches the CKI RabbitMQ delay queues.

name: foobar
protocol: sqs
queue_url: https://sqs.us-east-1.amazonaws.com/123456789012/my-queue
visibility_timeout: 600
wait_time_seconds: 20

Kafka Configuration

Routing keys

Traffic is split into two RabbitMQ namespaces so that consumers who want to receive messages from Kafka can subscribe to kafka.rx.<topic>, while producers who want to send messages to Kafka can use kafka.tx.<topic>.

RX Configuration (Kafka -> RabbitMQ)

Receives messages from Kafka topics and forwards them to the RabbitMQ exchange configured by RABBITMQ_PUBLISH_EXCHANGE. Outgoing routing keys use the kafka.rx. namespace (see Routing keys).

name: vezdur
protocol: kafka_rx
consumer_config:
  group.id: amqp-bridge-vezdur
  auto.offset.reset: earliest
  enable.auto.commit: false

TX Configuration (RabbitMQ -> Kafka)

Binds a queue to WEBHOOK_RECEIVER_EXCHANGE using ROUTING_KEYS, consumes messages, and produces to Kafka. Only routing keys matching kafka.tx.<topic> are accepted; the Kafka topic name is everything after that prefix (for example kafka.tx.cki.results → topic cki.results). See Routing keys.

RabbitMQ wildcard patterns (#, *) in ROUTING_KEYS only affect which messages reach the queue; the concrete routing key on each message must still follow kafka.tx.* for the bridge to derive the Kafka topic.

name: vezdur
protocol: kafka_tx
producer_config:
  message.send.max.retries: 3

Resulting messages

The messages after being forwarded will have the same body as the original ones, but the headers and routing key will have the following format.

Routing key

For AMQP 0.91 and AMQP 1.0, messages are published with the bridge name prepended: {name}.{routing_key}. For example, bridge foo and original topic tests.finished become foo.tests.finished.

SQS forwards using the message-topic attribute as the routing key (no bridge name prefix).

Kafka RX (kafka_rx) publishes with routing key kafka.rx.<kafka_topic> (no bridge name prefix). <kafka_topic> is the Kafka topic the record came from.

Headers

AMQP 1.0 Headers

AMQP 0.91 Headers

Kafka Headers (Kafka -> RabbitMQ)

Kafka message headers are passed through filtered_headers (CKI / x- prefixes dropped like other bridges). These headers are always set:

Kafka Headers (RabbitMQ -> Kafka)

Messages bridged from Kafka TX (RabbitMQ -> Kafka) do not add bridge headers; the original message body is forwarded as-is to the Kafka topic.