Google Cloud Pub/Sub Subscription Oldest Unacked Message Sentinel

id: gcp-pubsub-subscription-oldest-unacked-message-sentinel
namespace: company.cloud

inputs:
  - id: project_id
    type: STRING
    description: Google Cloud Project ID.
  - id: subscription_id
    type: STRING
    description: Cloud Pub/Sub subscription name to audit.
  - id: max_age_seconds_threshold
    type: INT
    defaults: 3600
    description: "Maximum acceptable age of oldest unacknowledged message in seconds
      (default 1 hour / 3600s)."

tasks:
  - id: query_pubsub_lag_metrics
    type: io.kestra.plugin.gcp.cli.GCloudCLI
    projectId: "{{ inputs.project_id }}"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT') }}"
    commands:
      - |
        START_TIME=$(date -u -v-15M +%Y-%m-%dT%H:%M:%SZ)
        FILTER="metric.type="pubsub.googleapis.com/subscription/oldest_unacked_message_age" AND resource.labels.subscription_id="{{ inputs.subscription_id }}""

        gcloud monitoring timeseries list \
          --filter="$FILTER" \
          --interval-start-time="$START_TIME" \
          --format=json > /tmp/pubsub_lag.json

        MAX_AGE=$(jq '[.[]?.points[]?.value.int64Value // 0] | max // 0' /tmp/pubsub_lag.json)

        echo "::set-output name=max_age::$MAX_AGE::"

  - id: evaluate_message_lag
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.query_pubsub_lag_metrics.vars.max_age | int >=
      inputs.max_age_seconds_threshold }}"
    then:
      - id: notify_streaming_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK') }}"
        payload: |
          {
            "channel": "#gcp-alerts",
            "attachments": [
              {
                "color": "danger",
                "title": "🚨 Google Cloud Pub/Sub Subscriber Lag Alert",
                "text": "Pub/Sub subscription `{{ inputs.subscription_id }}` in project `{{ inputs.project_id }}` has unacknowledged messages aging past *{{ outputs.query_pubsub_lag_metrics.vars.max_age }} seconds* (Threshold: {{ inputs.max_age_seconds_threshold }}s)! Impact: Downstream consumers (Dataflow, Cloud Functions, GKE workers) are lagging, crashed, or failing to acknowledge messages, threatening retention expiry data loss.",
                "footer": "Kestra Messaging Sentinel • Google Cloud Pub/Sub"
              }
            ]
          }

errors:
  - id: alert_pipeline_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "channel": "#gcp-alerts",
        "attachments": [
          {
            "color": "danger",
            "title": "⚠️ Google Cloud Pub/Sub Subscription Oldest Unacked Message Sentinel Pipeline Failure",
            "text": "Failed to query Cloud Monitoring for Pub/Sub subscription metrics. Check credentials.",
            "footer": "Kestra Workflow Failure Sentinel"
          }
        ]
      }

triggers:
  - id: periodic_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"
    disabled: true