id: kestra-concurrency-queue-backlog-collapse
namespace: company.platform
concurrency:
limit: 1
behavior: CANCEL
inputs:
- id: namespace
type: STRING
defaults: company.team
description: Namespace whose queued executions are checked.
- id: dry_run
type: BOOL
defaults: true
description: Only report the backlog and the plan. Set to false to kill the
duplicate queued runs.
- id: approval_above
type: INT
defaults: 20
min: 0
description: Wait for an approval when one run would kill more queued executions
than this.
- id: max_kills
type: INT
defaults: 100
min: 1
max: 1000
description: Upper bound on kills per run, oldest duplicates first. The next run
continues with the rest.
- id: page_size
type: INT
defaults: 500
min: 1
max: 1000
description: Number of queued executions read per run.
- id: kestra_url
type: URI
defaults: http://localhost:8080
description: Base URL of the Kestra API.
variables:
tenant: main
policy_key: queue-policy
policy_value: deduplicate
# Input: {flows, queued, running, key, value, max}. Groups the queued runs of opted-in flows by inputs,
# trigger payload (a Schedule's own dates and webhook headers are ignored), labels and replay origin,
# with object keys sorted at every level. Each group keeps its oldest run, which is next in line; the
# rest are duplicates. The selected duplicates are the oldest ones (bounded by max), listed newest first
# so the kills move away from the head of the queue.
plan_program: |
def norm: if type == "object" then (to_entries | sort_by(.key) | map(.value |= norm) | from_entries) elif type == "array" then map(norm) else . end;
.key as $k | .value as $v | .max as $max
| ([.flows[] | select([.labels[]? | select(.key == $k and .value == $v)] | length > 0) | .id]) as $optin
| (.running | map({flow: .flowId, id, since: .state.startDate})) as $running
| [.queued[] | {id, flow: .flowId, at: .state.startDate,
work: ({inputs: (.inputs // {}),
trigger: (if (.trigger.type // "") | startswith("io.kestra.plugin.core.trigger.Schedule") then null else ((.trigger.variables // null) | if type == "object" then del(.headers) else . end) end),
labels: ([(.labels // [])[] | select(.key | startswith("system.") | not)] | sort_by(.key, .value)),
replay_of: (if (.originalId // .id) != .id then .originalId else null end)} | norm | tojson)}]
| group_by(.flow)
| map(.[0].flow as $f
| (($optin | map(select(. == $f)) | length) > 0) as $in
| (group_by(.work) | map(sort_by(.at))) as $groups
| {flow: $f, opted_in: $in, queued: length, distinct_work: ($groups | length),
running: [$running[] | select(.flow == $f) | .id + " since " + .since],
duplicates: (if $in then [$groups[] | .[0].id as $keep | .[1:][] | {id, kept: $keep, at}] else [] end)})
| {flows: ., kills: ([.[].duplicates[]] | sort_by(.at) | .[:$max] | reverse)}
tasks:
- id: flows
type: io.kestra.plugin.kestra.flows.List
description: Flows of the namespace, to find the ones labelled with the policy.
kestraUrl: "{{ inputs.kestra_url }}"
tenantId: "{{ vars.tenant }}"
namespace: "{{ inputs.namespace }}"
auth:
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
- id: queued
type: io.kestra.plugin.core.http.Request
description: Queued executions of the namespace, oldest first, with their inputs
and trigger.
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant
}}/executions/search?page=1&size={{ inputs.page_size
}}&sort=state.startDate:asc&filters%5Bstate%5D%5BEQUALS%5D=QUEUED&filters\
%5Bnamespace%5D%5BEQUALS%5D={{ inputs.namespace }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
- id: running
type: io.kestra.plugin.core.http.Request
description: Running executions, reported as the runs that hold the concurrency slots.
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant
}}/executions/search?page=1&size=1000&sort=state.startDate:asc&filters%5B\
state%5D%5BEQUALS%5D=RUNNING&filters%5Bnamespace%5D%5BEQUALS%5D={{
inputs.namespace }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
- id: plan
type: io.kestra.plugin.core.output.OutputValues
values:
result: "{{ {'flows': outputs.flows.flows, 'queued':
fromJson(outputs.queued.body).results, 'running':
fromJson(outputs.running.body).results, 'key': vars.policy_key, 'value':
vars.policy_value, 'max': inputs.max_kills} | jq(vars.plan_program) |
first | toJson }}"
- id: report
type: io.kestra.plugin.core.log.Log
message: |-
{{ fromJson(outputs.queued.body).total }} queued executions in {{ inputs.namespace }} ({{ fromJson(outputs.queued.body).results | length }} read), {{ fromJson(outputs.plan.values.result).kills | length }} duplicate runs to kill{{ inputs.dry_run ? ' (dry run)' : '' }}.
{%- for f in fromJson(outputs.plan.values.result).flows %}
{{ f.flow }}: {{ f.queued }} queued, {{ f.distinct_work }} distinct input sets, {{ f.opted_in ? (f.duplicates | length) ~ ' duplicates' : 'not opted in, left alone' }}, slot held by {{ f.running is empty ? 'nothing running' : f.running | join(', ') }}
{%- endfor %}
- id: collapse
type: io.kestra.plugin.core.flow.If
condition: "{{ not inputs.dry_run and
(fromJson(outputs.plan.values.result).kills | length) > 0 }}"
then:
- id: approve
type: io.kestra.plugin.core.flow.Pause
description: Large collapses wait for a person. Without approval within 1 hour
the run is cancelled and nothing changes.
runIf: "{{ (fromJson(outputs.plan.values.result).kills | length) >
inputs.approval_above }}"
pauseDuration: PT1H
behavior: CANCEL
- id: kill_duplicates
type: io.kestra.plugin.core.flow.Loop
description: One duplicate at a time, newest first, so the kills stay away from
the run that is next in line.
values: "{{ {'kills': fromJson(outputs.plan.values.result).kills, 'run':
execution.id} | jq('.run as $r | [.kills[] | . + {run: $r} | tojson]')
| first }}"
concurrencyLimit: 1
tasks:
- id: recheck
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/executions/{{
fromJson(item.value).id }}"
options:
allowFailed: true
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: kept_state
type: io.kestra.plugin.core.http.Request
description: The run this one duplicates. If it was deleted, killed, cancelled
or failed meanwhile, this duplicate is the last copy of the work
and must stay.
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/executions/{{
fromJson(item.value).kept }}"
options:
allowFailed: true
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: still_queued
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.recheck.code == 200 and
fromJson(outputs.recheck.body).state.current == 'QUEUED' and
outputs.kept_state.code == 200 and ['QUEUED', 'RUNNING',
'SUCCESS'] contains
fromJson(outputs.kept_state.body).state.current }}"
then:
- id: kill
type: io.kestra.plugin.kestra.executions.Kill
description: Kill, not unqueue to CANCELLED. Kestra frees no concurrency slot
for a run killed while queued, whereas a queued run moved to
CANCELLED releases a slot it never held and lets another run
start next to the running one.
kestraUrl: "{{ inputs.kestra_url }}"
tenantId: "{{ vars.tenant }}"
executionId: "{{ fromJson(item.value).id }}"
auth:
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: wait_killed
type: io.kestra.plugin.core.flow.LoopUntil
condition: "{{ fromJson(outputs.killed_state.body).state.current == 'KILLED' }}"
failOnMaxReached: true
checkFrequency:
interval: PT1S
maxIterations: 30
tasks:
- id: killed_state
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/executions/{{
fromJson(item.value).id }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: mark
type: io.kestra.plugin.core.http.Request
description: Records which run it duplicated and which collapse run killed it.
The API replaces all non-system labels, so the existing ones
are sent back too.
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/executions/{{
fromJson(item.value).id }}/actions/labels"
method: POST
contentType: application/json
body: "{{ {'labels': fromJson(outputs.killed_state.body).labels ?? [], 'kept':
fromJson(item.value).kept, 'run': fromJson(item.value).run} |
jq('[.labels[] | select(.key | startswith(\"system.\") | not)]
+ [{key: \"queue-duplicate-of\", value: .kept}, {key:
\"queue-collapse-run\", value: .run}]') | first | toJson }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
retry:
type: constant
interval: PT5S
maxAttempts: 3
- id: started_before_kill
type: io.kestra.plugin.core.execution.Fail
description: The run left the queue between the re-check and the kill. Fails
loudly instead of reporting a clean collapse.
condition: "{{ fromJson(outputs.killed_state.body).state.histories | jq('[.[] |
select(.state == \"RUNNING\")] | length') | first > 0 }}"
errorMessage: "{{ fromJson(item.value).id }} had started RUNNING before the kill
landed; check its side effects."
else:
- id: left_alone
type: io.kestra.plugin.core.log.Log
level: WARN
message: "Not deduplicated: {{ fromJson(item.value).id }} is {{
outputs.recheck.code == 200 ?
fromJson(outputs.recheck.body).state.current : 'missing (HTTP
' ~ outputs.recheck.code ~ ')' }} and the run it duplicates
({{ fromJson(item.value).kept }}) is {{
outputs.kept_state.code == 200 ?
fromJson(outputs.kept_state.body).state.current : 'missing
(HTTP ' ~ outputs.kept_state.code ~ ')' }}."
- id: queued_after
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant
}}/executions/search?page=1&size=1000&filters%5Bstate%5D%5BEQUALS%5D=\
QUEUED&filters%5Bnamespace%5D%5BEQUALS%5D={{ inputs.namespace }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: running_after
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant
}}/executions/search?page=1&size=1000&filters%5Bstate%5D%5BEQUALS%5D=\
RUNNING&filters%5Bnamespace%5D%5BEQUALS%5D={{ inputs.namespace }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: verify
type: io.kestra.plugin.core.execution.Assert
description: A duplicate may only still be queued if the run it duplicates is
gone, which the loop logs as left alone.
conditions:
- "{{ {'queued': fromJson(outputs.queued_after.body).results,
'running': fromJson(outputs.running_after.body).results, 'kills':
fromJson(outputs.plan.values.result).kills} | jq('[.queued[].id] as
$q | ([.queued[].id] + [.running[].id]) as $alive | [.kills[] |
select(.id as $i | $q | index($i)) | select(.kept as $k | $alive |
index($k))] | length') | first == 0 }}"
errorMessage: "Duplicate runs are still queued next to the run they duplicate:
{{ {'queued': fromJson(outputs.queued_after.body).results, 'running':
fromJson(outputs.running_after.body).results, 'kills':
fromJson(outputs.plan.values.result).kills} | jq('[.queued[].id] as $q
| ([.queued[].id] + [.running[].id]) as $alive | [.kills[] |
select(.id as $i | $q | index($i)) | select(.kept as $k | $alive |
index($k)) | .id]') | first }}"
- id: deduplicated
type: io.kestra.plugin.core.http.Request
description: Runs this collapse actually killed and labelled.
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant
}}/executions/search?page=1&size=1000&filters%5Bnamespace%5D%5BEQUALS\
%5D={{ inputs.namespace
}}&filters%5Blabels%5D%5BEQUALS%5D%5Bqueue-collapse-run%5D={{
execution.id }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: outcome
type: io.kestra.plugin.core.output.OutputValues
values:
not_deduplicated: "{{ {'done': fromJson(outputs.deduplicated.body).results,
'kills': fromJson(outputs.plan.values.result).kills} | jq('[.done[]
| select(.state.current == \"KILLED\") | .id] as $d | [.kills[] |
select(.id as $i | $d | index($i) | not) | .id]') | first | toJson
}}"
- id: partial
type: io.kestra.plugin.core.flow.If
condition: "{{ fromJson(outputs.outcome.values.not_deduplicated) | length > 0 }}"
then:
- id: report_partial
type: io.kestra.plugin.core.log.Log
level: WARN
message: "Deduplicated {{ fromJson(outputs.deduplicated.body).total }} of {{
fromJson(outputs.plan.values.result).kills | length }} planned
duplicates. Left in place because the duplicate started, the run
it duplicates is gone or no longer eligible, or either was deleted
(see the not-deduplicated loop logs): {{
outputs.outcome.values.not_deduplicated }}"
- id: end_warning
type: io.kestra.plugin.core.execution.Exit
state: WARNING
else:
- id: summary
type: io.kestra.plugin.core.log.Log
message: "Deduplicated all {{ fromJson(outputs.plan.values.result).kills |
length }} planned duplicates; {{
fromJson(outputs.queued_after.body).total }} executions remain
queued in {{ inputs.namespace }}."
outputs:
- id: plan
type: JSON
value: "{{ outputs.plan.values.result ?? '{}' }}"
triggers:
- id: every_15_minutes
type: io.kestra.plugin.core.trigger.Schedule
cron: "*/15 * * * *"
inputs:
dry_run: false
