RabbitMQ

Synopsis

Creates a consumer that connects to one or more RabbitMQ brokers, declares or checks the exchange and the queue it is given, binds them, and consumes the queue. Supports multiple authentication methods, exchange types, TLS encryption, and multiple workers. A message is acknowledged only once its delivery to the target is confirmed (see Acknowledgement and Duplicates).

The same device runs under the amqp type for brokers that speak AMQP 0-9-1 but are not RabbitMQ. See AMQP.

For details, see Appendix.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: rabbitmq
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    protocol: <string>
    address: <string | string[]>
    port: <numeric>
    vhost: <string>
    username: <string>
    password: <string>
    authentication: <string>
    exchange:
      name: <string>
      type: <string>
      durable: <boolean>
      passive: <boolean>
      arguments: <object>
    queue:
      name: <string>
      key: <string | string[]>
      durable: <boolean>
      passive: <boolean>
      arguments: <object>
      consume_arguments: <object>
    selector: <string>
    prefetch_count: <numeric>
    reuse: <boolean>
    workers: <numeric>
    max_buffer_size: <numeric | string>
    content_encoding: <string>
    max_decompression_size: <numeric | string>
    connect_timeout: <numeric>
    heartbeat: <numeric>
    max_reconnect_interval: <numeric>
    tls:
      status: <boolean>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
      ca_name: <string>
      server_name: <string>
      min_tls_version: <string>
      max_tls_version: <string>
      insecure_skip_verify: <boolean>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
id	Y	-	Unique numeric identifier
name	Y	-	Device name
description	N	-	Optional description
type	Y	-	rabbitmq, or amqp for the same device under the protocol's name
tags	N	-	Optional tags
pipelines	N	-	Optional pre-processor pipelines
status	N	true	Enable/disable the device

Connection

Field	Required	Default	Description
protocol	N	0-9-1	The AMQP the device speaks: 0-9-1 or 1.0. Left out or empty, the device speaks AMQP 0-9-1, as every device configured before did. An unquoted YAML 1.0, which YAML reads as the number 1, is read as 1.0. Any other value ("1", "0.9.1", 2) is a Configuration error. See Protocol
address	N	localhost	The broker, or several brokers of one cluster: a list, or text with commas between them, 32 at most. A host may carry its own port (host:port, [v6]:port). A wildcard (0.0.0.0, ::), which the console wrote for devices created before it left a blank address out, is read as localhost, and the start logs a warning
port	N	5672, or 5671 with TLS	Port of every broker that has none of its own. 0 means the default
vhost	N	/	The virtual host
username	N	-	Authentication username. Left out under 0-9-1, the device logs in as the broker's default user guest, which RabbitMQ admits from localhost only. Under 1.0 see authentication
password	N	-	Authentication password
authentication	N	plain	Auth type: plain, amqplain or external under 0-9-1; plain, anonymous or external under 1.0. anonymous is refused under 0-9-1, and amqplain under 1.0. With external the client certificate authenticates the connection, and the broker maps its identity to a user that must exist: it needs TLS and a client certificate, either tls.cert_name / tls.key_name or the Director's cert.pem and key.pem. anonymous logs in with SASL ANONYMOUS, and a username beside it is not sent. Under 1.0, plain with no username also logs in anonymous, since AMQP 1.0 has no default user. The start logs a warning in both cases
connect_timeout	N	30	Seconds, 1–300, that a connect waits for the TCP, TLS and AMQP handshakes
heartbeat	N	10	Seconds between heartbeats, 0–3600. 0 takes the broker's (RabbitMQ: 60). Three silent intervals mean the connection is lost
max_reconnect_interval	N	30	Seconds, 1–3600. Reconnects start 1 s after a loss and the wait doubles up to this value

warning

A device with no username used to log in with an empty user, which every broker refuses. It now logs in as guest, as amqp091 does. Set username explicitly for any broker that is not on the Director's host.

Exchange

Field	Required	Default	Description
exchange.name	N	-	Exchange name, used exactly as written, spaces included. Left out, no exchange is declared and nothing is bound: the queue is consumed with the bindings it already has
exchange.type	N	topic	Exchange type (direct, fanout, topic, headers, or a plugin's type such as x-consistent-hash). header is accepted for headers
exchange.durable	N	true	The exchange survives a broker restart
exchange.passive	N	false	Check that the exchange exists instead of declaring it. A missing exchange is a Configuration error (NOT_FOUND). Lets a user without configure permission use an existing exchange
exchange.arguments	N	-	Declare arguments, as an object (for x-consistent-hash, hash-header or hash-property). Whole numbers are sent as integers

Supported exchange types:

Exchange Type	Description
direct	Exact routing key match
fanout	Broadcast to all bound queues
topic	Pattern-based routing using wildcards (*, #)
headers	Routing on message header values, ignoring the routing key
x-…	A plugin's exchange type registered on the broker, such as x-consistent-hash

note

x-custom is not an exchange type. A type the broker does not have — a plugin's type on a broker without that plugin — is refused, and the device shows Configuration error.

Queue

Field	Required	Default	Description
queue.name	Y	-	Queue name, used exactly as written, spaces included: a trimmed name is another queue
queue.key	N	-	One binding key, or a list of up to 64; the queue is bound to the exchange with each. A numeric key (key: 42) is bound as its digits. No key means no binding, and the start logs a warning; bind a fanout or headers exchange with key: ""
queue.durable	N	true	The queue survives a broker restart
queue.passive	N	false	Check that the queue exists instead of declaring it: it is consumed as it is, with its own arguments and bindings. A missing queue is a Configuration error (NOT_FOUND). Lets a user without configure permission consume an existing queue
queue.arguments	N	-	Declare arguments, as an object, such as x-queue-type: quorum, x-max-length, x-message-ttl, x-dead-letter-exchange, x-delivery-limit. Whole numbers are sent as integers, which is what the broker checks these arguments for
queue.consume_arguments	N	-	Consume arguments, as an object: x-priority (the consumer's priority), or x-stream-offset (where a stream is read from)
selector	N	-	protocol: "1.0" only. A filter the broker applies, such as region = 'eu', sent as the source filter apache.org:selector-filter:string. Artemis, ActiveMQ and Solace apply it; RabbitMQ queues have no JMS selector. Left out or empty, no filter is sent. Set under 0-9-1, it is a Configuration error

Under protocol: "1.0", queue.name is the source address the device consumes, and the other queue.* keys are refused, as every exchange.* key is. See Protocol.

TLS

Field	Required	Default	Description
tls.status	N	false	Enable TLS encryption
tls.cert_name	N	-	Client certificate for mutual TLS. Naming only one of cert_name and key_name takes the other's default file (cert.pem or key.pem). With neither, the Director's cert.pem and key.pem are presented when both files exist, and no client certificate is presented otherwise
tls.key_name	N	-	Client private key for mutual TLS. See tls.cert_name
tls.passphrase	N	-	Passphrase for an encrypted private key
tls.ca_name	N	-	CA bundle used to verify the broker certificate. When unset, the host trust store is used.
tls.server_name	N	-	SNI hostname override for the TLS handshake
tls.min_tls_version	N	tls1.2	Minimum accepted TLS version (tls1.0, tls1.1, tls1.2, tls1.3; "1.2" and "12" as text are also accepted)
tls.max_tls_version	N	-	Maximum accepted TLS version. When unset, the highest mutually supported version is negotiated.
tls.insecure_skip_verify	N	true	Skip broker certificate verification

Any other spelling of a TLS version — TLSv1.2, or an unquoted YAML 1.2, which is a number — is read as the default, and the start logs a warning. TLS without a client certificate used to be refused; it now connects to a broker that asks for none.

tls.status, tls.insecure_skip_verify and reuse are switches: a number is read as a switch (0 is false), and text that is not true or false is read as false, with a warning at start.

note

TLS material fields (cert_name, key_name, ca_name, client_ca_name) accept any of the following:

• File name — resolved relative to the service root directory. Nested paths such as certs/prod/server.pem are supported.
• Absolute path — honored only if it resolves inside the service root. Any path that escapes the root is refused.
• Inline PEM content — used verbatim when the value contains -----BEGIN.
• Environment variable — ${ENV_VAR}.
• Vault reference — $secret{id=...} or $secret{store=...,ref=...}.

warning

This device defaults tls.insecure_skip_verify to true, unlike every other message queue device. With tls.status: true and nothing further set, the connection is encrypted but the broker certificate is not verified. Set tls.insecure_skip_verify: false to verify it, and supply tls.ca_name when the broker presents a private CA.

Performance

Field	Required	Default	Description
reuse	N	true	Cap workers at the Director's threads
workers	N	4	Connections that consume the queue, 1–64. 0 or a value that is not a number is the default; below 0 is 1 and above 64 is 64, each with a warning at start. With reuse: true also capped at the Director's threads
max_buffer_size	N	128MB	Ingest buffer admission cap. Also sizes this listener's ingest buffers, once per worker: 3/8 of this value at start, growing to at most 3x
prefetch_count	N	10000	Messages each worker holds that it has not acknowledged, 1–65535. 0 or less, or a value that is not a number, is read as 10000; above 65535 is read as 65535, each with a warning at start. It also bounds throughput (see Acknowledgement and Duplicates)
content_encoding	N	identity	How a message body is read: identity (as it came), gzip, or auto (follows the message's own content encoding)
max_decompression_size	N	64MB	The most a body is decompressed to, 1KB–1GB

note

max_buffer_size is the ingest-buffer admission cap, and is distinct from any buffer_size field above, which sizes the network read buffer. Left unset it is 128MB, whose buffers take 48 MiB of heap per listener worker at start and grow to at most 384 MiB as bursts need the room — and with reuse: true a device runs one worker per listener. Size strings are binary, and 32MB and 32MiB are exact synonyms.

See Listener Memory Sizing for the arithmetic, the sizing table, and what happens when the cap is exhausted.

warning

Each worker has its own ingest stream, so in memory mode this device holds its buffers per worker: from 4 × 48 MiB = 192 MiB at start, up to 4 × 384 MiB = 1.5 GiB once its workers have burst, at the defaults of 4 workers and 128MB. Lower workers or max_buffer_size to fit more devices on one Director.

Operation

Protocol

AMQP 1.0 is a different wire protocol, not a newer version of AMQP 0-9-1. With protocol: "1.0" the device consumes brokers that speak only AMQP 1.0, such as ActiveMQ Artemis and Classic, Amazon MQ for ActiveMQ, and Solace. RabbitMQ 4 speaks both. A device with no protocol speaks AMQP 0-9-1 and is unchanged. For an Amazon MQ broker of either engine, the Amazon MQ device is the dedicated type, with IAM and OAuth 2.0 sign-in.

protocol	Result
left out, ""	0-9-1: unchanged behavior for every existing device
"0-9-1"	AMQP 0-9-1
"1.0"	AMQP 1.0
unquoted YAML 1.0 (read by YAML as the number 1)	read as 1.0
anything else ("1", "0.9.1", 2, …)	Configuration error at start

Under AMQP 1.0 there is no exchange to declare, no queue to declare and no binding to make: a worker attaches a receiving link to an address the broker already has. Each key means:

Key	0-9-1	1.0
queue.name	The queue declared and consumed	The source address the link attaches to. Required
prefetch_count	The prefetch window	The link credit: the same bounds (1–65535) and default (10000)
selector	Refused	Sent as the source filter apache.org:selector-filter:string
vhost	The virtual host	Other than /, the open frame's host name is vhost:<vhost>, which is how RabbitMQ 4 selects a virtual host. At /, it is the broker's host
authentication	plain, amqplain, external	plain, anonymous, external. amqplain is refused
exchange.*, queue.key, queue.durable, queue.passive, queue.arguments, queue.consume_arguments	As described above	Refused, and the error names the key. A key written empty, such as exchange.name: "", does not count as set
address, port, tls.*, username, password, workers, reuse, content_encoding, max_decompression_size, connect_timeout, max_reconnect_interval	As described above	The same meaning
heartbeat	Heartbeats both ways	The broker is asked to send a frame every heartbeat seconds, and a connection that brings nothing for twice that is lost. 0 asks for nothing

Settlement. Delivery is at-least-once, as under AMQP 0-9-1. Once its delivery is confirmed, a message is settled as Accepted. A message that can never be written (see Rejected Messages) is settled as Rejected, and the broker dead-letters or drops it, as its address is configured. A message not yet settled when its link closes, on a stop or a lost connection, is sent again by the broker.

Records. A message is one record. Its body is the message's data sections joined, or its value as text or binary; a value of any other type is written as JSON. _vmetric.event.request is the address, or <address>/<subject> when the message has a subject. The time is the message's creation-time, or the time it was received when it has none. The message's other properties and its application properties are not written.

States.

Broker condition	State
amqp:unauthorized-access, or a SASL failure	Authentication error
amqp:not-found for an address	Configuration error
amqp:not-found for a virtual host	Authentication error
A detached link, a closed connection, or amqp:resource-limit-exceeded	Connection error, reopened with the usual backoff

The first three are permanent, as their AMQP 0-9-1 counterparts are.

Brokers. Artemis creates an address on its first attach by default; where its address settings turn auto-creation off, create the address first. RabbitMQ 4 does not create one: declare the queue first, and use the address /queues/<queue>.

note

Azure Service Bus also speaks AMQP 1.0, but it is not reached through the amqp or rabbitmq devices.

Acknowledgement and Duplicates

Delivery is at-least-once. A message is acknowledged after the sender confirms its delivery to the target; until then a stop, a crash, a lost connection or a target that is down leaves it with the broker, which redelivers it. A message can therefore be written twice.

prefetch_count bounds each worker's rate to roughly the window divided by the delivery delay: at the default of 10,000 and a 5-second delay, about 2,000 messages per second per worker. Raise the window for a busy queue, or add workers.

Quorum queues count each return toward x-delivery-limit (20 by default in RabbitMQ 4), so a target that stays down across many restarts of the device can dead-letter messages.

With crash_resistance, broker redelivery can write a message twice after a crash. Memory mode is the recommended mode for this device: the broker is its durable buffer.

Rejected Messages

A message that can never be written is rejected without requeue:

• a body that is not the gzip it claims to be
• a body that decompresses past max_decompression_size
• a message with an unknown content encoding under content_encoding: auto
• a message larger than the ingest buffer

A rejected message goes to the queue's dead-letter exchange (x-dead-letter-exchange) if one is set, and is dropped otherwise.

A write that the ingest stream itself fails — its rotation ended, or its store failed — is not rejected: the message stays with the broker, and the device stops with Initialization error, to be started again.

Reconnects and States

A lost connection is reopened by its worker without restarting the device. The state is Connection error until every worker consumes again. A durable queue whose node is restarting (NOT_FOUND - home node … is down or inaccessible) is retried like an unreachable broker, not failed as a configuration.

A refusal that retrying cannot fix — the user, the permissions, the exchange or queue definition, or a certificate that fails on every broker — stops the device with the state that names it:

State	When
Connected	Every worker consumes, on the first connect and after a loss
Connection error	A broker cannot be reached or does not answer within connect_timeout, or a lost connection is being reopened
Certificate error	A broker presents a certificate the device does not trust, or that names another host, and no other broker took the connection
Authentication error	The broker refused the user, the password, the mechanism, the virtual host, or the user's permissions on the exchange or the queue (ACCESS_REFUSED), or the credentials could not be read
Configuration error	The configuration cannot run, or the broker refused the exchange or the queue as configured (PRECONDITION_FAILED, NOT_FOUND on a passive check, RESOURCE_LOCKED, an exchange type it does not have)
Initialization error	An ingest stream could not be opened, or took nothing more

Several Brokers

Several brokers in address are treated as nodes of one cluster. Each connect tries them in turn, starting from a random one, so the workers spread over them. Unreachable brokers, and brokers whose certificate fails, are skipped; a refused user or queue ends the attempt, since every node of a cluster would refuse the same.

Stopping

A stop runs in order: the broker is told to stop delivering, the queue flushes, acknowledgements land for up to 10 seconds, and then the connections close. The last flush is waited for however long it takes, and the 10 seconds start after it. A close waits 5 seconds at most for the broker, then closes the socket. A stop also ends a connect stuck waiting on a broker that does not answer.

Clusters of Directors

Every Director of a cluster runs the device's workers, and the broker shares the queue's messages between them, each message going to one consumer.

warning

A stream queue is not shared. A queue declared with x-queue-type: stream is read whole by every worker on every Director, from x-stream-offset on each connect. Read a stream with one worker on a Director that runs alone. The device logs a warning when it sees one.

Message Order

Order holds per worker only. For strict order, use one worker on one Director, or declare a classic or quorum queue with x-single-active-consumer: true. A stream queue does not honor single active consumer over AMQP 0-9-1, so a stream keeps to one worker on one Director.

Changes From Earlier Versions

These defaults and behaviors changed, and matter on upgrade:

• address defaults to localhost, where it was 0.0.0.0.
• port defaults to 5672, or 5671 with TLS, where it was required.
• A device with no username logs in as guest, where it logged in with no user and was refused.
• A message that cannot be written is rejected, where it was requeued and redelivered forever, blocking the queue.
• A stop acknowledges what was delivered before it closes the connections, where it closed them at once and the broker redelivered everything unacknowledged.
• prefetch_count above 65535 is read as 65535, where the device sent its low 16 bits (65536 meant 0, no bound at all).

Examples

Basic

Creating a simple RabbitMQ consumer with a direct exchange...

- id: 1 name: basic_rabbitmq type: rabbitmq properties: address: "rabbitmq.example.com" port: 5672 authentication: "plain" exchange: name: "logs" type: "direct" queue: name: "app_logs" key: "app.*"

Secure

Connecting with authentication and TLS encryption...

- id: 2 name: secure_rabbitmq type: rabbitmq properties: address: "rabbitmq.example.com" port: 5671 username: "consumer" password: "secret" authentication: "plain" exchange: name: "secure_logs" type: "topic" queue: name: "secure_app_logs" key: "secure.app.#" tls: status: true cert_name: "rabbitmq.crt" key_name: "rabbitmq.key"

High-Volume

Optimizing for high throughput with multi-worker mode...

- id: 3 name: performant_rabbitmq type: rabbitmq properties: address: "rabbitmq.example.com" port: 5672 authentication: "plain" exchange: name: "high_volume" type: "direct" queue: name: "high_volume_logs" key: "logs" reuse: true workers: 4

Topic Exchange

Pattern-based routing on a topic exchange...

- id: 4 name: topic_rabbitmq type: rabbitmq properties: address: "rabbitmq.example.com" port: 5672 authentication: "plain" exchange: name: "logs" type: "topic" queue: name: "filtered_logs" key: "app.*.error"

tip

Topic routing keys support * and # wildcards for single and multiple words respectively.

Pipelines

Applying custom processing to messages...

- id: 5 name: pipeline_rabbitmq type: rabbitmq pipelines: - json_parser - field_extractor properties: address: "rabbitmq.example.com" port: 5672 authentication: "plain" exchange: name: "raw_logs" type: "direct" queue: name: "processed_logs" key: "logs"

note

Pipelines are processed sequentially, and can modify or drop messages before ingestion.

AMQP 1.0

Consuming the orders address of an ActiveMQ Artemis broker over AMQP 1.0, keeping only the messages the selector matches...

- id: 6 name: artemis_orders type: amqp properties: protocol: "1.0" address: "artemis.example.com" username: "collector" password: "${ARTEMIS_PASSWORD}" queue: name: "orders" selector: "region = 'eu'"