id: kestra-execution-history-retention-purge-guard
namespace: company.team
description: |
Kestra keeps every execution it has ever run, so the backend database and the executions API get
slower the longer a cluster lives. This flow finds finished executions older than a retention
window, skips anything still RUNNING or PAUSED, skips a protected-namespace allowlist, then deletes
them behind a Pause approval with a per-run ceiling. After the purge it counts the aged backlog
again and fails the run if a targeted execution survived or if the backlog is still over the
ceiling. Deleting execution history is irreversible, so read-only is the default.
inputs:
- id: kestra_url
type: STRING
displayName: Target Kestra API URL
description: The Kestra whose execution history is purged. The default points
the flow at its own cluster.
defaults: "http://localhost:8080"
- id: retention_amount
type: INT
displayName: Retention window
description: Finished executions older than this are candidates for deletion.
defaults: 30
- id: retention_unit
type: SELECT
displayName: Retention unit
description: Unit for the retention window. Minutes and hours are there for a
dry run against a short window.
values:
- MINUTES
- HOURS
- DAYS
defaults: DAYS
- id: protected_namespaces
type: ARRAY
itemType: STRING
displayName: Namespaces never purged
description: >-
Execution history that must be kept whatever its age, for example an
audited namespace or the namespace holding this flow. Nothing in these
namespaces is ever deleted.
defaults: [ "system" ]
- id: scan_size
type: INT
displayName: Executions scanned per run
description: >-
How many aged executions one run pulls back. One page per run keeps the
scan bounded on a cluster with a long history. Whatever is left over is
picked up by the next run.
defaults: 500
- id: max_deletes
type: INT
displayName: Per-run delete ceiling
description: >-
Refuse to delete more than this many executions in one run. The rest are
deferred to the next run, so a first purge on an old cluster cannot wipe
years of history in one go.
defaults: 50
- id: backlog_ceiling
type: INT
displayName: Acceptable backlog after the purge
description: >-
After the purge, the run fails if this many aged executions are still
purgeable. That is the signal to raise max_deletes or to look at why
history is growing faster than it is pruned.
defaults: 500
- id: delete_logs
type: BOOL
displayName: Delete logs too
description: Remove the log lines of each deleted execution. Set false to keep
logs after the execution row is gone.
defaults: true
- id: delete_metrics
type: BOOL
displayName: Delete metrics too
defaults: true
- id: delete_storage
type: BOOL
displayName: Delete internal storage files too
description: Remove the internal storage files the execution produced. This is
usually the bulk of the space.
defaults: true
- 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 delete. 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, before any execution
is deleted.
defaults: true
tasks:
- id: window
type: io.kestra.plugin.core.output.OutputValues
description: >-
Work out the retention cutoff once, in UTC, so every query, the
client-side filter and the message all use the same instant.
values:
cutoff: "{{ now(timeZone='UTC') | dateAdd(-1 * inputs.retention_amount,
inputs.retention_unit) | date(\"yyyy-MM-dd'T'HH:mm:ss.SSS'Z'\",
timeZone='UTC') }}"
cutoff_compare: "{{ now(timeZone='UTC') | dateAdd(-1 * inputs.retention_amount,
inputs.retention_unit) | date(\"yyyy-MM-dd'T'HH:mm:ss\", timeZone='UTC')
}}"
- id: aged_before
type: io.kestra.plugin.kestra.executions.Count
description: Read-only. Count the finished executions on the target that started
before the cutoff.
kestraUrl: "{{ inputs.kestra_url }}"
auth:
username: "{{ inputs.kestra_user }}"
password: "{{ inputs.kestra_password }}"
startDate: "1970-01-01T00:00:00.000Z"
endDate: "{{ outputs.window.values.cutoff }}"
- id: scan
type: io.kestra.plugin.kestra.executions.Query
description: >-
Read-only. Fetch one page of the executions that started before the
cutoff. State is filtered in the next task rather than here, because the
plugin turns a states list into one EQUALS filter per state and the search
API combines them with AND, which matches nothing. startDate is pinned to
the epoch because the API refuses a window whose end is not after its
start.
kestraUrl: "{{ inputs.kestra_url }}"
auth:
username: "{{ inputs.kestra_user }}"
password: "{{ inputs.kestra_password }}"
fetchType: FETCH
page: 1
size: "{{ inputs.scan_size }}"
startDate: "1970-01-01T00:00:00.000Z"
endDate: "{{ outputs.window.values.cutoff }}"
- id: select
type: io.kestra.plugin.core.output.OutputValues
description: >-
Filter client-side rather than trusting the server filters. Keep only
terminated executions that started before the cutoff, drop every protected
namespace, drop this flow and this execution, oldest first, then cut the
list at the per-run ceiling.
values:
candidates: "{{ (outputs.scan.rows ?? [[]]) | jq('(.[0] // []) | [.[] | {id,
flowId, ns: .namespace, state: (.state.current // \"\"), started:
(.state.startDate // \"\")}] | map(select(.state == \"SUCCESS\" or
.state == \"WARNING\" or .state == \"FAILED\" or .state == \"KILLED\" or
.state == \"CANCELLED\" or .state == \"SKIPPED\" or .state ==
\"RETRIED\")) | map(select(.started != \"\" and (.started[0:19]) < \"' ~
outputs.window.values.cutoff_compare ~ '\")) | map(select(([.ns] - ' ~
(inputs.protected_namespaces | jq('.') | first | toJson) ~ ') | length >
0)) | map(select(.id != \"' ~ execution.id ~ '\" and .flowId != \"' ~
flow.id ~ '\")) | sort_by(.started)') | first | toJson }}"
- id: counts
type: io.kestra.plugin.core.output.OutputValues
description: Split the filtered list into what this run may delete and what it defers.
values:
purgeable: "{{ outputs.select.values.candidates | jq('length') | first }}"
purge_ids: "{{ outputs.select.values.candidates | jq('[.[].id] | .[0:' ~
inputs.max_deletes ~ ']') | first | toJson }}"
purge_count: "{{ outputs.select.values.candidates | jq('[.[0:' ~
inputs.max_deletes ~ '] | .[].id] | length') | first }}"
deferred: "{{ outputs.select.values.candidates | jq('[.[' ~ inputs.max_deletes ~
':] | .[].id] | length') | first }}"
detail: "{{ outputs.select.values.candidates | jq('[.[0:' ~ inputs.max_deletes ~
'] | .[] | .ns + \"/\" + .flowId + \" \" + .id + \" \" + .state + \"
started \" + .started] | join(\"; \")') | first }}"
- id: summary
type: io.kestra.plugin.core.log.Log
message: >-
Retention purge on {{ inputs.kestra_url }}: cutoff {{
outputs.window.values.cutoff }} ({{ inputs.retention_amount }} {{
inputs.retention_unit }}), {{ outputs.aged_before.count }} finished
execution(s) older than it on the instance, {{
outputs.counts.values.purgeable }} purgeable after the protected
namespaces {{ inputs.protected_namespaces }} and the state filter, {{
outputs.counts.values.purge_count }} selected for this run and {{
outputs.counts.values.deferred }} deferred. {{
outputs.counts.values.detail }}
- id: route
type: io.kestra.plugin.core.flow.If
condition: "{{ (outputs.counts.values.purge_count | number) > 0 }}"
then:
- id: alert
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{{ {'text': 'Kestra retention purge on ' ~ inputs.kestra_url ~ ': ' ~ outputs.counts.values.purgeable ~ ' finished execution(s) older than ' ~ outputs.window.values.cutoff ~ ' are purgeable, ' ~ outputs.counts.values.purge_count ~ ' selected and ' ~ outputs.counts.values.deferred ~ ' deferred (' ~ outputs.counts.values.detail ~ ').' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Deleting them behind approval.' : ' Nothing deleted (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: approval
type: io.kestra.plugin.core.flow.Pause
description: >-
Hold for a human to approve the delete. Execution history cannot
be restored, so no answer inside the window keeps everything.
pauseDuration: PT1H
behavior: RESUME
onResume:
- id: approved
type: BOOL
required: false
description: Set true to delete the listed executions. Anything else keeps them.
- id: note
type: STRING
required: false
description: Optional reason, included in the confirmation log.
- id: approved_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.approval.onResume.approved ?? false }}"
then:
- id: remediate_purge
type: io.kestra.plugin.core.flow.Loop
description: >-
One delete at a time. Parallel deletes of logs, metrics and
storage files hammer the same backend tables. A partial
failure is also easier to read in order.
concurrencyLimit: 1
values: "{{ outputs.counts.values.purge_ids }}"
tasks:
- id: delete
type: io.kestra.plugin.kestra.executions.Delete
description: >-
Delete one execution with its logs, metrics and storage
files. The plugin refuses an execution that is not
terminated, which is the second guard behind the state
filter.
allowFailure: true
kestraUrl: "{{ inputs.kestra_url }}"
executionId: "{{ item.value }}"
deleteLogs: "{{ inputs.delete_logs }}"
deleteMetrics: "{{ inputs.delete_metrics }}"
deleteStorage: "{{ inputs.delete_storage }}"
auth:
username: "{{ inputs.kestra_user }}"
password: "{{ inputs.kestra_password }}"
- id: settle
type: io.kestra.plugin.core.flow.Sleep
duration: PT5S
- id: aged_after
type: io.kestra.plugin.kestra.executions.Count
description: Count the aged finished executions again. The drop against
aged_before is the proof the delete landed.
kestraUrl: "{{ inputs.kestra_url }}"
auth:
username: "{{ inputs.kestra_user }}"
password: "{{ inputs.kestra_password }}"
startDate: "1970-01-01T00:00:00.000Z"
endDate: "{{ outputs.window.values.cutoff }}"
- id: reverify
type: io.kestra.plugin.kestra.executions.Query
description: >-
Read the target again over the same window, so the result
cannot come from the delete task's own status.
kestraUrl: "{{ inputs.kestra_url }}"
auth:
username: "{{ inputs.kestra_user }}"
password: "{{ inputs.kestra_password }}"
fetchType: FETCH
page: 1
size: "{{ inputs.scan_size }}"
startDate: "1970-01-01T00:00:00.000Z"
endDate: "{{ outputs.window.values.cutoff }}"
- id: verify_counts
type: io.kestra.plugin.core.output.OutputValues
description: >-
Which targeted ids survived plus how much aged history is
still purgeable. The state, window and namespace filters are
the same as the ones that built the list, so the two numbers
are comparable.
values:
survivors: "{{ (outputs.reverify.rows ?? [[]]) | jq('(.[0] // []) | [.[].id] as
$live | ' ~ outputs.counts.values.purge_ids ~ ' - (' ~
outputs.counts.values.purge_ids ~ ' - $live)') | first |
toJson }}"
survivor_count: "{{ (outputs.reverify.rows ?? [[]]) | jq('(.[0] // []) |
[.[].id] as $live | (' ~ outputs.counts.values.purge_ids ~ '
- (' ~ outputs.counts.values.purge_ids ~ ' - $live)) |
length') | first }}"
still_purgeable: "{{ (outputs.reverify.rows ?? [[]]) | jq('(.[0] // []) | [.[] |
{id, flowId, ns: .namespace, state: (.state.current //
\"\"), started: (.state.startDate // \"\")}] |
map(select(.state == \"SUCCESS\" or .state == \"WARNING\" or
.state == \"FAILED\" or .state == \"KILLED\" or .state ==
\"CANCELLED\" or .state == \"SKIPPED\" or .state ==
\"RETRIED\")) | map(select(.started != \"\" and
(.started[0:19]) < \"' ~
outputs.window.values.cutoff_compare ~ '\")) |
map(select(([.ns] - ' ~ (inputs.protected_namespaces |
jq('.') | first | toJson) ~ ') | length > 0)) |
map(select(.id != \"' ~ execution.id ~ '\" and .flowId !=
\"' ~ flow.id ~ '\")) | length') | first }}"
- id: landed_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ (outputs.verify_counts.values.survivor_count | number) > 0 }}"
then:
- id: delete_did_not_land
type: io.kestra.plugin.core.execution.Fail
errorMessage: "Delete did not land: {{
outputs.verify_counts.values.survivor_count }} targeted
execution(s) are still in the history ({{
outputs.verify_counts.values.survivors }}). The API user
may lack permission to delete. An execution may also have
been restarted so it is no longer terminated."
else:
- id: backlog_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ (outputs.verify_counts.values.still_purgeable | number) >
inputs.backlog_ceiling }}"
then:
- id: backlog_over_ceiling
type: io.kestra.plugin.core.execution.Fail
errorMessage: "Deleted {{ outputs.counts.values.purge_count }} execution(s),
aged history went from {{ outputs.aged_before.count }}
to {{ outputs.aged_after.count }}, but {{
outputs.verify_counts.values.still_purgeable }} aged
execution(s) are still purgeable, over the
backlog_ceiling of {{ inputs.backlog_ceiling }}. Raise
max_deletes or run more often."
else:
- id: purged
type: io.kestra.plugin.core.log.Log
message: "Deleted and verified {{ outputs.counts.values.purge_count }}
execution(s): {{ outputs.counts.values.purge_ids }}.
Aged history on {{ inputs.kestra_url }} went from {{
outputs.aged_before.count }} to {{
outputs.aged_after.count }}, {{
outputs.verify_counts.values.still_purgeable }} still
purgeable. {{ outputs.approval.onResume.note ?? '' }}"
else:
- id: declined
type: io.kestra.plugin.core.log.Log
message: "Approval not granted, so no execution was deleted. {{
outputs.counts.values.purgeable }} aged execution(s) are still
in the history on {{ inputs.kestra_url }}."
else:
- id: clean
type: io.kestra.plugin.core.log.Log
message: "Nothing to purge on {{ inputs.kestra_url }}. No finished execution
outside the protected namespaces started before {{
outputs.window.values.cutoff }}."
errors:
- id: alert_flow_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{{ {'text': 'Kestra execution history retention purge guard failed on ' ~ inputs.kestra_url ~ ' (execution ' ~ execution.id ~ '). The delete may not have landed. The aged backlog may still be over the ceiling. The target may be unreachable. Check the logs before retrying.'} | toJson }}
triggers:
- id: daily
type: io.kestra.plugin.core.trigger.Schedule
disabled: true
description: Enable once the dry run output looks right on your cluster. The
trigger inputs keep it read-only.
cron: "0 3 * * *"
inputs:
dry_run: true
auto_remediate: false
