Nightly digest of failed Kestra executions with their ERROR log lines per namespace

id: kestra-failed-execution-error-digest
namespace: company.team

description: |
  Nightly digest of what failed on a Kestra instance, through its own API: execution and failure counts per namespace
  for the window, and the ERROR lines of the recent failed executions collected into one JSON Lines file, with the
  first error of each run in Slack. Nothing is modified on the instance. The companion flow
  `kestra-instance-snapshot-and-drift-manifest` backs up the flows themselves.

labels:
  team: platform
  purpose: error-digest

inputs:
  - id: namespace_prefix
    type: STRING
    defaults: ""
    description: Only namespaces starting with this prefix are covered. Empty
      (default) means every namespace of the tenant.

  - id: exclude_namespaces
    type: JSON
    defaults: [ "system" ]
    description: Namespaces left out even when they match the prefix. `system` holds
      Kestra's own flows and is skipped by default.

  - id: window_hours
    type: INT
    defaults: 24
    description: Window, ending at the schedule date (or now for a manual run), for
      the counts and the digest. 1 to 720.

  - id: max_failed_per_namespace
    type: INT
    defaults: 10
    description: How many recent FAILED executions per namespace have their ERROR
      lines fetched. Each one costs one API call for its logs. 1 to 100.

  - id: notify
    type: BOOL
    defaults: true
    description: Post the digest to Slack when at least one execution failed in the
      window. The same text is always in `log_summary`.

  - id: kestra_api_url
    type: STRING
    defaults: http://localhost:8080
    description: Base URL of this Kestra instance's API, as seen from the worker,
      without a trailing slash.

variables:
  tenant: main
  state_key: kestra_error_digest

checks:
  - when: "{{ inputs.window_hours >= 1 and inputs.window_hours <= 720 }}"
    message: window_hours must be between 1 and 720 (30 days).
    style: ERROR
    behavior: BLOCK_EXECUTION
  - when: "{{ inputs.max_failed_per_namespace >= 1 and
      inputs.max_failed_per_namespace <= 100 }}"
    message: max_failed_per_namespace must be between 1 and 100, each failed
      execution costs one API call for its logs.
    style: ERROR
    behavior: BLOCK_EXECUTION
  - when: "{{ inputs.exclude_namespaces | jq('type == \"array\" and all(.[]; type ==
      \"string\")') | first }}"
    message: exclude_namespaces must be a JSON array of namespace names, for example
      ["system", "company.sandbox"].
    style: ERROR
    behavior: BLOCK_EXECUTION
  - when: "{{ {'p': inputs.namespace_prefix} | jq('.p | test(\"^[a-z0-9._-]*$\")') |
      first }}"
    message: namespace_prefix may only contain lowercase letters, digits, dots,
      dashes and underscores.
    style: ERROR
    behavior: BLOCK_EXECUTION
  - when: "{{ {'u': inputs.kestra_api_url | trim} | jq('.u |
      test(\"^https?://.+[^/]$\")') | first }}"
    message: kestra_api_url must be an http(s) URL without a trailing slash, for
      example http://localhost:8080.
    style: ERROR
    behavior: BLOCK_EXECUTION

concurrency:
  limit: 1
  behavior: CANCEL

triggers:
  - id: nightly
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "45 2 * * *"
    description: "Every night at 02:45, after the flows snapshot.
      `recoverMissedSchedules: LAST` writes one digest after a scheduler outage
      instead of one per missed night."
    recoverMissedSchedules: LAST

tasks:
  - id: discover_namespaces
    type: io.kestra.plugin.kestra.namespaces.List
    description: "Every namespace of the tenant under the prefix, including
      namespaces that only exist because a flow lives in them (`existingOnly:
      false`)."
    kestraUrl: "{{ inputs.kestra_api_url }}"
    tenantId: "{{ vars.tenant }}"
    auth:
      {
        username: "{{ secret('KESTRA_API_USERNAME') }}",
        password: "{{ secret('KESTRA_API_PASSWORD') }}"
      }
    prefix: "{{ inputs.namespace_prefix }}"
    existingOnly: false

  - id: scope
    type: io.kestra.plugin.core.output.OutputValues
    description: Namespaces to cover and the time window. The window ends at the
      schedule date so a recovered run covers its own night.
    values:
      namespaces: "{{ {'all': outputs.discover_namespaces.namespaces ?? [],
        'excluded': inputs.exclude_namespaces} | jq('.excluded as $x | [.all[] |
        select(. as $n | ($x | index($n)) == null)] | sort') | first | toJson
        }}"
      window_end: "{{ trigger.date ?? execution.startDate }}"
      window_start: "{{ (trigger.date ?? execution.startDate) | dateAdd(0 -
        inputs.window_hours, 'HOURS') }}"
      run_url: "{{ kestra.url ?? inputs.kestra_api_url }}/ui/{{ vars.tenant
        }}/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}"

  - id: nothing_in_scope
    type: io.kestra.plugin.core.flow.If
    description: A prefix that matches nothing is a configuration mistake, not an
      empty digest.
    condition: "{{ fromJson(outputs.scope.values.namespaces) | length == 0 }}"
    then:
      - id: refuse_empty_scope
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "No namespace matches prefix '{{ inputs.namespace_prefix }}' after
          excluding {{ inputs.exclude_namespaces | toJson }}. Found: {{
          (outputs.discover_namespaces.namespaces ?? []) | toJson }}."

  - id: failures
    type: io.kestra.plugin.core.flow.Loop
    description: |
      One iteration per namespace, two at a time. `transmitFailed: false`: a namespace whose API calls fail is recorded in
      its row and dead-lettered at the end, the other namespaces are still counted.
    values: "{{ outputs.scope.values.namespaces }}"
    concurrencyLimit: 2
    transmitFailed: false
    tasks:
      - id: count_runs
        type: io.kestra.plugin.kestra.executions.Count
        kestraUrl: "{{ inputs.kestra_api_url }}"
        tenantId: "{{ vars.tenant }}"
        auth:
          {
            username: "{{ secret('KESTRA_API_USERNAME') }}",
            password: "{{ secret('KESTRA_API_PASSWORD') }}"
          }
        namespaces: [ "{{ item.value }}" ]
        startDate: "{{ outputs.scope.values.window_start }}"
        endDate: "{{ outputs.scope.values.window_end }}"

      - id: count_failed
        type: io.kestra.plugin.kestra.executions.Count
        kestraUrl: "{{ inputs.kestra_api_url }}"
        tenantId: "{{ vars.tenant }}"
        auth:
          {
            username: "{{ secret('KESTRA_API_USERNAME') }}",
            password: "{{ secret('KESTRA_API_PASSWORD') }}"
          }
        namespaces: [ "{{ item.value }}" ]
        states: [ FAILED ]
        startDate: "{{ outputs.scope.values.window_start }}"
        endDate: "{{ outputs.scope.values.window_end }}"

      - id: recent_failures
        type: io.kestra.plugin.kestra.executions.Query
        description: "The FAILED executions of the window, at most
          `max_failed_per_namespace` (`page: 1` with `size`). On 2.0.5 the API
          returned them oldest first, so with more failures than the limit the
          digest covers the start of the window; `digest_truncated` in the row
          says so."
        runIf: "{{ (outputs.count_failed.count ?? 0) > 0 }}"
        kestraUrl: "{{ inputs.kestra_api_url }}"
        tenantId: "{{ vars.tenant }}"
        auth:
          {
            username: "{{ secret('KESTRA_API_USERNAME') }}",
            password: "{{ secret('KESTRA_API_PASSWORD') }}"
          }
        namespace: "{{ item.value }}"
        states: [ FAILED ]
        startDate: "{{ outputs.scope.values.window_start }}"
        endDate: "{{ outputs.scope.values.window_end }}"
        fetchType: FETCH
        page: 1
        size: "{{ inputs.max_failed_per_namespace }}"
    outputs:
      - id: row
        type: JSON
        value: |
          {{ {
            'namespace': item.value,
            'executions': outputs.count_runs.count ?? null,
            'failed': outputs.count_failed.count ?? null,
            'digest_truncated': (outputs.count_failed.count ?? 0) > inputs.max_failed_per_namespace,
            'recent_failures': (outputs.recent_failures.rows ?? []) | jq('[flatten[] | {namespace, flowId, id, started: .state.startDate}]') | first,
            'ok': outputs.count_runs.count is defined and outputs.count_failed.count is defined
          } | toJson }}

  - id: error_digest
    type: io.kestra.plugin.core.flow.Loop
    description: |
      One iteration per recent failed execution, across all namespaces. `logs.Fetch` reads another execution's logs by
      `namespace`, `flowId` and `executionId` into an ION file. Two things observed on Kestra 2.0.5 shape this loop: the
      task's `level: ERROR` still returned DEBUG and WARN lines, and its rows carry `taskId` and `executionId` as
      `{present: true}`, which also makes `FilterItems` fail on the file. So the file is converted to JSON Lines, the
      ERROR lines are picked with `jq` and rewritten with the execution id, one small file per execution.
    values: "{{ (outputs.failures.outputs ?? []) | jq('[.[].outputs.row |
      .recent_failures[]?]') | first | toJson }}"
    concurrencyLimit: 2
    transmitFailed: false
    tasks:
      - id: fetch_logs
        type: io.kestra.plugin.kestra.logs.Fetch
        kestraUrl: "{{ inputs.kestra_api_url }}"
        tenantId: "{{ vars.tenant }}"
        auth:
          {
            username: "{{ secret('KESTRA_API_USERNAME') }}",
            password: "{{ secret('KESTRA_API_PASSWORD') }}"
          }
        namespace: "{{ fromJson(item.value).namespace }}"
        flowId: "{{ fromJson(item.value).flowId }}"
        executionId: "{{ fromJson(item.value).id }}"
        level: ERROR

      - id: to_jsonl
        type: io.kestra.plugin.serdes.json.IonToJson
        runIf: "{{ (outputs.fetch_logs.size ?? 0) > 0 }}"
        from: "{{ outputs.fetch_logs.uri }}"

      - id: pick_errors
        type: io.kestra.plugin.core.output.OutputValues
        description: Keeps namespace, flowId, timestamp and message of each ERROR line
          and adds the execution id.
        runIf: "{{ outputs.to_jsonl.uri is defined }}"
        values:
          errors: >-
            {{ {'id': fromJson(item.value).id, 'lines':
            read(outputs.to_jsonl.uri) | trim | split('\n')}
               | jq('.id as $id | [.lines[] | select(length > 0) | fromjson | select(.level == "ERROR") | {namespace, flowId, executionId: $id, timestamp, message}]')
               | first | toJson }}

      - id: write_errors
        type: io.kestra.plugin.core.storage.Write
        description: The first 200 ERROR lines of this execution as JSON Lines;
          `error_lines` in the row is the full count.
        runIf: "{{ fromJson(outputs.pick_errors.values.errors ?? '[]') | length > 0 }}"
        extension: .jsonl
        content: "{% for e in fromJson(outputs.pick_errors.values.errors) %}{% if
          loop.index < 200 %}{{ e | toJson }}

          {% endif %}{% endfor %}"
    outputs:
      - id: row
        type: JSON
        value: |
          {{ {
            'namespace': fromJson(item.value).namespace,
            'flow_id': fromJson(item.value).flowId,
            'execution_id': fromJson(item.value).id,
            'started': fromJson(item.value).started,
            'log_lines': outputs.fetch_logs.size ?? null,
            'error_lines': fromJson(outputs.pick_errors.values.errors ?? '[]') | length,
            'first_error': (fromJson(outputs.pick_errors.values.errors ?? '[]') | length) > 0 ? (fromJson(outputs.pick_errors.values.errors)[0].message | abbreviate(300)) : null,
            'uri': outputs.write_errors.uri ?? null,
            'ok': outputs.fetch_logs.size is defined
          } | toJson }}

  - id: summary
    type: io.kestra.plugin.core.output.OutputValues
    values:
      namespaces: "{{ fromJson(outputs.scope.values.namespaces) | length }}"
      executions: "{{ (outputs.failures.outputs ?? []) |
        jq('[.[].outputs.row.executions // 0] | add // 0') | first }}"
      failed_executions: "{{ (outputs.failures.outputs ?? []) |
        jq('[.[].outputs.row.failed // 0] | add // 0') | first }}"
      failed_namespaces: "{{ (outputs.failures.outputs ?? []) | jq('[.[].outputs.row |
        select(.ok != true) | .namespace]') | first | toJson }}"
      truncated_namespaces: "{{ (outputs.failures.outputs ?? []) |
        jq('[.[].outputs.row | select(.digest_truncated == true) | .namespace]')
        | first | toJson }}"
      error_lines: "{{ (outputs.error_digest.outputs ?? []) |
        jq('[.[].outputs.row.error_lines] | add // 0') | first }}"
      digest_files: "{{ (outputs.error_digest.outputs ?? []) |
        jq('[.[].outputs.row.uri | select(. != null)]') | first | toJson }}"
      digest_rows: "{{ (outputs.error_digest.outputs ?? []) | jq('[.[].outputs.row |
        del(.uri, .ok)] | sort_by(.namespace, .started)') | first | toJson }}"
      unreadable_logs: "{{ (outputs.error_digest.outputs ?? []) | jq('[.[].outputs.row
        | select(.ok != true) | .execution_id]') | first | toJson }}"

  - id: digest_file
    type: io.kestra.plugin.core.flow.If
    description: One JSON Lines file of every ERROR line of the window, empty when
      there was none, so the `error_digest` output always exists.
    condition: "{{ fromJson(outputs.summary.values.digest_files) | length > 0 }}"
    then:
      - id: concat_digest
        type: io.kestra.plugin.core.storage.Concat
        files: "{{ outputs.summary.values.digest_files }}"
        extension: .jsonl
    else:
      - id: empty_digest
        type: io.kestra.plugin.core.storage.Write
        content: ""
        extension: .jsonl

  - id: label
    type: io.kestra.plugin.core.execution.Labels
    description: Filter the Executions page by nights with failures without opening a run.
    labels:
      digest_failed_executions: "{{ outputs.summary.values.failed_executions }}"
      digest_error_lines: "{{ outputs.summary.values.error_lines }}"
      digest_truncated: "{{ fromJson(outputs.summary.values.truncated_namespaces) |
        length > 0 ? 'yes' : 'no' }}"

  - id: log_summary
    type: io.kestra.plugin.core.log.Log
    message: |
      Failures {{ outputs.scope.values.window_start }} to {{ outputs.scope.values.window_end }}: {{ outputs.summary.values.executions }} executions in {{ outputs.summary.values.namespaces }} namespace(s), {{ outputs.summary.values.failed_executions }} failed, {{ outputs.summary.values.error_lines }} ERROR lines from {{ fromJson(outputs.summary.values.digest_rows) | length }} failed execution(s){{ fromJson(outputs.summary.values.truncated_namespaces) | length > 0 ? ' (digest truncated for ' ~ (fromJson(outputs.summary.values.truncated_namespaces) | join(', ')) ~ ', raise max_failed_per_namespace)' : '' }}.
      {% for r in fromJson(outputs.summary.values.digest_rows) %}{{ r.namespace }}/{{ r.flow_id }} {{ r.execution_id }}: {{ r.error_lines }} ERROR line(s){{ r.first_error != null ? ', first: ' ~ r.first_error : '' }}
      {% endfor %}

  - id: notify_digest
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: One message per night with failures, the first error of each failed
      run and the execution link for the full digest file. allowFailure so a
      missing `SLACK_WEBHOOK_URL` does not fail a digest that is otherwise
      complete.
    runIf: "{{ inputs.notify and (outputs.summary.values.failed_executions | number)
      > 0 }}"
    allowFailure: true
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :rotating_light: {{ outputs.summary.values.failed_executions }} failed execution(s) in the last {{ inputs.window_hours }}h across {{ outputs.summary.values.namespaces }} namespace(s), {{ outputs.summary.values.error_lines }} ERROR lines.
      {% for r in fromJson(outputs.summary.values.digest_rows) %}{% if loop.index < 20 %}`{{ r.namespace }}/{{ r.flow_id }}` {{ r.execution_id }}: {{ r.first_error ?? 'no ERROR line logged' }}
      {% endif %}{% endfor %}{{ fromJson(outputs.summary.values.digest_rows) | length > 20 ? '... and ' ~ ((fromJson(outputs.summary.values.digest_rows) | length) - 20) ~ ' more in the digest file.' : '' }}
      {{ outputs.scope.values.run_url }}

  - id: dead_letter
    type: io.kestra.plugin.core.flow.If
    description: Namespaces whose counts failed are kept 7 days in KV and the run
      ends FAILED, with the digest of the other namespaces intact. Executions
      whose logs could not be read only appear in `unreadable_logs`.
    condition: "{{ fromJson(outputs.summary.values.failed_namespaces) | length > 0 }}"
    then:
      - id: remember_failed
        type: io.kestra.plugin.core.kv.Set
        key: "{{ vars.state_key }}_failed_namespaces"
        kvType: JSON
        ttl: P7D
        value: "{{ {'execution_id': execution.id, 'window_end':
          outputs.scope.values.window_end, 'namespaces':
          fromJson(outputs.summary.values.failed_namespaces)} | toJson }}"
      - id: report_partial_failure
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "Digest incomplete: {{
          fromJson(outputs.summary.values.failed_namespaces) | join(', ') }}
          failed (see the Loop iterations of `failures`)."
    else:
      - id: clear_failed
        type: io.kestra.plugin.core.kv.Delete
        key: "{{ vars.state_key }}_failed_namespaces"
        errorOnMissing: false

errors:
  - id: first_error
    type: io.kestra.plugin.core.output.OutputValues
    description: Read once, at the top of the branch, because `errorLogs()` also
      returns lines logged by the branch itself.
    values:
      task: "{{ (errorLogs() | length) > 0 ? (errorLogs()[0].taskId ?? 'none') :
        'none' }}"
      message: "{{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'no error
        line was logged' }}"
  - id: log_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: "Error digest failed in task {{ outputs.first_error.values.task }}: {{
      outputs.first_error.values.message }} {{ outputs.scope.values.run_url ??
      execution.id }}"
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: allowFailure so a missing `SLACK_WEBHOOK_URL` does not hide the
      original error; the same text is in the ERROR log above.
    allowFailure: true
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: ":rotating_light: Kestra error digest failed in `{{
      outputs.first_error.values.task }}`: {{ outputs.first_error.values.message
      }} {{ outputs.scope.values.run_url ?? execution.id }}"

finally:
  - id: record_run
    type: io.kestra.plugin.core.kv.Set
    description: Always written, also after a failure, so another flow can check
      when the last digest ran and what it counted.
    key: "{{ vars.state_key }}_last_run"
    kvType: JSON
    ttl: P90D
    value: "{{ {'execution_id': execution.id, 'started_at': execution.startDate,
      'window_end': outputs.scope.values.window_end ?? null, 'executions':
      ((outputs.summary.values.executions ?? '') == '' ? null :
      (outputs.summary.values.executions | number)), 'failed_executions':
      ((outputs.summary.values.failed_executions ?? '') == '' ? null :
      (outputs.summary.values.failed_executions | number)), 'error_lines':
      ((outputs.summary.values.error_lines ?? '') == '' ? null :
      (outputs.summary.values.error_lines | number)), 'failed_namespaces':
      fromJson(outputs.summary.values.failed_namespaces ?? '[]')} | toJson }}"

outputs:
  - id: error_digest
    type: FILE
    description: JSON Lines, every ERROR line of the recent failed executions
      (namespace, flowId, executionId, timestamp, message).
    value: "{{ outputs.concat_digest.uri ?? outputs.empty_digest.uri }}"
  - id: error_digest_rows
    type: JSON
    description: One row per failed execution with its ERROR line count and first
      error message.
    value: "{{ fromJson(outputs.summary.values.digest_rows) }}"
  - id: namespaces
    type: JSON
    description: One row per namespace with execution and failure counts and whether
      the digest was truncated.
    value: "{{ (outputs.failures.outputs ?? []) | jq('[.[].outputs.row |
      del(.recent_failures)]') | first }}"