Restart failed Kestra executions on the fixed flow revision with a canary

id: kestra-incident-fix-forward-replay
namespace: company.team

concurrency:
  limit: 1
  behavior: FAIL

inputs:
  - id: flow_id
    type: STRING
    required: true
    displayName: Flow that was fixed
    description: The flow whose failed executions should run again. They restart on
      its latest revision.

  - id: flow_namespace
    type: STRING
    displayName: Flow namespace
    defaults: company.team

  - id: failed_task_id
    type: STRING
    displayName: Task that failed in the incident
    description: One incident is the executions that failed at this task. Leave
      empty to list the incidents of the flow and stop.
    defaults: ""

  - id: since_hours
    type: INT
    displayName: Look back (hours)
    defaults: 72
    min: 1
    max: 720

  - id: canary_size
    type: INT
    displayName: Canary executions
    description: How many executions restart first. The rest start only if every
      canary succeeds.
    defaults: 1
    min: 1
    max: 10

  - id: max_replays
    type: INT
    displayName: Most executions per run
    defaults: 20
    min: 1
    max: 100

  - id: kestra_url
    type: STRING
    displayName: Kestra URL
    description: The instance that runs the flow. Calls use the KESTRA_API_USERNAME
      and KESTRA_API_PASSWORD secrets.
    defaults: http://localhost:8080

  - id: auto_remediate
    type: BOOL
    displayName: Allow restarts
    description: Restarts happen only when this is true, dry_run is false and a
      person approves.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Report the plan and change nothing.
    defaults: true

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Post a failure alert to Slack. Needs the SLACK_WEBHOOK_URL secret.
    defaults: false

tasks:
  - id: flow
    type: io.kestra.plugin.core.http.Request
    description: Read the flow. Its current revision is the one the failed
      executions restart on.
    uri: "{{ inputs.kestra_url }}/api/v1/main/flows/{{ inputs.flow_namespace }}/{{
      inputs.flow_id }}"
    options:
      auth:
        {
          type: BASIC,
          username: "{{ secret('KESTRA_API_USERNAME') }}",
          password: "{{ secret('KESTRA_API_PASSWORD') }}"
        }

  - id: window
    type: io.kestra.plugin.kestra.executions.Query
    description: >-
      Every execution of the flow in the look-back window, in any state. A
      restarted execution carries the labels system.restarted and
      system.correlationId (the id it was restarted from), so a failed execution
      that was restarted before is recognised without a second query.
    kestraUrl: "{{ inputs.kestra_url }}"
    tenantId: main
    auth:
      {
        username: "{{ secret('KESTRA_API_USERNAME') }}",
        password: "{{ secret('KESTRA_API_PASSWORD') }}"
      }
    namespace: "{{ inputs.flow_namespace }}"
    flowId: "{{ inputs.flow_id }}"
    timeRange: "PT{{ inputs.since_hours }}H"
    fetchType: FETCH
    page: 1
    size: 1000

  - id: scan
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      A restart creates a new execution whose parentId is the execution it was
      restarted from, so the restarts of one run of the flow form a chain and
      only the end of a chain can still need a restart. Candidates are the chain
      ends that FAILED on an older revision, oldest first. A restarted tip that
      failed again on the latest revision stops the run, so a bad fix is not
      retried in a loop.
    values:
      latest: "{{ fromJson(outputs.flow.body).revision }}"
      executions: "{{ outputs.window.rows | jq('[flatten[]] | length') | first }}"
      failed_replays: >-
        {{ {'rows': outputs.window.rows, 'latest':
        fromJson(outputs.flow.body).revision} | jq('
          .latest as $l | (.rows | flatten) as $r | ([$r[] | .parentId // empty]) as $parents
          | [$r[] | select(.id as $i | $parents | index($i) | not)
            | select(.state.current == "FAILED" and .flowRevision == $l and any(.labels[]?; .key == "system.restarted")) | .id]') | first | toJson }}
      candidates: >-
        {{ {'rows': outputs.window.rows, 'latest':
        fromJson(outputs.flow.body).revision} | jq('
          def corr: ((.labels // []) | map(select(.key == "system.correlationId")) | first | .value) // .id;
          def ts: (.state.startDate | sub("Z$"; "") | split(".")) as $p | $p[0] + "." + ((($p[1] // "0") + "000000000")[0:9]);
          .latest as $l | (.rows | flatten) as $r | ([$r[] | .parentId // empty]) as $parents
          | [$r[] | select(.id as $i | $parents | index($i) | not)
            | select(.state.current == "FAILED" and .flowRevision < $l) | {id, corr: corr, start: ts}]
          | sort_by(.start) | .[:100]') | first | toJson }}

  - id: guards
    type: io.kestra.plugin.core.flow.If
    description: Refuse a window the query cannot show whole. Refuse a fix that
      already failed once.
    condition: "{{ outputs.scan.values.executions == '1000' or
      fromJson(outputs.scan.values.failed_replays) | length > 0 }}"
    then:
      - id: refuse
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: >-
          {{ outputs.scan.values.executions == '1000'
            ? 'The window holds 1000 executions or more, so the query cannot show it whole. Lower since_hours.'
            : 'A restarted execution already failed on revision ' ~ outputs.scan.values.latest ~ ' (' ~ fromJson(outputs.scan.values.failed_replays) | join(', ') ~ '). Fix the flow again and deploy a new revision first.' }}

  - id: has_candidates
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.scan.values.candidates) | length > 0 }}"
    then:
      - id: read
        type: io.kestra.plugin.core.flow.Loop
        description: The query result has no task runs, so each candidate is read once
          to see which task failed.
        values: "{{ outputs.scan.values.candidates }}"
        concurrencyLimit: 5
        transmitFailed: false
        tasks:
          - id: execution
            type: io.kestra.plugin.core.http.Request
            uri: "{{ inputs.kestra_url }}/api/v1/main/executions/{{ fromJson(item.value).id
              }}"
            options:
              auth:
                {
                  type: BASIC,
                  username: "{{ secret('KESTRA_API_USERNAME') }}",
                  password: "{{ secret('KESTRA_API_PASSWORD') }}"
                }
        outputs:
          - id: row
            type: JSON
            value: >-
              {{ {'id': fromJson(item.value).id, 'corr':
              fromJson(item.value).corr, 'start': fromJson(item.value).start,
                  'task': outputs.execution.body | jq('[.taskRunList[] | select(.state.current == "FAILED")] as $f | [$f[] | select(.id as $i | all($f[]; .parentTaskRunId != $i)) | .taskId] | first') | first} | toJson }}

      - id: incidents
        type: io.kestra.plugin.core.output.OutputValues
        values:
          groups: "{{ (outputs.read.outputs ?? []) | jq('[.[].outputs.row] |
            group_by(.task) | map({task: .[0].task, executions: length}) |
            sort_by(-.executions)') | first | toJson }}"
          selected: >-
            {{ {'rows': (outputs.read.outputs ?? []), 't':
            inputs.failed_task_id, 'n': inputs.max_replays} | jq('
              .t as $t | .n as $n | [.rows[].outputs.row | select(.task == $t) | {id, corr, start}] | sort_by(.start) | .[:$n]') | first | toJson }}

      - id: stages
        type: io.kestra.plugin.core.output.OutputValues
        description: The selected executions in start order. The first canary_size
          restart as canary, the rest as the wave.
        values:
          canary: "{{ {'ids': fromJson(outputs.incidents.values.selected), 'c':
            inputs.canary_size} | jq('.ids[:.c]') | first | toJson }}"
          wave: "{{ {'ids': fromJson(outputs.incidents.values.selected), 'c':
            inputs.canary_size} | jq('.ids[.c:]') | first | toJson }}"

      - id: plan
        type: io.kestra.plugin.core.log.Log
        message: >-
          Flow {{ inputs.flow_namespace }}.{{ inputs.flow_id }}, latest revision
          {{ outputs.scan.values.latest }}. {{
          fromJson(outputs.scan.values.candidates) | length }} failed
          execution(s) of older revisions have not been restarted. Incidents by
          failed task: {{ outputs.incidents.values.groups }}. {{
          inputs.failed_task_id == '' ? 'No failed_task_id given, nothing is
          selected.' : 'Selected for task ' ~ inputs.failed_task_id ~ ': ' ~
          (outputs.incidents.values.selected | jq('map(.id)') | first | toJson)
          ~ '.' }}

      - id: route
        type: io.kestra.plugin.core.flow.If
        condition: "{{ fromJson(outputs.incidents.values.selected) | length > 0 and
          inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: hold
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              {{ fromJson(outputs.incidents.values.selected) | length }}
              execution(s) of {{ inputs.flow_id }} restart on revision {{
              outputs.scan.values.latest }}, the first {{ inputs.canary_size }}
              as canary. Approve to start.
            pauseDuration: PT2H
            behavior: FAIL
            onResume:
              - id: approved
                type: BOOL
                defaults: false

          - id: apply
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.hold.onResume.approved ?? false }}"
            then:
              - id: same_revision
                type: io.kestra.plugin.core.http.Request
                description: The approval covered one revision. Stop if the flow was saved again
                  during the wait.
                uri: "{{ inputs.kestra_url }}/api/v1/main/flows/{{ inputs.flow_namespace }}/{{
                  inputs.flow_id }}"
                options:
                  auth:
                    {
                      type: BASIC,
                      username: "{{ secret('KESTRA_API_USERNAME') }}",
                      password: "{{ secret('KESTRA_API_PASSWORD') }}"
                    }

              - id: revision_changed
                type: io.kestra.plugin.core.flow.If
                condition: "{{ fromJson(outputs.same_revision.body).revision ~ '' !=
                  outputs.scan.values.latest }}"
                then:
                  - id: stop_changed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: "The flow moved from revision {{ outputs.scan.values.latest }} to
                      {{ fromJson(outputs.same_revision.body).revision }} during
                      the approval. Nothing was restarted. Run again."

              - id: restart_canary
                type: io.kestra.plugin.core.flow.Loop
                description: >-
                  One execution at a time. Each is read again right before its
                  restart: it must still be FAILED and must not have a restarted
                  twin yet, otherwise it is skipped and the next run reconciles
                  it.
                values: "{{ outputs.stages.values.canary }}"
                concurrencyLimit: 1
                tasks:
                  - id: current_canary
                    type: io.kestra.plugin.core.http.Request
                    uri: "{{ inputs.kestra_url }}/api/v1/main/executions/search"
                    params:
                      filters[namespace][EQUALS]: "{{ inputs.flow_namespace }}"
                      filters[flowId][EQUALS]: "{{ inputs.flow_id }}"
                      filters[labels][EQUALS][system.correlationId]: "{{ fromJson(item.value).corr }}"
                    options:
                      auth:
                        {
                          type: BASIC,
                          username: "{{ secret('KESTRA_API_USERNAME') }}",
                          password: "{{ secret('KESTRA_API_PASSWORD') }}"
                        }
                  - id: check_canary
                    type: io.kestra.plugin.core.output.OutputValues
                    values:
                      eligible: >-
                        {{ {'rows':
                        fromJson(outputs.current_canary.body).results, 'id':
                        fromJson(item.value).id} | jq('
                          .id as $i | .rows as $r | ([$r[] | select(.id == $i)] | first) as $me
                          | $me != null and $me.state.current == "FAILED" and all($r[]; .parentId != $i)') | first }}
                  - id: go_canary
                    type: io.kestra.plugin.core.flow.If
                    condition: "{{ outputs.check_canary.values.eligible == 'true' }}"
                    then:
                      - id: call_canary
                        type: io.kestra.plugin.core.http.Request
                        method: POST
                        uri: "{{ inputs.kestra_url }}/api/v1/main/executions/{{ fromJson(item.value).id
                          }}/actions/restart?revision={{
                          outputs.scan.values.latest }}"
                        options:
                          auth:
                            {
                              type: BASIC,
                              username: "{{ secret('KESTRA_API_USERNAME') }}",
                              password: "{{ secret('KESTRA_API_PASSWORD') }}"
                            }
                outputs:
                  - id: row
                    type: JSON
                    value: "{{ {'id': fromJson(item.value).id, 'corr': fromJson(item.value).corr,
                      'start': fromJson(item.value).start, 'restarted':
                      outputs.check_canary.values.eligible == 'true'} | toJson
                      }}"

              - id: wait_canary
                type: io.kestra.plugin.core.flow.LoopUntil
                description: Wait until every restarted execution of this stage has ended.
                condition: >-
                  {{ {'rows': fromJson(outputs.poll_canary.body).results,
                  'items': (outputs.restart_canary.outputs ?? []) |
                  jq('[.[].outputs.row | select(.restarted)]') | first} | jq('
                    .rows as $r | all(.items[]; . as $x | ([$r[] | select(.parentId == $x.id)] | first) as $t
                      | $t != null and ($t.state.current as $s | ["SUCCESS","WARNING","FAILED","KILLED","CANCELLED"] | any(. == $s)))') | first }}
                checkFrequency:
                  interval: PT5S
                  maxDuration: PT30M
                failOnMaxReached: false
                tasks:
                  - id: poll_canary
                    type: io.kestra.plugin.core.http.Request
                    uri: "{{ inputs.kestra_url }}/api/v1/main/executions/search"
                    params:
                      size: "1000"
                      filters[namespace][EQUALS]: "{{ inputs.flow_namespace }}"
                      filters[flowId][EQUALS]: "{{ inputs.flow_id }}"
                      filters[labels][EQUALS][system.restarted]: "true"
                    options:
                      auth:
                        {
                          type: BASIC,
                          username: "{{ secret('KESTRA_API_USERNAME') }}",
                          password: "{{ secret('KESTRA_API_PASSWORD') }}"
                        }

              - id: verdict_canary
                type: io.kestra.plugin.core.output.OutputValues
                description: Read the restarted executions of this stage once more. Only SUCCESS
                  and WARNING count as good.
                values:
                  restarted: "{{ (outputs.restart_canary.outputs ?? []) | jq('[.[].outputs.row |
                    select(.restarted)] | length') | first }}"
                  twins: >-
                    {{ {'rows': fromJson(outputs.poll_canary.body).results,
                    'items': (outputs.restart_canary.outputs ?? []) |
                    jq('[.[].outputs.row | select(.restarted)]') | first} | jq('
                      .rows as $r | [.items[] | . as $x | {original: $x.id, twin: ([$r[] | select(.parentId == $x.id)] | first | {id, state: .state.current})}]') | first | toJson }}
                  succeeded: >-
                    {{ {'rows': fromJson(outputs.poll_canary.body).results,
                    'items': (outputs.restart_canary.outputs ?? []) |
                    jq('[.[].outputs.row | select(.restarted)]') | first} | jq('
                      .rows as $r | [.items[] | . as $x | ([$r[] | select(.parentId == $x.id)] | first) | select(. != null and (.state.current == "SUCCESS" or .state.current == "WARNING"))] | length') | first }}

              - id: gate_canary
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.verdict_canary.values.restarted == '0' or
                  outputs.verdict_canary.values.succeeded !=
                  outputs.verdict_canary.values.restarted }}"
                then:
                  - id: stop_canary
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      {{ outputs.verdict_canary.values.restarted == '0' ? 'No
                      canary could be restarted, because the executions changed
                      after the plan. Nothing else was restarted. Run again.' :
                      'The canary did not succeed on revision ' ~
                      outputs.scan.values.latest ~ ': ' ~
                      outputs.verdict_canary.values.twins ~ '. The rest of the
                      incident was not restarted.' }}
              - id: restart_wave
                type: io.kestra.plugin.core.flow.Loop
                description: >-
                  One execution at a time. Each is read again right before its
                  restart: it must still be FAILED and must not have a restarted
                  twin yet, otherwise it is skipped and the next run reconciles
                  it.
                values: "{{ outputs.stages.values.wave }}"
                concurrencyLimit: 1
                tasks:
                  - id: current_wave
                    type: io.kestra.plugin.core.http.Request
                    uri: "{{ inputs.kestra_url }}/api/v1/main/executions/search"
                    params:
                      filters[namespace][EQUALS]: "{{ inputs.flow_namespace }}"
                      filters[flowId][EQUALS]: "{{ inputs.flow_id }}"
                      filters[labels][EQUALS][system.correlationId]: "{{ fromJson(item.value).corr }}"
                    options:
                      auth:
                        {
                          type: BASIC,
                          username: "{{ secret('KESTRA_API_USERNAME') }}",
                          password: "{{ secret('KESTRA_API_PASSWORD') }}"
                        }
                  - id: check_wave
                    type: io.kestra.plugin.core.output.OutputValues
                    values:
                      eligible: >-
                        {{ {'rows': fromJson(outputs.current_wave.body).results,
                        'id': fromJson(item.value).id} | jq('
                          .id as $i | .rows as $r | ([$r[] | select(.id == $i)] | first) as $me
                          | $me != null and $me.state.current == "FAILED" and all($r[]; .parentId != $i)') | first }}
                  - id: go_wave
                    type: io.kestra.plugin.core.flow.If
                    condition: "{{ outputs.check_wave.values.eligible == 'true' }}"
                    then:
                      - id: call_wave
                        type: io.kestra.plugin.core.http.Request
                        method: POST
                        uri: "{{ inputs.kestra_url }}/api/v1/main/executions/{{ fromJson(item.value).id
                          }}/actions/restart?revision={{
                          outputs.scan.values.latest }}"
                        options:
                          auth:
                            {
                              type: BASIC,
                              username: "{{ secret('KESTRA_API_USERNAME') }}",
                              password: "{{ secret('KESTRA_API_PASSWORD') }}"
                            }
                outputs:
                  - id: row
                    type: JSON
                    value: "{{ {'id': fromJson(item.value).id, 'corr': fromJson(item.value).corr,
                      'start': fromJson(item.value).start, 'restarted':
                      outputs.check_wave.values.eligible == 'true'} | toJson }}"

              - id: wait_wave
                type: io.kestra.plugin.core.flow.LoopUntil
                description: Wait until every restarted execution of this stage has ended.
                condition: >-
                  {{ {'rows': fromJson(outputs.poll_wave.body).results, 'items':
                  (outputs.restart_wave.outputs ?? []) | jq('[.[].outputs.row |
                  select(.restarted)]') | first} | jq('
                    .rows as $r | all(.items[]; . as $x | ([$r[] | select(.parentId == $x.id)] | first) as $t
                      | $t != null and ($t.state.current as $s | ["SUCCESS","WARNING","FAILED","KILLED","CANCELLED"] | any(. == $s)))') | first }}
                checkFrequency:
                  interval: PT5S
                  maxDuration: PT30M
                failOnMaxReached: false
                tasks:
                  - id: poll_wave
                    type: io.kestra.plugin.core.http.Request
                    uri: "{{ inputs.kestra_url }}/api/v1/main/executions/search"
                    params:
                      size: "1000"
                      filters[namespace][EQUALS]: "{{ inputs.flow_namespace }}"
                      filters[flowId][EQUALS]: "{{ inputs.flow_id }}"
                      filters[labels][EQUALS][system.restarted]: "true"
                    options:
                      auth:
                        {
                          type: BASIC,
                          username: "{{ secret('KESTRA_API_USERNAME') }}",
                          password: "{{ secret('KESTRA_API_PASSWORD') }}"
                        }

              - id: verdict_wave
                type: io.kestra.plugin.core.output.OutputValues
                description: Read the restarted executions of this stage once more. Only SUCCESS
                  and WARNING count as good.
                values:
                  restarted: "{{ (outputs.restart_wave.outputs ?? []) | jq('[.[].outputs.row |
                    select(.restarted)] | length') | first }}"
                  twins: >-
                    {{ {'rows': fromJson(outputs.poll_wave.body).results,
                    'items': (outputs.restart_wave.outputs ?? []) |
                    jq('[.[].outputs.row | select(.restarted)]') | first} | jq('
                      .rows as $r | [.items[] | . as $x | {original: $x.id, twin: ([$r[] | select(.parentId == $x.id)] | first | {id, state: .state.current})}]') | first | toJson }}
                  succeeded: >-
                    {{ {'rows': fromJson(outputs.poll_wave.body).results,
                    'items': (outputs.restart_wave.outputs ?? []) |
                    jq('[.[].outputs.row | select(.restarted)]') | first} | jq('
                      .rows as $r | [.items[] | . as $x | ([$r[] | select(.parentId == $x.id)] | first) | select(. != null and (.state.current == "SUCCESS" or .state.current == "WARNING"))] | length') | first }}

              - id: gate_wave
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.verdict_wave.values.succeeded !=
                  outputs.verdict_wave.values.restarted }}"
                then:
                  - id: stop_wave
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      Not every restarted execution succeeded on revision {{
                      outputs.scan.values.latest }}: {{
                      outputs.verdict_wave.values.twins }}.
              - id: replayed
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Restarted {{ outputs.verdict_canary.values.restarted }} canary
                  and {{ outputs.verdict_wave.values.restarted }} wave
                  execution(s) of {{ inputs.flow_id }} on revision {{
                  outputs.scan.values.latest }}. The restarted executions are {{
                  outputs.verdict_canary.values.twins }} and {{
                  outputs.verdict_wave.values.twins }}. The failed originals
                  keep their FAILED state.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: Not approved. Nothing was restarted.
        else:
          - id: no_restart
            type: io.kestra.plugin.core.log.Log
            message: >-
              {{ fromJson(outputs.incidents.values.selected) | length == 0
                ? 'No failed execution is selected' ~ (inputs.failed_task_id == '' ? ': give failed_task_id to pick an incident.' : ' for task ' ~ inputs.failed_task_id ~ '.') ~ ' Nothing was restarted.'
                : 'Plan only. Set auto_remediate to true and dry_run to false, then approve, to restart the selected executions.' }}
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: No failed execution of an older revision is waiting. Nothing to
          restart.

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    runIf: "{{ inputs.notify_slack }}"
    description: A refused window, a failed canary or a failed restart must reach a person.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Incident replay for ' ~ inputs.flow_namespace ~ '.' ~ inputs.flow_id ~ ' stopped (execution ' ~ execution.id ~ '). Open the execution for the reason.'} | toJson }}

triggers:
  - id: weekday_triage
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "0 7 * * 1-5"
    description: Lists the incidents of one flow every weekday. Disabled and pinned
      to the safe mode.
    inputs:
      flow_id: orders-nightly
      dry_run: true
      auto_remediate: false

After a bad deploy a flow leaves dozens of failed executions behind. You fix the flow and save a new revision while the failed runs still sit at the old one. This flow finds them, groups them by the task that failed and restarts one group on the latest revision. A canary goes first. The rest only starts if the canary succeeded on the new revision.

How it works

flow reads the latest revision. window lists the executions of the flow in the look-back window. scan keeps the end of each restart chain that FAILED on an older revision.
read fetches each candidate once, because the search result has no task runs. incidents groups them by the task that failed and selects the oldest max_replays of the task in failed_task_id.
With the defaults the run logs the plan and stops. With auto_remediate true and dry_run false, hold (Pause) waits for an approval.
same_revision reads the flow again. If it was saved during the wait, nothing is restarted. Every restart reads its execution first and is skipped when the execution is no longer FAILED or already has a restarted twin.
restart_canary calls POST /executions/{id}/actions/restart?revision=N. The twin continues from the failed task on revision N and the original stays FAILED. wait_canary polls until the twins end. gate_canary fails the run unless every canary succeeded.
restart_wave does the same for the rest of the incident, one execution at a time.

What you get

A list of incidents with their counts, a restart of one incident that cannot touch another and a stop at the first sign that the fix does not work. A canary that fails again on the new revision ends the run. The next run refuses to start while that failed twin is the newest execution of its chain.

Who it's for

Teams that run Kestra for pipelines and fix a bug after many runs have failed.

Why orchestrate this with Kestra

The restart, the wait and the verdict all use the Kestra API, so one run holds the plan, the approval and the proof. Pause carries the approval. concurrency with FAIL keeps a second run from restarting the same executions.

Tested end to end

13 cases on Kestra 2.0.6 and 2.0.5 with plugin-kestra 2.1.1 and 2.1.0. Covered: nothing to restart, a list of two incidents, a plan with seven candidates of two incidents, a declined approval, a canary and a wave that restart five executions, a rerun, a canary that fails on the new revision, a rerun after that failure, an execution restarted during the approval, a flow saved during the approval, a second run during the approval, a fix in two steps where only the failed restart is restarted again and a run killed after the canary. Each case checks the twins and the originals through the API.

Prerequisites

A Kestra user that can read and restart executions of the flow.

Secrets

KESTRA_API_USERNAME and KESTRA_API_PASSWORD for the API. SLACK_WEBHOOK_URL only when notify_slack is true.

Inputs

flow_id is required. failed_task_id picks the incident. canary_size defaults to 1 and max_replays to 20. since_hours defaults to 72. The window query shows 1000 executions at most. The run refuses a window that holds that many. auto_remediate defaults to false and dry_run to true.

Quick start

Run with flow_id only and read the incidents in the plan log.
Set failed_task_id. The run shows the executions it would restart and stops.
Set auto_remediate true and dry_run false, then approve the Pause.

How to extend

Add a task after replayed that posts the twin ids to your incident channel. Raise canary_size for a bigger incident. To restart one execution at a given revision without a canary, call the restart endpoint directly.

Replay one incident's failed executions on the fixed revision with a canary

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?

Replay one incident's failed executions on the fixed revision with a canary

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

Kestra instance retention policy with dry run, approval and audit CSV

Release Kestra Flows from Git with Smoke Tests and Automatic Rollback

Kestra Schedule Trigger Health Sentinel

Nightly Kestra flows backup to S3 with a SHA-256 drift manifest per namespace

Promote Kestra flows from staging to production with approval, snapshot and rollback

Move a Kestra namespace to another instance with a drained cutover and rollback

New to Kestra?