id: kestra-instance-snapshot-and-drift-manifest
namespace: company.team
description: |
Nightly snapshot of the flows of a Kestra instance through its own API: one flows ZIP per namespace plus a SHA-256
manifest of every flow definition, compared with the previous night (added, removed, changed, revision only), with
the ZIPs and the manifest copied to an S3 bucket so the backup survives losing the instance. Nothing is modified on
the instance. The ERROR lines of recent failed executions are the job of the companion flow
`kestra-failed-execution-error-digest`.
labels:
team: platform
purpose: instance-snapshot
inputs:
- id: namespace_prefix
type: STRING
defaults: ""
description: Only namespaces starting with this prefix are snapshotted. Empty
(default) means every namespace of the tenant.
- id: exclude_namespaces
type: JSON
defaults: [ "system" ]
description: Namespaces left out even when they match the prefix. `system` holds
Kestra's own flows and is skipped by default.
- id: alert_on_drift
type: BOOL
defaults: true
description: Send a Slack message when a flow was added, removed or changed
since the previous snapshot.
- id: manifest_ttl
type: DURATION
defaults: P90D
description: How long the manifest is kept in KV when the flow stops running.
Each run renews it.
- 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.
- id: s3_bucket
type: STRING
defaults: ""
description: Bucket that receives the flows ZIPs and the manifest. Empty
(default) keeps them in internal storage only, which does not survive
losing the instance; the run logs a WARN and gets the label
`snapshot_offsite` = `skipped`.
- id: s3_prefix
type: STRING
defaults: kestra-snapshots
description: Key prefix in the bucket. Objects land under
`<prefix>/<yyyy-MM-dd>/<namespace>.zip` and
`<prefix>/<yyyy-MM-dd>/manifest.json`, so one night is one folder.
- id: s3_region
type: STRING
defaults: us-east-1
description: Region of the bucket.
- id: s3_endpoint
type: STRING
defaults: ""
description: Endpoint override for an S3-compatible store (MinIO, Ceph,
LocalStack), for example `http://minio:9000`. Empty (default) means AWS
S3.
variables:
tenant: main
state_key: kestra_snapshot
drift_jq: |
.prev as $p | .curr as $c | .keep as $k
| ($c + ($p | with_entries(select(.key as $n | ($k | index($n)) != null)))) as $new
| { manifest: $new,
namespaces: [ (($new | keys) + ($p | keys) | unique)[] as $ns
| ($new[$ns] // {}) as $cn | ($p[$ns] // {}) as $pn
| { namespace: $ns,
flows: ($cn | length),
added: [ $cn | keys[] | select($pn[.] == null) ],
removed: [ $pn | keys[] | select($cn[.] == null) ],
changed: [ $cn | keys[] | select($pn[.] != null and $pn[.].hash != $cn[.].hash) ],
revision_only: [ $cn | keys[] | select($pn[.] != null and $pn[.].hash == $cn[.].hash and $pn[.].revision != $cn[.].revision) ] }
| . + { total: ((.added | length) + (.removed | length) + (.changed | length)) } ] }
| . + { total: ([.namespaces[].total] | add // 0) }
checks:
- when: "{{ inputs.exclude_namespaces | jq('type == \"array\" and all(.[]; type ==
\"string\")') | first }}"
message: exclude_namespaces must be a JSON array of namespace names, for example
["system", "company.sandbox"].
style: ERROR
behavior: BLOCK_EXECUTION
- when: "{{ {'p': inputs.namespace_prefix} | jq('.p | test(\"^[a-z0-9._-]*$\")') |
first }}"
message: namespace_prefix may only contain lowercase letters, digits, dots,
dashes and underscores.
style: ERROR
behavior: BLOCK_EXECUTION
- when: "{{ {'u': inputs.kestra_api_url | trim} | jq('.u |
test(\"^https?://.+[^/]$\")') | first }}"
message: kestra_api_url must be an http(s) URL without a trailing slash, for
example http://localhost:8080.
style: ERROR
behavior: BLOCK_EXECUTION
- when: "{{ {'e': inputs.s3_endpoint | trim} | jq('.e == \"\" or (.e |
test(\"^https?://.+[^/]$\"))') | first }}"
message: s3_endpoint must be empty (AWS S3) or an http(s) URL without a trailing
slash, for example http://minio:9000.
style: ERROR
behavior: BLOCK_EXECUTION
concurrency:
limit: 1
behavior: CANCEL
sla:
- id: max_runtime
type: MAX_DURATION
duration: PT30M
behavior: FAIL
labels:
snapshot_sla: breached
triggers:
- id: nightly
type: io.kestra.plugin.core.trigger.Schedule
cron: "15 2 * * *"
description: "Every night at 02:15. `recoverMissedSchedules: LAST` takes one
snapshot after a scheduler outage instead of one per missed night."
recoverMissedSchedules: LAST
tasks:
- id: previous_manifest
type: io.kestra.plugin.core.kv.Get
description: Last night's manifest. Missing on the first run, which then only
records a baseline.
key: "{{ vars.state_key }}_manifest"
errorOnMissing: false
- id: discover_namespaces
type: io.kestra.plugin.kestra.namespaces.List
description: "Every namespace of the tenant under the prefix, including
namespaces that only exist because a flow lives in them (`existingOnly:
false`)."
kestraUrl: "{{ inputs.kestra_api_url }}"
tenantId: "{{ vars.tenant }}"
auth:
{
username: "{{ secret('KESTRA_API_USERNAME') }}",
password: "{{ secret('KESTRA_API_PASSWORD') }}"
}
prefix: "{{ inputs.namespace_prefix }}"
existingOnly: false
- id: scope
type: io.kestra.plugin.core.output.OutputValues
description: Namespaces to snapshot and the snapshot date. `taken_at` is the
schedule date, so a recovered run is filed under its own night, and `day`
names the S3 folder.
values:
namespaces: "{{ {'all': outputs.discover_namespaces.namespaces ?? [],
'excluded': inputs.exclude_namespaces} | jq('.excluded as $x | [.all[] |
select(. as $n | ($x | index($n)) == null)] | sort') | first | toJson
}}"
taken_at: "{{ trigger.date ?? execution.startDate }}"
day: "{{ (trigger.date ?? execution.startDate) | date('yyyy-MM-dd') }}"
offsite: "{{ (inputs.s3_bucket | trim) != '' }}"
run_url: "{{ kestra.url ?? inputs.kestra_api_url }}/ui/{{ vars.tenant
}}/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}"
- id: nothing_in_scope
type: io.kestra.plugin.core.flow.If
description: A prefix that matches nothing is a configuration mistake, not an
empty snapshot.
condition: "{{ fromJson(outputs.scope.values.namespaces) | length == 0 }}"
then:
- id: refuse_empty_scope
type: io.kestra.plugin.core.execution.Fail
errorMessage: "No namespace matches prefix '{{ inputs.namespace_prefix }}' after
excluding {{ inputs.exclude_namespaces | toJson }}. Found: {{
(outputs.discover_namespaces.namespaces ?? []) | toJson }}."
- id: no_offsite_copy
type: io.kestra.plugin.core.log.Log
description: Said once, loudly, because a ZIP that only lives in the instance's
internal storage is not a backup of that instance.
runIf: "{{ outputs.scope.values.offsite != 'true' }}"
level: WARN
message: "s3_bucket is empty: the flows ZIPs and the manifest stay in internal
storage only and will not survive losing this instance."
- id: snapshot
type: io.kestra.plugin.core.flow.Loop
description: |
One iteration per namespace, two at a time. `transmitFailed: false`: a namespace whose export fails is recorded in
its row and dead-lettered at the end, the other namespaces are still snapshotted.
values: "{{ outputs.scope.values.namespaces }}"
concurrencyLimit: 2
transmitFailed: false
tasks:
- id: list_flows
type: io.kestra.plugin.core.http.Download
description: |
Flow definitions of exactly this namespace (not its children), saved to internal storage. `flows.List` would be
the natural task, but it returns every definition as a task output and a namespace with 140 flows produced a
2.5 MB output, over the 1 MB queue message limit of 2.0.5, so the iteration failed. A downloaded file has no such limit.
uri: "{{ inputs.kestra_api_url | trim }}/api/v1/{{ vars.tenant }}/flows/{{
item.value }}"
options:
auth:
{
type: BASIC,
username: "{{ secret('KESTRA_API_USERNAME') }}",
password: "{{ secret('KESTRA_API_PASSWORD') }}"
}
timeout: PT2M
retry: { type: exponential, interval: PT5S, maxInterval: PT1M, maxAttempts: 3 }
- id: flow_hashes
type: io.kestra.plugin.core.output.OutputValues
description: |
SHA-256 of each flow definition as the API returns it, with `revision` and `updated` removed, so the hash only
moves when the definition moves. A new revision with the same hash (a re-save, a comment edit) is reported as
`revision_only`. `read()` loads the downloaded file, the rendered list is a few hundred bytes per flow.
values:
flows: >-
[{% for f in fromJson(read(outputs.list_flows.uri)) %}{{ {'id':
f.id, 'revision': f.revision, 'disabled': f.disabled, 'hash': (f |
jq('del(.revision, .updated, .deleted, .draft)') | first | toJson |
sha256)} | toJson }}{% if not loop.last %},{% endif %}{% endfor %}]
- id: export_flows
type: io.kestra.plugin.kestra.flows.Export
description: |
ZIP of `<namespace>-<flow>.yml` sources, the restore artifact (import it from the Flows page or with the
`/flows/import` API). `namespace` is a prefix filter, so the ZIP of `company` also holds `company.team`.
Skipped for namespaces that own no flow.
runIf: "{{ fromJson(outputs.flow_hashes.values.flows) | length > 0 }}"
kestraUrl: "{{ inputs.kestra_api_url }}"
tenantId: "{{ vars.tenant }}"
auth:
{
username: "{{ secret('KESTRA_API_USERNAME') }}",
password: "{{ secret('KESTRA_API_PASSWORD') }}"
}
namespace: "{{ item.value }}"
- id: upload_zip
type: io.kestra.plugin.aws.s3.Upload
description: |
The ZIP as `<prefix>/<day>/<namespace>.zip`. `endpointOverride`, `forcePathStyle` and `compatibilityMode` only
take effect when `s3_endpoint` is set, so the same task writes to AWS S3 or to MinIO. Skipped with the bucket
empty or when the namespace owns no flow.
runIf: "{{ outputs.scope.values.offsite == 'true' and
outputs.export_flows.flowsZip is defined }}"
endpointOverride: "{{ (inputs.s3_endpoint | trim) != '' ? inputs.s3_endpoint : null }}"
forcePathStyle: "{{ (inputs.s3_endpoint | trim) != '' }}"
compatibilityMode: "{{ (inputs.s3_endpoint | trim) != '' }}"
region: "{{ inputs.s3_region }}"
accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
bucket: "{{ inputs.s3_bucket }}"
key: "{{ inputs.s3_prefix }}/{{ outputs.scope.values.day }}/{{ item.value
}}.zip"
from: "{{ outputs.export_flows.flowsZip }}"
retry: { type: exponential, interval: PT5S, maxInterval: PT1M, maxAttempts: 3 }
outputs:
- id: row
type: JSON
value: |
{{ {
'namespace': item.value,
'flow_count': fromJson(outputs.flow_hashes.values.flows ?? '[]') | length,
'flows': fromJson(outputs.flow_hashes.values.flows ?? '[]'),
'zip': outputs.export_flows.flowsZip ?? null,
's3_key': outputs.upload_zip.key ?? null,
'ok': outputs.flow_hashes.values.flows is defined
and (fromJson(outputs.flow_hashes.values.flows ?? '[]') | length == 0 or outputs.export_flows.flowsZip is defined)
and (outputs.scope.values.offsite != 'true' or outputs.export_flows.flowsZip is not defined or outputs.upload_zip.key is defined)
} | toJson }}
- id: drift
type: io.kestra.plugin.core.output.OutputValues
description: |
Compares the previous manifest with tonight's, flow by flow (`vars.drift_jq`). Namespaces whose iteration failed keep
last night's entry instead of showing every flow as removed; namespaces that left the scope are reported as removed.
values:
baseline: "{{ (outputs.previous_manifest.value ?? '') == '' }}"
failed_namespaces: "{{ (outputs.snapshot.outputs ?? []) | jq('[.[].outputs.row |
select(.ok != true) | .namespace]') | first | toJson }}"
report: >-
{{ {
'prev': outputs.previous_manifest.value.namespaces ?? {},
'curr': (outputs.snapshot.outputs ?? []) | jq('[.[].outputs.row | select(.ok == true) | {(.namespace): ([.flows[] | {(.id): {hash, revision}}] | add // {})}] | add // {}') | first,
'keep': (outputs.snapshot.outputs ?? []) | jq('[.[].outputs.row | select(.ok != true) | .namespace]') | first
} | jq(vars.drift_jq) | first | toJson }}
- id: save_manifest
type: io.kestra.plugin.core.kv.Set
description: Tonight's manifest becomes tomorrow's baseline. The TTL only
matters if the flow stops running.
key: "{{ vars.state_key }}_manifest"
kvType: JSON
ttl: "{{ inputs.manifest_ttl }}"
value: "{{ {'taken_at': outputs.scope.values.taken_at, 'execution_id':
execution.id, 'namespaces':
fromJson(outputs.drift.values.report).manifest} | toJson }}"
- id: offsite_manifest
type: io.kestra.plugin.core.flow.If
description: The manifest and tonight's drift next to the ZIPs, so a restore
knows which ZIP holds which flows and what changed that night.
condition: "{{ outputs.scope.values.offsite == 'true' }}"
then:
- id: write_manifest
type: io.kestra.plugin.core.storage.Write
extension: .json
content: "{{ {'taken_at': outputs.scope.values.taken_at, 'execution_id':
execution.id, 'namespaces':
fromJson(outputs.drift.values.report).manifest, 'drift':
fromJson(outputs.drift.values.report).namespaces, 'zips':
(outputs.snapshot.outputs ?? []) | jq('[.[].outputs.row |
select(.s3_key != null) | {(.namespace): .s3_key}] | add // {}') |
first} | toJson }}"
- id: upload_manifest
type: io.kestra.plugin.aws.s3.Upload
endpointOverride: "{{ (inputs.s3_endpoint | trim) != '' ? inputs.s3_endpoint : null }}"
forcePathStyle: "{{ (inputs.s3_endpoint | trim) != '' }}"
compatibilityMode: "{{ (inputs.s3_endpoint | trim) != '' }}"
region: "{{ inputs.s3_region }}"
accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
bucket: "{{ inputs.s3_bucket }}"
key: "{{ inputs.s3_prefix }}/{{ outputs.scope.values.day }}/manifest.json"
from: "{{ outputs.write_manifest.uri }}"
retry: { type: exponential, interval: PT5S, maxInterval: PT1M, maxAttempts: 3 }
- id: summary
type: io.kestra.plugin.core.output.OutputValues
values:
drift_total: "{{ fromJson(outputs.drift.values.report).total }}"
namespaces: "{{ fromJson(outputs.scope.values.namespaces) | length }}"
flows: "{{ (outputs.snapshot.outputs ?? []) | jq('[.[].outputs.row.flow_count] |
add // 0') | first }}"
zips: "{{ (outputs.snapshot.outputs ?? []) | jq('[.[].outputs.row.zip | select(.
!= null)] | length') | first }}"
uploaded: "{{ (outputs.snapshot.outputs ?? []) | jq('[.[].outputs.row.s3_key |
select(. != null)] | length') | first }}"
s3_objects: >-
{{ {'rows': outputs.snapshot.outputs ?? [], 'manifest':
outputs.upload_manifest.key ?? ''}
| jq('[.rows[].outputs.row.s3_key | select(. != null)] + (if .manifest != "" then [.manifest] else [] end)') | first | toJson }}
- id: label
type: io.kestra.plugin.core.execution.Labels
description: Filter the Executions page by drift or by missing offsite copies
without opening a run.
labels:
snapshot_drift: "{{ outputs.summary.values.drift_total }}"
snapshot_manifest: "{{ outputs.drift.values.baseline == 'true' ? 'baseline' : 'compared' }}"
snapshot_offsite: "{{ outputs.scope.values.offsite != 'true' ? 'skipped' :
(outputs.upload_manifest.key is defined and
outputs.summary.values.uploaded == outputs.summary.values.zips ?
'complete' : 'partial') }}"
- id: log_summary
type: io.kestra.plugin.core.log.Log
message: |
Snapshot {{ outputs.scope.values.taken_at }}: {{ outputs.summary.values.namespaces }} namespace(s), {{ outputs.summary.values.flows }} flows, {{ outputs.summary.values.zips }} ZIP(s), {{ outputs.scope.values.offsite == 'true' ? (outputs.summary.values.uploaded ~ ' ZIP(s) and the manifest copied to s3://' ~ inputs.s3_bucket ~ '/' ~ inputs.s3_prefix ~ '/' ~ outputs.scope.values.day ~ '/') : 'no offsite copy (s3_bucket empty)' }}.
Drift since previous snapshot{{ outputs.drift.values.baseline == 'true' ? ' (none, first run, baseline recorded)' : '' }}: {{ outputs.summary.values.drift_total }}.
{% for n in fromJson(outputs.drift.values.report).namespaces %}{{ n.namespace }}: {{ n.flows }} flows, added {{ n.added | toJson }}, removed {{ n.removed | toJson }}, changed {{ n.changed | toJson }}, revision only {{ n.revision_only | toJson }}
{% endfor %}
- id: drift_alert
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.alert_on_drift and outputs.drift.values.baseline != 'true'
and (outputs.summary.values.drift_total | number) > 0 }}"
then:
- id: notify_drift
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: allowFailure so a missing `SLACK_WEBHOOK_URL` does not fail a
snapshot that is otherwise complete; the same text is in
`log_summary`.
allowFailure: true
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: |
:mag: Kestra flow drift since the previous snapshot: {{ outputs.summary.values.drift_total }} flow(s).
{% for n in fromJson(outputs.drift.values.report).namespaces %}{% if n.total > 0 %}`{{ n.namespace }}`: added {{ n.added | toJson }}, removed {{ n.removed | toJson }}, changed {{ n.changed | toJson }}
{% endif %}{% endfor %}{{ outputs.scope.values.run_url }}
- id: dead_letter
type: io.kestra.plugin.core.flow.If
description: Namespaces whose iteration failed are kept 7 days in KV and the run
ends FAILED, with the other namespaces' snapshot intact.
condition: "{{ fromJson(outputs.drift.values.failed_namespaces) | length > 0 }}"
then:
- id: remember_failed
type: io.kestra.plugin.core.kv.Set
key: "{{ vars.state_key }}_failed_namespaces"
kvType: JSON
ttl: P7D
value: "{{ {'execution_id': execution.id, 'taken_at':
outputs.scope.values.taken_at, 'namespaces':
fromJson(outputs.drift.values.failed_namespaces)} | toJson }}"
- id: report_partial_failure
type: io.kestra.plugin.core.execution.Fail
errorMessage: "Snapshot incomplete: {{
fromJson(outputs.drift.values.failed_namespaces) | join(', ') }}
failed (see the Loop iterations of `snapshot`). The manifest kept last
night's entry for them."
else:
- id: clear_failed
type: io.kestra.plugin.core.kv.Delete
key: "{{ vars.state_key }}_failed_namespaces"
errorOnMissing: false
errors:
- id: first_error
type: io.kestra.plugin.core.output.OutputValues
description: Read once, at the top of the branch, because `errorLogs()` also
returns lines logged by the branch itself.
values:
task: "{{ (errorLogs() | length) > 0 ? (errorLogs()[0].taskId ?? 'none') :
'none' }}"
message: "{{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'no error
line was logged' }}"
- id: log_failure
type: io.kestra.plugin.core.log.Log
level: ERROR
message: "Snapshot failed in task {{ outputs.first_error.values.task }}: {{
outputs.first_error.values.message }} {{ outputs.scope.values.run_url ??
execution.id }}"
- id: remember_failure
type: io.kestra.plugin.core.kv.Set
description: Last failure next to the manifest, so whoever reads the manifest
sees why the newest run is missing.
key: "{{ vars.state_key }}_last_failure"
kvType: JSON
ttl: P7D
value: "{{ {'execution_id': execution.id, 'failed_at': execution.startDate,
'task': outputs.first_error.values.task, 'message':
outputs.first_error.values.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: ":rotating_light: Kestra instance snapshot failed in `{{
outputs.first_error.values.task }}`: {{ outputs.first_error.values.message
}} {{ outputs.scope.values.run_url ?? execution.id }}"
finally:
- id: record_run
type: io.kestra.plugin.core.kv.Set
description: Always written, also after a failure, so a dashboard or another
flow can check when the last snapshot ran and how far it got.
key: "{{ vars.state_key }}_last_run"
kvType: JSON
ttl: "{{ inputs.manifest_ttl }}"
value: "{{ {'execution_id': execution.id, 'started_at': execution.startDate,
'namespaces': ((outputs.summary.values.namespaces ?? '') == '' ? null :
(outputs.summary.values.namespaces | number)), 'flows':
((outputs.summary.values.flows ?? '') == '' ? null :
(outputs.summary.values.flows | number)), 'drift':
((outputs.summary.values.drift_total ?? '') == '' ? null :
(outputs.summary.values.drift_total | number)), 's3_objects':
fromJson(outputs.summary.values.s3_objects ?? '[]'), 'failed_namespaces':
fromJson(outputs.drift.values.failed_namespaces ?? '[]')} | toJson }}"
afterExecution:
- id: notify_outcome
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: "Runs once the final state is known. On 2.0.5 an SLA breach fails
the running task and kills the tasks still pending, the Slack task of
`errors` included, so this is the place that reports it (label
`snapshot_sla: breached`)."
runIf: "{{ execution.state != 'FAILED' or (labels.snapshot_sla ?? '') ==
'breached' }}"
allowFailure: true
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Kestra instance snapshot ended {{ execution.state }}{{
(labels.snapshot_sla ?? '') == 'breached' ? ' because the max_runtime SLA
was breached' : '' }}: {{ outputs.summary.values.flows ?? 'n/a' }} flows
in {{ outputs.summary.values.namespaces ?? 'n/a' }} namespace(s), drift {{
outputs.summary.values.drift_total ?? 'n/a' }}, {{
outputs.summary.values.uploaded ?? 'n/a' }} ZIP(s) copied to S3. {{
outputs.scope.values.run_url ?? execution.id }}"
outputs:
- id: manifest
type: JSON
description: Tonight's manifest, namespace -> flow -> {hash, revision}; the same
value is stored in KV.
value: "{{ fromJson(outputs.drift.values.report).manifest }}"
- id: drift
type: JSON
description: Per namespace added, removed, changed and revision_only flow ids.
value: "{{ fromJson(outputs.drift.values.report).namespaces }}"
- id: snapshot
type: JSON
description: One row per namespace with the flows ZIP URI in internal storage
and its S3 key.
value: "{{ (outputs.snapshot.outputs ?? []) | jq('[.[].outputs.row |
del(.flows)]') | first }}"
- id: s3_objects
type: JSON
description: Keys written to the bucket tonight (ZIPs and `manifest.json`);
empty when `s3_bucket` is empty.
value: "{{ fromJson(outputs.summary.values.s3_objects) }}"
