id: kestra-namespace-move-drained-cutover
namespace: company.platform
description: >-
Moves one namespace (flows, namespace files and KV pairs) from one Kestra
instance to another, for example from a 1.3 instance to a fresh 2.0 one. The
target validates every flow before anything is written, the source is frozen
and drained before the target takes over, a failed cutover rolls itself back,
and a KV ledger lets an interrupted run resume. The same file runs on Kestra
0.21 through 2.0.
labels:
blueprint: kestra-namespace-move
concurrency:
limit: 1
behavior: QUEUE
inputs:
- id: mode
type: SELECT
values:
- STAGE
- CUTOVER
- ROLLBACK
defaults: STAGE
description: >-
STAGE copies the namespace to the target with every flow disabled and can
be repeated to copy what changed. CUTOVER asks for approval, freezes the
source, waits for its executions to finish, copies the last changes and
enables the flows on the target. ROLLBACK disables them on the target and
enables them again on the source.
- id: dry_run
type: SELECT
values:
- "true"
- "false"
defaults: "true"
description: >-
"true" computes and reports what the selected mode would change and writes
nothing. It is a SELECT and not a BOOL input because BOOL does not exist
before Kestra 0.23.
- id: namespace
type: STRING
defaults: company.billing
validator: ^[a-zA-Z0-9][a-zA-Z0-9._-]*$
description: Namespace to move. Child namespaces are not included, move them one
at a time.
- id: source_url
type: STRING
defaults: http://kestra-old:8080
validator: ^https?://[^\s/]+(/[^\s]*[^\s/])?$
description: Base URL of the instance that runs the namespace today, without a
trailing slash.
- id: target_url
type: STRING
defaults: http://kestra-new:8080
validator: ^https?://[^\s/]+(/[^\s]*[^\s/])?$
description: Base URL of the instance that takes the namespace over, without a
trailing slash.
- id: source_tenant
type: STRING
required: false
validator: ^[a-zA-Z0-9_-]*$
description: Tenant id on the source (Enterprise Edition). Leave it empty for
open source Kestra.
- id: target_tenant
type: STRING
required: false
validator: ^[a-zA-Z0-9_-]*$
description: Tenant id on the target (Enterprise Edition). Leave it empty for
open source Kestra.
- id: on_conflict
type: SELECT
values:
- STOP
- OVERWRITE
defaults: STOP
description: >-
What to do with a target flow, file or KV pair that differs from the
source and was not written by this flow. STOP refuses to write anything.
OVERWRITE replaces it; the previous version stays in the target exports
attached to this execution.
- id: drain_timeout_minutes
type: INT
defaults: 30
description: How long CUTOVER waits for the running executions of the namespace
on the source before it rolls back.
- id: smoke_test_flow
type: STRING
required: false
description: >-
Id of a flow of the namespace, without required inputs, that CUTOVER runs
on the target once it is enabled. If that run does not end in SUCCESS, the
cutover rolls back. Leave it empty to skip.
- id: smoke_test_timeout_minutes
type: INT
defaults: 10
description: How long CUTOVER waits for the smoke test execution to finish.
variables:
# jq programs live here as literal blocks, so Pebble never parses their quotes and backslashes.
checks_jq: |
. as $i
| [ (if $i.src == $i.tgt then "source_url and target_url are the same address" else empty end),
(if ($i.own == $i.ns) or ($i.own | startswith($i.ns + ".")) then
"this flow runs in " + $i.own + ", inside the namespace it would move: run it from another namespace"
else empty end),
(if $i.drain < 1 then "drain_timeout_minutes must be at least 1" else empty end),
(if $i.smoke_timeout < 1 then "smoke_test_timeout_minutes must be at least 1" else empty end) ]
instance_jq: |
. as $i
| ($i.cfg.version // "0.0.0") as $ver
| ($ver | split(".") | .[0:2] | map(tonumber)) as $v
| ($i.url + "/api/v1" + (if (($i.tenant // "") | length) > 0 then "/" + $i.tenant else "" end)) as $api
| (["CREATED", "RUNNING", "PAUSED", "RESTARTED", "KILLING", "QUEUED", "RETRYING"]
+ (if $v >= [0, 24] then ["BREAKPOINT"] else [] end)
+ (if $v >= [1, 1] then ["SUBMITTED", "RESUBMITTED"] else [] end)) as $states
| {
url: $i.url,
api: $api,
version: $ver,
edition: ($i.cfg.edition // "OSS"),
uuid: ($i.cfg.uuid // ""),
supported: ($v >= [0, 21]),
unsafe_names: (if $v >= [1, 3] then null
elif $v >= [1, 2] then "[ %\\[\\]^{}|`\"<>\\\\]"
else "[ %#?:\\[\\]^{}|`\"<>\\\\]" end),
kv_list_url: (if $v >= [1, 1]
then $api + "/kv?page=1&size=1000&filters%5Bnamespace%5D%5BEQUALS%5D=" + $i.ns
else $api + "/namespaces/" + $i.ns + "/kv" end),
pending_url: (if $v >= [0, 22]
then $api + "/executions/search?page=1&size=50&filters%5Bnamespace%5D%5BEQUALS%5D=" + $i.ns
+ "&filters%5Bstate%5D%5BIN%5D=" + ($states | join(","))
else $api + "/executions/search?page=1&size=50&namespace=" + $i.ns
+ ($states | map("&state=" + .) | join("")) end),
triggers_url: (if $v >= [0, 22]
then $api + "/triggers/search?page=1&size=1000&filters%5Bnamespace%5D%5BEQUALS%5D=" + $i.ns
else $api + "/triggers/search?page=1&size=1000&namespace=" + $i.ns end)
}
pending_jq: |
.ns as $ns
| (.page.results // []) as $r
| ([$r[] | select(.namespace == $ns) | {id: .id, flowId: .flowId, state: .state.current}]) as $mine
| (((.page.total // 0) - ($r | length)) | if . > 0 then . else 0 end) as $hidden
| $mine + (if $hidden > 0
then [{id: "", flowId: "(" + ($hidden | tostring) + " more not listed)", state: "UNKNOWN", more: $hidden}]
else [] end)
pending_count_jq: |
map(.more // 1) | add // 0
flows_jq: |
[.[] | {namespace: .namespace, id: .id, disabled: (.disabled // false),
triggers: [(.triggers // [])[] | {id: .id, type: .type, key: (.key // null)}]}]
freeze_jq: |
[.[] | select((.disabled // false) | not) | {namespace: .namespace, id: .id}]
flow_ids_jq: |
[.[] | {namespace: .namespace, id: .id}]
by_ids_jq: |
.ns as $ns | .ids | map({namespace: $ns, id: .})
not_enabled_jq: |
.flows as $f
| [.ids[] | . as $id | select(([$f[] | select(.id == $id and ((.disabled // false) | not))] | length) == 0)]
page_complete_jq: |
if type == "object" then ((.total // 0) <= ((.results // []) | length)) else true end
# Line based on purpose: only column 0 holds top-level keys and document markers, so no regex anchor
# semantics are involved. A leading --- (yamllint wants one) would split the multi-document validation body.
clean_jq: |
.s | gsub("\r\n"; "\n") | split("\n") | map(select(test("^---[ \t]*(#.*)?$") | not))
| join("\n") | sub("\\A\\s+"; "") | sub("\\s+\\z"; "")
doc_jq: |
(.s | gsub("\r\n"; "\n") | split("\n") | map(select(test("^---[ \t]*(#.*)?$") | not))) as $lines
| {
id: (($lines | map(select(startswith("id:"))) | .[0] // "")
| sub("^id:[ \t]*"; "") | sub("[ \t]+#.*$"; "") | gsub("[\"' \t]"; "")),
norm: ($lines | map(select(startswith("disabled:") | not)) | join("\n") | sub("\\A\\s+"; "") | sub("\\s+\\z"; "")),
secrets: ([$lines[] | scan("secret\\(\\s*['\"]([^'\"]+)['\"]") | .[0]] | unique)
}
# Every top-level disabled line is replaced by one disabled true at the end, and the end of the document is kept
# byte for byte, so a block scalar that ends the flow keeps its final line break.
inject_jq: |
(.s | gsub("\r\n"; "\n") | split("\n")
| map(select((test("^---[ \t]*(#.*)?$") or startswith("disabled:")) | not))) as $all
| ($all | length) as $n
| (if ($n > 0) and ($all[$n - 1] == "") then $all[0:$n - 1] else $all end) + ["disabled: true"]
| join("\n") + "\n"
ns_keep_jq: |
keys | map(select((startswith("_flows/") or (ascii_downcase | endswith(".zip"))) | not))
ns_zip_jq: |
keys | map(select((startswith("_flows/") | not) and (ascii_downcase | endswith(".zip"))))
# ArchiveDecompress stores "a b.md" as "a_b.md", and in a storage URI that is not encoded a # or ? ends the path,
# so two namespace files can land on the same internal storage file.
collisions_jq: |
def groups: with_entries(select((.key | startswith("_flows/") or (ascii_downcase | endswith(".zip"))) | not))
| to_entries | map({key: .key, u: (.value | tostring | sub("[#?].*$"; ""))})
| group_by(.u) | map(select(length > 1) | map(.key) | sort);
{source: ((.s // {}) | groups), target: ((.t // {}) | groups)}
# A KV pair whose updateDate on both sides matches the ledger is unchanged since the last copy: every write,
# even of the same value, moves updateDate, so its value is not read again.
kv_keys_jq: |
. as $i
| ($i.ledger // {}) as $L
| ([($i.s | if type == "object" then (.results // []) else . end)[] | select((.namespace // $i.ns) == $i.ns)
| {key: .key, exp: (.expirationDate // null), up: (.updateDate // null), src: true}]
+ [($i.t | if type == "object" then (.results // []) else . end)[] | select((.namespace // $i.ns) == $i.ns)
| {key: .key, exp: (.expirationDate // null), up: (.updateDate // null), src: false}])
| group_by(.key)
| map((map(select(.src)) | .[0]) as $s | (map(select(.src | not)) | .[0]) as $t
| {key: .[0].key, src: ($s != null), tgt: ($t != null), exp: ($s.exp // null), texp: ($t.exp // null),
sup: ($s.up // null), tup: ($t.up // null)})
| map(. + {known: (.src and .tgt and ((($L.kv // {})[.key] // null) != null)
and ((($L.kv_dates // {})[.key] // {}) == {s: .sup, t: .tup}))})
kv_read_jq: |
map(select(.src and (.known | not)))
kv_dates_jq: |
. as $i
| ([($i.s | if type == "object" then (.results // []) else . end)[] | select((.namespace // $i.ns) == $i.ns)
| {key: .key, value: (.updateDate // null)}] | from_entries) as $sd
| ([($i.t | if type == "object" then (.results // []) else . end)[] | select((.namespace // $i.ns) == $i.ns)
| {key: .key, value: (.updateDate // null)}] | from_entries) as $td
| [$i.keys[] | {key: ., value: {s: ($sd[.] // null), t: ($td[.] // null)}}] | from_entries
kv_merge_jq: |
.prev + ([{key: .key, value: .entry}] | from_entries)
plan_jq: |
def classify: if .t == null then "NEW" elif .s == .t then "IN_SYNC" elif (.m != null) and (.m == .t) then "UPDATE" else "CONFLICT" end;
def writes: (.st == "NEW") or (.st == "UPDATE") or ((.st == "CONFLICT") and (.policy == "OVERWRITE"));
def counts: group_by(.status) | map({key: .[0].status, value: length}) | from_entries;
. as $i
| ($i.ledger // {}) as $L
| ($i.src_docs | map({key: .id, value: .}) | from_entries) as $sd
| ($i.tgt_docs | map({key: .id, value: .}) | from_entries) as $td
| ($i.src_docs | map({key: (.idx | tostring), value: .id}) | from_entries) as $doc_by_idx
| ($i.validation | map({key: (.flow // $doc_by_idx[(.index | tostring)] // ""), value: .}) | from_entries) as $val
| ($i.kv // {}) as $read
| ([$i.kv_keys[] | .key as $k | . as $m
| {key: $k, value: (if ($read[$k] // null) != null then $read[$k]
elif ($m.src | not) then {src_type: null, src_sha: null, src_exp: null, tgt_type: null, tgt_sha: null}
else {src_type: null, src_sha: (($L.kv // {})[$k]), src_exp: $m.exp, tgt_type: null, tgt_sha: (($L.kv // {})[$k])} end)}]
| from_entries) as $kvall
| ($i.src_files | map({key: .path, value: .sha}) | from_entries) as $sf
| ($i.tgt_files | map({key: .path, value: .sha}) | from_entries) as $tf
| ([$i.src_triggers[] | (.triggerContext // .state // {})
| select(.namespace == $i.ns and (.disabled // false) and ((.sourceDisabled // false) | not))
| (.flowId + "." + .triggerId)]) as $disabled_triggers
| [ $i.src_flows[] | .id as $id
| (($val[$id].constraints // null) | if . == null then "" else tostring end) as $c
| (if ($c | length) > 0 then "INVALID"
else ({s: ($sd[$id].sha // null), t: ($td[$id].sha // null), m: (($L.flows // {})[$id] // null)} | classify) end) as $st
| {
id: $id,
key: ($sd[$id].key // null),
status: $st,
write: ({st: $st, policy: $i.on_conflict} | writes),
enabled_on_source: ((.disabled // false) | not),
constraints: (if $st == "INVALID" then $c else null end),
notices: ((($val[$id].warnings // []) | map(tostring))
+ (($val[$id].deprecationPaths // []) | map("deprecated property " + tostring))),
triggers: [(.triggers // [])[] | {id: .id, type: .type, key: (.key // null)}],
secrets: ($sd[$id].secrets // [])
} ] as $flows
| ($flows | map({key: .id, value: true}) | from_entries) as $on_source
| [ $i.tgt_flows[] | select(($on_source[.id] // false) | not)
| {id: .id, status: (if (($L.flows // {})[.id] // null) != null then "STALE" else "TARGET_ONLY" end),
enabled_on_target: ((.disabled // false) | not)} ] as $flows_extra
| [ $sf | to_entries[] | .key as $p
| (if ($i.target.unsafe_names != null) and ($p | test($i.target.unsafe_names)) then "UNSUPPORTED_NAME"
else ({s: .value, t: ($tf[$p] // null), m: (($L.files // {})[$p] // null)} | classify) end) as $st
| {path: $p, status: $st,
write: (($st != "UNSUPPORTED_NAME") and ({st: $st, policy: $i.on_conflict} | writes))} ] as $files
| [ $tf | keys[] | select(($sf[.] // null) == null)
| {path: ., status: (if (($L.files // {})[.] // null) != null then "STALE" else "TARGET_ONLY" end)} ] as $files_extra
| [ $kvall | to_entries[] | select(.value.src_sha != null) | .key as $k | .value as $e
| (if $e.src_sha == "EXPIRING" then "EXPIRING"
else ({s: $e.src_sha, t: $e.tgt_sha, m: (($L.kv // {})[$k] // null)} | classify) end) as $st
| {key: $k, status: $st, type: $e.src_type, target_type: $e.tgt_type, expires_at: $e.src_exp,
write: (($st != "EXPIRING") and ({st: $st, policy: $i.on_conflict} | writes))} ] as $kv
| [ $kvall | to_entries[] | select(.value.src_sha == null)
| {key: .key, status: (if (($L.kv // {})[.key] // null) != null then "STALE" else "TARGET_ONLY" end)} ] as $kv_extra
| [ $flows[] | .id as $fid | .triggers[] | select(.type | endswith(".Webhook"))
| {flow: $fid, trigger: .id, key: .key,
new_url: ($i.target.api + "/executions/webhook/" + $i.ns + "/" + $fid + "/" + (.key // "<key>"))} ] as $webhooks
| ([$flows[] | .triggers[] | select(.type | endswith(".Schedule"))] | length) as $schedules
| ([$flows[] | .triggers[] | select(.type | test("Realtime"))] | length) as $realtime
| ([$flows[] | .secrets[]] | unique) as $secrets
| ($i.pending | map(select(.state == "PAUSED"))) as $paused
| {
namespace: $i.ns,
mode: $i.mode,
dry_run: $i.dry,
source: {url: $i.source.url, version: $i.source.version},
target: {url: $i.target.url, version: $i.target.version},
counts: {
flows: ($flows | counts),
files: ($files | counts),
kv: ($kv | counts),
kv_values_read: ($read | length),
pending_executions: ($i.pending | map(.more // 1) | add // 0)
},
blockers: (
[$flows[] | select(.status == "INVALID") | "flow " + .id + " is rejected by the target: " + .constraints]
+ (if $i.on_conflict == "STOP" then
[$flows[] | select(.status == "CONFLICT") | "target flow " + .id + " differs from the source and was not written by this flow"]
+ [$files[] | select(.status == "CONFLICT") | "target file " + .path + " differs from the source and was not written by this flow"]
+ [$kv[] | select(.status == "CONFLICT") | "target KV pair " + .key + " differs from the source and was not written by this flow"]
else [] end)
+ ($i.collisions.source | map("namespace files " + join(", ") + " land on the same file when the source export is unpacked; rename all but one on the source"))
+ ($i.collisions.target | map("namespace files " + join(", ") + " land on the same file when the target export is unpacked; rename all but one on the target"))
+ (if $i.kv_complete then [] else
["the source holds more KV pairs in " + $i.ns + " than one page of its KV API returns (1000)"] end)
+ (if $i.kv_target_complete then [] else
["the target holds more KV pairs in " + $i.ns + " than one page of its KV API returns (1000), so conflicts cannot be checked"] end)
+ (if (($flows | length) + ($files | length) + ($kv | length)) == 0 then
["namespace " + $i.ns + " has no flow, file or KV pair on the source"] else [] end)),
warnings: (
[$flows[] | select((.notices | length) > 0) | "flow " + .id + ": " + (.notices | join(", "))]
+ [$flows_extra[] | select(.status == "STALE") | "flow " + .id + " was copied earlier but no longer exists on the source; delete it on the target if it is obsolete"]
+ [$flows_extra[] | select(.status == "TARGET_ONLY" and .enabled_on_target) | "flow " + .id + " exists only on the target and is enabled there"]
+ [$files_extra[] | select(.status == "STALE") | "file " + .path + " was copied earlier but no longer exists on the source"]
+ [$kv_extra[] | select(.status == "STALE") | "KV pair " + .key + " was copied earlier but no longer exists on the source"]
+ [$kv[] | select((.type != null) and (.target_type != null) and (.target_type != .type))
| "KV pair " + .key + " is a " + .type + " on the source and a " + .target_type + " on the target"]
+ [$kv[] | select(.status == "EXPIRING") | "KV pair " + .key + " expires within a minute and is not copied"]
+ ($i.zips | map("namespace file " + . + " is skipped because the namespace file API unpacks uploaded .zip files; copy it by hand"))
+ [$files[] | select(.status == "UNSUPPORTED_NAME") | "namespace file " + .path + " is skipped: Kestra "
+ $i.target.version + " cannot store this name unchanged; rename it on the source"]
+ ($disabled_triggers | map("trigger " + . + " is disabled on the source only; disable it on the target after the cutover"))
+ (if ($paused | length) > 0 then
["paused executions wait for a human and hold the drain: " + ($paused | map(.flowId + "/" + .id) | join(", "))]
else [] end)
+ (if $schedules > 0 then
["schedules due between the freeze and the enable run on neither instance (" + ($schedules | tostring) + " Schedule triggers)"]
else [] end)
+ (if $realtime > 0 then
["realtime triggers stop consuming on the source at the freeze and start on the target at the enable"]
else [] end)
+ (if ($webhooks | length) > 0 then ["webhook callers must switch to the target URLs listed in webhooks"] else [] end)
+ (if ($secrets | length) > 0 then ["check that these secrets exist on the target: " + ($secrets | join(", "))] else [] end)),
writes: {
flows: [$flows[] | select(.write) | .id],
flow_items: [$flows[] | select(.write) | {id: .id, key: .key, exists: (.status != "NEW")}],
files: [$files[] | select(.write) | .path],
kv: [$kv[] | select(.write) | {key: .key, expires_at: .expires_at}]
},
freeze: [$flows[] | select(.enabled_on_source) | .id],
pending: $i.pending,
webhooks: $webhooks,
secrets: $secrets,
shas: {
flows: ([$flows[] | select(.status != "INVALID") | {key: .id, value: $sd[.id].sha}] | from_entries),
files: ([$files[] | select(.status != "UNSUPPORTED_NAME") | {key: .path, value: $sf[.path]}] | from_entries),
kv: ([$kv[] | select(.status != "EXPIRING") | {key: .key, value: $kvall[.key].src_sha}] | from_entries)
},
flows: $flows,
files: $files,
kv: $kv,
extra: {flows: $flows_extra, files: $files_extra, kv: $kv_extra}
}
# After a successful write every source object is in sync with the target, so the ledger takes all their hashes.
ledger_stage_jq: |
. as $i
| ($i.ledger // {}) as $L
| $L + {
namespace: $i.ns,
source: $i.source,
target: $i.target,
flows: (($L.flows // {}) + $i.shas.flows),
files: (($L.files // {}) + $i.shas.files),
kv: (($L.kv // {}) + $i.shas.kv),
kv_dates: (($L.kv_dates // {}) + $i.kv_dates),
phase: (if (($L.phase // "") == "FROZEN") or (($L.phase // "") == "DRAINED") then $L.phase else "STAGED" end),
staged_at: $i.now,
staged_by: $i.execution,
history: ((($L.history // []) + [{at: $i.now, execution: $i.execution, mode: $i.mode, event: "staged",
flows: ($i.written.flows | length), files: ($i.written.files | length), kv: ($i.written.kv | length)}]) | .[-20:])
}
ledger_set_jq: |
(.ledger // {}) as $L
| $L + .set + {history: ((($L.history // []) + [.event]) | .[-20:])}
tasks:
- id: check_inputs
type: io.kestra.plugin.core.output.OutputValues
description: Collects every input problem, so a single run reports all of them.
values:
problems: >-
{{ {'src': inputs.source_url, 'tgt': inputs.target_url, 'ns':
inputs.namespace, 'own': flow.namespace, 'drain':
inputs.drain_timeout_minutes, 'smoke_timeout':
inputs.smoke_test_timeout_minutes} | jq(vars.checks_jq) | first | toJson
}}
- id: stop_on_input_problems
type: io.kestra.plugin.core.execution.Fail
condition: "{{ fromJson(outputs.check_inputs.values.problems) | length > 0 }}"
errorMessage: "Invalid inputs, nothing was read or written: {{
fromJson(outputs.check_inputs.values.problems) | join('; ') }}"
- id: set_labels
type: io.kestra.plugin.core.execution.Labels
description: Makes every run of a move easy to find in the execution list.
labels:
kestra-move-namespace: "{{ inputs.namespace }}"
kestra-move-mode: "{{ inputs.mode }}{{ inputs.dry_run == 'true' ? '-dry-run' : '' }}"
- id: discover
type: io.kestra.plugin.core.flow.Parallel
description: >-
Reads both instance configurations and the ledger. /api/v1/configs carries
the version that decides which API dialect the rest of the flow speaks to
each instance.
tasks:
- id: source_config
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.source_url }}/api/v1/configs"
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
options:
allowFailed: true
allowFailure: true
timeout: PT1M
retry:
type: exponential
interval: PT2S
maxInterval: PT20S
maxDuration: PT1M
- id: target_config
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.target_url }}/api/v1/configs"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
options:
allowFailed: true
allowFailure: true
timeout: PT1M
retry:
type: exponential
interval: PT2S
maxInterval: PT20S
maxDuration: PT1M
- id: read_ledger
type: io.kestra.plugin.core.kv.Get
key: "kestra-move.{{ inputs.namespace }}"
errorOnMissing: false
- id: stop_if_unreachable
type: io.kestra.plugin.core.execution.Fail
description: >-
The two config reads tolerate failure so that a refused connection, a
wrong URL and a rejected secret all end here with one explanation.
condition: "{{ (outputs.source_config.code ?? 0) != 200 or
(outputs.target_config.code ?? 0) != 200 }}"
errorMessage: >-
Cannot read /api/v1/configs (source: {{ (outputs.source_config.code ?? 0)
== 0 ? 'no answer' : 'HTTP ' ~ outputs.source_config.code }}, target: {{
(outputs.target_config.code ?? 0) == 0 ? 'no answer' : 'HTTP ' ~
outputs.target_config.code }}). Check source_url and target_url and that
this Kestra can reach them; on HTTP 401 or 403 check the
KESTRA_SOURCE_AUTHORIZATION and KESTRA_TARGET_AUTHORIZATION secrets.
- id: instances
type: io.kestra.plugin.core.output.OutputValues
description: Turns each instance version into the API paths and query dialect
that this version understands.
values:
source: >-
{{ {'cfg': fromJson(outputs.source_config.body), 'url':
inputs.source_url, 'tenant': inputs.source_tenant ?? '', 'ns':
inputs.namespace} | jq(vars.instance_jq) | first | toJson }}
target: >-
{{ {'cfg': fromJson(outputs.target_config.body), 'url':
inputs.target_url, 'tenant': inputs.target_tenant ?? '', 'ns':
inputs.namespace} | jq(vars.instance_jq) | first | toJson }}
ledger: "{{ (outputs.read_ledger.value ?? {}) | toJson }}"
- id: stop_if_incompatible
type: io.kestra.plugin.core.execution.Fail
description: Refuses unsupported versions, a target that is the source under
another URL, and a ledger of another move.
condition: >-
{{ not fromJson(outputs.instances.values.source).supported or not
fromJson(outputs.instances.values.target).supported or
fromJson(outputs.instances.values.source).uuid ==
fromJson(outputs.instances.values.target).uuid or
((fromJson(outputs.instances.values.ledger).target.uuid ??
fromJson(outputs.instances.values.target).uuid)
!= fromJson(outputs.instances.values.target).uuid) }}
errorMessage: >-
Source {{ fromJson(outputs.instances.values.source).version }} and target
{{ fromJson(outputs.instances.values.target).version }} cannot be used:
both must be Kestra 0.21 or later, must be two different instances (source
uuid {{ fromJson(outputs.instances.values.source).uuid }}, target uuid {{
fromJson(outputs.instances.values.target).uuid }}), and the ledger
kestra-move.{{ inputs.namespace }} must belong to this target (it names {{
fromJson(outputs.instances.values.ledger).target.uuid ?? 'none' }}).
- id: stop_on_phase
type: io.kestra.plugin.core.execution.Fail
description: Keeps the ledger state machine consistent, so STAGE never runs in
the middle of a cutover.
condition: >-
{{ inputs.dry_run == 'false' and (
(inputs.mode == 'STAGE' and (['FROZEN', 'DRAINED', 'CUT_OVER'] contains (fromJson(outputs.instances.values.ledger).phase ?? '')))
or (inputs.mode == 'CUTOVER' and (fromJson(outputs.instances.values.ledger).phase ?? '') == 'CUT_OVER')) }}
errorMessage: >-
The ledger says {{ inputs.namespace }} is {{
fromJson(outputs.instances.values.ledger).phase ?? 'new' }}. A cutover in
progress is resumed with mode CUTOVER and undone with mode ROLLBACK; a
finished cutover is undone with mode ROLLBACK before it can be staged or
cut over again.
- id: read_state
type: io.kestra.plugin.core.flow.Parallel
description: >-
Lists the flows of the namespace on both sides and the source executions
that are not finished. The answers go to internal storage, because the
list of a large namespace can exceed the 1 MiB output limit of Kestra 2.0.
tasks:
- id: source_flows
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.source).api }}/flows/{{
inputs.namespace }}"
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
allowFailure: true
timeout: PT2M
- id: target_flows
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.target).api }}/flows/{{
inputs.namespace }}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
allowFailure: true
timeout: PT2M
- id: source_pending
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.source).pending_url }}"
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
allowFailure: true
timeout: PT2M
- id: stop_on_access
type: io.kestra.plugin.core.execution.Fail
condition: >-
{{ (outputs.source_flows.uri ?? '') == '' or (outputs.target_flows.uri ??
'') == '' or (outputs.source_pending.uri ?? '') == '' }}
errorMessage: >-
Reading {{ inputs.namespace }} failed for:{% if (outputs.source_flows.uri
?? '') == '' %} the source flows{% endif %}{% if (outputs.target_flows.uri
?? '') == '' %} the target flows{% endif %}{% if
(outputs.source_pending.uri ?? '') == '' %} the source executions{% endif
%}. The failed task above logs the HTTP status. On 401 or 403 the
authorization secret lacks read access to the namespace; on 404 check the
tenant inputs.
- id: state
type: io.kestra.plugin.core.output.OutputValues
description: >-
Keeps what the next steps need from the lists. Executions are filtered
again on the exact namespace, because Kestra 0.21 treats the namespace
filter as a prefix.
values:
src_flows: "{{ fromJson(read(outputs.source_flows.uri)) | jq(vars.flows_jq) |
first | toJson }}"
tgt_flows: "{{ fromJson(read(outputs.target_flows.uri)) | jq(vars.flows_jq) |
first | toJson }}"
pending: >-
{{ {'ns': inputs.namespace, 'page':
fromJson(read(outputs.source_pending.uri))} | jq(vars.pending_jq) |
first | toJson }}
freeze: "{{ fromJson(read(outputs.source_flows.uri)) | jq(vars.freeze_jq) |
first | toJson }}"
phase: "{{ fromJson(outputs.instances.values.ledger).phase ?? 'NEW' }}"
- id: stop_if_smoke_flow_unknown
type: io.kestra.plugin.core.execution.Fail
description: Catches a typo in smoke_test_flow before the approval, while
nothing has changed yet.
condition: >-
{{ inputs.mode == 'CUTOVER' and (inputs.smoke_test_flow ?? '') != '' and
not ((fromJson(outputs.state.values.src_flows) | jq('map(.id)') | first)
contains inputs.smoke_test_flow) }}
errorMessage: >-
smoke_test_flow {{ inputs.smoke_test_flow }} is not a flow of {{
inputs.namespace }} on the source. Nothing was changed. Set it to a flow
of the namespace that has no required inputs, or leave it empty.
- id: cutover_freeze
type: io.kestra.plugin.core.flow.If
description: The only step that changes the source. It runs once per cutover; a
resumed cutover skips it.
condition: >-
{{ inputs.mode == 'CUTOVER' and inputs.dry_run == 'false' and not
(['FROZEN', 'DRAINED'] contains outputs.state.values.phase) }}
then:
- id: cutover_brief
type: io.kestra.plugin.core.log.Log
message: >-
Approval needed to cut {{ inputs.namespace }} over from {{
inputs.source_url }} ({{
fromJson(outputs.instances.values.source).version }}) to {{
inputs.target_url }} ({{
fromJson(outputs.instances.values.target).version }}). {{
fromJson(outputs.state.values.freeze) | length }} enabled flows will
be disabled on the source: {{ fromJson(outputs.state.values.freeze) |
jq('map(.id) | join(", ")') | first }}. {{
fromJson(outputs.state.values.pending) | jq(vars.pending_count_jq) |
first }} source executions are not finished yet and will be waited for
up to {{ inputs.drain_timeout_minutes }} minutes. Last STAGE: {{
fromJson(outputs.instances.values.ledger).staged_at ?? 'never' }}.
Resume with decision CUTOVER to continue or ABORT to stop without
changes.
- id: approve_cutover
type: io.kestra.plugin.core.flow.Pause
onResume:
- id: decision
type: SELECT
values:
- CUTOVER
- ABORT
defaults: ABORT
description: CUTOVER freezes the source now. ABORT ends this execution without
any change.
- id: abort_if_declined
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.approve_cutover.onResume.decision != 'CUTOVER' }}"
errorMessage: The cutover of {{ inputs.namespace }} was declined at approval.
Nothing was changed.
- id: freeze_source
type: io.kestra.plugin.core.http.Request
description: >-
Disables the flows on the source. A disabled flow keeps its running
executions but starts no new one, from any trigger or caller.
runIf: "{{ fromJson(outputs.state.values.freeze) | length > 0 }}"
uri: "{{ fromJson(outputs.instances.values.source).api }}/flows/disable/by-ids"
method: POST
contentType: application/json
body: "{{ outputs.state.values.freeze }}"
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: record_freeze
type: io.kestra.plugin.core.kv.Set
description: Written right after the freeze, so a crash at any later point
leaves enough state to resume or roll back.
key: "kestra-move.{{ inputs.namespace }}"
kvType: JSON
overwrite: true
value: >-
{{ {'ledger': fromJson(outputs.instances.values.ledger), 'set': {
'namespace': inputs.namespace, 'source': {'url': inputs.source_url,
'uuid': fromJson(outputs.instances.values.source).uuid, 'version':
fromJson(outputs.instances.values.source).version}, 'target': {'url':
inputs.target_url, 'uuid':
fromJson(outputs.instances.values.target).uuid, 'version':
fromJson(outputs.instances.values.target).version}, 'phase': 'FROZEN',
'frozen': fromJson(outputs.state.values.freeze) | jq('map(.id)') |
first, 'frozen_at': now(), 'cutover_execution': execution.id},
'event': {'at': now(), 'execution': execution.id, 'mode': 'CUTOVER',
'event': 'approved and froze the source'}} | jq(vars.ledger_set_jq) |
first | toJson }}
- id: cutover_drain
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.mode == 'CUTOVER' and inputs.dry_run == 'false' }}"
then:
- id: claim_cutover
type: io.kestra.plugin.core.kv.Set
description: >-
A CUTOVER that finds the ledger FROZEN or DRAINED resumes an
interrupted one. It takes the ledger over, so a failure from here on
rolls back like in the original run.
runIf: "{{ ['FROZEN', 'DRAINED'] contains outputs.state.values.phase }}"
key: "kestra-move.{{ inputs.namespace }}"
kvType: JSON
overwrite: true
value: >-
{{ {'ledger': kv('kestra-move.' ~ inputs.namespace), 'set':
{'cutover_execution': execution.id}, 'event': {'at': now(),
'execution': execution.id, 'mode': 'CUTOVER', 'event': 'resumed a
cutover left ' ~ outputs.state.values.phase}} | jq(vars.ledger_set_jq)
| first | toJson }}
- id: drain_started
type: io.kestra.plugin.core.output.OutputValues
values:
at: "{{ now() | timestamp }}"
deadline: "{{ (now() | timestamp) + inputs.drain_timeout_minutes * 60 }}"
- id: drain
type: io.kestra.plugin.core.flow.LoopUntil
description: >-
Polls the source until no execution of the namespace is unfinished, or
until the deadline. The wait between polls is a Sleep task: Kestra 1.0
to 1.2 keep the concurrency slot of a run killed while LoopUntil
waits.
condition: >-
{{ (fromJson(outputs.drain_check.values.pending) | length) == 0 or
(now() | timestamp) >= (outputs.drain_started.values.deadline |
number) }}
failOnMaxReached: false
checkFrequency:
maxIterations: 100000
interval: PT0.1S
tasks:
- id: drain_poll
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.source).pending_url }}"
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
timeout: PT1M
retry:
type: exponential
interval: PT2S
maxInterval: PT20S
maxDuration: PT2M
- id: drain_check
type: io.kestra.plugin.core.output.OutputValues
values:
pending: >-
{{ {'ns': inputs.namespace, 'page':
fromJson(read(outputs.drain_poll.uri))} | jq(vars.pending_jq) |
first | toJson }}
- id: drain_pause
type: io.kestra.plugin.core.flow.Sleep
runIf: >-
{{ (fromJson(outputs.drain_check.values.pending) | length) > 0 and
(now() | timestamp) < (outputs.drain_started.values.deadline |
number) }}
duration: PT5S
- id: stop_if_not_drained
type: io.kestra.plugin.core.execution.Fail
condition: "{{ (fromJson(outputs.drain_check.values.pending) | length) > 0 }}"
errorMessage: >-
{{ fromJson(outputs.drain_check.values.pending) |
jq(vars.pending_count_jq) | first }} executions of {{ inputs.namespace
}} were still unfinished on the source after {{
inputs.drain_timeout_minutes }} minutes: {{
fromJson(outputs.drain_check.values.pending) | jq('map(.flowId + "/" +
.id + " " + .state) | join(", ")') | first }}. The cutover rolls back.
Finish or kill them, then run CUTOVER again.
- id: record_drain
type: io.kestra.plugin.core.kv.Set
key: "kestra-move.{{ inputs.namespace }}"
kvType: JSON
overwrite: true
value: >-
{{ {'ledger': kv('kestra-move.' ~ inputs.namespace), 'set': {'phase':
'DRAINED', 'drained_at': now()}, 'event': {'at': now(), 'execution':
execution.id, 'mode': 'CUTOVER', 'event': 'drained the source'}} |
jq(vars.ledger_set_jq) | first | toJson }}
- id: inventory
type: io.kestra.plugin.core.flow.If
description: >-
Exports both sides. In CUTOVER this runs after the drain, so KV pairs that
the last executions wrote are copied too.
condition: "{{ inputs.mode != 'ROLLBACK' }}"
then:
- id: export
type: io.kestra.plugin.core.flow.Parallel
tasks:
- id: source_flows_zip
type: io.kestra.plugin.core.http.Download
runIf: "{{ fromJson(outputs.state.values.src_flows) | length > 0 }}"
uri: "{{ fromJson(outputs.instances.values.source).api }}/flows/export/by-ids"
method: POST
contentType: application/json
body: "{{ fromJson(outputs.state.values.src_flows) | jq(vars.flow_ids_jq) |
first | toJson }}"
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
timeout: PT5M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT5M
- id: target_flows_zip
type: io.kestra.plugin.core.http.Download
runIf: "{{ fromJson(outputs.state.values.tgt_flows) | length > 0 }}"
uri: "{{ fromJson(outputs.instances.values.target).api }}/flows/export/by-ids"
method: POST
contentType: application/json
body: "{{ fromJson(outputs.state.values.tgt_flows) | jq(vars.flow_ids_jq) |
first | toJson }}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT5M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT5M
- id: source_files_zip
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.source).api }}/namespaces/{{
inputs.namespace }}/files/export"
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
timeout: PT5M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT5M
- id: target_files_zip
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.target).api }}/namespaces/{{
inputs.namespace }}/files/export"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT5M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT5M
- id: source_kv_list
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.source).kv_list_url }}"
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: target_kv_list
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.target).kv_list_url }}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: source_triggers
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.source).triggers_url }}"
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: unpack
type: io.kestra.plugin.core.flow.Parallel
description: The exports are ZIP archives; plugin-compress turns each of them
into one file per flow or namespace file.
tasks:
- id: source_flow_files
type: io.kestra.plugin.compress.ArchiveDecompress
runIf: "{{ fromJson(outputs.state.values.src_flows) | length > 0 }}"
from: "{{ outputs.source_flows_zip.uri }}"
algorithm: ZIP
- id: target_flow_files
type: io.kestra.plugin.compress.ArchiveDecompress
runIf: "{{ fromJson(outputs.state.values.tgt_flows) | length > 0 }}"
from: "{{ outputs.target_flows_zip.uri }}"
algorithm: ZIP
- id: source_ns_files
type: io.kestra.plugin.compress.ArchiveDecompress
from: "{{ outputs.source_files_zip.uri }}"
algorithm: ZIP
allowFailure: true
- id: target_ns_files
type: io.kestra.plugin.compress.ArchiveDecompress
from: "{{ outputs.target_files_zip.uri }}"
algorithm: ZIP
allowFailure: true
- id: stop_if_unpack_failed
type: io.kestra.plugin.core.execution.Fail
description: >-
Kestra 1.3 and older cannot unpack namespace file names with
characters such as %, [ or ], and 0.23 and older also names with a
space. The run stops here with that reason instead of a storage error.
condition: "{{ (outputs.source_ns_files.files ?? '') == '' or
(outputs.target_ns_files.files ?? '') == '' }}"
errorMessage: >-
The namespace files of the {{ (outputs.source_ns_files.files ?? '') ==
'' ? 'source' : 'target' }}{{ ((outputs.source_ns_files.files ?? '')
== '' and (outputs.target_ns_files.files ?? '') == '') ? ' and the
target' : '' }} could not be unpacked on the Kestra instance that runs
this flow (the failed task above logs why). Kestra 1.3 and older
cannot unpack file names with characters such as %, [ or ], and 0.23
and older also names with a space. Run this flow on Kestra 2.0, or
rename the file. Nothing was written to the target.
- id: fingerprint
type: io.kestra.plugin.core.output.OutputValues
description: >-
Hashes every flow without its top-level disabled line, and every
namespace file with its size, because read() decodes a binary file as
text. Exports up to 1.3 also hold the flows under _flows/, which are
skipped because importing them would create enabled flows.
values:
source_docs: >-
[{% for e in (outputs.source_flow_files.files ?? {}) %}{% set d =
{'s': read(e.value)} | jq(vars.doc_jq) | first %}{{ loop.index > 0 ?
',' : '' }}{{ {'idx': loop.index, 'key': e.key, 'id': d.id, 'sha':
d.norm | sha256, 'secrets': d.secrets} | toJson }}{% endfor %}]
target_docs: >-
[{% for e in (outputs.target_flow_files.files ?? {}) %}{% set d =
{'s': read(e.value)} | jq(vars.doc_jq) | first %}{{ loop.index > 0 ?
',' : '' }}{{ {'idx': loop.index, 'key': e.key, 'id': d.id, 'sha':
d.norm | sha256} | toJson }}{% endfor %}]
source_files: >-
[{% for k in (outputs.source_ns_files.files | jq(vars.ns_keep_jq) |
first) %}{{ loop.index > 0 ? ',' : '' }}{{ {'path': k, 'sha':
(read(outputs.source_ns_files.files[k]) | sha256) ~ ':' ~
fileSize(outputs.source_ns_files.files[k])} | toJson }}{% endfor %}]
target_files: >-
[{% for k in (outputs.target_ns_files.files | jq(vars.ns_keep_jq) |
first) %}{{ loop.index > 0 ? ',' : '' }}{{ {'path': k, 'sha':
(read(outputs.target_ns_files.files[k]) | sha256) ~ ':' ~
fileSize(outputs.target_ns_files.files[k])} | toJson }}{% endfor %}]
zips: "{{ outputs.source_ns_files.files | jq(vars.ns_zip_jq) | first | toJson
}}"
file_collisions: >-
{{ {'s': outputs.source_ns_files.files, 't':
outputs.target_ns_files.files} | jq(vars.collisions_jq) | first |
toJson }}
kv_keys: >-
{{ {'ns': inputs.namespace, 's':
fromJson(read(outputs.source_kv_list.uri)), 't':
fromJson(read(outputs.target_kv_list.uri)), 'ledger':
kv('kestra-move.' ~ inputs.namespace, errorOnMissing=false) ?? {}} |
jq(vars.kv_keys_jq) | first | toJson }}
kv_to_read: >-
{{ {'ns': inputs.namespace, 's':
fromJson(read(outputs.source_kv_list.uri)), 't':
fromJson(read(outputs.target_kv_list.uri)), 'ledger':
kv('kestra-move.' ~ inputs.namespace, errorOnMissing=false) ?? {}} |
jq(vars.kv_keys_jq) | first | jq(vars.kv_read_jq) | first | toJson
}}
kv_complete: "{{ fromJson(read(outputs.source_kv_list.uri)) |
jq(vars.page_complete_jq) | first }}"
kv_target_complete: "{{ fromJson(read(outputs.target_kv_list.uri)) |
jq(vars.page_complete_jq) | first }}"
- id: validate_on_target
type: io.kestra.plugin.core.http.Request
description: >-
Sends every source flow to the target validator before anything is
written. A 1.3 flow that uses a construct removed in 2.0, such as
ForEach or pluginDefaults, comes back with its constraint here.
runIf: "{{ fromJson(outputs.fingerprint.values.source_docs) | length > 0 }}"
uri: "{{ fromJson(outputs.instances.values.target).api }}/flows/validate"
method: POST
contentType: application/x-yaml
body: |-
{% for d in fromJson(outputs.fingerprint.values.source_docs) %}{% if loop.index > 0 %}
---
{% endif %}{{ {'s': read(outputs.source_flow_files.files[d.key])} | jq(vars.clean_jq) | first }}{% endfor %}
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT5M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT5M
- id: kv_fingerprint
type: io.kestra.plugin.core.flow.LoopUntil
description: >-
Reads the KV values that changed since the last copy on both sides
into internal storage and keeps only their type and hash, so values
never appear in task outputs or logs.
runIf: "{{ fromJson(outputs.fingerprint.values.kv_to_read) | length > 0 }}"
condition: "{{ outputs.kv_item.values.last == 'true' }}"
failOnMaxReached: true
checkFrequency:
maxIterations: 100000
interval: PT0.1S
tasks:
- id: kv_item
type: io.kestra.plugin.core.output.OutputValues
values:
key: "{{
fromJson(outputs.fingerprint.values.kv_to_read)[outputs.kv_fingerprint.\
iterationCount - 1].key }}"
src: "{{
fromJson(outputs.fingerprint.values.kv_to_read)[outputs.kv_fingerprint.\
iterationCount - 1].src }}"
tgt: "{{
fromJson(outputs.fingerprint.values.kv_to_read)[outputs.kv_fingerprint.\
iterationCount - 1].tgt }}"
exp: "{{
fromJson(outputs.fingerprint.values.kv_to_read)[outputs.kv_fingerprint.\
iterationCount - 1].exp ?? '' }}"
read_src: >-
{{ not
fromJson(outputs.fingerprint.values.kv_to_read)[outputs.kv_fingerprint.iterationCount
- 1].src ? false :
((fromJson(outputs.fingerprint.values.kv_to_read)[outputs.kv_fingerprint.iterationCount
- 1].exp ?? '') == '' ? true :
(((fromJson(outputs.fingerprint.values.kv_to_read)[outputs.kv_fingerprint.iterationCount
- 1].exp | timestamp) - (now() | timestamp)) > 60)) }}
read_tgt: >-
{{ not
fromJson(outputs.fingerprint.values.kv_to_read)[outputs.kv_fingerprint.iterationCount
- 1].tgt ? false :
((fromJson(outputs.fingerprint.values.kv_to_read)[outputs.kv_fingerprint.iterationCount
- 1].texp ?? '') == '' ? true :
(((fromJson(outputs.fingerprint.values.kv_to_read)[outputs.kv_fingerprint.iterationCount
- 1].texp | timestamp) - (now() | timestamp)) > 60)) }}
last: "{{ outputs.kv_fingerprint.iterationCount >=
(fromJson(outputs.fingerprint.values.kv_to_read) | length) }}"
- id: kv_source_value
type: io.kestra.plugin.core.http.Download
runIf: "{{ outputs.kv_item.values.read_src == 'true' }}"
uri: "{{ fromJson(outputs.instances.values.source).api }}/namespaces/{{
inputs.namespace }}/kv/{{ outputs.kv_item.values.key }}"
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
timeout: PT1M
retry:
type: exponential
interval: PT1S
maxInterval: PT10S
maxDuration: PT1M
- id: kv_target_value
type: io.kestra.plugin.core.http.Download
runIf: "{{ outputs.kv_item.values.read_tgt == 'true' }}"
uri: "{{ fromJson(outputs.instances.values.target).api }}/namespaces/{{
inputs.namespace }}/kv/{{ outputs.kv_item.values.key }}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT1M
retry:
type: exponential
interval: PT1S
maxInterval: PT10S
maxDuration: PT1M
- id: kv_note
type: io.kestra.plugin.core.kv.Set
description: >-
Hashes the value only. A pair whose value matches but whose type
differs on the target is reported as a warning instead of being
rewritten. A source pair that expires within a minute is marked
EXPIRING and not copied.
key: "kestra-move.{{ inputs.namespace }}.scratch"
kvType: JSON
overwrite: true
ttl: P1D
value: >-
{% set src = outputs.kv_item.values.read_src == 'true' ?
fromJson(read(outputs.kv_source_value.uri)) : null %}{% set tgt =
outputs.kv_item.values.read_tgt == 'true' ?
fromJson(read(outputs.kv_target_value.uri)) : null %}{{ { 'prev':
outputs.kv_fingerprint.iterationCount == 1 ? {} :
(kv('kestra-move.' ~ inputs.namespace ~ '.scratch',
errorOnMissing=false) ?? {}), 'key': outputs.kv_item.values.key,
'entry': { 'src_type': src == null ? null : src.type, 'src_sha':
outputs.kv_item.values.src != 'true' ? null : (src == null ?
'EXPIRING' : ((src.value | toJson) | sha256)), 'src_uri': src ==
null ? null : outputs.kv_source_value.uri, 'src_exp':
outputs.kv_item.values.exp == '' ? null :
outputs.kv_item.values.exp, 'tgt_type': tgt == null ? null :
tgt.type, 'tgt_sha': tgt == null ? null : ((tgt.value | toJson) |
sha256)}} | jq(vars.kv_merge_jq) | first | toJson }}
- id: plan
type: io.kestra.plugin.core.output.OutputValues
description: >-
Classifies every flow, file and KV pair as NEW, IN_SYNC, UPDATE,
CONFLICT, INVALID, EXPIRING, UNSUPPORTED_NAME, STALE or TARGET_ONLY.
values:
plan: >-
{{ {'ns': inputs.namespace, 'mode': inputs.mode, 'dry':
inputs.dry_run, 'on_conflict': inputs.on_conflict, 'source':
fromJson(outputs.instances.values.source), 'target':
fromJson(outputs.instances.values.target), 'ledger':
kv('kestra-move.' ~ inputs.namespace, errorOnMissing=false) ?? {},
'src_flows': fromJson(outputs.state.values.src_flows), 'tgt_flows':
fromJson(outputs.state.values.tgt_flows), 'src_docs':
fromJson(outputs.fingerprint.values.source_docs), 'tgt_docs':
fromJson(outputs.fingerprint.values.target_docs), 'validation':
fromJson(outputs.validate_on_target.body ?? '[]'), 'src_files':
fromJson(outputs.fingerprint.values.source_files), 'tgt_files':
fromJson(outputs.fingerprint.values.target_files), 'zips':
fromJson(outputs.fingerprint.values.zips), 'collisions':
fromJson(outputs.fingerprint.values.file_collisions), 'kv_keys':
fromJson(outputs.fingerprint.values.kv_keys), 'kv':
(fromJson(outputs.fingerprint.values.kv_to_read) | length) > 0 ?
kv('kestra-move.' ~ inputs.namespace ~ '.scratch') : {},
'kv_complete': outputs.fingerprint.values.kv_complete == 'true',
'kv_target_complete': outputs.fingerprint.values.kv_target_complete
== 'true', 'src_triggers':
fromJson(read(outputs.source_triggers.uri)).results ?? [],
'pending': fromJson(outputs.state.values.pending)} |
jq(vars.plan_jq) | first | toJson }}
- id: report
type: io.kestra.plugin.core.log.Log
message: >-
{{ inputs.mode }}{{ inputs.dry_run == 'true' ? ' (dry run, nothing
written)' : '' }} of {{ inputs.namespace }} from {{ inputs.source_url
}} ({{ fromJson(outputs.plan.values.plan).source.version }}) to {{
inputs.target_url }} ({{
fromJson(outputs.plan.values.plan).target.version }}). Flows {{
fromJson(outputs.plan.values.plan).counts.flows | toJson }}, files {{
fromJson(outputs.plan.values.plan).counts.files | toJson }}, KV pairs
{{ fromJson(outputs.plan.values.plan).counts.kv | toJson }},
unfinished source executions {{
fromJson(outputs.plan.values.plan).counts.pending_executions }}. To
write: {{ fromJson(outputs.plan.values.plan).writes |
jq('del(.flow_items)') | first | toJson }}. Blockers: {{
fromJson(outputs.plan.values.plan).blockers | length == 0 ? 'none' :
fromJson(outputs.plan.values.plan).blockers | join(' | ') }}.
Warnings: {{ fromJson(outputs.plan.values.plan).warnings | length == 0
? 'none' : fromJson(outputs.plan.values.plan).warnings | join(' | ')
}}. Webhooks to re-point: {{
fromJson(outputs.plan.values.plan).webhooks | toJson }}.
- id: stop_on_blockers
type: io.kestra.plugin.core.execution.Fail
description: In CUTOVER a blocker found after the freeze fails the run, and the
errors branch unfreezes the source.
condition: "{{ inputs.dry_run == 'false' and
(fromJson(outputs.plan.values.plan).blockers | length) > 0 }}"
errorMessage: "Nothing was written to the target: {{
fromJson(outputs.plan.values.plan).blockers | join('; ') }}"
- id: write_target
type: io.kestra.plugin.core.flow.If
description: Writes only what the plan marked, so a repeated STAGE copies the
delta and rewrites nothing else.
condition: "{{ inputs.dry_run == 'false' and inputs.mode != 'ROLLBACK' }}"
then:
- id: write_flows
type: io.kestra.plugin.core.flow.LoopUntil
description: >-
Writes one flow per request with a top-level disabled true, so no
trigger of the copy can fire before the cutover enables it. It uses
the single-flow endpoints because the namespace endpoint of Kestra
0.21 to 1.3 splits its body on every "---", including the ones inside
scripts and descriptions.
runIf: "{{ fromJson(outputs.plan.values.plan).writes.flows | length > 0 }}"
condition: "{{ outputs.flow_to_write.values.last == 'true' }}"
failOnMaxReached: true
checkFrequency:
maxIterations: 100000
interval: PT0.1S
tasks:
- id: flow_to_write
type: io.kestra.plugin.core.output.OutputValues
values:
id: "{{
fromJson(outputs.plan.values.plan).writes.flow_items[outputs.write_flows\
.iterationCount - 1].id }}"
key: "{{
fromJson(outputs.plan.values.plan).writes.flow_items[outputs.write_flow\
s.iterationCount - 1].key }}"
exists: "{{
fromJson(outputs.plan.values.plan).writes.flow_items[outputs.write_f\
lows.iterationCount - 1].exists }}"
last: "{{ outputs.write_flows.iterationCount >=
(fromJson(outputs.plan.values.plan).writes.flow_items | length)
}}"
- id: create_flow
type: io.kestra.plugin.core.http.Request
runIf: "{{ outputs.flow_to_write.values.exists == 'false' }}"
uri: "{{ fromJson(outputs.instances.values.target).api }}/flows"
method: POST
contentType: application/x-yaml
body: "{{ {'s':
read(outputs.source_flow_files.files[outputs.flow_to_write.values\
.key])} | jq(vars.inject_jq) | first }}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT20S
maxDuration: PT2M
- id: update_flow
type: io.kestra.plugin.core.http.Request
runIf: "{{ outputs.flow_to_write.values.exists == 'true' }}"
uri: "{{ fromJson(outputs.instances.values.target).api }}/flows/{{
inputs.namespace }}/{{ outputs.flow_to_write.values.id }}"
method: PUT
contentType: application/x-yaml
body: "{{ {'s':
read(outputs.source_flow_files.files[outputs.flow_to_write.values\
.key])} | jq(vars.inject_jq) | first }}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT20S
maxDuration: PT2M
- id: write_files
type: io.kestra.plugin.core.flow.LoopUntil
description: >-
Uploads one file per iteration. The end of the loop is decided inside
the iteration, because on Kestra 0.21 and 1.0 to 1.3 the condition
already sees the next iteration count.
runIf: "{{ fromJson(outputs.plan.values.plan).writes.files | length > 0 }}"
condition: "{{ outputs.file_to_write.values.last == 'true' }}"
failOnMaxReached: true
checkFrequency:
maxIterations: 100000
interval: PT0.1S
tasks:
- id: file_to_write
type: io.kestra.plugin.core.output.OutputValues
values:
path: "{{
fromJson(outputs.plan.values.plan).writes.files[outputs.write_files.it\
erationCount - 1] }}"
last: "{{ outputs.write_files.iterationCount >=
(fromJson(outputs.plan.values.plan).writes.files | length) }}"
- id: upload_file
type: io.kestra.plugin.core.http.Request
uri: >-
{{ fromJson(outputs.instances.values.target).api }}/namespaces/{{
inputs.namespace }}/files?path={{ ('/' ~
outputs.file_to_write.values.path) | urlencode | replace({'+':
'%20'}) }}
method: POST
contentType: multipart/form-data
formData:
fileContent: "{{
outputs.source_ns_files.files[outputs.file_to_write.values.path]
}}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT20S
maxDuration: PT2M
- id: write_kv
type: io.kestra.plugin.core.flow.LoopUntil
description: >-
Copies each KV pair with the time to live it has left on the source at
the moment of the write. The KV API infers the type from the body,
read as Amazon Ion: a quoted value keeps STRING, NUMBER, BOOLEAN, JSON
and DURATION, and an unquoted ISO date or UTC timestamp is stored as
DATE or DATETIME.
runIf: "{{ fromJson(outputs.plan.values.plan).writes.kv | length > 0 }}"
condition: "{{ outputs.kv_to_write.values.last == 'true' }}"
failOnMaxReached: true
checkFrequency:
maxIterations: 100000
interval: PT0.1S
tasks:
- id: kv_to_write
type: io.kestra.plugin.core.output.OutputValues
values:
key: "{{
fromJson(outputs.plan.values.plan).writes.kv[outputs.write_kv.iteration\
Count - 1].key }}"
ttl: >-
{{
(fromJson(outputs.plan.values.plan).writes.kv[outputs.write_kv.iterationCount
- 1].expires_at ?? '') == '' ? '' :
((fromJson(outputs.plan.values.plan).writes.kv[outputs.write_kv.iterationCount
- 1].expires_at | timestamp) - (now() | timestamp)) }}
last: "{{ outputs.write_kv.iterationCount >=
(fromJson(outputs.plan.values.plan).writes.kv | length) }}"
- id: put_kv
type: io.kestra.plugin.core.http.Request
runIf: "{{ outputs.kv_to_write.values.ttl == '' }}"
uri: "{{ fromJson(outputs.instances.values.target).api }}/namespaces/{{
inputs.namespace }}/kv/{{ outputs.kv_to_write.values.key }}"
method: PUT
contentType: text/plain
body: >-
{% set v = fromJson(read(kv('kestra-move.' ~ inputs.namespace ~
'.scratch')[outputs.kv_to_write.values.key].src_uri)) %}{{ (v.type
== 'DATE' or v.type == 'DATETIME') ? v.value : (v.value | toJson)
}}
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT1M
retry:
type: exponential
interval: PT1S
maxInterval: PT10S
maxDuration: PT1M
- id: put_kv_with_ttl
type: io.kestra.plugin.core.http.Request
runIf: "{{ outputs.kv_to_write.values.ttl == '' ? false :
((outputs.kv_to_write.values.ttl | number) > 0) }}"
uri: "{{ fromJson(outputs.instances.values.target).api }}/namespaces/{{
inputs.namespace }}/kv/{{ outputs.kv_to_write.values.key }}"
method: PUT
contentType: text/plain
body: >-
{% set v = fromJson(read(kv('kestra-move.' ~ inputs.namespace ~
'.scratch')[outputs.kv_to_write.values.key].src_uri)) %}{{ (v.type
== 'DATE' or v.type == 'DATETIME') ? v.value : (v.value | toJson)
}}
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
ttl: "PT{{ outputs.kv_to_write.values.ttl }}S"
timeout: PT1M
retry:
type: exponential
interval: PT1S
maxInterval: PT10S
maxDuration: PT1M
- id: target_kv_after
type: io.kestra.plugin.core.http.Download
description: Lists the target KV pairs again, because the writes above moved
their updateDate.
uri: "{{ fromJson(outputs.instances.values.target).kv_list_url }}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: record_stage
type: io.kestra.plugin.core.kv.Set
description: Records the hash of everything now in sync, which is how the next
run tells its own copies from foreign edits.
key: "kestra-move.{{ inputs.namespace }}"
kvType: JSON
overwrite: true
value: >-
{{ {'ledger': kv('kestra-move.' ~ inputs.namespace,
errorOnMissing=false) ?? {}, 'ns': inputs.namespace, 'mode':
inputs.mode, 'now': now(), 'execution': execution.id, 'source':
{'url': inputs.source_url, 'uuid':
fromJson(outputs.instances.values.source).uuid, 'version':
fromJson(outputs.instances.values.source).version}, 'target': {'url':
inputs.target_url, 'uuid':
fromJson(outputs.instances.values.target).uuid, 'version':
fromJson(outputs.instances.values.target).version}, 'shas':
fromJson(outputs.plan.values.plan).shas, 'kv_dates': {'ns':
inputs.namespace, 'keys': fromJson(outputs.plan.values.plan).shas.kv |
keys, 's': fromJson(read(outputs.source_kv_list.uri)), 't':
fromJson(read(outputs.target_kv_after.uri))} | jq(vars.kv_dates_jq) |
first, 'written': {'flows':
fromJson(outputs.plan.values.plan).writes.flows, 'files':
fromJson(outputs.plan.values.plan).writes.files, 'kv':
fromJson(outputs.plan.values.plan).writes.kv}} |
jq(vars.ledger_stage_jq) | first | toJson }}
- id: cutover_enable
type: io.kestra.plugin.core.flow.If
description: Hands the triggers over. The source was frozen first, so no
schedule, webhook or consumer runs on both sides.
condition: "{{ inputs.mode == 'CUTOVER' and inputs.dry_run == 'false' }}"
then:
- id: enable_target
type: io.kestra.plugin.core.http.Request
runIf: "{{ (kv('kestra-move.' ~ inputs.namespace).frozen ?? []) | length > 0 }}"
uri: "{{ fromJson(outputs.instances.values.target).api }}/flows/enable/by-ids"
method: POST
contentType: application/json
body: >-
{{ {'ns': inputs.namespace, 'ids': kv('kestra-move.' ~
inputs.namespace).frozen ?? []} | jq(vars.by_ids_jq) | first | toJson
}}
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: record_cutover
type: io.kestra.plugin.core.kv.Set
key: "kestra-move.{{ inputs.namespace }}"
kvType: JSON
overwrite: true
value: >-
{{ {'ledger': kv('kestra-move.' ~ inputs.namespace), 'set': {'phase':
'CUT_OVER', 'enabled': kv('kestra-move.' ~ inputs.namespace).frozen ??
[], 'cut_over_at': now()}, 'event': {'at': now(), 'execution':
execution.id, 'mode': 'CUTOVER', 'event': 'enabled the target'}} |
jq(vars.ledger_set_jq) | first | toJson }}
- id: target_after
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.target).api }}/flows/{{
inputs.namespace }}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: stop_if_not_enabled
type: io.kestra.plugin.core.execution.Fail
description: Reads the target back instead of trusting the enable call.
condition: >-
{{ ({'ids': kv('kestra-move.' ~ inputs.namespace).frozen ?? [],
'flows': fromJson(read(outputs.target_after.uri))} |
jq(vars.not_enabled_jq) | first | length) > 0 }}
errorMessage: >-
These flows of {{ inputs.namespace }} are not enabled on the target
after the enable call: {{ {'ids': kv('kestra-move.' ~
inputs.namespace).frozen ?? [], 'flows':
fromJson(read(outputs.target_after.uri))} | jq(vars.not_enabled_jq) |
first | join(', ') }}. The cutover rolls back.
- id: smoke_test
type: io.kestra.plugin.core.flow.If
condition: "{{ (inputs.smoke_test_flow ?? '') != '' }}"
then:
- id: smoke_start
type: io.kestra.plugin.core.http.Request
uri: "{{ fromJson(outputs.instances.values.target).api }}/executions/{{
inputs.namespace }}/{{ inputs.smoke_test_flow }}?wait=false"
method: POST
contentType: multipart/form-data
formData:
kestra_move_smoke_test: "{{ execution.id }}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
options:
allowFailed: true
timeout: PT1M
- id: stop_if_smoke_not_started
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.smoke_start.code != 200 }}"
errorMessage: >-
The target refused to start the smoke test flow {{
inputs.smoke_test_flow }} (HTTP {{ outputs.smoke_start.code }}: {{
(outputs.smoke_start.body ?? '') | abbreviate(300) }}). The
cutover rolls back.
- id: smoke_started
type: io.kestra.plugin.core.output.OutputValues
values:
id: "{{ fromJson(outputs.smoke_start.body).id }}"
deadline: "{{ (now() | timestamp) + inputs.smoke_test_timeout_minutes * 60 }}"
- id: smoke_wait
type: io.kestra.plugin.core.flow.LoopUntil
condition: >-
{{ (['SUCCESS', 'WARNING', 'FAILED', 'KILLED', 'CANCELLED',
'SKIPPED', 'RETRIED'] contains outputs.smoke_check.values.state)
or (now() | timestamp) >= (outputs.smoke_started.values.deadline |
number) }}
failOnMaxReached: false
checkFrequency:
maxIterations: 100000
interval: PT0.1S
tasks:
- id: smoke_poll
type: io.kestra.plugin.core.http.Download
uri: "{{ fromJson(outputs.instances.values.target).api }}/executions/{{
outputs.smoke_started.values.id }}"
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT1M
retry:
type: exponential
interval: PT2S
maxInterval: PT20S
maxDuration: PT2M
- id: smoke_check
type: io.kestra.plugin.core.output.OutputValues
values:
state: "{{ fromJson(read(outputs.smoke_poll.uri)).state.current }}"
- id: smoke_pause
type: io.kestra.plugin.core.flow.Sleep
runIf: >-
{{ not (['SUCCESS', 'WARNING', 'FAILED', 'KILLED',
'CANCELLED', 'SKIPPED', 'RETRIED'] contains
outputs.smoke_check.values.state) and (now() | timestamp) <
(outputs.smoke_started.values.deadline | number) }}
duration: PT5S
- id: stop_if_smoke_failed
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.smoke_check.values.state != 'SUCCESS' }}"
errorMessage: >-
Smoke test {{ inputs.smoke_test_flow }} ended {{
outputs.smoke_check.values.state }} on the target (execution {{
outputs.smoke_started.values.id }}). The cutover rolls back.
- id: rollback
type: io.kestra.plugin.core.flow.If
description: Disables the target copies before enabling the source again, so the
two never run together.
condition: "{{ inputs.mode == 'ROLLBACK' }}"
then:
- id: rollback_plan
type: io.kestra.plugin.core.log.Log
message: >-
ROLLBACK{{ inputs.dry_run == 'true' ? ' (dry run, nothing written)' :
'' }} of {{ inputs.namespace }}: the ledger is {{
outputs.state.values.phase }}. {% if ['FROZEN', 'DRAINED', 'CUT_OVER']
contains outputs.state.values.phase %}Disable on the target: {{
(fromJson(outputs.instances.values.ledger).enabled ?? []) | join(', ')
}}. Enable on the source: {{
(fromJson(outputs.instances.values.ledger).frozen ?? []) | join(', ')
}}.{% else %}Nothing to undo.{% endif %}
- id: rollback_apply
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.dry_run == 'false' and (['FROZEN', 'DRAINED', 'CUT_OVER']
contains outputs.state.values.phase) }}"
then:
- id: rollback_target
type: io.kestra.plugin.core.http.Request
runIf: "{{ (fromJson(outputs.instances.values.ledger).enabled ?? []) | length >
0 }}"
uri: "{{ fromJson(outputs.instances.values.target).api }}/flows/disable/by-ids"
method: POST
contentType: application/json
body: >-
{{ {'ns': inputs.namespace, 'ids':
fromJson(outputs.instances.values.ledger).enabled ?? []} |
jq(vars.by_ids_jq) | first | toJson }}
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: rollback_source
type: io.kestra.plugin.core.http.Request
runIf: "{{ (fromJson(outputs.instances.values.ledger).frozen ?? []) | length > 0
}}"
uri: "{{ fromJson(outputs.instances.values.source).api }}/flows/enable/by-ids"
method: POST
contentType: application/json
body: >-
{{ {'ns': inputs.namespace, 'ids':
fromJson(outputs.instances.values.ledger).frozen ?? []} |
jq(vars.by_ids_jq) | first | toJson }}
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: record_rollback
type: io.kestra.plugin.core.kv.Set
key: "kestra-move.{{ inputs.namespace }}"
kvType: JSON
overwrite: true
value: >-
{{ {'ledger': fromJson(outputs.instances.values.ledger), 'set':
{'phase': 'ROLLED_BACK', 'enabled': [], 'frozen': [],
'rolled_back_at': now()}, 'event': {'at': now(), 'execution':
execution.id, 'mode': 'ROLLBACK', 'event': 'rolled back on
request'}} | jq(vars.ledger_set_jq) | first | toJson }}
errors:
- id: auto_rollback
type: io.kestra.plugin.core.flow.If
description: >-
Runs when CUTOVER fails after the freeze, for example a drain timeout, a
blocker found on the frozen source or a failed smoke test. A killed
execution does not reach this branch; its ledger lets ROLLBACK or CUTOVER
take over.
condition: >-
{{ inputs.mode == 'CUTOVER' and inputs.dry_run == 'false' and
(outputs.instances is defined) and (['FROZEN', 'DRAINED', 'CUT_OVER']
contains ((kv('kestra-move.' ~ inputs.namespace, errorOnMissing=false) ??
{}).phase ?? '')) and ((kv('kestra-move.' ~ inputs.namespace,
errorOnMissing=false) ?? {}).cutover_execution ?? '') == execution.id }}
then:
- id: undo_target
type: io.kestra.plugin.core.http.Request
runIf: "{{ (kv('kestra-move.' ~ inputs.namespace).enabled ?? []) | length > 0
}}"
uri: "{{ fromJson(outputs.instances.values.target).api }}/flows/disable/by-ids"
method: POST
contentType: application/json
body: >-
{{ {'ns': inputs.namespace, 'ids': kv('kestra-move.' ~
inputs.namespace).enabled ?? []} | jq(vars.by_ids_jq) | first | toJson
}}
headers:
Authorization: "{{ secret('KESTRA_TARGET_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: undo_freeze
type: io.kestra.plugin.core.http.Request
runIf: "{{ (kv('kestra-move.' ~ inputs.namespace).frozen ?? []) | length > 0 }}"
uri: "{{ fromJson(outputs.instances.values.source).api }}/flows/enable/by-ids"
method: POST
contentType: application/json
body: >-
{{ {'ns': inputs.namespace, 'ids': kv('kestra-move.' ~
inputs.namespace).frozen ?? []} | jq(vars.by_ids_jq) | first | toJson
}}
headers:
Authorization: "{{ secret('KESTRA_SOURCE_AUTHORIZATION') }}"
timeout: PT2M
retry:
type: exponential
interval: PT2S
maxInterval: PT30S
maxDuration: PT3M
- id: record_auto_rollback
type: io.kestra.plugin.core.kv.Set
key: "kestra-move.{{ inputs.namespace }}"
kvType: JSON
overwrite: true
value: >-
{{ {'ledger': kv('kestra-move.' ~ inputs.namespace), 'set': {'phase':
'ROLLED_BACK', 'enabled': [], 'frozen': [], 'rolled_back_at': now()},
'event': {'at': now(), 'execution': execution.id, 'mode': 'CUTOVER',
'event': 'rolled back automatically'}} | jq(vars.ledger_set_jq) |
first | toJson }}
- id: explain_rollback
type: io.kestra.plugin.core.log.Log
level: ERROR
message: >-
The cutover of {{ inputs.namespace }} failed after the source was
frozen and was rolled back: the source flows run again and the target
copies are disabled. The task that failed explains why. Fix it and run
CUTOVER again.
finally:
- id: drop_scratch
type: io.kestra.plugin.core.kv.Delete
key: "kestra-move.{{ inputs.namespace }}.scratch"
errorOnMissing: false
- id: final_state
type: io.kestra.plugin.core.log.Log
message: >-
Ledger kestra-move.{{ inputs.namespace }} in {{ flow.namespace }}: {{
(kv('kestra-move.' ~ inputs.namespace, errorOnMissing=false) ?? {}).phase
?? 'not written' }}.
outputs:
- id: plan
type: JSON
description: The full plan; the writes, blockers, warnings and webhooks keys are
the ones to read first.
value: "{{ outputs.plan.values.plan ?? '{}' }}"
triggers:
- id: nightly_drift_report
type: io.kestra.plugin.core.trigger.Schedule
description: >-
Enable it while both instances coexist to get a nightly dry-run STAGE
report of what changed on the source since the last copy. It never writes.
disabled: true
cron: "0 6 * * *"
inputs:
mode: STAGE
dry_run: "true"
namespace: company.billing
source_url: http://kestra-old:8080
target_url: http://kestra-new:8080
