Kestra stuck execution reaper

id: kestra-stuck-execution-reaper
namespace: company.team

description: |
  Find Kestra executions that have been stuck in RUNNING past a threshold, the dead-worker
  zombies that hold a concurrency slot forever. List them in dry run. On auto_remediate, behind a
  Pause approval, it kills each one with the Kestra plugin, then re-queries and fails if any
  targeted execution is still RUNNING.

inputs:
  - id: kestra_url
    type: STRING
    displayName: Target Kestra API URL
    description: The Kestra whose executions are reaped. The default points the flow
      at its own cluster.
    defaults: "http://localhost:8080"

  - id: namespace
    type: STRING
    displayName: Namespace to reap
    description: Only executions in exactly this namespace are considered. A safe
      narrow scope.
    defaults: company.team

  - id: flow_id
    type: STRING
    displayName: Flow to reap
    description: Restrict to one flow in the namespace. Leave empty to consider
      every flow in the namespace.
    defaults: ""

  - id: max_running_minutes
    type: INT
    displayName: Stuck threshold (minutes)
    description: An execution RUNNING longer than this is treated as stuck.
    defaults: 30

  - id: exclude_flows
    type: ARRAY
    itemType: STRING
    displayName: Flows never reaped
    description: Flow ids that are allowed to run long, for example a known long batch job.
    defaults: []

  - id: max_kills
    type: INT
    displayName: Kill ceiling
    description: Refuse to kill more than this many executions in one run. A larger
      number of zombies is treated as an incident for a human to look at.
    defaults: 25

  - id: kestra_user
    type: STRING
    displayName: Kestra API username
    defaults: admin@kestra.io

  - id: kestra_password
    type: STRING
    displayName: Kestra API password
    defaults: "{{ secret('KESTRA_PASSWORD') }}"

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Allow the kill. Still requires dry_run false and a Pause approval.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for any execution to
      be killed.
    defaults: true

variables:
  cutoff: "{{ now(timeZone='UTC') | dateAdd(-1 * inputs.max_running_minutes,
    'MINUTES') | date(\"yyyy-MM-dd'T'HH:mm:ss'Z'\", timeZone='UTC') }}"
  select_program: "{{ '[.results[] | {id, flowId, ns: .namespace, start:
    .state.startDate}]' ~ ' | map(select(.flowId != \"' ~ flow.id ~ '\"))' ~ ' |
    map(select(.id != \"' ~ execution.id ~ '\"))' ~ ' | map(select(\"' ~
    inputs.flow_id ~ '\" == \"\" or .flowId == \"' ~ inputs.flow_id ~ '\"))' ~ '
    | map(select(([.flowId] - ' ~ (inputs.exclude_flows | jq('.') | first |
    toJson) ~ ') | length > 0))' }}"

tasks:
  - id: detect
    type: io.kestra.plugin.core.http.Request
    description: Read-only. Search the target Kestra for executions in the namespace
      that are still RUNNING and started before the stuck threshold.
    uri: "{{ inputs.kestra_url }}/api/v1/main/executions/search"
    method: GET
    contentType: application/json
    options:
      auth:
        type: BASIC
        username: "{{ inputs.kestra_user }}"
        password: "{{ inputs.kestra_password }}"
    params:
      filters[state][EQUALS]: RUNNING
      filters[namespace][EQUALS]: "{{ inputs.namespace }}"
      filters[startDate][LESS_THAN_OR_EQUAL_TO]: "{{ render(vars.cutoff) }}"
      size: "200"

  - id: select
    type: io.kestra.plugin.core.output.OutputValues
    description: Drop this reaper's own execution and flow, apply the flow filter
      and the exclusion list, then keep the stuck executions to act on.
    values:
      victims: "{{ outputs.detect.body | jq(render(vars.select_program)) | first |
        toJson }}"

  - id: counts
    type: io.kestra.plugin.core.output.OutputValues
    values:
      victim_ids: "{{ outputs.select.values.victims | jq('[.[].id]') | first | toJson }}"
      victim_count: "{{ outputs.select.values.victims | jq('length') | first }}"
      detail: "{{ outputs.select.values.victims | jq('map(.flowId + \" \" + .id + \"
        since \" + .start) | join(\"; \")') | first }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Reaper on {{ inputs.kestra_url }} namespace {{ inputs.namespace }}: {{
      outputs.detect.body | jq('.total') | first }} execution(s) stuck RUNNING
      over {{ inputs.max_running_minutes }} minute(s), {{
      outputs.counts.values.victim_count }} selected after exclusions. {{
      outputs.counts.values.detail }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.counts.values.victim_count | number) > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Kestra reaper on ' ~ inputs.kestra_url ~ ' namespace ' ~ inputs.namespace ~ ': ' ~ outputs.counts.values.victim_count ~ ' execution(s) stuck RUNNING over ' ~ inputs.max_running_minutes ~ ' min (' ~ outputs.counts.values.detail ~ ').' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Killing them behind approval.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ' dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: ceiling
            type: io.kestra.plugin.core.flow.If
            condition: "{{ (outputs.counts.values.victim_count | number) > inputs.max_kills
              }}"
            then:
              - id: too_many
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "Refusing to kill {{ outputs.counts.values.victim_count }}
                  executions, over the max_kills ceiling of {{ inputs.max_kills
                  }}. Investigate the cluster, then raise max_kills if the kill
                  is right."
            else:
              - id: approval
                type: io.kestra.plugin.core.flow.Pause
                description: Hold for a human to approve the kill. No answer within the window
                  keeps everything.
                pauseDuration: PT1H
                behavior: RESUME
                onResume:
                  - id: approved
                    type: BOOL
                    required: false
                    description: Set true to kill the listed executions. Anything else keeps them.
                  - id: note
                    type: STRING
                    required: false
                    description: Optional reason, included in the confirmation.

              - id: approved_gate
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.approval.onResume.approved ?? false }}"
                then:
                  - id: reap
                    type: io.kestra.plugin.core.flow.Loop
                    description: One kill at a time, so a slow API call cannot pile concurrent kills
                      onto the target.
                    concurrencyLimit: 1
                    values: "{{ outputs.counts.values.victim_ids }}"
                    tasks:
                      - id: kill
                        type: io.kestra.plugin.kestra.executions.Kill
                        allowFailure: true
                        kestraUrl: "{{ inputs.kestra_url }}"
                        executionId: "{{ item.value }}"
                        auth:
                          username: "{{ inputs.kestra_user }}"
                          password: "{{ inputs.kestra_password }}"

                  - id: settle
                    type: io.kestra.plugin.core.flow.Sleep
                    duration: PT5S

                  - id: reverify
                    type: io.kestra.plugin.core.http.Request
                    description: Read the target again and keep the ids that are still RUNNING.
                    uri: "{{ inputs.kestra_url }}/api/v1/main/executions/search"
                    method: GET
                    contentType: application/json
                    options:
                      auth:
                        type: BASIC
                        username: "{{ inputs.kestra_user }}"
                        password: "{{ inputs.kestra_password }}"
                    params:
                      filters[state][EQUALS]: RUNNING
                      filters[namespace][EQUALS]: "{{ inputs.namespace }}"
                      size: "200"

                  - id: verify_counts
                    type: io.kestra.plugin.core.output.OutputValues
                    values:
                      still_ids: "{{ outputs.reverify.body | jq('[.results[].id] as $run | ' ~
                        outputs.counts.values.victim_ids ~ ' - (' ~
                        outputs.counts.values.victim_ids ~ ' - $run)') | first |
                        toJson }}"
                      still_count: "{{ outputs.reverify.body | jq('[.results[].id] as $run | (' ~
                        outputs.counts.values.victim_ids ~ ' - (' ~
                        outputs.counts.values.victim_ids ~ ' - $run)) | length')
                        | first }}"

                  - id: verify_gate
                    type: io.kestra.plugin.core.flow.If
                    condition: "{{ (outputs.verify_counts.values.still_count | number) > 0 }}"
                    then:
                      - id: kill_failed
                        type: io.kestra.plugin.core.execution.Fail
                        errorMessage: "Kill did not land: {{ outputs.verify_counts.values.still_count }}
                          targeted execution(s) still RUNNING ({{
                          outputs.verify_counts.values.still_ids }}). The API
                          user may lack permission to kill. A worker may also
                          have re-queued them."
                    else:
                      - id: reaped
                        type: io.kestra.plugin.core.log.Log
                        message: "Killed and verified {{ outputs.counts.values.victim_count }} stuck
                          execution(s): {{ outputs.counts.values.victim_ids }}.
                          None of them is RUNNING any more."
                else:
                  - id: declined
                    type: io.kestra.plugin.core.log.Log
                    message: "Approval not granted, so nothing was killed. {{
                      outputs.counts.values.victim_count }} execution(s) are
                      still RUNNING."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "No execution in namespace {{ inputs.namespace }} has been RUNNING over
          {{ inputs.max_running_minutes }} minute(s)."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Kestra stuck execution reaper failed on ' ~ inputs.kestra_url ~ ' (execution ' ~ execution.id ~ '). The target may be unreachable or the API user may be wrong. Check the logs.'} | toJson }}

triggers:
  - id: hourly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode.
    cron: "15 * * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

When a worker dies mid task, its execution can sit in RUNNING for ever. Nothing fails and no alert fires. The execution keeps holding a concurrency slot, so the next scheduled run queues behind a job that will never finish. A failure alert cannot catch this, because the execution never failed. This flow finds those stuck executions. It is the piece most monitors leave out: it does the kill, behind an approval, then proves the kill landed.

It talks to the Kestra API of whichever instance you point kestra_url at. The default is the cluster the flow runs on, so it reaps itself.

This blueprint was created by zkasuran.

Safety

Killing executions is destructive, so the scope is narrow by default and the kill is gated.

It only looks at one namespace. The flow_id input can narrow it to a single flow.
It never kills its own execution or any other run of this reaper flow.
exclude_flows lists flows that are allowed to run long.
A kill runs only when all of these hold: auto_remediate is true, dry_run is false, the count is at or under max_kills, a human approves the Pause. Any missing step keeps everything.
After the kills it reads the API again and fails the run if any targeted execution is still RUNNING.

How it works

detect (io.kestra.plugin.core.http.Request, read-only) calls GET /api/v1/main/executions/search with filters[state][EQUALS]=RUNNING, filters[namespace][EQUALS] and filters[startDate][LESS_THAN_OR_EQUAL_TO] set to now minus the threshold. The modern filters[...] form is used on purpose: the flat state=/namespace= form is ignored on 2.0.5 and returns everything, which is the wrong answer for a kill.
select and counts drop this reaper's own flow and execution, apply the flow_id filter and exclude_flows, then emit the ids and a one-line detail of what is stuck.
summary logs how many are stuck and how many were selected.
route (io.kestra.plugin.core.flow.If):
- nothing stuck: clean logs it.
- stuck found: alert posts to Slack.
  - remediate_gate runs only when auto_remediate is true and dry_run is false.
  - ceiling fails the run rather than killing when more than max_kills are stuck.
  - approval (io.kestra.plugin.core.flow.Pause) waits for a human. No answer keeps everything.
  - reap (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) kills each id with io.kestra.plugin.kestra.executions.Kill.
  - reverify reads the API again and verify_gate fails if any targeted id is still RUNNING.
errors posts to Slack when the API cannot be reached.
hourly (io.kestra.plugin.core.trigger.Schedule), shipped disabled and in safe mode.

What you get

The executions in a namespace that have been RUNNING longer than you allow, with the flow id and the time each started.
A kill that is scoped, approved and capped, so it cannot run away.
A second read of the API as proof, so a kill that silently failed shows up as a failed run rather than a false all-clear.

Who it's for

Platform teams whose workers crash or get evicted and leave executions hanging.
Anyone who has seen a schedule stop firing because a dead run still holds the concurrency slot.
Teams who want the remediation step a plain stuck-execution monitor stops short of.

Why orchestrate this with Kestra

The reaper is itself a Kestra flow, so it needs no extra infrastructure to watch Kestra. It runs on a schedule, reports to Slack, holds the kill behind an approval, kills one execution at a time, then keeps a full execution history of every decision. Point kestra_url at any instance to reap a remote cluster from one place.

Tested end to end

On Kestra 2.0.5 OSS. To keep real executions safe the reaper pointed at a second Kestra 2.0.5 as the target, with a local HTTP sink standing in for Slack. The target held one throwaway flow whose only task sleeps for 30 minutes, so every run of it sits in RUNNING. I seeded several and drove each path. Before and after counts come from the target's own executions API, not only the flow log.

Run	Inputs	Result
Clean	threshold 120 min	0 stuck, clean branch, run SUCCESS
Detect and report	threshold 0, `dry_run: true`	4 stuck found and listed, Slack alert sent, nothing killed, the 4 still RUNNING afterwards
Remediate	threshold 0, `auto_remediate: true`, `dry_run: false`	4 found, Pause approved, 4 killed, the re-query found 0 still RUNNING, the target API confirmed all 4 KILLED. RUNNING went from 4 to 0
Ceiling	threshold 0, `max_kills: 2` with 3 stuck	run failed before any kill, the 3 left RUNNING
Re-verify gate	kill step disabled in a copy	the 3 stayed RUNNING, the re-query caught them, the run failed instead of reporting success
Target down	wrong port	detect failed, the errors block posted to Slack

Prerequisites

A Kestra API user that can search and kill executions in the namespace.
kestra_url reachable from the Kestra worker. The default http://localhost:8080 works when the flow reaps the cluster it runs on.

Secrets

KESTRA_PASSWORD: the API password, used as the default for the kestra_password input.
SLACK_WEBHOOK_URL: used by alert and the errors block.
In Kestra OSS, provide each one as a base64-encoded environment variable prefixed SECRET_.

Inputs

Input	Default	Purpose
`kestra_url`	`http://localhost:8080`	Target Kestra API, the cluster to reap
`namespace`	`company.team`	The one namespace considered
`flow_id`	empty	Narrow to a single flow, empty means every flow in the namespace
`max_running_minutes`	`30`	A run older than this while RUNNING is stuck
`exclude_flows`	`[]`	Flows allowed to run long
`max_kills`	`25`	Refuse to kill more than this in one run
`kestra_user`	`admin@kestra.io`	API username
`kestra_password`	`{{ secret('KESTRA_PASSWORD') }}`	API password
`auto_remediate`	`false`	Allow the kill
`dry_run`	`true`	Must be `false` for a kill to run

Quick start

Add the KESTRA_PASSWORD and SLACK_WEBHOOK_URL secrets.
Set namespace, then point kestra_user at a user that can kill executions.
Run it once with the defaults and read the list of stuck executions.
When you want it to act, run with auto_remediate: true and dry_run: false, then approve the Pause.
Enable the hourly trigger, which always runs in safe mode.

How to extend

Open an incident ticket per stuck execution, instead of or alongside the Slack alert.
Restart the owning flow after a kill, so the work actually gets done.
Loop over several kestra_url values to reap more than one cluster.
Match the namespace by prefix to cover child namespaces.

Reap Kestra executions stuck in RUNNING

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Reap Kestra executions stuck in RUNNING

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Copacetic Container Vulnerability Hotpatch Engine

Kubernetes CoreDNS Request Latency Spike and Upstream RTT Sentinel

HashiCorp Consul Raft Leader Stability and Quorum Sentinel

Corosync Totem Token Timeout and Quorum Loss Sentinel

Audit the Operating System Inside Your Container Images for End of Life

Scheduled Container Registry Vulnerability Audit with Trivy and Slack Alerts

New to Kestra?