Triage and Replay Failed Kestra Executions

id: kestra-failed-execution-dlq-replay-engine
namespace: company.team

description: |
  Find failed workflow executions in a namespace, alert Slack with the details,
  and optionally restart them one by one behind a Pause approval gate.

inputs:
  - id: kestra_url
    type: STRING
    displayName: Target Kestra API URL
    description: Base URL of the Kestra cluster whose failed executions are triaged
      and replayed.
    defaults: "http://localhost:8080"

  - id: namespace
    type: STRING
    displayName: Target Namespace
    description: Namespace scope to inspect for failed executions.
    defaults: company.team

  - id: max_replays
    type: INT
    displayName: Maximum Replays Per Run
    description: Upper bound on the number of failed executions to restart in a
      single execution.
    defaults: 5

  - id: auto_replay
    type: BOOL
    displayName: Enable Auto-Replay
    description: When true and dry run is false, prompts for human approval and
      restarts candidates.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry Run Mode
    description: When true, discovers and reports failed executions without
      triggering restarts.
    defaults: true

  - id: exclude_flows
    type: ARRAY
    itemType: STRING
    displayName: Excluded Flow IDs
    description: List of flow IDs to skip during triage, preventing automated
      restarts of known flaky tasks.
    defaults: []

  - id: api_token
    type: STRING
    displayName: Kestra API Token / Authorization Header
    description: Optional bearer token or credentials for authenticating against
      secured Kestra clusters. Leave empty for unauthenticated local instances.
    defaults: ""

tasks:
  - id: query_failed_executions
    type: io.kestra.plugin.core.http.Request
    description: Queries the Kestra 2.0 REST API for executions currently in FAILED
      state within the namespace.
    uri: "{{ inputs.kestra_url }}/api/v1/executions/search"
    method: GET
    headers: "{{ (inputs.api_token is not empty) ? {'Authorization':
      (inputs.api_token contains ' ' ? inputs.api_token : 'Bearer ' ~
      inputs.api_token)} : {} }}"
    params:
      "filters[state][EQUALS]": "FAILED"
      "filters[namespace][EQUALS]": "{{ inputs.namespace }}"
      "size": "25"

  - id: triage_failures
    type: io.kestra.plugin.core.output.OutputValues
    description: Evaluates API results, excludes self-runs and ignored flows, and
      caps candidates to max_replays.
    values:
      candidates: >-
        {{
          outputs.query_failed_executions.body
          | jq('[.results[]? | select(.id != "' ~ execution.id ~ '" and .flowId != "' ~ flow.id ~ '") | select(.flowId as $f | ' ~ (inputs.exclude_flows | toJson) ~ ' | index($f) | not)] | .[0:' ~ inputs.max_replays ~ '] | map({id: .id, flowId: .flowId, namespace: .namespace, startDate: .state.startDate})')
        }}

  - id: check_candidate_count
    type: io.kestra.plugin.core.flow.If
    description: Branches based on whether eligible failed executions were discovered.
    condition: "{{ (outputs.triage_failures.values.candidates | length) > 0 }}"
    then:
      - id: notify_triage_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Sends failure details to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          :warning: *Kestra Failed Executions Alert*
          *Namespace:* `{{ inputs.namespace }}`
          *Failed Candidates Found:* `{{ outputs.triage_failures.values.candidates | length }}` (capped at {{ inputs.max_replays }})
          *Mode:* `{{ inputs.auto_replay ? (inputs.dry_run ? 'Dry Run (Simulation)' : 'Active Replay Armed') : 'Audit Only' }}`
          *Executions:*
          {{ outputs.triage_failures.values.candidates | jq('.[] | "• `" + .flowId + "` (ID: `" + .id + "`, Started: " + .startDate + ")"') | join('\n') }}

      - id: evaluate_replay_permission
        type: io.kestra.plugin.core.flow.If
        description: Checks whether auto_replay is enabled and dry run is disabled.
        condition: "{{ inputs.auto_replay and not inputs.dry_run }}"
        then:
          - id: approve_replay
            type: io.kestra.plugin.core.flow.Pause
            description: Pauses for approval before restarting executions.
            onResume:
              - id: approved
                type: BOOL
                description: Confirm restart of all triaged candidate executions.
                defaults: true

          - id: restart_candidates
            type: io.kestra.plugin.core.flow.Loop
            description: Restarts failed executions one at a time.
            values: "{{ outputs.triage_failures.values.candidates }}"
            concurrencyLimit: 1
            tasks:
              - id: call_restart_api
                type: io.kestra.plugin.core.http.Request
                description: Calls Kestra restart endpoint to re-execute failed task runs.
                uri: "{{ inputs.kestra_url }}/api/v1/executions/{{ item.value.id
                  }}/actions/restart"
                method: POST
                headers: "{{ (inputs.api_token is not empty) ? {'Authorization':
                  (inputs.api_token contains ' ' ? inputs.api_token : 'Bearer '
                  ~ inputs.api_token)} : {} }}"

          - id: settling_delay
            type: io.kestra.plugin.core.flow.Sleep
            description: Waits 15 seconds for restarted executions to initialize.
            duration: PT15S

          - id: verify_post_restart_states
            type: io.kestra.plugin.core.http.Request
            description: Queries the executions API to check if any runs are still failed.
            uri: "{{ inputs.kestra_url }}/api/v1/executions/search"
            method: GET
            headers: "{{ (inputs.api_token is not empty) ? {'Authorization':
              (inputs.api_token contains ' ' ? inputs.api_token : 'Bearer ' ~
              inputs.api_token)} : {} }}"
            params:
              "filters[state][EQUALS]": "FAILED"
              "filters[namespace][EQUALS]": "{{ inputs.namespace }}"
              "size": "25"

          - id: notify_resolution_slack
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Sends restart summary to Slack with remaining failure count.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: |
              :white_check_mark: *Kestra Restart Summary*
              *Namespace:* `{{ inputs.namespace }}`
              *Restarted Count:* `{{ outputs.triage_failures.values.candidates | length }}`
              *Remaining Failures in Namespace:* `{{ outputs.verify_post_restart_states.body | jq('.total // 0') }}`
              *Verification:* Completed sequential replay run.
    else:
      - id: log_clean_state
        type: io.kestra.plugin.core.log.Log
        description: Logs when no failed executions are detected.
        message: "No failed executions found in namespace {{ inputs.namespace }}."

triggers:
  - id: hourly_failure_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodic hourly check. Disabled by default.
    cron: "0 * * * *"
    disabled: true

errors:
  - id: alert_on_flow_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts Slack if this workflow encounters an error.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :x: *Kestra Replay Flow Failed*
      *Execution:* `{{ execution.id }}`
      *Flow:* `{{ flow.id }}`
      *Namespace:* `{{ flow.namespace }}`
      Check task execution logs in Kestra UI.