Kestra Cluster Execution Backlog and Worker Starvation Watchdog

id: kestra-cluster-backlog-and-worker-watchdog
namespace: company.sre

description: |
  Automated SRE reliability watchdog that monitors Kestra cluster queue latency, worker thread
  utilization, and execution backlog health. Detects worker starvation where executions remain stuck
  in QUEUED state, identifies runaway zombie executions exceeding runtime SLOs, and dispatches
  remediation alerts to on-call infrastructure engineers.

inputs:
  - id: seed_demo_data
    type: BOOL
    defaults: true
    description: When true, simulates cluster execution metrics with queued lag and
      zombie task edge cases for zero-setup local execution.

  - id: max_queued_seconds_threshold
    type: INT
    defaults: 120
    description: Maximum seconds an execution can remain in QUEUED state before
      triggering worker starvation alert.

  - id: max_running_hours_threshold
    type: INT
    defaults: 6
    description: Maximum hours an execution can run before being flagged as a
      potential zombie process.

  - id: kestra_api_url
    type: STRING
    defaults: http://localhost:8080
    description: Internal Kestra REST API base URL reachable from the worker network.

tasks:
  - id: collect_cluster_metrics
    type: io.kestra.plugin.scripts.python.Script
    description: Ingests execution queue metrics via Kestra REST API or synthesizes
      cluster telemetry in demo mode.
    containerImage: python:3.11-slim
    script: |
      import json
      from kestra import Kestra

      demo_mode = {{ inputs.seed_demo_data }}
      queued_threshold = {{ inputs.max_queued_seconds_threshold }}
      running_threshold_hours = {{ inputs.max_running_hours_threshold }}

      stalled_queued_executions = []
      runaway_zombies = []
      total_active = 0
      total_queued = 0

      if demo_mode:
          total_active = 42
          total_queued = 8
          # Simulate stalled queued execution (340s in queue > 120s threshold)
          stalled_queued_executions = [
              {
                  "execution_id": "exec_3k9f01ab92",
                  "flow_id": "nightly-order-aggregation",
                  "namespace": "company.analytics",
                  "queued_seconds": 340
              }
          ]
          # Simulate runaway zombie execution (8.5h runtime > 6h threshold)
          runaway_zombies = [
              {
                  "execution_id": "exec_88a1cd99ef",
                  "flow_id": "legacy-ftp-sync",
                  "namespace": "company.integrations",
                  "duration_hours": 8.5
              }
          ]
      else:
          try:
              import urllib.request
              from datetime import datetime, timezone
              api_url = "{{ inputs.kestra_api_url }}/api/v1/executions/search?page=1&size=100"
              req = urllib.request.Request(api_url, headers={"Accept": "application/json"})
              with urllib.request.urlopen(req, timeout=15) as resp:
                  payload = json.loads(resp.read().decode("utf-8"))
                  executions = payload.get("results", [])
                  now = datetime.now(timezone.utc)
                  for ex in executions:
                      state = ex.get("state", {}).get("current", "")
                      if state == "RUNNING":
                          total_active += 1
                          start_str = ex.get("state", {}).get("startDate")
                          if start_str:
                              start_dt = datetime.fromisoformat(start_str.replace("Z", "+00:00"))
                              hours = (now - start_dt).total_seconds() / 3600.0
                              if hours > running_threshold_hours:
                                  runaway_zombies.append({
                                      "execution_id": ex.get("id"),
                                      "flow_id": ex.get("flowId"),
                                      "namespace": ex.get("namespace"),
                                      "duration_hours": round(hours, 1)
                                  })
                      elif state == "QUEUED":
                          total_queued += 1
                          start_str = ex.get("state", {}).get("startDate")
                          if start_str:
                              start_dt = datetime.fromisoformat(start_str.replace("Z", "+00:00"))
                              secs = (now - start_dt).total_seconds()
                              if secs > queued_threshold:
                                  stalled_queued_executions.append({
                                      "execution_id": ex.get("id"),
                                      "flow_id": ex.get("flowId"),
                                      "namespace": ex.get("namespace"),
                                      "queued_seconds": int(secs)
                                  })
          except Exception as e:
              print(f"API query fallback: {e}")

      worker_starvation = len(stalled_queued_executions) > 0
      has_zombies = len(runaway_zombies) > 0
      cluster_degraded = worker_starvation or has_zombies

      Kestra.outputs({
          "total_active_executions": total_active,
          "total_queued_executions": total_queued,
          "stalled_queued_count": len(stalled_queued_executions),
          "runaway_zombies_count": len(runaway_zombies),
          "worker_starvation_detected": worker_starvation,
          "zombies_detected": has_zombies,
          "cluster_degraded": cluster_degraded,
          "stalled_queued_details": json.dumps(stalled_queued_executions),
          "runaway_zombies_details": json.dumps(runaway_zombies)
      })

  - id: evaluate_cluster_health_gate
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether worker starvation or runaway zombie executions
      require SRE intervention.
    condition: "{{ outputs.collect_cluster_metrics.vars.cluster_degraded }}"
    then:
      - id: notify_sre_cluster_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority cluster degradation alert card to SRE on-call.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "🚨 *Kestra Cluster Degradation Alert*\n• *Active Executions:* {{ outputs.collect_cluster_metrics.vars.total_active_executions }}\n• *Queued Executions:* {{ outputs.collect_cluster_metrics.vars.total_queued_executions }}\n• *Starved Queued Tasks:* {{ outputs.collect_cluster_metrics.vars.stalled_queued_count }} (Exceeded {{ inputs.max_queued_seconds_threshold }}s threshold)\n• *Runaway Zombies:* {{ outputs.collect_cluster_metrics.vars.runaway_zombies_count }} (Exceeded {{ inputs.max_running_hours_threshold }}h threshold)\n• *Action Required:* Inspect worker pool capacity and thread concurrency."
          }

      - id: log_degraded_cluster_state
        type: io.kestra.plugin.core.log.Log
        description: Records detailed cluster diagnostic telemetry in execution history.
        message: "Cluster health degraded: Starved Tasks: {{
          outputs.collect_cluster_metrics.vars.stalled_queued_details }} |
          Runaway Zombies: {{
          outputs.collect_cluster_metrics.vars.runaway_zombies_details }}"
    else:
      - id: log_nominal_cluster_state
        type: io.kestra.plugin.core.log.Log
        description: Records healthy cluster telemetry when queue metrics are within
          nominal operating limits.
        message: "Cluster queue health nominal (Active: {{
          outputs.collect_cluster_metrics.vars.total_active_executions }},
          Queued: {{
          outputs.collect_cluster_metrics.vars.total_queued_executions }})."

triggers:
  - id: periodic_cluster_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Regular 15-minute diagnostic audit of cluster execution queues.
      Shipped disabled by default.
    cron: "*/15 * * * *"
    disabled: true

  - id: cluster_sre_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook for external Prometheus or Datadog alert ingestion.
    key: "{{ secret('SRE_WEBHOOK_KEY') }}"

errors:
  - id: alert_watchdog_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts on-call Slack if the watchdog workflow fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "⚠️ Cluster watchdog execution {{ execution.id }} failed on flow {{ flow.id }}."
      }