AMQP Queue Depth Sentinel and Backlog Audit with Kestra

id: amqp-queue-depth-sentinel
namespace: company.team

inputs:
  - id: queue_name
    type: STRING
    defaults: "orders.queue"
    displayName: "Target AMQP Queue Name"
    description: "The RabbitMQ or AMQP queue to probe and audit for unconsumed
      message accumulation."

  - id: max_records_probe
    type: INT
    defaults: 50
    displayName: "Probe Sample Limit"
    description: "Maximum number of unacknowledged or pending messages to sample
      during the audit pass."

  - id: backlog_threshold
    type: INT
    defaults: 10
    displayName: "Backlog Warning Threshold"
    description: "Number of unconsumed messages in sample above which an alert or
      remediation signal is triggered."

  - id: remediation_action
    type: SELECT
    values:
      - "ALERT_ONLY"
      - "PURGE_EXPIRED"
    defaults: "ALERT_ONLY"
    displayName: "Remediation Mode"
    description: "Action to perform when queue backlog exceeds threshold: notify
      Slack only or log quarantine action."

  - id: dry_run
    type: STRING
    defaults: "true"
    displayName: "Dry Run Mode"
    description: "When set to true, safe remediation actions will only log actions
      without mutating queue state."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 */6 * * *"
    disabled: true

tasks:
  - id: probe_queue
    type: io.kestra.plugin.amqp.Consume
    description: "Sample pending messages from the target AMQP queue within a
      10-second window to audit backlog depth."
    host: "{{ secret('RABBITMQ_HOST') }}"
    port: "{{ secret('RABBITMQ_PORT') }}"
    username: "{{ secret('RABBITMQ_USERNAME') }}"
    password: "{{ secret('RABBITMQ_PASSWORD') }}"
    virtualHost: "{{ secret('RABBITMQ_VHOST') }}"
    queue: "{{ inputs.queue_name }}"
    maxRecords: "{{ inputs.max_records_probe }}"
    maxDuration: "PT10S"
    serdeType: "STRING"
    autoAck: false

  - id: evaluate_backlog
    type: io.kestra.plugin.core.flow.Switch
    description: "Branch execution based on whether sampled message count exceeds
      the configured backlog threshold."
    value: "{{ (outputs.probe_queue.count | default(0) > inputs.backlog_threshold) |
      string }}"
    cases:
      "true":
        - id: dispatch_backlog_alert
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Notify SRE and Messaging Ops teams of queue backlog threshold
            breach via Slack incoming webhook."
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "AMQP Queue Backlog Alert: Queue '{{ inputs.queue_name }}' sampled {{ outputs.probe_queue.count }} messages, exceeding backlog threshold of {{ inputs.backlog_threshold }}. Remediation mode: {{ inputs.remediation_action }} (Dry Run: {{ inputs.dry_run }})."
            }

        - id: evaluate_remediation
          type: io.kestra.plugin.core.flow.If
          description: "Check whether automatic quarantine remediation should be logged
            based on remediation mode and dry-run toggle."
          condition: "{{ inputs.remediation_action == 'PURGE_EXPIRED' and inputs.dry_run
            == 'false' }}"
          then:
            - id: log_quarantine_remediation
              type: io.kestra.plugin.core.log.Log
              description: "Record active quarantine remediation action taken for bloated
                queue backlog."
              message: "Quarantine remediation active for queue {{ inputs.queue_name }}.
                Sampled {{ outputs.probe_queue.count }} backlog messages."

      "false":
        - id: log_queue_healthy
          type: io.kestra.plugin.core.log.Log
          description: "Record healthy audit result when queue message count is within
            acceptable operational thresholds."
          message: "AMQP queue {{ inputs.queue_name }} is healthy. Sampled {{
            outputs.probe_queue.count | default(0) }} messages (threshold: {{
            inputs.backlog_threshold }})."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: "Final post-audit summary confirming completion of the AMQP queue
      depth audit cycle."
    message: "AMQP queue depth audit completed for queue '{{ inputs.queue_name }}'.
      Execution ID: {{ execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Flow-level error handler triggering an emergency alert when the
      AMQP queue depth audit encounters an operational failure."
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "CRITICAL: AMQP Queue Depth Sentinel failed during audit of queue '{{ inputs.queue_name }}'. Execution ID: {{ execution.id }}."
      }

Maintain message broker health by auditing AMQP queue depth before consumer lag causes memory pressure or downstream delivery timeouts. This blueprint uses Kestra's AMQP plugin to sample unconsumed messages from a target queue, evaluates backlog metrics against configurable threshold rules, and dispatches automated Slack notifications and safe quarantine logging when queue depth breaches operational SLAs.

How it works

probe_queue (io.kestra.plugin.amqp.Consume) samples pending messages from the designated AMQP queue (inputs.queue_name) up to max_records_probe within a 10-second polling window (PT10S).
evaluate_backlog (io.kestra.plugin.core.flow.Switch) checks whether the consumed message count (outputs.probe_queue.count) exceeds the backlog_threshold value.
If the count exceeds the threshold, dispatch_backlog_alert (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) posts a notification to Slack detailing the queue name, message count, and remediation status.
evaluate_remediation (io.kestra.plugin.core.flow.If) checks whether remediation_action is set to PURGE_EXPIRED and dry_run is set to false, logging active quarantine remediation if enabled.
If the count is within threshold, log_queue_healthy (io.kestra.plugin.core.log.Log) records a healthy audit status.
log_audit_summary (io.kestra.plugin.core.log.Log) confirms completion of the audit pass with full execution lineage.
A flow-level errors block (alert_flow_failure) catches broker connection failures or unhandled task errors, dispatching an incident notification to Slack.

What you get

Automated, non-destructive queue depth sampling across RabbitMQ and AMQP brokers.
Configurable backlog threshold evaluation with dry-run protection.
Instant Slack alerts detailing queue backlog spikes before consumer exhaustion occurs.
Centralized execution lineage and audit history inside Kestra.

Who it's for

Site Reliability Engineers (SREs) and Messaging Platform engineers maintaining RabbitMQ clusters.
Data Engineers running real-time streaming pipelines who need automated consumer lag alerting.
System Administrators looking to eliminate opaque custom queue monitoring scripts.

Why orchestrate this with Kestra

Monitoring RabbitMQ queue depth with custom cron scripts or external exporter plugins leaves monitoring isolated from your primary workflow orchestrator. Kestra provides declarative YAML management, built-in secret management, native Slack notification blocks, and execution logging in a single pane of glass. By running queue health audits directly inside Kestra, SREs gain end-to-end visibility and can chain downstream recovery flows seamlessly.

Prerequisites

A running RabbitMQ or AMQP 0-9-1 compatible message broker.
An active AMQP queue with read permissions.
A configured Slack Incoming Webhook URL.

Quick start

Import this blueprint into your Kestra instance, configure your AMQP broker connection secrets (RABBITMQ_HOST, RABBITMQ_PORT, etc.), and run an initial execution to sample queue backlog metrics.

How to extend

Extend this flow by chaining container auto-scaling triggers or copying dead-letter payloads into object storage (e.g., AWS S3 or Google Cloud Storage) when queue backlog thresholds are breached.

Secrets

RABBITMQ_HOST: Hostname or IP address of the RabbitMQ broker.
RABBITMQ_PORT: Port number of the RabbitMQ broker (e.g. 5672).
RABBITMQ_USERNAME: Authentication username for AMQP broker access.
RABBITMQ_PASSWORD: Authentication password for AMQP broker access.
RABBITMQ_VHOST: Virtual host path (e.g. / or /production).
SLACK_WEBHOOK: Slack Incoming Webhook URL for alerting.

Note on OSS Secrets: In open-source (OSS) Kestra deployments without an external secret manager, set secrets via environment variables formatted as SECRET_RABBITMQ_HOST, SECRET_RABBITMQ_PORT, SECRET_RABBITMQ_USERNAME, SECRET_RABBITMQ_PASSWORD, SECRET_RABBITMQ_VHOST, and SECRET_SLACK_WEBHOOK.

Inputs

queue_name (STRING, default: "orders.queue"): Name of the target AMQP queue.
max_records_probe (INT, default: 50): Maximum messages to sample during probe pass.
backlog_threshold (INT, default: 10): Backlog count threshold for triggering alerts.
remediation_action (SELECT, default: "ALERT_ONLY", values: ["ALERT_ONLY", "PURGE_EXPIRED"]): Remediation strategy when threshold is breached.
dry_run (STRING, default: "true"): Safety flag ensuring remediation logic is executed in dry-run mode unless explicitly disabled.

Outputs

Sampled message count: {{ outputs.probe_queue.count }}
Internal payload URI: {{ outputs.probe_queue.uri }}

Pitfalls

Port Conflicts: Ensure host port 5672 or 15672 does not conflict with existing local database or broker containers.
Virtual Host Scoping: AMQP queues are scoped to specific virtual hosts. Verify RABBITMQ_VHOST matches your queue setup.
Credential Permissions: The configured user must have read and consume permissions on the specified queue.