Audit SNS Fanout Failures Before Subscribers Drift

id: sns-fanout-failure-audit
namespace: company.team
description: |
  Audit SNS topic fanout health the moment delivery metrics land, so
  subscribers that silently fail or get filtered out are caught before they
  drift out of sync. A monitoring pipeline POSTs per subscriber delivery
  counters to a keyed webhook, Python computes the failure rate for every
  subscriber against a configurable threshold, a Loop walks the subscriber
  list and alerts on each over threshold entry, and a final If escalates the
  worst offender and fails the run. Healthy fanouts are logged, and the worst
  subscriber, the totals and the full per subscriber table are published as
  outputs.

triggers:
  - id: on_metrics_publish
    type: io.kestra.plugin.core.trigger.Webhook
    description: The monitoring pipeline or a scheduled exporter POSTs per
      subscriber SNS delivery metrics here so the fanout is audited as soon as a
      window closes.
    key: "{{ secret('SNS_METRICS_WEBHOOK_KEY') }}"
  - id: daily_fanout_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily 07:00 UTC net that rescores the last day of delivery metrics,
      shipped disabled until the webhook path is trusted in your environment.
    cron: "0 7 * * *"
    disabled: true

inputs:
  - id: topic_arn
    type: STRING
    defaults: "arn:aws:sns:us-east-1:123456789012:orders.fanout"
    description: ARN of the SNS topic under audit, echoed in the alerts when the
      posted metrics omit it.
  - id: failure_rate_threshold_pct
    type: FLOAT
    defaults: 10.0
    description: Per subscriber failure rate percentage that flags a subscriber as
      over threshold for alerting.
  - id: min_failed_deliveries
    type: INT
    defaults: 5
    description: Minimum absolute failed deliveries a subscriber needs before its
      rate is allowed to flag, so tiny denominators do not page anyone.

tasks:
  # --- 1. fanout score ---
  - id: audit_fanout
    type: io.kestra.plugin.scripts.python.Script
    description: Read the posted delivery metrics or the bundled example window,
      compute the failure rate per subscriber, flag entries over both the rate
      and the absolute floor, and emit the worst subscriber, the totals and the
      full table through the outputs protocol.
    containerImage: python:3.12-slim
    env:
      SNS_METRICS_JSON: "{{ trigger.body ?? '' }}"
      FAILURE_RATE_THRESHOLD_PCT: "{{ inputs.failure_rate_threshold_pct }}"
      MIN_FAILED_DELIVERIES: "{{ inputs.min_failed_deliveries }}"
    script: |
      import json
      import os

      DEFAULT_METRICS = """
      {
        "topic_arn": "arn:aws:sns:us-east-1:123456789012:orders.fanout",
        "window_hours": 24,
        "subscribers": [
          {"subscriber": "orders-processor-lambda", "protocol": "lambda", "delivered": 18420, "failed": 12, "filtered": 340},
          {"subscriber": "orders-dlq-queue", "protocol": "sqs", "delivered": 18700, "failed": 0, "filtered": 0},
          {"subscriber": "finance-mailing-list", "protocol": "email", "delivered": 46, "failed": 8, "filtered": 0},
          {"subscriber": "analytics-firehose", "protocol": "firehose", "delivered": 12004, "failed": 2, "filtered": 88}
        ]
      }
      """


      def as_int(value, fallback=0):
          try:
              return int(value)
          except (TypeError, ValueError):
              try:
                  return int(float(value))
              except (TypeError, ValueError):
                  return fallback


      def as_float(value, fallback=0.0):
          try:
              return float(value)
          except (TypeError, ValueError):
              return fallback


      raw = (os.environ.get("SNS_METRICS_JSON") or "").strip() or DEFAULT_METRICS.strip()
      threshold = as_float(os.environ.get("FAILURE_RATE_THRESHOLD_PCT"), 10.0)
      min_failed = max(0, as_int(os.environ.get("MIN_FAILED_DELIVERIES"), 5))

      outputs = {
          "audit_ok": False,
          "topic": "unknown",
          "subscribers": [],
          "worst_subscriber": "none",
          "worst_failure_rate": 0.0,
          "total_failed": 0,
          "over_threshold_count": 0,
      }

      try:
          payload = json.loads(raw)
      except ValueError:
          payload = None

      if isinstance(payload, dict):
          subscribers = payload.get("subscribers")
          if not isinstance(subscribers, list):
              subscribers = []
          rows = []
          total_failed = 0
          over = 0
          worst_name = "none"
          worst_rate = 0.0
          for sub in subscribers:
              if not isinstance(sub, dict):
                  continue
              delivered = max(0, as_int(sub.get("delivered")))
              failed = max(0, as_int(sub.get("failed")))
              filtered = max(0, as_int(sub.get("filtered")))
              total_failed += failed
              denominator = delivered + failed
              rate = (failed * 100.0 / denominator) if denominator > 0 else 0.0
              flagged = rate >= threshold and failed >= min_failed
              if flagged:
                  over += 1
              name = str(sub.get("subscriber") or sub.get("arn") or "unknown")
              rows.append({
                  "subscriber": name,
                  "protocol": str(sub.get("protocol") or "unknown"),
                  "delivered": delivered,
                  "failed": failed,
                  "filtered": filtered,
                  "failure_rate_pct": round(rate, 2),
                  "over_threshold": flagged,
              })
              if rate > worst_rate:
                  worst_rate = round(rate, 2)
                  worst_name = name
          outputs.update({
              "audit_ok": True,
              "topic": str(payload.get("topic_arn") or "unknown"),
              "subscribers": rows,
              "worst_subscriber": worst_name,
              "worst_failure_rate": worst_rate,
              "total_failed": total_failed,
              "over_threshold_count": over,
          })

      print("audit_ok=%s subscribers=%d over=%d worst=%s rate=%.2f%% failed=%d" % (
          outputs["audit_ok"], len(outputs["subscribers"]), outputs["over_threshold_count"],
          outputs["worst_subscriber"], outputs["worst_failure_rate"], outputs["total_failed"]))
      print("::" + json.dumps({"outputs": outputs}) + "::")

  # --- 2. per subscriber sweep ---
  - id: review_subscribers
    type: io.kestra.plugin.core.flow.Loop
    description: Walk the scored subscriber rows one at a time so each over
      threshold entry gets its own alert with the exact counters instead of one
      blended message.
    concurrencyLimit: 1
    values: "{{ outputs.audit_fanout.vars.subscribers }}"
    tasks:
      - id: flag_subscriber
        type: io.kestra.plugin.core.flow.If
        description: Branch on the per subscriber over threshold flag the scorer already
          computed, keeping the rate and floor policy in one place.
        condition: "{{ item.value.over_threshold }}"
        then:
          - id: alert_subscriber
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post the failing subscriber, its protocol, the failed and delivered
              counts and its failure rate so the owning team can repair the
              endpoint.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "SNS fanout subscriber {{ item.value.subscriber }} ({{ item.value.protocol }}) on {{ inputs.topic_arn }} is over threshold - {{ item.value.failed }} failed of {{ item.value.delivered }} delivered plus {{ item.value.failed }} failed, a {{ item.value.failure_rate_pct }} percent failure rate against the {{ inputs.failure_rate_threshold_pct }} percent limit. {{ item.value.filtered }} delivery(ies) were filtered by policy. Execution {{ execution.id }}."
              }
        else:
          - id: log_subscriber_ok
            type: io.kestra.plugin.core.log.Log
            description: Record the healthy subscriber for the window so the execution
              history carries the full per subscriber trail.
            message: "SNS subscriber {{ item.value.subscriber }} healthy - {{
              item.value.failed }} failed of {{ item.value.delivered }}
              delivered, rate {{ item.value.failure_rate_pct }} percent under
              threshold."

  # --- 3. escalation gate ---
  - id: escalate_worst
    type: io.kestra.plugin.core.flow.If
    description: Branch on the audit verdict so a fanout with any over threshold
      subscriber escalates the worst offender and fails the run, while a healthy
      fanout is logged.
    condition: "{{ outputs.audit_fanout.vars.audit_ok and
      outputs.audit_fanout.vars.over_threshold_count > 0 }}"
    then:
      - id: alert_worst
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the worst subscriber, its rate, the over threshold count and
          the window failure total so the on call sees the full blast radius in
          one message.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "SNS fanout audit ESCALATION for {{ inputs.topic_arn }} - worst subscriber {{ outputs.audit_fanout.vars.worst_subscriber }} at {{ outputs.audit_fanout.vars.worst_failure_rate }} percent failure rate, {{ outputs.audit_fanout.vars.over_threshold_count }} subscriber(s) over the {{ inputs.failure_rate_threshold_pct }} percent threshold, {{ outputs.audit_fanout.vars.total_failed }} total failed deliveries in the window. Subscribers are silently out of sync until repaired. Execution {{ execution.id }}."
          }
      - id: fail_audit
        type: io.kestra.plugin.core.execution.Fail
        description: Turn the execution red so the pipeline watching this audit stops
          instead of assuming the fanout delivered everything.
        errorMessage: "SNS fanout audit found {{
          outputs.audit_fanout.vars.over_threshold_count }} subscriber(s) over
          threshold, worst {{ outputs.audit_fanout.vars.worst_subscriber }} at
          {{ outputs.audit_fanout.vars.worst_failure_rate }} percent."
    else:
      - id: log_healthy_fanout
        type: io.kestra.plugin.core.log.Log
        description: Record the healthy verdict with the subscriber count and the
          failure total so a quiet window still leaves evidence the audit ran.
        message: "SNS fanout audit healthy for {{ inputs.topic_arn }} - {{
          outputs.audit_fanout.vars.subscribers | length }} subscriber(s)
          checked, none over threshold, {{
          outputs.audit_fanout.vars.total_failed }} failed deliveries in the
          window."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: worst_subscriber
    type: STRING
    description: Subscriber with the highest failure rate in the window, for example
      finance-mailing-list, or none when every rate is zero or nothing could be
      scored.
    value: "{{ outputs.audit_fanout.vars.worst_subscriber }}"
  - id: worst_failure_rate
    type: FLOAT
    description: Failure rate percentage of the worst subscriber, for example 14.81
      when eight of fifty four deliveries failed.
    value: "{{ outputs.audit_fanout.vars.worst_failure_rate }}"
  - id: total_failed
    type: INT
    description: Sum of failed deliveries across every subscriber in the scored window.
    value: "{{ outputs.audit_fanout.vars.total_failed }}"
  - id: over_threshold_count
    type: INT
    description: Number of subscribers whose failure rate met both the rate
      threshold and the absolute delivery floor.
    value: "{{ outputs.audit_fanout.vars.over_threshold_count }}"
  - id: subscribers
    type: ARRAY
    description: Full per subscriber table with delivered, failed, filtered, failure
      rate and the over threshold flag for each entry.
    value: "{{ outputs.audit_fanout.vars.subscribers }}"