id: kestra-unused-flow-retirement
namespace: company.platform
concurrency:
limit: 1
behavior: QUEUE
inputs:
- id: namespace
type: STRING
defaults: company.team
description: Namespace whose unused flows are retired. Child namespaces are not
included.
- id: idle_period
type: DURATION
defaults: P90D
description: A flow is a candidate when it has not run and has not been edited
for this long.
- id: quarantine_period
type: DURATION
defaults: P14D
description: How long a candidate stays with its triggers disabled before it can
be deleted.
- id: max_flows
type: INT
defaults: 20
description: Maximum number of new candidates quarantined per run.
- id: dry_run
type: BOOL
defaults: true
description: Only report candidates. Nothing is disabled, recorded or deleted.
- id: restore_flow_id
type: STRING
required: false
description: Set to restore one quarantined or retired flow of the namespace
instead of running a review.
- id: kestra_url
type: URI
defaults: http://localhost:8080
description: Base URL of the Kestra API.
variables:
tenant: main
tasks:
- id: check_namespace
type: io.kestra.plugin.core.execution.Assert
description: Never review the namespace that holds this flow, or one of its parents.
conditions:
- "{{ inputs.namespace != flow.namespace and not (flow.namespace |
startsWith(inputs.namespace ~ '.')) }}"
errorMessage: "namespace {{ inputs.namespace }} must not contain this flow ({{
flow.namespace }}.{{ flow.id }})"
- id: mode
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.restore_flow_id is not empty }}"
then:
- id: quarantine_record
type: io.kestra.plugin.core.kv.Get
key: "retire-quarantine_{{ inputs.namespace }}_{{ inputs.restore_flow_id }}"
- id: archive_record
type: io.kestra.plugin.core.kv.Get
key: "retire-archive_{{ inputs.namespace }}_{{ inputs.restore_flow_id }}"
- id: check_record
type: io.kestra.plugin.core.execution.Assert
conditions:
- "{{ outputs.quarantine_record.value is defined or
outputs.archive_record.value is defined }}"
errorMessage: "No quarantine or archive record for {{ inputs.namespace }}.{{
inputs.restore_flow_id }}, nothing to restore"
- id: recreate_flow
type: io.kestra.plugin.core.http.Request
description: Re-creates a deleted flow from the source saved before its deletion.
runIf: "{{ outputs.archive_record.value is defined }}"
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows"
method: POST
contentType: application/x-yaml
body: "{{ outputs.archive_record.value.source }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: enable_triggers
type: io.kestra.plugin.kestra.triggers.Toggle
kestraUrl: "{{ inputs.kestra_url }}"
tenantId: "{{ vars.tenant }}"
namespace: "{{ inputs.namespace }}"
flowId: "{{ inputs.restore_flow_id }}"
enabled: true
auth:
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: verify_restored
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows/{{ inputs.namespace
}}/{{ inputs.restore_flow_id }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: forget_quarantine
type: io.kestra.plugin.core.kv.Delete
key: "retire-quarantine_{{ inputs.namespace }}_{{ inputs.restore_flow_id }}"
- id: forget_ready_record
type: io.kestra.plugin.core.kv.Delete
description: Without this, a later approval would delete the flow that was just
restored.
key: "retire-ready_{{ inputs.namespace }}_{{ inputs.restore_flow_id }}"
- id: forget_archive
type: io.kestra.plugin.core.kv.Delete
key: "retire-archive_{{ inputs.namespace }}_{{ inputs.restore_flow_id }}"
else:
- id: flows
type: io.kestra.plugin.kestra.flows.List
kestraUrl: "{{ inputs.kestra_url }}"
tenantId: "{{ vars.tenant }}"
namespace: "{{ inputs.namespace }}"
auth:
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: still_quarantined
type: io.kestra.plugin.core.kv.GetKeys
prefix: "retire-quarantine_{{ inputs.namespace }}_"
- id: candidates
type: io.kestra.plugin.core.output.OutputValues
description: Flows not already quarantined and not labelled retirement=keep,
capped at max_flows.
values:
ids: "{{ {'flows': outputs.flows.flows, 'keys': outputs.still_quarantined.keys}
| jq('(.keys | map(ltrimstr(\"retire-quarantine_' ~ inputs.namespace
~ '_\"))) as $q | [.flows[] | select(.id as $id | $q | index($id) |
not) | select([(.labels // [])[] | select(.key == \"retirement\" and
.value == \"keep\")] | length == 0) | .id] | .[0:' ~
inputs.max_flows ~ ']') | first | toJson }}"
- id: review_candidates
type: io.kestra.plugin.core.flow.Loop
description: Quarantines each candidate that has not run or changed during the
idle period and that no other flow depends on.
runIf: "{{ outputs.candidates.values.ids != '[]' }}"
values: "{{ outputs.candidates.values.ids }}"
tasks:
- id: flow_source
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows/{{ inputs.namespace
}}/{{ item.value }}?source=true"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: recent_runs
type: io.kestra.plugin.kestra.executions.Count
kestraUrl: "{{ inputs.kestra_url }}"
tenantId: "{{ vars.tenant }}"
namespaces:
- "{{ inputs.namespace }}"
flowId: "{{ item.value }}"
startDate: "{{ now() | dateAdd(-(inputs.idle_period | string |
substringBefore('.') | number), 'SECONDS') }}"
auth:
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: dependents
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows/{{ inputs.namespace
}}/{{ item.value }}/dependencies"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: assessment
type: io.kestra.plugin.core.output.OutputValues
values:
runs: "{{ outputs.recent_runs.count }}"
edited_recently: "{{ (now() | timestamp) -
(fromJson(outputs.flow_source.body).updated | timestamp) <
(inputs.idle_period | string | substringBefore('.') | number)
}}"
blockers: "{{ fromJson(outputs.dependents.body) | jq('[.edges[] | select(.target
== \"' ~ vars.tenant ~ '_' ~ inputs.namespace ~ '_' ~ item.value
~ '\" or (.source == \"' ~ vars.tenant ~ '_' ~ inputs.namespace
~ '_' ~ item.value ~ '\" and .relation == \"FLOW_TRIGGER\"))] |
length') | first }}"
- id: is_unused
type: io.kestra.plugin.core.flow.If
condition: "{{ (outputs.assessment.values.runs | number) == 0 and
outputs.assessment.values.edited_recently == 'false' and
(outputs.assessment.values.blockers | number) == 0 }}"
then:
- id: report_candidate
type: io.kestra.plugin.core.log.Log
message: "{{ item.value }} is unused: no run and no edit within the idle period,
and no dependent flow.{{ inputs.dry_run ? ' Dry run, nothing
changed.' : ' Quarantining it.' }}"
- id: disable_triggers
type: io.kestra.plugin.kestra.triggers.Toggle
runIf: "{{ not inputs.dry_run }}"
kestraUrl: "{{ inputs.kestra_url }}"
tenantId: "{{ vars.tenant }}"
namespace: "{{ inputs.namespace }}"
flowId: "{{ item.value }}"
enabled: false
auth:
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: quarantine
type: io.kestra.plugin.core.kv.Set
runIf: "{{ not inputs.dry_run }}"
key: "retire-quarantine_{{ inputs.namespace }}_{{ item.value }}"
kvType: JSON
value: "{{ {'flowId': item.value, 'namespace': inputs.namespace,
'quarantinedAt': now(), 'source':
fromJson(outputs.flow_source.body).source} | toJson }}"
else:
- id: report_kept
type: io.kestra.plugin.core.log.Log
message: "{{ item.value }} kept: {{ outputs.assessment.values.runs }} run(s) in
the idle period, edited recently: {{
outputs.assessment.values.edited_recently }}, {{
outputs.assessment.values.blockers }} dependent flow(s)."
- id: quarantined
type: io.kestra.plugin.core.kv.GetKeys
prefix: "retire-quarantine_{{ inputs.namespace }}_"
- id: review_quarantined
type: io.kestra.plugin.core.flow.Loop
description: For each quarantined flow, cancel the retirement if it ran or
gained a dependent, or mark it ready once the quarantine period is
over.
runIf: "{{ not inputs.dry_run and (outputs.quarantined.keys | length) > 0 }}"
values: "{{ outputs.quarantined.keys }}"
tasks:
- id: record
type: io.kestra.plugin.core.kv.Get
key: "{{ item.value }}"
errorOnMissing: true
- id: runs_in_quarantine
type: io.kestra.plugin.kestra.executions.Count
kestraUrl: "{{ inputs.kestra_url }}"
tenantId: "{{ vars.tenant }}"
namespaces:
- "{{ inputs.namespace }}"
flowId: "{{ outputs.record.value.flowId }}"
startDate: "{{ outputs.record.value.quarantinedAt }}"
auth:
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: dependents_now
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows/{{ inputs.namespace
}}/{{ outputs.record.value.flowId }}/dependencies"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: verdict
type: io.kestra.plugin.core.output.OutputValues
values:
used: "{{ outputs.runs_in_quarantine.count > 0 }}"
blockers: "{{ fromJson(outputs.dependents_now.body) | jq('[.edges[] |
select(.target == \"' ~ vars.tenant ~ '_' ~ inputs.namespace ~
'_' ~ outputs.record.value.flowId ~ '\" or (.source == \"' ~
vars.tenant ~ '_' ~ inputs.namespace ~ '_' ~
outputs.record.value.flowId ~ '\" and .relation ==
\"FLOW_TRIGGER\"))] | length') | first }}"
elapsed: "{{ (now() | timestamp) - (outputs.record.value.quarantinedAt |
timestamp) >= (inputs.quarantine_period | string |
substringBefore('.') | number) }}"
- id: decide
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.verdict.values.used == 'true' or
(outputs.verdict.values.blockers | number) > 0 }}"
then:
- id: cancel_retirement
type: io.kestra.plugin.kestra.triggers.Toggle
description: The flow is still in use, so its triggers come back on.
kestraUrl: "{{ inputs.kestra_url }}"
tenantId: "{{ vars.tenant }}"
namespace: "{{ inputs.namespace }}"
flowId: "{{ outputs.record.value.flowId }}"
enabled: true
auth:
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: drop_quarantine
type: io.kestra.plugin.core.kv.Delete
key: "{{ item.value }}"
- id: drop_ready
type: io.kestra.plugin.core.kv.Delete
description: A flow already waiting for approval must not be deleted once it is
used again.
key: "retire-ready_{{ inputs.namespace }}_{{ outputs.record.value.flowId }}"
- id: log_cancel
type: io.kestra.plugin.core.log.Log
level: WARN
message: "Retirement of {{ outputs.record.value.flowId }} cancelled: {{
outputs.runs_in_quarantine.count }} run(s) and {{
outputs.verdict.values.blockers }} dependent flow(s) since
quarantine."
else:
- id: mark_ready
type: io.kestra.plugin.core.kv.Set
runIf: "{{ outputs.verdict.values.elapsed == 'true' }}"
key: "retire-ready_{{ inputs.namespace }}_{{ outputs.record.value.flowId }}"
kvType: JSON
value: "{{ outputs.record.value | toJson }}"
- id: ready
type: io.kestra.plugin.core.kv.GetKeys
prefix: "retire-ready_{{ inputs.namespace }}_"
- id: retire_ready
type: io.kestra.plugin.core.flow.If
condition: "{{ not inputs.dry_run and (outputs.ready.keys | length) > 0 }}"
then:
- id: approval
type: io.kestra.plugin.core.flow.Pause
description: "Resume to delete the flows listed in the outputs of the `ready`
task. Without approval within a day, nothing is deleted and they
stay quarantined."
pauseDuration: P1D
behavior: CANCEL
- id: delete_ready
type: io.kestra.plugin.core.flow.Loop
values: "{{ outputs.ready.keys }}"
tasks:
- id: ready_record
type: io.kestra.plugin.core.kv.Get
key: "{{ item.value }}"
errorOnMissing: true
- id: archive
type: io.kestra.plugin.core.kv.Set
description: Keeps the source for 180 days so the flow can be restored, written
before the delete.
key: "retire-archive_{{ inputs.namespace }}_{{ outputs.ready_record.value.flowId
}}"
kvType: JSON
value: "{{ outputs.ready_record.value | toJson }}"
ttl: P180D
- id: delete_flow
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows/{{ inputs.namespace
}}/{{ outputs.ready_record.value.flowId }}"
method: DELETE
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: check_deleted
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows/{{ inputs.namespace
}}/{{ outputs.ready_record.value.flowId }}"
options:
allowFailed: true
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: assert_deleted
type: io.kestra.plugin.core.execution.Assert
conditions:
- "{{ outputs.check_deleted.code == 404 }}"
errorMessage: "{{ outputs.ready_record.value.flowId }} still exists after the
delete (HTTP {{ outputs.check_deleted.code }})"
- id: forget_ready
type: io.kestra.plugin.core.kv.Delete
key: "{{ item.value }}"
- id: forget_quarantined
type: io.kestra.plugin.core.kv.Delete
key: "retire-quarantine_{{ inputs.namespace }}_{{
outputs.ready_record.value.flowId }}"
triggers:
- id: weekly_review
type: io.kestra.plugin.core.trigger.Schedule
cron: "0 7 * * 1"
inputs:
dry_run: false
