id: kestra-subflow-interface-change-gate
namespace: company.platform
concurrency:
limit: 1
behavior: QUEUE
inputs:
- id: shared_namespace
type: STRING
defaults: company.shared
description: Namespace of the reusable flows that other flows call as subflows.
Only these flows are checked and deployed.
- id: caller_namespace_prefix
type: STRING
defaults: company
description: Root namespace to scan for callers. Every flow in it and its child
namespaces is checked.
- id: candidate_source
type: STRING
required: false
description: YAML of a new version of one flow from shared_namespace. Leave
empty to audit the live flows instead.
- id: allow_breaking
type: BOOL
defaults: false
description: Wait for an approval instead of failing when the candidate breaks
existing callers.
- id: dry_run
type: BOOL
defaults: true
description: Only check the candidate. Set to false to deploy it when the check passes.
- id: max_flows
type: INT
defaults: 1000
min: 1
max: 10000
description: Upper bound on the number of caller flows scanned. The run fails
rather than skip callers above it.
- id: kestra_url
type: URI
defaults: http://localhost:8080
description: Base URL of the Kestra API.
variables:
tenant: main
# Input: {library, callers, candidate, shared}. Output: one record per call site into the shared namespace.
interface_check: |
def key: .namespace + "/" + .id;
def iface: {disabled: (.disabled == true), inputs: (.inputs // []), outputs: [(.outputs // [])[] | .id]};
def among($list; $x): ($list | map(select(. == $x)) | length) > 0;
.shared as $shared | .candidate as $cand
| ([.library[] | {key: key, value: iface}] | from_entries) as $live
| (if $cand == null then $live else $live + {($cand | key): ($cand | iface)} end) as $iface
| ([.callers[] | select($cand == null or key != ($cand | key))] + (if $cand == null then [] else [$cand] end)) as $flows
| [ $flows[] as $f
| ($f | tostring) as $text
| $f | .. | objects | select(.type? == "io.kestra.plugin.core.flow.Subflow")
| . as $t
| {caller: ($f | key), task: $t.id, subflow: ($t.namespace + "/" + $t.flowId)} as $site
| if (($t.namespace + $t.flowId) | contains("{")) then
[$site + {severity: "UNVERIFIED", problem: "namespace or flowId is an expression, check this call by hand"}]
elif $t.namespace != $shared then []
elif $iface[$site.subflow] == null then
[$site + {severity: "BREAKING", problem: "the subflow does not exist"}]
else
$iface[$site.subflow] as $i
| ([($t.inputs // {}) | keys[]]) as $passed
| ([$i.inputs[] | .id]) as $declared
| ($live[$site.subflow].inputs // []) as $before
| [ (if $i.disabled then $site + {severity: "BREAKING", problem: "the subflow is disabled"} else empty end),
($i.inputs[] | select(.required != false and .defaults == null) | .id
| select(among($passed; .) | not)
| $site + {severity: "BREAKING", problem: ("required input " + . + " is not passed")}),
($passed[] | select(among($declared; .) | not)
| $site + {severity: "WARNING", problem: ("passes input " + . + " that the subflow does not declare, the value is ignored")}),
($i.inputs[] | . as $new | select(among($passed; $new.id))
| ($before[] | select(.id == $new.id and .type != $new.type))
| $site + {severity: "WARNING", problem: ("input " + $new.id + " changes type from " + .type + " to " + $new.type)}),
([$text | scan("outputs[.]" + $t.id + "[.]outputs[.]([A-Za-z0-9_-]+)") | .[0]] | unique[]
| select(among($i.outputs; .) | not)
| $site + {severity: "BREAKING", problem: ("reads output " + . + " that the subflow does not return")}) ]
| if length == 0 then [$site + {severity: "OK", problem: ""}] else . end
end
| .[] ]
tasks:
- id: candidate_checks
type: io.kestra.plugin.core.flow.If
description: Validates the candidate with Kestra's own flow validator before
anything else uses it.
condition: "{{ inputs.candidate_source is not empty }}"
then:
- id: validate_candidate
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows/validate"
method: POST
contentType: application/x-yaml
body: "{{ inputs.candidate_source }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
- id: check_candidate
type: io.kestra.plugin.core.execution.Assert
description: One valid flow from the shared namespace, and never this gate itself.
conditions:
- "{{ (fromJson(outputs.validate_candidate.body) | length) == 1 }}"
- "{{ (fromJson(outputs.validate_candidate.body) | first).constraints
is empty }}"
- "{{ (fromJson(outputs.validate_candidate.body) | first).namespace ==
inputs.shared_namespace }}"
- "{{ (fromJson(outputs.validate_candidate.body) | first).flow !=
flow.id or inputs.shared_namespace != flow.namespace }}"
errorMessage: "The candidate must be exactly one valid flow of namespace {{
inputs.shared_namespace }}: {{ outputs.validate_candidate.body }}"
- id: live_candidate
type: io.kestra.plugin.core.http.Request
description: Current source of the flow the candidate replaces. A 404 means the
candidate is a new flow.
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows/{{
inputs.shared_namespace }}/{{
(fromJson(outputs.validate_candidate.body) | first).flow
}}?source=true"
options:
allowFailed: true
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: check_live_lookup
type: io.kestra.plugin.core.execution.Assert
conditions:
- "{{ outputs.live_candidate.code == 200 or
outputs.live_candidate.code == 404 }}"
errorMessage: "Could not read the live version of the candidate (HTTP {{
outputs.live_candidate.code }})"
- id: library
type: io.kestra.plugin.kestra.flows.List
description: Current inputs and outputs of every flow in the shared namespace.
kestraUrl: "{{ inputs.kestra_url }}"
tenantId: "{{ vars.tenant }}"
namespace: "{{ inputs.shared_namespace }}"
auth:
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
- id: callers
type: io.kestra.plugin.core.http.Request
description: Every flow under the caller prefix, with its tasks, in one bounded page.
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant
}}/flows/search?page=1&size={{ inputs.max_flows
}}&filters%5Bnamespace%5D%5BPREFIX%5D={{ inputs.caller_namespace_prefix
}}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
- id: check_scan_complete
type: io.kestra.plugin.core.execution.Assert
conditions:
- "{{ (fromJson(outputs.callers.body)).total <= inputs.max_flows }}"
errorMessage: "{{ (fromJson(outputs.callers.body)).total }} flows match {{
inputs.caller_namespace_prefix }}, more than max_flows ({{
inputs.max_flows }}). Raise max_flows so no caller is skipped."
- id: analysis
type: io.kestra.plugin.core.output.OutputValues
values:
findings: "{{ {'library': outputs.library.flows, 'callers':
(fromJson(outputs.callers.body)).results, 'candidate':
(inputs.candidate_source is empty ? null :
yaml(inputs.candidate_source)), 'shared': inputs.shared_namespace} |
jq(vars.interface_check) | first | toJson }}"
- id: verdict
type: io.kestra.plugin.core.output.OutputValues
description: BLOCKED and AUDIT_BROKEN fail the run. UNCHANGED makes a repeated
deploy a no-op.
values:
sites: "{{ outputs.analysis.values.findings | jq('[.[] | .caller + \"/\" +
.task] | unique | length') | first }}"
broken_sites: "{{ outputs.analysis.values.findings | jq('[.[] | select(.severity
== \"BREAKING\") | .caller + \"/\" + .task] | unique | length') | first
}}"
breaking: "{{ outputs.analysis.values.findings | jq('[.[] | select(.severity ==
\"BREAKING\")] | length') | first }}"
warnings: "{{ outputs.analysis.values.findings | jq('[.[] | select(.severity ==
\"WARNING\" or .severity == \"UNVERIFIED\")] | length') | first }}"
decision: >-
{%- set breaking = outputs.analysis.values.findings | jq('[.[] |
select(.severity == "BREAKING")] | length') | first -%} {%- if
inputs.candidate_source is empty -%}{{ breaking > 0 ? 'AUDIT_BROKEN' :
'AUDIT_OK' }} {%- elseif breaking > 0 and not inputs.allow_breaking
-%}BLOCKED {%- elseif outputs.live_candidate.code == 200 and
(fromJson(outputs.live_candidate.body)).source | trim ==
inputs.candidate_source | trim -%}UNCHANGED {%- elseif inputs.dry_run
-%}CHECKED {%- elseif breaking > 0 -%}APPROVAL {%- else -%}DEPLOY{%-
endif -%}
- id: report
type: io.kestra.plugin.core.log.Log
level: "{{ (outputs.verdict.values.breaking | number) > 0 ? 'ERROR' :
((outputs.verdict.values.warnings | number) > 0 ? 'WARN' : 'INFO') }}"
message: |-
{{ outputs.verdict.values.decision }}: {{ outputs.verdict.values.sites }} call sites into {{ inputs.shared_namespace }}, {{ outputs.verdict.values.breaking }} breaking problems in {{ outputs.verdict.values.broken_sites }} of them, {{ outputs.verdict.values.warnings }} findings to review.
{%- for f in fromJson(outputs.analysis.values.findings) %}{% if f.severity != 'OK' %}
{{ f.severity }} {{ f.caller }} task {{ f.task }} -> {{ f.subflow }}: {{ f.problem }}{% endif %}{% endfor %}
- id: gate
type: io.kestra.plugin.core.flow.Switch
value: "{{ outputs.verdict.values.decision }}"
cases:
BLOCKED:
- id: block_change
type: io.kestra.plugin.core.execution.Fail
errorMessage: "Not deployed: the candidate breaks {{
outputs.verdict.values.broken_sites }} call sites. Fix the callers
first, or rerun with allow_breaking to ask for approval."
AUDIT_BROKEN:
- id: fail_audit
type: io.kestra.plugin.core.execution.Fail
errorMessage: "{{ outputs.verdict.values.broken_sites }} call sites into {{
inputs.shared_namespace }} fail at runtime. See the report task for
each caller and task."
APPROVAL:
- id: approve_breaking_change
type: io.kestra.plugin.core.flow.Pause
description: Resume to deploy a change that breaks the listed callers. Without
approval within 8 hours, the run is cancelled and nothing is
deployed.
pauseDuration: PT8H
behavior: CANCEL
defaults:
- id: nothing_blocks
type: io.kestra.plugin.core.log.Log
message: "{{ outputs.verdict.values.decision }}: no call site blocks this run."
- id: release
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.verdict.values.decision == 'DEPLOY' or
outputs.verdict.values.decision == 'APPROVAL' }}"
then:
- id: deploy
type: io.kestra.plugin.core.http.Request
description: Updates the flow, or creates it when it is new.
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows{{
outputs.live_candidate.code == 200 ? '/' ~ inputs.shared_namespace ~
'/' ~ (fromJson(outputs.validate_candidate.body) | first).flow : ''
}}"
method: "{{ outputs.live_candidate.code == 200 ? 'PUT' : 'POST' }}"
contentType: application/x-yaml
body: "{{ inputs.candidate_source }}"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: deployed
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows/{{
inputs.shared_namespace }}/{{
(fromJson(outputs.validate_candidate.body) | first).flow
}}?source=true"
options:
auth:
type: BASIC
username: "{{ secret('KESTRA_USERNAME') }}"
password: "{{ secret('KESTRA_PASSWORD') }}"
- id: check_deployed
type: io.kestra.plugin.core.execution.Assert
conditions:
- "{{ (fromJson(outputs.deployed.body)).source | trim ==
inputs.candidate_source | trim }}"
errorMessage: "The live source of {{ (fromJson(outputs.validate_candidate.body)
| first).flow }} does not match the candidate after the deploy.
Another change may have landed at the same time."
- id: summary
type: io.kestra.plugin.core.log.Log
message: "Deployed {{ inputs.shared_namespace }}.{{
(fromJson(outputs.validate_candidate.body) | first).flow }} revision
{{ (fromJson(outputs.deployed.body)).revision }} ({{
outputs.verdict.values.broken_sites }} breaking call sites approved)."
outputs:
- id: decision
type: STRING
value: "{{ outputs.verdict.values.decision ?? 'NOT_EVALUATED' }}"
- id: findings
type: JSON
value: "{{ outputs.analysis.values.findings ?? '[]' }}"
triggers:
- id: nightly_audit
type: io.kestra.plugin.core.trigger.Schedule
cron: "0 6 * * *"
