Argo Workflows Stuck Running Termination Guard

id: argo-workflows-stuck-running-termination-guard
namespace: company.infrastructure

description: |
  Audit Argo Workflows for workflows stuck in the Running phase far longer than
  reasonable, a known failure mode when a workflow pod is evicted, OOMKilled, or
  its node is drained and the controller loses track of it, leaving the
  workflow object Running forever with no further progress. Pause for operator
  approval, terminate the stuck workflow via the Argo Server API, and verify
  it actually left the Running phase.

inputs:
  - id: argo_server_url
    type: STRING
    displayName: Argo Server URL
    description: Base HTTP(S) URL of the Argo Server API (default port 2746), e.g.
      https://argo-server.argo.svc.cluster.local:2746.
    defaults: "https://localhost:2746"

  - id: argo_namespace
    type: STRING
    displayName: Argo Namespace
    description: Kubernetes namespace to audit for workflows.
    defaults: "argo"

  - id: argo_bearer_token
    type: STRING
    displayName: Argo Server Bearer Token
    description: A Kubernetes ServiceAccount bearer token (or equivalent) authorized
      to list and terminate workflows in the target namespace.
    defaults: "{{ secret('ARGO_BEARER_TOKEN') }}"

  - id: exclude_workflow_name_prefixes
    type: ARRAY
    itemType: STRING
    displayName: Excluded Workflow Name Prefixes
    description: Workflow name prefixes that must never be flagged or terminated,
      for example a known-legitimate long-running training job. Empty by
      default.
    defaults: []

  - id: max_running_hours
    type: INT
    displayName: Max Running Hours
    description: Hours a workflow may remain in the Running phase before it is
      flagged as stuck.
    defaults: 24

  - id: max_terminations_per_run
    type: INT
    displayName: Max Terminations Per Run
    description: Safety ceiling on the number of workflows terminated in a single
      execution pass. When more workflows qualify, the longest-running ones are
      prioritized and the rest deferred to the next run.
    defaults: 1

  - id: dry_run
    type: BOOL
    displayName: Dry Run Mode
    description: When true, audits workflow running time and logs the proposed
      terminations without mutating any workflow.
    defaults: true

  - id: auto_remediate
    type: BOOL
    displayName: Auto Remediate
    description: When true and dry_run is false, pauses for operator approval and
      then calls the Argo Server terminate API on verified stuck workflows.
    defaults: false

triggers:
  - id: periodic_stuck_workflow_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Audit for workflows stuck in the Running phase every hour. Shipped
      disabled so the flow can be verified manually in safe mode first.
    cron: "0 * * * *"
    disabled: true

  - id: on_demand_audit_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP endpoint invoked by a cluster alerting rule or
      incident runbook.
    key: "{{ secret('ARGO_GUARD_WEBHOOK_KEY') }}"

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.execution.Assert
    description: Assert operational thresholds satisfy sanity constraints before
      calling the Argo Server API.
    conditions:
      - "{{ inputs.max_running_hours >= 1 }}"
      - "{{ inputs.max_terminations_per_run >= 1 }}"
    errorMessage: "Invalid configuration: max_running_hours and
      max_terminations_per_run must both be at least 1."

  - id: list_workflows
    type: io.kestra.plugin.core.http.Request
    description: List workflows in the target namespace via the Argo Server REST API.
    uri: "{{ inputs.argo_server_url }}/api/v1/workflows/{{ inputs.argo_namespace }}"
    method: GET
    headers:
      Accept: application/json
    options:
      auth:
        type: BEARER
        token: "{{ inputs.argo_bearer_token }}"

  - id: triage_workflows
    type: io.kestra.plugin.scripts.python.Script
    description: Classify workflows stuck in the Running phase beyond
      max_running_hours, excluding protected name prefixes and applying the
      max_terminations_per_run safety cap.
    containerImage: python:3.11-slim
    env:
      WORKFLOWS_BODY: "{{ outputs.list_workflows.body is defined ?
        outputs.list_workflows.body : '{}' }}"
      EXCLUDE_WORKFLOW_NAME_PREFIXES: "{{ inputs.exclude_workflow_name_prefixes | toJson }}"
      MAX_RUNNING_HOURS: "{{ inputs.max_running_hours }}"
      MAX_TERMINATIONS_PER_RUN: "{{ inputs.max_terminations_per_run }}"
      EXECUTION_ID: "{{ execution.id }}"
    script: |
      import json
      import os
      from datetime import datetime, timezone

      try:
          from kestra import Kestra
      except ImportError:
          Kestra = None

      now = datetime.now(timezone.utc)
      now_iso = now.isoformat()
      workflows_body_raw = os.environ.get("WORKFLOWS_BODY", "{}")
      exclude_prefixes = tuple(json.loads(os.environ.get("EXCLUDE_WORKFLOW_NAME_PREFIXES", "[]")))
      max_running_hours = float(os.environ.get("MAX_RUNNING_HOURS", "24"))
      max_terminations_per_run = int(os.environ.get("MAX_TERMINATIONS_PER_RUN", "1"))
      execution_id = os.environ.get("EXECUTION_ID", "manual")

      def parse_k8s_time(value):
          if not value:
              return None
          try:
              # metav1.Time serializes as RFC3339 with a trailing "Z".
              return datetime.fromisoformat(value.replace("Z", "+00:00"))
          except Exception:
              return None

      try:
          workflows_data = json.loads(workflows_body_raw) if workflows_body_raw.strip() else {}
      except Exception:
          workflows_data = {}

      items = workflows_data.get("items") or []
      if not isinstance(items, list):
          items = []

      termination_candidates = []
      excluded_count = 0
      healthy_count = 0

      for wf in items:
          if not isinstance(wf, dict):
              continue
          metadata = wf.get("metadata") or {}
          status = wf.get("status") or {}
          name = str(metadata.get("name", ""))
          namespace = str(metadata.get("namespace", ""))
          phase = str(status.get("phase", ""))

          if not name:
              continue
          if phase != "Running":
              continue
          if name.startswith(exclude_prefixes):
              excluded_count += 1
              continue

          started_at = parse_k8s_time(status.get("startedAt"))
          if started_at is None:
              # No startedAt on a Running workflow would itself be unusual; skip
              # rather than guess, since there is nothing real to compute age from.
              continue

          running_hours = (now - started_at).total_seconds() / 3600.0

          if running_hours >= max_running_hours:
              termination_candidates.append({
                  "name": name,
                  "namespace": namespace,
                  "running_hours": round(running_hours, 1),
                  "started_at": status.get("startedAt"),
                  "progress": str(status.get("progress", "")),
                  "message": str(status.get("message", "")),
              })
          else:
              healthy_count += 1

      # Cap at max_terminations_per_run, longest-running first.
      termination_candidates.sort(key=lambda c: c["running_hours"], reverse=True)
      selected_candidates = termination_candidates[:max_terminations_per_run]
      deferred_candidates = termination_candidates[max_terminations_per_run:]

      has_candidates = len(selected_candidates) > 0
      overall_status = "WARNING" if has_candidates else ("INFO" if deferred_candidates else "HEALTHY")

      summary_lines = [
          f"- `{c['namespace']}/{c['name']}`: Running for {c['running_hours']}h (started {c['started_at']})"
          for c in selected_candidates
      ]
      if deferred_candidates:
          summary_lines.append(f"Deferred by max_terminations_per_run cap: {len(deferred_candidates)} additional stuck workflow(s) not included this pass.")

      summary_text = (
          f"Argo Workflows Stuck Running Guard: {overall_status} | "
          f"Audited {len(items)} workflow(s) ({excluded_count} excluded). "
          f"Termination candidates: {len(selected_candidates)}. Deferred: {len(deferred_candidates)}. "
          f"Healthy: {healthy_count}."
      )
      print(summary_text)

      audit_manifest = {
          "overall_status": overall_status,
          "workflows_audited": len(items),
          "excluded_count": excluded_count,
          "termination_candidate_count": len(selected_candidates),
          "deferred_count": len(deferred_candidates),
          "healthy_count": healthy_count,
          "termination_candidates": selected_candidates,
          "execution_id": execution_id,
          "evaluated_at": now_iso,
          "summary": summary_text,
      }

      output_data = {
          "has_termination_candidates": has_candidates,
          "termination_candidate_count": len(selected_candidates),
          "overall_status": overall_status,
          "termination_candidates": selected_candidates,
          "summary": summary_text,
          "summary_details": "\n".join(summary_lines) if summary_lines else "None",
          "audit_manifest": audit_manifest,
      }

      if Kestra:
          Kestra.outputs(output_data)
      else:
          print("::" + json.dumps({"outputs": output_data}) + "::")

  - id: evaluate_stuck_workflows
    type: io.kestra.plugin.core.flow.If
    description: Branch execution based on whether any workflow has been Running
      beyond max_running_hours.
    condition: "{{ outputs.triage_workflows.vars.has_termination_candidates == true }}"
    then:
      - id: dispatch_slack_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Deliver triage notification to Slack naming workflows stuck in the
          Running phase.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "attachments": [
              {
                "color": "warning",
                "title": ":warning: Argo Workflows Stuck Running Alert",
                "text": "*Status:* `{{ outputs.triage_workflows.vars.overall_status }}`\n*Namespace:* `{{ inputs.argo_namespace }}`\n*Termination Candidates:* {{ outputs.triage_workflows.vars.termination_candidate_count }}\n*Mode:* auto_remediate={{ inputs.auto_remediate }}, dry_run={{ inputs.dry_run }}\n*Execution:* `{{ execution.id }}`\n\n*Details:*\n{{ outputs.triage_workflows.vars.summary_details }}\n\n*Summary:* {{ outputs.triage_workflows.vars.summary }}"
              }
            ]
          }

      - id: remediation_mode_gate
        type: io.kestra.plugin.core.flow.If
        description: Gate automated remediation behind auto_remediate and dry_run
          configuration flags.
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: pause_for_operator_approval
            type: io.kestra.plugin.core.flow.Pause
            description: Pause workflow execution for operator review before terminating
              stuck Argo Workflows. Resume to apply or kill to abort; no timeout
              is set so this waits indefinitely for a human decision.
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: Set true to confirm the Argo Server terminate call should be issued
                  against the flagged stuck workflows.

          - id: branch_on_approval
            type: io.kestra.plugin.core.flow.If
            description: Execute termination only if explicit human approval was provided
              during the pause gate.
            condition: "{{ outputs.pause_for_operator_approval.onResume.approved == true }}"
            then:
              - id: terminate_stuck_workflows
                type: io.kestra.plugin.core.flow.Loop
                description: Iterate sequentially over termination candidates, terminate each
                  via the Argo Server API, and verify it left the Running phase.
                concurrencyLimit: 1
                values: "{{ outputs.triage_workflows.vars.termination_candidates | toJson }}"
                tasks:
                  - id: terminate_workflow
                    type: io.kestra.plugin.core.http.Request
                    description: Call the Argo Server terminate endpoint for this workflow.
                    uri: "{{ inputs.argo_server_url }}/api/v1/workflows/{{
                      fromJson(item.value).namespace }}/{{
                      fromJson(item.value).name }}/terminate"
                    method: PUT
                    contentType: application/json
                    body: "{}"
                    options:
                      auth:
                        type: BEARER
                        token: "{{ inputs.argo_bearer_token }}"

                  - id: settle_termination
                    type: io.kestra.plugin.core.flow.Sleep
                    description: Allow the Argo controller time to process the termination before
                      re-checking the workflow's phase.
                    duration: PT10S

                  - id: verify_workflow_terminated
                    type: io.kestra.plugin.core.http.Request
                    description: Re-fetch this workflow to confirm it has left the Running phase.
                    uri: "{{ inputs.argo_server_url }}/api/v1/workflows/{{
                      fromJson(item.value).namespace }}/{{
                      fromJson(item.value).name }}"
                    method: GET
                    headers:
                      Accept: application/json
                    options:
                      auth:
                        type: BEARER
                        token: "{{ inputs.argo_bearer_token }}"

                  - id: assert_workflow_not_running
                    type: io.kestra.plugin.core.flow.If
                    description: Fail loudly if the workflow still reports the Running phase after
                      the terminate call.
                    condition: "{{ (fromJson(outputs.verify_workflow_terminated.body).status.phase
                      ?? 'Running') == 'Running' }}"
                    then:
                      - id: fail_still_running
                        type: io.kestra.plugin.core.execution.Fail
                        description: Abort with explicit error details identifying the workflow that did
                          not terminate.
                        errorMessage: "Argo Workflow {{ fromJson(item.value).namespace }}/{{
                          fromJson(item.value).name }} still reports phase
                          Running after a terminate call. Check Argo controller
                          health and whether the bearer token has permission to
                          terminate workflows in this namespace."
                    else:
                      - id: log_workflow_terminated
                        type: io.kestra.plugin.core.log.Log
                        description: Record successful termination for this workflow.
                        message: "Terminated stuck workflow {{ fromJson(item.value).namespace }}/{{
                          fromJson(item.value).name }} (was Running for {{
                          fromJson(item.value).running_hours }}h)."

              - id: save_recovery_audit_kv
                type: io.kestra.plugin.core.kv.Set
                description: Persist the remediation audit manifest into Kestra KV for incident
                  review. Not needed for the detection logic itself (workflow
                  running time is monotonic and a terminated workflow naturally
                  leaves the Running phase it was flagged in), but kept for
                  compliance history.
                key: "argo_workflows_termination_audit_{{ execution.id }}"
                value: "{{ outputs.triage_workflows.vars.audit_manifest | toJson }}"
                kvType: JSON

              - id: notify_resolution_slack
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                description: Dispatch resolution confirmation to Slack once all approved stuck
                  workflows have been terminated and verified.
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                payload: |
                  {
                    "attachments": [
                      {
                        "color": "#2EB67D",
                        "title": ":white_check_mark: Argo Workflows Stuck Workflows Terminated",
                        "text": "*Namespace:* `{{ inputs.argo_namespace }}`\n*Workflows Terminated:* `{{ outputs.triage_workflows.vars.termination_candidate_count }}`\n*Execution:* `{{ execution.id }}`"
                      }
                    ]
                  }
            else:
              - id: log_approval_declined
                type: io.kestra.plugin.core.log.Log
                description: Record that the operator declined approval for terminating stuck
                  workflows.
                message: "Operator declined approval for Argo Workflows termination. Flagged
                  workflows were left running."
        else:
          - id: log_dry_run_plan
            type: io.kestra.plugin.core.log.Log
            description: Log the planned termination without mutating any workflow.
            message: |
              Dry run active or auto_remediate disabled (dry_run={{ inputs.dry_run }}, auto_remediate={{ inputs.auto_remediate }}).
              Planned termination for {{ outputs.triage_workflows.vars.termination_candidate_count }} workflow(s):
              {{ outputs.triage_workflows.vars.termination_candidates | toJson }}
    else:
      - id: log_healthy_state
        type: io.kestra.plugin.core.log.Log
        description: Confirm nominal state when no workflow has been Running beyond the
          configured threshold.
        message: "Argo Workflows nominal in namespace {{ inputs.argo_namespace }}. No
          workflow Running for {{ inputs.max_running_hours }}+ hours."

errors:
  - id: alert_on_workflow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Dispatch emergency alert to Slack if the guard encounters an
      unhandled execution error.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "attachments": [
          {
            "color": "danger",
            "title": ":rotating_light: Argo Workflows Stuck Running Guard Execution Failed",
            "text": "Failed execution `{{ execution.id }}` while auditing or terminating workflows in `{{ inputs.argo_namespace }}` on `{{ inputs.argo_server_url }}`. Check Argo Server connectivity, bearer token validity, and worker logs."
          }
        ]
      }

outputs:
  - id: workflows_audited
    type: INT
    value: "{{ outputs.triage_workflows.vars.audit_manifest.workflows_audited ?? 0 }}"

  - id: termination_candidate_count
    type: INT
    value: "{{ outputs.triage_workflows.vars.termination_candidate_count ?? 0 }}"

  - id: status
    type: STRING
    value: "{{ (outputs.triage_workflows.vars.termination_candidate_count ?? 0) > 0
      ? ((inputs.auto_remediate and not inputs.dry_run and
      (outputs.pause_for_operator_approval.onResume.approved ?? false)) ?
      'REMEDIATED' : 'STUCK_WORKFLOWS_DETECTED') : 'HEALTHY' }}"

Argo Workflows executes each step of a workflow as a Kubernetes Pod. When that pod is evicted by the node's kubelet under memory pressure, killed by a node drain during a cluster upgrade, or otherwise removed outside of Argo's own control, the workflow controller can lose track of it. The workflow object is left in the Running phase indefinitely: not failed, not succeeded, just silently stalled, with no further node transitions and no natural end. Nothing in Argo itself pages anyone when this happens, and a stuck workflow can sit there consuming a slot in dashboards and metrics until someone notices it manually.

This blueprint provides an automated guard that lists workflows in a namespace via the Argo Server REST API, flags any still in the Running phase after max_running_hours has elapsed since status.startedAt, requests human approval, calls Argo's terminate API, and re-fetches the workflow to confirm it actually left the Running phase.

How it works

Pre-flight Assertion: validate_inputs (io.kestra.plugin.core.execution.Assert) checks that the running-hours threshold and the per-run termination cap are both positive.
Workflow Discovery: list_workflows (io.kestra.plugin.core.http.Request) calls GET /api/v1/workflows/{namespace} on the Argo Server API, authenticated with a Bearer token.
Triage: triage_workflows (io.kestra.plugin.scripts.python.Script) filters for status.phase == "Running", excludes any workflow whose name matches exclude_workflow_name_prefixes, and flags the remainder whose elapsed time since status.startedAt is at or beyond max_running_hours. Candidates beyond max_terminations_per_run are deferred to a later run, prioritizing the longest-running ones first. Running time is monotonic and a terminated workflow naturally leaves the Running phase it was flagged in, so unlike a flapping reachability signal, this detection needs no cross-run state to avoid false positives.
Slack Triage Alert: When stuck workflows exist, dispatch_slack_alert posts the affected workflow names, namespace, and hours running.
Human Approval Gate: When auto_remediate is enabled and dry_run is false, pause_for_operator_approval (io.kestra.plugin.core.flow.Pause) halts execution until an operator resumes with approved: true.
Terminate and Verify: terminate_stuck_workflows (io.kestra.plugin.core.flow.Loop) calls PUT /api/v1/workflows/{namespace}/{name}/terminate per approved workflow, waits briefly, re-fetches the workflow, and fails loudly via io.kestra.plugin.core.execution.Fail if it still reports phase Running.
Audit & Resolution: The audit manifest is archived in Kestra KV for compliance history, and a resolution card is posted to Slack.
Failure Handling: A flow-level errors block alerts Slack if Argo Server connectivity, bearer token validity, or scripting fails unexpectedly.

What you get

Automated detection of Argo Workflows silently stuck in the Running phase after a pod eviction or node drain.
A per-run termination cap that prioritizes the longest-stuck workflow first instead of terminating a whole batch at once.
Human-in-the-loop approval before any workflow is terminated.
Post-termination verification that fails loudly instead of assuming the terminate call worked.
A name-prefix exclusion list so legitimately long-running workflows are never touched.

Who it's for

Platform Engineers and SREs operating Argo Workflows for CI, data, or ML pipelines on Kubernetes.
Teams running node auto-scaling or spot/preemptible node pools, where pod eviction is a routine occurrence that can orphan a workflow.
On-call engineers who want stuck workflows surfaced and cleared without manually scanning argo list output.

Why orchestrate this with Kestra

Argo Workflows has no built-in alert for a workflow stuck in the Running phase, and nothing automatically terminates one even when its pods are long gone. Kestra adds declarative scheduling, a threshold check against the real startedAt field, an interactive approval gate, and verified, auditable termination on top of the raw Argo Server API.

Prerequisites

A reachable Argo Server (Argo Workflows 3.x or higher) with its API port (default 2746) reachable from the Kestra worker.
A Kubernetes ServiceAccount bearer token (or equivalent) with permission to list and terminate workflows in the target namespace.
A Slack Incoming Webhook URL for triage and resolution notifications.

Secrets

ARGO_BEARER_TOKEN: A Kubernetes ServiceAccount bearer token authorized against the Argo Server API.
SLACK_WEBHOOK_URL: Slack Incoming Webhook endpoint for triage and resolution alerts.
ARGO_GUARD_WEBHOOK_KEY: Secret authentication token protecting the on-demand webhook trigger.

Quick start

Configure ARGO_BEARER_TOKEN, SLACK_WEBHOOK_URL, and ARGO_GUARD_WEBHOOK_KEY in your Kestra namespace secrets.
Import this blueprint and set argo_server_url and argo_namespace to match your cluster.
Run a manual execution with the default dry_run: true to audit currently Running workflows without mutating anything.
Review the execution logs and Slack triage card, and set exclude_workflow_name_prefixes for any legitimately long-running workflows.
Run with auto_remediate: true and dry_run: false to arm remediation; when stuck workflows are detected, approve the pause_for_operator_approval task with approved: true and observe the verified termination.
Enable the periodic_stuck_workflow_audit schedule trigger for continuous monitoring.

How to extend

Node-Level Confirmation: Before terminating, add a Kubernetes API call checking whether the workflow's pods actually still exist, to distinguish a truly orphaned workflow from one that is merely slow.
Auto-Resubmit: After a verified termination, chain a call to POST /api/v1/workflows/{namespace}/{name}/resubmit to automatically retry the work under a new workflow name, when the underlying job is safe to rerun.
PagerDuty Escalation: Add io.kestra.plugin.pagerduty.PagerDutyAlert when a workflow has been stuck for multiples of max_running_hours, signaling a recurring eviction pattern worth deeper investigation.

Terminate Argo Workflows Stuck in the Running Phase After Approval

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Terminate Argo Workflows Stuck in the Running Phase After Approval

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

More Data Blueprints

New to Kestra?