Amazon SQS Message Age and Consumer Processing Lag Sentinel

id: aws-sqs-approximate-age-of-oldest-message-sentinel
namespace: company.messaging
description: Audits Amazon SQS queues for ApproximateAgeOfOldestMessage metrics
  via CloudWatch, detects consumer lag backlogs exceeding SLA thresholds, logs
  an automated GitHub incident ticket, and alerts Slack.
inputs:
  - id: aws_region
    type: STRING
    displayName: AWS Region
    defaults: us-east-1
  - id: queue_name
    type: STRING
    displayName: SQS Queue Name
    defaults: order-processing-queue
  - id: max_age_seconds_threshold
    type: INT
    displayName: Max Age Threshold (Seconds)
    defaults: 1800
  - id: target_repository
    type: STRING
    displayName: Incident Tracking Repository
    defaults: my-org/messaging-platform
triggers:
  - id: periodic_sqs_age_check
    type: io.kestra.plugin.core.trigger.Schedule
    cron: '*/15 * * * *'
    disabled: true
tasks:
  - id: probe_sqs_age_metric
    type: io.kestra.plugin.aws.cli.AwsCLI
    description: Queries CloudWatch for SQS ApproximateAgeOfOldestMessage metric
      over the last 15 minutes
    accessKeyId: '{{ secret(''AWS_ACCESS_KEY_ID'') }}'
    secretKeyId: '{{ secret(''AWS_SECRET_ACCESS_KEY'') }}'
    region: '{{ inputs.aws_region }}'
    commands:
      - aws cloudwatch get-metric-data --metric-data-queries
        '[{"Id":"m1","MetricStat":{"Metric":{"Namespace":"AWS/SQS","MetricName":"ApproximateAgeOfOldestMessage","Dimensions":[{"Name":"QueueName","Value":"{{
        inputs.queue_name
        }}"}]},"Period":900,"Stat":"Maximum"},"ReturnData":true}]' --start-time
        $(date -u -v-15M +%Y-%m-%dT%H:%M:%SZ) --end-time $(date -u
        +%Y-%m-%dT%H:%M:%SZ) --output json | tr -d '\n' | xargs -0 -I {} echo
        '::{"outputs":{}}::'
  - id: evaluate_message_age
    type: io.kestra.plugin.core.debug.Return
    description: Evaluates maximum message age against SLA threshold
    format:
      queue: '{{ inputs.queue_name }}'
      oldest_age_seconds: '{{
        outputs.probe_sqs_age_metric.vars.MetricDataResults[0].Values[0] ?? 0
        }}'
      threshold_seconds: '{{ inputs.max_age_seconds_threshold }}'
      is_sla_breached: '{{
        (outputs.probe_sqs_age_metric.vars.MetricDataResults[0].Values[0] ?? 0)
        >= inputs.max_age_seconds_threshold }}'
  - id: branch_on_age_breach
    type: io.kestra.plugin.core.flow.If
    description: Branches based on whether message age breached SLA threshold
    condition: '{{ outputs.evaluate_message_age.value.is_sla_breached == true }}'
    then:
      - id: create_sqs_incident_issue
        type: io.kestra.plugin.github.issues.Create
        description: Logs a prioritized GitHub issue tracking consumer processing latency
        oauthToken: '{{ secret(''GITHUB_TOKEN'') }}'
        repository: '{{ inputs.target_repository }}'
        title: '[SQS Alert] Message Processing Latency SLA Breached on {{
          inputs.queue_name }}'
        body: "## 🚨 Amazon SQS Message Age SLA Breach


          The messaging sentinel detected messages languishing unprocessed in
          queue:

          - **Queue Name:** `{{ inputs.queue_name }}`

          - **Oldest Message Age:** **{{
          outputs.evaluate_message_age.value.oldest_age_seconds }} seconds**

          - **SLA Threshold:** `{{ inputs.max_age_seconds_threshold }} seconds`


          Please verify consumer worker container health and database write
          contention.

          *Generated automatically by Kestra SQS Sentinel.*"
      - id: alert_messaging_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority alert to messaging Slack channel
        url: '{{ secret(''SLACK_WEBHOOK'') }}'
        payload: "{{ {'text': '🚨 *Amazon SQS Consumer Lag SLA Breach*\\n*Queue:* `' ~
          inputs.queue_name ~ '`\\n*Oldest Message:* *' ~
          outputs.evaluate_message_age.value.oldest_age_seconds ~ 's*
          (Threshold: `' ~ inputs.max_age_seconds_threshold ~ 's`)\\n*Incident
          Ticket:* <' ~ (outputs.create_sqs_incident_issue.issueUrl ?? '#') ~
          '|View GitHub Issue>'} | toJson }}"
    else:
      - id: log_queue_healthy
        type: io.kestra.plugin.core.log.Log
        message: SQS queue {{ inputs.queue_name }} message latency is within SLA ({{
          outputs.evaluate_message_age.value.oldest_age_seconds }}s < {{
          inputs.max_age_seconds_threshold }}s).
  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final audit status log
    message: 'SQS age audit complete for {{ inputs.queue_name }}. Age: {{
      outputs.evaluate_message_age.value.oldest_age_seconds }}s.'
errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts Slack on execution failure
    url: '{{ secret(''SLACK_WEBHOOK'') }}'
    payload: "{{ {'text': '⚠️ *Amazon SQS Sentinel Failed* in execution `' ~
      execution.id ~ '`.'} | toJson }}"