Audit RabbitMQ AMQP Message TTL Expiration and Dead-Letter Backlog

In RabbitMQ and AMQP architectures, messages published with per-message Time-To-Live (TTL) or queues configured with message expiration are intended to be processed or dropped within a deterministic timeframe. However, RabbitMQ only evaluates per-message TTL when a message reaches the head of the queue. If an unexpired or stuck message blocks the head, newer expired messages remain accumulated behind it, silently consuming broker memory. Furthermore, when messages finally expire and route to a Dead-Letter Queue (DLQ), consumer drop-offs can cause unmonitored DLQ backlog spikes.

This blueprint provides an automated Day 2 Operations sentinel flow that samples DLQ backlogs and message expiration volumes on a scheduled cadence. It classifies queue health against configurable warning thresholds, alerts on-call engineers via Slack upon backlog accumulation, and logs actionable diagnostic remediation telemetry.

How it works

1. schedule (io.kestra.plugin.core.trigger.Schedule) initiates the audit cycle every 6 hours (disabled by default for safe initial setup).
2. probe_dlq_backlog (io.kestra.plugin.amqp.Consume) pulls up to probe_sample_limit messages from dlq_queue_name without acknowledging them (autoAck: false), ensuring messages remain intact on the broker while measuring backlog volume.
3. evaluate_threshold (io.kestra.plugin.core.flow.Switch) evaluates whether outputs.probe_dlq_backlog.count exceeds inputs.expired_threshold.
4. When within normal limits (HEALTHY), log_healthy_status records the successful pass.
5. When the limit is exceeded (BREACH), alert_threshold_breach dispatches an operational alert to Slack using io.kestra.plugin.slack.notifications.SlackIncomingWebhook, and log_remediation_action logs the chosen remediation mode.
6. log_audit_completion writes a final audit recap for execution tracking and observability.
7. The flow-level errors block intercepts broker connectivity failures or authentication errors and notifies Slack.

What you get

• Early detection of expired message accumulation before broker memory exhaustion.
• Identification of head-of-line queue blocking and lagging consumer groups.
• Zero message disruption during probe passes using non-acknowledging inspection.
• Automated Slack incident notifications with execution context and queue metadata.
• Flow-level error alerting safeguarding audit pipeline reliability.

Who it's for

• Platform Engineers and Site Reliability Engineers managing RabbitMQ or AMQP message brokers.
• Event-Driven Architecture (EDA) teams building asynchronous order or event streaming systems.
• DevOps engineers tasked with broker observability and DLQ backlog hygiene.

Why orchestrate this with Kestra

Monitoring RabbitMQ TTL and DLQ accumulation with standalone cron scripts requires managing custom credentials, error handling, retry logic, and webhook clients in isolated codebases. Kestra provides a declarative workflow engine with native AMQP consumption, centralized secret management, conditional branching, execution history, and integrated Slack alerting without writing or maintaining custom polling daemons.

Prerequisites

• A running RabbitMQ broker (3.8+ or compatible AMQP 0-9-1 broker).
• An AMQP queue configured with a Dead-Letter Exchange (x-dead-letter-exchange) and DLQ.
• A Slack Incoming Webhook URL configured in Kestra secrets.

Local Docker Container Setup

To run RabbitMQ locally for development and testing while avoiding host port 8080 conflicts:

docker run -d --name rabbitmq-dev -p 5672:5672 -p 15672:15672 -e RABBITMQ_DEFAULT_USER=guest -e RABBITMQ_DEFAULT_PASS=guest rabbitmq:3-management

If Kestra is running inside a Docker network, connect the RabbitMQ container to Kestra network:

docker network ls
docker network connect YOUR_KESTRA_NETWORK rabbitmq-dev

Secrets

• RABBITMQ_HOST: Hostname or container name of the RabbitMQ broker (e.g. rabbitmq-dev or localhost).
• RABBITMQ_PORT: AMQP port (e.g. 5672).
• RABBITMQ_USERNAME: RabbitMQ administrative or monitoring user.
• RABBITMQ_PASSWORD: Password for the RabbitMQ user.
• RABBITMQ_VHOST: Virtual host name (default /).
• SLACK_WEBHOOK: Incoming Webhook URL for Slack alerting.

Note for Kestra Open-Source (OSS): Secrets are provided as base64-encoded environment variables formatted as SECRET_<NAME> on the Kestra server or worker environment.

Inputs

Name	Type	Default	Allowed Values	Description
queue_name	STRING	"orders.queue"	Any valid queue name	Primary AMQP queue configured with TTL
dlq_queue_name	STRING	"orders.dlq"	Any valid queue name	Dead-letter queue capturing expired messages
probe_sample_limit	INT	50	Positive integer	Maximum messages to sample during audit pass
expired_threshold	INT	5	Positive integer	Threshold of expired messages triggering alert
remediation_mode	SELECT	"ALERT_ONLY"	"ALERT_ONLY", "QUARANTINE_LOG"	Operational remediation action mode
dry_run	SELECT	"true"	"true", "false"	Dry run simulation flag

Outputs

• {{ outputs.probe_dlq_backlog.count }}: Total number of messages sampled from the DLQ.
• {{ outputs.probe_dlq_backlog.uri }}: Internal storage URI containing the sampled message payload batch.

Quick start

1. Configure RABBITMQ_HOST, RABBITMQ_PORT, RABBITMQ_USERNAME, RABBITMQ_PASSWORD, RABBITMQ_VHOST, and SLACK_WEBHOOK in your Kestra namespace.
2. Import this blueprint into your Kestra instance.
3. Run a manual execution with dry_run: "true" to verify connectivity and DLQ sampling.
4. Enable the schedule trigger to run automated recurring audits.

How to extend

• Add automated queue dead-letter routing reconfiguration using io.kestra.plugin.amqp.CreateQueue or QueueBind.
• Integrate with PagerDuty or Opsgenie using additional notification tasks in the BREACH branch.
• Archive expired message payloads to object storage (Amazon S3 or MinIO) for regulatory compliance before queue purge.

Pitfalls

• Head-of-Line Blocking: Per-message TTL does not drop expired messages if an unexpired message precedes them at the queue head. Always monitor both the primary queue and the DLQ.
• Port Conflicts: Ensure RabbitMQ management port (15672) and AMQP port (5672) do not conflict with other local services.
• Credential Scope: Ensure the Kestra AMQP user has read permissions on both the target queue and its dead-letter queue.

Links

• Kestra AMQP Plugin Documentation
• RabbitMQ Time-To-Live (TTL) Guide
• RabbitMQ Dead Letter Exchanges
• Kestra Slack Notification Plugin