id: namespace-release-freeze-calendar-controller
namespace: company.team
description: |
Release freeze calendar for a Kestra namespace. One `ScheduleOnDates` trigger fires at every freeze start and switches
off the Schedule triggers of the target namespace, recording what it switched off in KV. A second `ScheduleOnDates`
trigger fires at every freeze end and switches back on only those. In-flight executions are listed and, behind an
approval, killed. Every run leaves an audit CSV.
labels:
team: platform
purpose: release-freeze
inputs:
- id: phase
type: SELECT
values: [ preview, freeze, thaw ]
defaults: preview
description: |
preview (default) lists what a freeze would switch off and what is in flight, changes nothing. freeze switches the
in-scope Schedule triggers off and records them. thaw switches back on what the matching freeze run recorded.
The two triggers pass freeze and thaw themselves; manual runs pick one here.
- id: window
type: STRING
defaults: ""
description: |
Name of a window in `variables.freeze_calendar`, for manual freeze or thaw runs. Triggered runs leave it empty:
the window is found from the trigger date, which must match a `start` (freeze) or an `end` (thaw) of the calendar.
- id: target_namespace
type: STRING
defaults: company.team
description: Namespace whose Schedule triggers are frozen. Exactly this
namespace, child namespaces are not included.
- id: flow_pattern
type: STRING
defaults: ".*"
description: Regular expression tested against each flow id of the target
namespace. Only matching flows are frozen (`^etl-` freezes the ETL flows
only).
- id: exclude_flows
type: JSON
defaults: "[]"
description: Flow ids never frozen, for instance the alerting flows that must
keep running during a freeze. This controller excludes itself.
- id: inflight_policy
type: SELECT
values: [ report, approve_kill ]
defaults: report
description: |
What a freeze run does with executions already running in the frozen scope. report lists them. approve_kill pauses
for a human; the executions are killed only when the approver sets `kill_inflight` to true.
- id: approval_timeout
type: DURATION
defaults: PT2H
description: How long the approve_kill pause waits. Without an answer nothing is
killed and the run continues.
- id: record_ttl
type: DURATION
defaults: P90D
description: Lifetime of the KV record that lists what the freeze run switched
off. Must outlive the longest freeze window.
- 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
schedule_type_prefix: io.kestra.plugin.core.trigger.Schedule
inflight_states: [ CREATED, QUEUED, RUNNING, PAUSED, RETRYING ]
date_format: "yyyy-MM-dd'T'HH:mm'Z'"
freeze_calendar:
- name: black-friday-2026
start: "2026-11-26T00:00:00+05:30"
end: "2026-11-30T09:00:00+05:30"
- name: year-end-2026
start: "2026-12-18T18:00:00+05:30"
end: "2027-01-04T09:00:00+05:30"
checks:
- when: "{{ inputs.exclude_flows | jq('type == \"array\"') | first }}"
message: exclude_flows must be a JSON array of flow ids, for example
["alerting", "on-call-pager"].
style: ERROR
behavior: BLOCK_EXECUTION
- when: "{{ {'ns': inputs.target_namespace} | jq('.ns |
test(\"^[a-z0-9][a-z0-9._-]*$\")') | first }}"
message: target_namespace must be a single namespace such as company.team
(lowercase letters, digits, dots, dashes). A malformed value would make
the trigger search match nothing or everything.
style: ERROR
behavior: BLOCK_EXECUTION
concurrency:
limit: 1
behavior: QUEUE
sla:
- id: max_runtime
type: MAX_DURATION
duration: PT12H
behavior: FAIL
labels:
freeze_sla: breached
triggers:
- id: freeze_start
type: io.kestra.plugin.core.trigger.ScheduleOnDates
description: |
Fires once at each `start` of the calendar, read from `variables.freeze_calendar` so the two triggers and the window
lookup share one list. `recoverMissedSchedules: LAST` still freezes if the scheduler was down at that minute.
Shipped disabled: enable it after replacing the sample dates, because it switches off other flows' triggers.
disabled: true
dates: "{{ vars.freeze_calendar | jq('[.[].start]') | first }}"
recoverMissedSchedules: LAST
inputs:
phase: freeze
- id: freeze_end
type: io.kestra.plugin.core.trigger.ScheduleOnDates
description: Fires once at each `end` of the calendar and restores what the
matching freeze run recorded.
disabled: true
dates: "{{ vars.freeze_calendar | jq('[.[].end]') | first }}"
recoverMissedSchedules: LAST
inputs:
phase: thaw
tasks:
- id: normalize
type: io.kestra.plugin.core.output.OutputValues
description: |
Calendar dates may carry any offset and the trigger date arrives in the server zone, so both are rewritten in UTC
at minute precision before they are compared. `selector` is the jq predicate that finds the window for this run.
values:
anchor_utc: "{{ (trigger.date ?? execution.startDate) | date(vars.date_format,
timeZone='UTC') }}"
calendar_utc: >-
[{% for w in vars.freeze_calendar %}{"name": "{{ w.name }}", "start":
"{{ w.start | date(vars.date_format, timeZone='UTC') }}", "end": "{{
w.end | date(vars.date_format, timeZone='UTC') }}"}{% if not loop.last
%}, {% endif %}{% endfor %}]
selector: >-
{% set a = (trigger.date ?? execution.startDate) |
date(vars.date_format, timeZone='UTC') %}{{ inputs.window != '' ? '.name
== "' ~ inputs.window ~ '"' : (inputs.phase == 'freeze' ? '.start == "'
~ a ~ '"' : (inputs.phase == 'thaw' ? '.end == "' ~ a ~ '"' : '(.start
== "' ~ a ~ '" or .end == "' ~ a ~ '")')) }}
state_key: "release_freeze_{{ inputs.target_namespace | replace({'.': '_'}) }}"
run_url: "{{ kestra.url ?? inputs.kestra_api_url }}/ui/{{ vars.tenant
}}/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}"
- id: resolve
type: io.kestra.plugin.core.output.OutputValues
description: The calendar window of this run. Empty when the trigger date
matches no start or end, or a manual run named an unknown window.
values:
window: "{{ outputs.normalize.values.calendar_utc | jq('[.[] | select(' ~
outputs.normalize.values.selector ~ ')] | first // {}') | first | toJson
}}"
window_name: "{{ outputs.normalize.values.calendar_utc | jq('[.[] | select(' ~
outputs.normalize.values.selector ~ ')] | first // {} | .name // \"\"')
| first }}"
- id: guard_window
type: io.kestra.plugin.core.flow.If
description: A freeze or thaw must belong to a calendar window, otherwise the KV
record and the audit cannot say which freeze this was. Preview runs
continue with a blank window.
condition: "{{ inputs.phase != 'preview' and outputs.resolve.values.window_name
== '' }}"
then:
- id: unknown_window
type: io.kestra.plugin.core.execution.Fail
errorMessage: "No calendar window for this {{ inputs.phase }} run: {{
inputs.window != '' ? 'window \"' ~ inputs.window ~ '\" is not in
variables.freeze_calendar' : 'the date ' ~
outputs.normalize.values.anchor_utc ~ ' matches no ' ~ (inputs.phase
== 'freeze' ? 'start' : 'end') ~ ' in variables.freeze_calendar.
Manual runs need the window input; triggered runs mean the calendar
changed after the trigger was scheduled' }} (known windows: {{
vars.freeze_calendar | jq('[.[].name] | join(\", \")') | first }})."
- id: inventory
type: io.kestra.plugin.core.http.Request
description: |
Every trigger of the target namespace with its definition and its live state in one call. `state.disabled` is the
flag the UI toggle and the Toggle task change; `state.sourceDisabled` is `disabled: true` in the flow source, which
no API call can clear. The inventory is read again after the toggles, so the record holds what really changed.
uri: "{{ inputs.kestra_api_url }}/api/v1/{{ vars.tenant }}/triggers/search"
params:
"filters[namespace][EQUALS]": "{{ inputs.target_namespace }}"
size: "500"
method: GET
timeout: PT1M
retry: { type: exponential, interval: PT5S, maxInterval: PT1M, maxAttempts: 3 }
options:
auth:
{
type: BASIC,
username: "{{ secret('KESTRA_API_USERNAME') }}",
password: "{{ secret('KESTRA_API_PASSWORD') }}"
}
- id: inventory_complete
type: io.kestra.plugin.core.execution.Assert
description: A freeze that only saw the first page would leave triggers running
and the thaw would never know about them.
conditions:
- "{{ fromJson(outputs.inventory.body).total <= 500 }}"
errorMessage: "{{ fromJson(outputs.inventory.body).total }} triggers in {{
inputs.target_namespace }}, more than the 500 this flow reads in one page.
Narrow target_namespace or add pagination before freezing."
- id: scope
type: io.kestra.plugin.core.output.OutputValues
description: |
One row per trigger: flow, trigger, type, both disabled flags and the next run. `in_scope` is the flow filter
(pattern, exclusions, never this controller), `schedule` is true for Schedule and ScheduleOnDates triggers.
Flow, polling and other triggers stay untouched during a freeze (Webhook triggers have no state entry at all).
values:
triggers: >-
{{ fromJson(outputs.inventory.body).results | jq('[.[] | {flow:
.state.flowId, trigger: .state.triggerId, type: .trigger.type, disabled:
(.state.disabled == true), source_disabled: (.state.sourceDisabled ==
true), next_run: (.state.nextEvaluationDate // "")} | . + {schedule:
(.type | startswith("' ~ vars.schedule_type_prefix ~ '")), in_scope:
((.flow | test(' ~ (inputs.flow_pattern | toJson) ~ ')) and .flow != "'
~ flow.id ~ '" and ((' ~ (inputs.exclude_flows | toJson) ~ ' - [.flow] |
length) == (' ~ (inputs.exclude_flows | toJson) ~ ' | length)))}]') |
first | toJson }}
- id: select
type: io.kestra.plugin.core.output.OutputValues
description: |
`to_disable` is what a freeze switches off: in scope, a schedule, and currently on. Triggers already off are listed
as `left_alone` and never recorded, so the thaw does not switch on something an operator had switched off on purpose.
values:
to_disable: "{{ outputs.scope.values.triggers | jq('[.[] | select(.in_scope and
.schedule and (.disabled or .source_disabled | not))]') | first | toJson
}}"
left_alone: "{{ outputs.scope.values.triggers | jq('[.[] | select(.in_scope and
.schedule and (.disabled or .source_disabled))]') | first | toJson }}"
other_count: "{{ outputs.scope.values.triggers | jq('[.[] | select(.in_scope and
(.schedule | not))] | length') | first }}"
flows_in_scope: "{{ outputs.scope.values.triggers | jq('[.[] | select(.in_scope)
| .flow] | unique | length') | first }}"
- id: previous_record
type: io.kestra.plugin.core.kv.Get
description: The record of a freeze that is still in force, if any. Read for
every phase, missing is a normal answer.
key: "{{ outputs.normalize.values.state_key }}"
errorOnMissing: false
- id: plan_report
type: io.kestra.plugin.core.log.Log
message: |
{{ inputs.phase | upper }} window={{ outputs.resolve.values.window_name != '' ? outputs.resolve.values.window_name : '(none)' }} namespace={{ inputs.target_namespace }} pattern={{ inputs.flow_pattern }} anchor={{ outputs.normalize.values.anchor_utc }}
{{ outputs.select.values.flows_in_scope }} flow(s) in scope. Schedule triggers on: {{ fromJson(outputs.select.values.to_disable) | length }}, already off (left alone): {{ fromJson(outputs.select.values.left_alone) | length }}, non-schedule triggers untouched: {{ outputs.select.values.other_count }}.
{% for t in fromJson(outputs.select.values.to_disable) %} on : {{ t.flow }}/{{ t.trigger }} next run {{ t.next_run }}
{% endfor %}{% for t in fromJson(outputs.select.values.left_alone) %} off: {{ t.flow }}/{{ t.trigger }}{{ t.source_disabled ? ' (disabled in flow source)' : '' }}
{% endfor %}Existing freeze record: {{ outputs.previous_record.value is defined ? (outputs.previous_record.value.window ~ ' frozen at ' ~ outputs.previous_record.value.frozen_at ~ ' by ' ~ outputs.previous_record.value.execution_id) : 'none' }}.
- id: apply
type: io.kestra.plugin.core.flow.Switch
description: The only place that writes to other flows' triggers. preview does
nothing here.
value: "{{ inputs.phase }}"
cases:
preview:
- id: preview_only
type: io.kestra.plugin.core.log.Log
message: "Preview: nothing was switched. Run with phase=freeze to apply."
freeze:
- id: refuse_double_freeze
type: io.kestra.plugin.core.flow.If
description: Two freeze runs without a thaw would overwrite the record and the
thaw would restore only the second, shorter list.
condition: "{{ outputs.previous_record.value is defined }}"
then:
- id: already_frozen
type: io.kestra.plugin.core.execution.Fail
errorMessage: "{{ inputs.target_namespace }} is already frozen by window {{
outputs.previous_record.value.window }} (execution {{
outputs.previous_record.value.execution_id }}, {{
outputs.previous_record.value.frozen_at }}). Run phase=thaw for
that window first, or delete KV key {{
outputs.normalize.values.state_key }} if the record is stale.
Nothing was switched."
- id: disable
type: io.kestra.plugin.core.flow.Loop
description: |
One Toggle per trigger, sequential. A failing toggle does not stop the others (`transmitFailed: false`); the
re-read below is the proof of what happened. On 2.0.5 Toggle logs "0 triggers found to toggle" even when it
worked, because the bulk endpoint answers `totalItems` and the task reads `count`; do not trust that line.
values: "{{ outputs.select.values.to_disable }}"
concurrencyLimit: 1
transmitFailed: false
tasks:
- id: switch_off
type: io.kestra.plugin.kestra.triggers.Toggle
kestraUrl: "{{ inputs.kestra_api_url }}"
tenantId: "{{ vars.tenant }}"
auth:
username: "{{ secret('KESTRA_API_USERNAME') }}"
password: "{{ secret('KESTRA_API_PASSWORD') }}"
namespace: "{{ inputs.target_namespace }}"
flowId: "{{ fromJson(item.value).flow }}"
trigger: "{{ fromJson(item.value).trigger }}"
enabled: false
thaw:
- id: require_record
type: io.kestra.plugin.core.flow.If
description: Without a record there is nothing safe to switch on. Guessing from
the current state would also re-enable triggers someone switched off
by hand.
condition: "{{ outputs.previous_record.value is not defined }}"
then:
- id: nothing_to_restore
type: io.kestra.plugin.core.execution.Fail
errorMessage: "No freeze record under KV key {{
outputs.normalize.values.state_key }}: the freeze run for {{
inputs.target_namespace }} never ran, already thawed, or the
record expired (record_ttl {{ inputs.record_ttl }}). Switch
triggers on by hand if needed; this run changed nothing."
- id: restore
type: io.kestra.plugin.core.flow.Loop
description: Switches on exactly the triggers the freeze run recorded, nothing
else.
values: "{{ (outputs.previous_record.value.disabled ?? []) | toJson }}"
concurrencyLimit: 1
transmitFailed: false
tasks:
- id: switch_on
type: io.kestra.plugin.kestra.triggers.Toggle
kestraUrl: "{{ inputs.kestra_api_url }}"
tenantId: "{{ vars.tenant }}"
auth:
username: "{{ secret('KESTRA_API_USERNAME') }}"
password: "{{ secret('KESTRA_API_PASSWORD') }}"
namespace: "{{ inputs.target_namespace }}"
flowId: "{{ fromJson(item.value).flow }}"
trigger: "{{ fromJson(item.value).trigger }}"
enabled: true
- id: reread
type: io.kestra.plugin.core.http.Request
description: Same query as inventory, after the toggles. The audit, the record
and the final verdict all come from this read, not from the Toggle task.
uri: "{{ inputs.kestra_api_url }}/api/v1/{{ vars.tenant }}/triggers/search"
params:
"filters[namespace][EQUALS]": "{{ inputs.target_namespace }}"
size: "500"
method: GET
timeout: PT1M
retry: { type: exponential, interval: PT5S, maxInterval: PT1M, maxAttempts: 3 }
options:
auth:
{
type: BASIC,
username: "{{ secret('KESTRA_API_USERNAME') }}",
password: "{{ secret('KESTRA_API_PASSWORD') }}"
}
- id: verify
type: io.kestra.plugin.core.output.OutputValues
description: |
`expected` is what this phase meant to change (freeze: to_disable, thaw: the record). Each row gets its state
after the toggles; `mismatch` lists the ones that did not land. `rows` is the trigger part of the audit CSV; triggers
that were already off and are not in `expected` are reported as left alone.
values:
expected: "{{ inputs.phase == 'thaw' ? ((outputs.previous_record.value.disabled
?? []) | toJson) : outputs.select.values.to_disable }}"
after: "{{ fromJson(outputs.reread.body).results | jq('[.[] | {key:
(.state.flowId + \"/\" + .state.triggerId), off: ((.state.disabled ==
true) or (.state.sourceDisabled == true))}]') | first | toJson }}"
want_off: "{{ inputs.phase == 'freeze' }}"
rows: >-
{{ {'phase': inputs.phase, 'expected': fromJson(inputs.phase == 'thaw' ?
((outputs.previous_record.value.disabled ?? []) | toJson) :
outputs.select.values.to_disable), 'left':
fromJson(outputs.select.values.left_alone), 'after':
fromJson(outputs.reread.body).results | jq('[.[] | {key: (.state.flowId
+ "/" + .state.triggerId), off: ((.state.disabled == true) or
(.state.sourceDisabled == true))}]') | first} | jq('.phase as $p |
.after as $a | ([.expected[] | . as $e | ([$a[] | select(.key ==
($e.flow + "/" + $e.trigger)) | .off] | first) as $off | {kind:
"trigger", flow: $e.flow, item: $e.trigger, type: ($e.type | split(".")
| last), before: (if $p == "thaw" then "disabled" else "enabled" end),
after: (if $off == null then "missing" elif $off then "disabled" else
"enabled" end), action: (if $off == null then (if $p == "thaw" then
"gone_nothing_to_restore" else "MISMATCH_trigger_missing" end) elif $p
== "preview" then "would_disable" elif $p == "freeze" then (if $off then
"disabled" else "MISMATCH_still_enabled" end) else (if $off then
"MISMATCH_still_disabled" else "restored" end) end)}]) + ([.expected[] |
.flow + "/" + .trigger] as $ek | [.left[] | select(($ek - [.flow + "/" +
.trigger] | length) == ($ek | length)) | {kind: "trigger", flow: .flow,
item: .trigger, type: (.type | split(".") | last), before: "disabled",
after: "disabled", action: (if .source_disabled then
"left_alone_source_disabled" else "left_alone_already_disabled"
end)}])') | first | toJson }}
- id: outcome
type: io.kestra.plugin.core.output.OutputValues
values:
changed: "{{ outputs.verify.values.rows | jq('[.[] | select(.action ==
\"disabled\" or .action == \"restored\") | {flow, trigger: .item,
type}]') | first | toJson }}"
mismatch: "{{ outputs.verify.values.rows | jq('[.[] | select(.action |
startswith(\"MISMATCH\")) | .flow + \"/\" + .item]') | first | toJson
}}"
- id: persist
type: io.kestra.plugin.core.flow.Switch
description: The record is written right after the toggles and before the
in-flight handling, so a run killed during the approval pause still leaves
a complete record for the thaw.
value: "{{ inputs.phase }}"
cases:
freeze:
- id: write_record
type: io.kestra.plugin.core.kv.Set
description: "Only triggers confirmed off by the re-read are recorded.
`overwrite: false` is a second guard against a double freeze."
key: "{{ outputs.normalize.values.state_key }}"
kvType: JSON
overwrite: false
ttl: "{{ inputs.record_ttl }}"
value: |
{{ {'window': outputs.resolve.values.window_name, 'namespace': inputs.target_namespace, 'flow_pattern': inputs.flow_pattern, 'frozen_at': outputs.normalize.values.anchor_utc, 'execution_id': execution.id, 'disabled': fromJson(outputs.outcome.values.changed), 'left_alone': fromJson(outputs.select.values.left_alone) | jq('[.[] | .flow + "/" + .trigger]') | first} | toJson }}
thaw:
- id: close_record
type: io.kestra.plugin.core.flow.If
description: The record is deleted only when every recorded trigger is back on.
Otherwise it stays, and the next thaw run retries exactly the
missing ones.
condition: "{{ fromJson(outputs.outcome.values.mismatch) | length == 0 }}"
then:
- id: delete_record
type: io.kestra.plugin.core.kv.Delete
key: "{{ outputs.normalize.values.state_key }}"
else:
- id: keep_record
type: io.kestra.plugin.core.log.Log
level: WARN
message: "Record {{ outputs.normalize.values.state_key }} kept: {{
fromJson(outputs.outcome.values.mismatch) | length }} trigger(s)
still off. Run phase=thaw again once the cause is fixed."
defaults:
- id: nothing_to_persist
type: io.kestra.plugin.core.log.Log
message: "Preview: no record written{{ outputs.previous_record.value is defined
? \", the existing record is untouched\" : \"\" }}."
- id: inflight
type: io.kestra.plugin.core.http.Request
description: |
Executions of the target namespace that are not finished. Switching triggers off stops new runs, it does not stop
running ones, so a freeze lists them. Skipped on thaw. This is a plain API call rather than `executions.Query`:
that task adds one EQUALS filter per entry of `states`, so two or more states match nothing (checked against
the plugin source and against a running execution during the tests).
runIf: "{{ inputs.phase != 'thaw' }}"
uri: "{{ inputs.kestra_api_url }}/api/v1/{{ vars.tenant }}/executions/search"
params:
"filters[namespace][EQUALS]": "{{ inputs.target_namespace }}"
"filters[state][IN]": "{{ vars.inflight_states | join(',') }}"
size: "200"
method: GET
timeout: PT1M
retry: { type: exponential, interval: PT5S, maxInterval: PT1M, maxAttempts: 3 }
options:
auth:
{
type: BASIC,
username: "{{ secret('KESTRA_API_USERNAME') }}",
password: "{{ secret('KESTRA_API_PASSWORD') }}"
}
- id: inflight_scope
type: io.kestra.plugin.core.output.OutputValues
description: Only executions of flows in scope count, and never this execution.
Empty on thaw, where the search is skipped.
values:
executions: >-
{{ ((outputs.inflight.body is defined ?
fromJson(outputs.inflight.body).results : []) | jq('[.[] | select(.id !=
"' ~ execution.id ~ '" and (.flowId | test(' ~ (inputs.flow_pattern |
toJson) ~ '))) | {id, flowId, state: .state.current, since:
.state.startDate}]') | first) | toJson }}
- id: inflight_gate
type: io.kestra.plugin.core.flow.If
description: A human decides about running work; the default policy only reports it.
condition: "{{ inputs.phase == 'freeze' and inputs.inflight_policy ==
'approve_kill' and (fromJson(outputs.inflight_scope.values.executions) |
length) > 0 }}"
then:
- id: approve_kill
type: io.kestra.plugin.core.flow.Pause
description: "Triggers are already off at this point. `behavior: RESUME` with
the default false means a timeout kills nothing."
pauseDuration: "{{ inputs.approval_timeout }}"
behavior: RESUME
onResume:
- id: kill_inflight
type: BOOL
displayName: Kill the in-flight executions listed by plan_report and
inflight_scope? They end in state KILLED.
defaults: false
- id: reason
type: STRING
defaults: ""
- id: decision
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.approve_kill.onResume.kill_inflight ?? false }}"
then:
- id: kill
type: io.kestra.plugin.core.flow.Loop
description: One Kill per execution; an execution that finished during the pause
makes its Kill fail, which is recorded, not fatal.
values: "{{ outputs.inflight_scope.values.executions }}"
concurrencyLimit: 1
transmitFailed: false
tasks:
- id: kill_one
type: io.kestra.plugin.kestra.executions.Kill
kestraUrl: "{{ inputs.kestra_api_url }}"
tenantId: "{{ vars.tenant }}"
auth:
username: "{{ secret('KESTRA_API_USERNAME') }}"
password: "{{ secret('KESTRA_API_PASSWORD') }}"
executionId: "{{ fromJson(item.value).id }}"
propagateKill: true
else:
- id: left_running
type: io.kestra.plugin.core.log.Log
message: "In-flight executions left running ({{
(outputs.approve_kill.onResume.reason ?? '') != '' ?
outputs.approve_kill.onResume.reason : 'no decision or
kill_inflight=false' }}): {{
fromJson(outputs.inflight_scope.values.executions) | jq('[.[] |
.flowId + \"/\" + .id] | join(\", \")') | first }}"
- id: inflight_rows
type: io.kestra.plugin.core.output.OutputValues
description: Execution part of the audit CSV. `approved` is false when there was
no pause.
values:
rows: >-
{{ {'execs': fromJson(outputs.inflight_scope.values.executions),
'killed': (outputs.approve_kill.onResume.kill_inflight ?? false),
'phase': inputs.phase} | jq('.phase as $p | .killed as $k | [.execs[] |
{kind: "execution", flow: .flowId, item: .id, type: "execution", before:
.state, after: (if $k then "kill_requested" else .state end), action:
(if $p == "preview" then "reported" elif $k then "killed" else
"left_running" end)}]') | first | toJson }}
- id: mismatch_gate
type: io.kestra.plugin.core.flow.If
description: Fails last, after the record and the in-flight handling, so a
partial freeze is still fully recorded.
condition: "{{ fromJson(outputs.outcome.values.mismatch) | length > 0 }}"
then:
- id: incomplete
type: io.kestra.plugin.core.execution.Fail
errorMessage: "{{ inputs.phase | upper }} incomplete: {{
fromJson(outputs.outcome.values.mismatch) | length }} trigger(s) did
not change state: {{ fromJson(outputs.outcome.values.mismatch) |
join(', ') }}. The API user may lack permission, or the trigger is
disabled in the flow source. {{ inputs.phase == 'freeze' ? 'The record
holds the triggers that did switch off.' : 'The record was kept; run
phase=thaw again.' }}"
errors:
- id: log_failure
type: io.kestra.plugin.core.log.Log
level: ERROR
message: "Freeze controller {{ inputs.phase }} failed in task {{ ((errorLogs() |
length) > 0 and (errorLogs()[0].taskId ?? '') != '') ?
errorLogs()[0].taskId : 'none' }}: {{ (errorLogs() | length) > 0 ?
errorLogs()[0].message : 'no task error was logged, which is how an SLA
breach ends the run' }} {{ outputs.normalize.values.run_url ??
execution.id }}"
- id: dead_letter
type: io.kestra.plugin.core.kv.Set
description: Last failure per namespace, kept 30 days next to the freeze record
so the operator who looks at the record sees why the matching run failed.
key: "{{ outputs.normalize.values.state_key ?? ('release_freeze_' ~
(inputs.target_namespace | replace({'.': '_'}))) }}_last_error"
kvType: JSON
ttl: P30D
value: "{{ {'phase': inputs.phase, 'window': outputs.resolve.values.window_name
?? '', 'execution_id': execution.id, 'failed_at': execution.startDate,
'task': (errorLogs() | length) > 0 ? (errorLogs()[0].taskId ?? '') : '',
'message': (errorLogs() | length) > 0 ? errorLogs()[0].message : ''} |
toJson }}"
- 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: ":snowflake: Freeze controller {{ inputs.phase }} for {{
inputs.target_namespace }} FAILED in `{{ ((errorLogs() | length) > 0 and
(errorLogs()[0].taskId ?? '') != '') ? errorLogs()[0].taskId : 'none' }}`:
{{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'no task error
was logged (SLA breach)' }} {{ outputs.normalize.values.run_url ??
execution.id }}"
finally:
- id: write_audit
type: io.kestra.plugin.core.storage.Write
description: Always written, also after a refusal or a partial failure. Rows
that do not exist yet (a run that failed before the toggles) are simply
absent; the header is always there.
extension: .csv
content: |
run_execution_id,phase,window,namespace,kind,flow_id,item_id,type,before,after,action
{{ (fromJson(outputs.verify.values.rows ?? '[]') | merge(fromJson(outputs.inflight_rows.values.rows ?? '[]'))) | jq('[.[] | ["' ~ execution.id ~ '", "' ~ inputs.phase ~ '", "' ~ (outputs.resolve.values.window_name ?? '') ~ '", "' ~ inputs.target_namespace ~ '", .kind, .flow, .item, .type, .before, .after, .action] | @csv] | join("\n")') | first }}
- id: log_audit
type: io.kestra.plugin.core.log.Log
message: "Audit CSV {{ outputs.write_audit.uri }}: {{
fromJson(outputs.verify.values.rows ?? '[]') | length }} trigger row(s),
{{ fromJson(outputs.inflight_rows.values.rows ?? '[]') | length }}
execution row(s), {{ fromJson(outputs.outcome.values.mismatch ?? '[]') |
length }} mismatch(es)."
afterExecution:
- id: notify_outcome
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Runs once the final state is known. An SLA breach fails the run at
once and skips `finally`, so this is the only place that reports it.
Ordinary failures were reported by `errors`.
runIf: "{{ execution.state != 'FAILED' or (labels.freeze_sla ?? '') ==
'breached' }}"
allowFailure: true
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: ":snowflake: Freeze controller {{ inputs.phase }} window {{
outputs.resolve.values.window_name ?? '?' }} on {{ inputs.target_namespace
}} ended {{ execution.state }}{{ (labels.freeze_sla ?? '') == 'breached' ?
' (12-hour SLA breached)' : '' }}: {{
fromJson(outputs.outcome.values.changed ?? '[]') | length }} trigger(s) {{
inputs.phase == 'thaw' ? 'restored' : 'switched off' }}, {{
fromJson(outputs.inflight_scope.values.executions ?? '[]') | length }} in
flight. {{ outputs.normalize.values.run_url ?? execution.id }}"
outputs:
- id: audit_csv
type: FILE
value: "{{ outputs.write_audit.uri }}"
- id: summary
type: JSON
value: "{{ {'phase': inputs.phase, 'window': outputs.resolve.values.window_name
?? '', 'namespace': inputs.target_namespace, 'changed':
fromJson(outputs.outcome.values.changed ?? '[]'), 'mismatch':
fromJson(outputs.outcome.values.mismatch ?? '[]'), 'left_alone':
fromJson(outputs.select.values.left_alone ?? '[]') | length, 'inflight':
fromJson(outputs.inflight_scope.values.executions ?? '[]') | length} |
toJson }}"
