AMQP 1.0 Dead-Letter Queue and Canary Delivery Sentinel

id: amqp1-dead-letter-quarantine-and-poison-message-sentinel
namespace: company.infrastructure

inputs:
  - id: amqp_host
    type: STRING
    defaults: localhost
    description: Hostname or IP address of the AMQP 1.0 broker (e.g. ActiveMQ
      Artemis, RabbitMQ 4, or Azure Service Bus).

  - id: amqp_port
    type: INT
    defaults: 5672
    description: Listening TCP port for AMQP 1.0 connections.

  - id: dlq_address
    type: STRING
    defaults: "orders.dlq"
    description: Target AMQP 1.0 address or queue holding dead-letter and poison messages.

  - id: canary_address
    type: STRING
    defaults: "sentinel.canary"
    description: Target AMQP 1.0 address used to verify broker write availability
      and link establishment.

  - id: sample_limit
    type: INT
    defaults: 25
    description: Maximum number of dead-letter records to ingest into Kestra
      internal storage per audit cycle.

  - id: dlq_critical_threshold
    type: INT
    defaults: 10
    description: Number of accumulated dead-letter messages that triggers a critical
      incident alert.

triggers:
  - id: periodic_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 */4 * * *"
    disabled: true
    inputs:
      amqp_host: localhost
      amqp_port: 5672
      dlq_address: "orders.dlq"
      canary_address: "sentinel.canary"
      sample_limit: 25
      dlq_critical_threshold: 10

tasks:
  - id: probe_canary_delivery
    type: io.kestra.plugin.amqp.Amqp1Publish
    host: "{{ inputs.amqp_host }}"
    port: "{{ inputs.amqp_port }}"
    username: "{{ secret('AMQP_USERNAME') }}"
    password: "{{ secret('AMQP_PASSWORD') }}"
    address: "{{ inputs.canary_address }}"
    serdeType: JSON
    from:
      - data:
          canary: true
          executionId: "{{ execution.id }}"
          timestamp: "{{ now() }}"
        applicationProperties:
          origin: "kestra-sentinel"
          probeType: "synthetic-heartbeat"
    allowFailure: true
    description: Publishes a synthetic heartbeat message to verify AMQP 1.0 broker
      connectivity, link opening, and message acceptance.

  - id: probe_dlq_backlog
    type: io.kestra.plugin.amqp.Amqp1Consume
    host: "{{ inputs.amqp_host }}"
    port: "{{ inputs.amqp_port }}"
    username: "{{ secret('AMQP_USERNAME') }}"
    password: "{{ secret('AMQP_PASSWORD') }}"
    address: "{{ inputs.dlq_address }}"
    serdeType: STRING
    maxRecords: "{{ inputs.sample_limit }}"
    maxDuration: PT10S
    pollDuration: PT3S
    autoAck: false
    allowFailure: true
    description: Consumes up to sample_limit dead-letter messages from the DLQ
      address into Kestra internal storage for inspection.

  - id: classify_broker_posture
    type: io.kestra.plugin.core.debug.Return
    format: >-
      {{
        (outputs.probe_canary_delivery is not defined or outputs.probe_dlq_backlog is not defined) ? 'BROKER_UNREACHABLE' :
        (outputs.probe_canary_delivery.messagesCount == 0) ? 'CANARY_DELIVERY_FAILED' :
        (outputs.probe_dlq_backlog.count >= inputs.dlq_critical_threshold) ? 'DLQ_BACKLOG_BREACH' :
        (outputs.probe_dlq_backlog.count > 0) ? 'DLQ_WARNING' :
        'HEALTHY'
      }}
    description: Evaluates canary publish confirmation and DLQ backlog count to
      determine overall operational posture.

  - id: route_posture_action
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.classify_broker_posture.value }}"
    cases:
      DLQ_BACKLOG_BREACH:
        - id: alert_dlq_critical_breach
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: "{{ {'text': '🚨 *CRITICAL: AMQP 1.0 DLQ Backlog Breach*

            • *Host:* `' ~ inputs.amqp_host ~ ':' ~ inputs.amqp_port ~ '`

            • *DLQ Address:* `' ~ inputs.dlq_address ~ '`

            • *Sampled Messages:* ' ~ outputs.probe_dlq_backlog.count ~ '
            (Threshold: ' ~ inputs.dlq_critical_threshold ~ ')

            • *Storage URI:* `' ~ (outputs.probe_dlq_backlog.uri ?? 'None') ~ '`

            • *Execution ID:* `' ~ execution.id ~ '`

            *Action Required:* Investigate poisoned payloads and downstream
            consumer exceptions.'} | toJson }}"

      CANARY_DELIVERY_FAILED:
        - id: alert_canary_failure
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: "{{ {'text': '⚠️ *WARNING: AMQP 1.0 Canary Delivery Failed*

            • *Host:* `' ~ inputs.amqp_host ~ ':' ~ inputs.amqp_port ~ '`

            • *Canary Address:* `' ~ inputs.canary_address ~ '`

            • *Status:* Broker rejected or failed to acknowledge synthetic
            message.

            • *Execution ID:* `' ~ execution.id ~ '`'} | toJson }}"

      BROKER_UNREACHABLE:
        - id: alert_broker_unreachable
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: "{{ {'text': '🔥 *ALERT: AMQP 1.0 Broker Unreachable*

            • *Host:* `' ~ inputs.amqp_host ~ ':' ~ inputs.amqp_port ~ '`

            • *Error:* Failed to establish TCP connection or authenticate
            against AMQP 1.0 broker.

            • *Execution ID:* `' ~ execution.id ~ '`'} | toJson }}"

      DLQ_WARNING:
        - id: log_dlq_warning
          type: io.kestra.plugin.core.log.Log
          message: "AMQP 1.0 DLQ address {{ inputs.dlq_address }} holds {{
            outputs.probe_dlq_backlog.count }} dead-letter records. Below
            critical threshold ({{ inputs.dlq_critical_threshold }})."

      HEALTHY:
        - id: log_broker_healthy
          type: io.kestra.plugin.core.log.Log
          message: "AMQP 1.0 broker {{ inputs.amqp_host }}:{{ inputs.amqp_port }} is
            healthy. Canary delivery confirmed and DLQ address {{
            inputs.dlq_address }} is empty."
    description: Routes operational alerts or audit logs based on the classified posture.

  - id: audit_sentinel_run
    type: io.kestra.plugin.core.log.Log
    message: "AMQP 1.0 sentinel execution {{ execution.id }} complete. Verdict: {{
      outputs.classify_broker_posture.value }}, DLQ count: {{
      outputs.probe_dlq_backlog.count ?? 0 }}, Canary published: {{
      outputs.probe_canary_delivery.messagesCount ?? 0 }}."
    description: Records an immutable audit log entry capturing cycle metrics and
      the final operational verdict.

errors:
  - id: alert_execution_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: "{{ {'text': '❌ *AMQP 1.0 Sentinel Execution Error*

      Flow failed during execution `' ~ execution.id ~ '` against broker `' ~
      inputs.amqp_host ~ ':' ~ inputs.amqp_port ~ '`. Check Kestra run logs for
      stack trace.'} | toJson }}"