Gate a Kestra upgrade against every live flow

id: kestra-upgrade-compatibility-gate
namespace: company.team

description: |
  Compare every live flow in a Kestra namespace with a candidate Kestra version before an upgrade.
  The gate validates the current and candidate representations, reports only regressions and can
  quarantine candidate-invalid schedules after an approval by disabling their triggers and recording
  the prior state in KV for reversal.

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: kestra_url
    type: STRING
    displayName: Current Kestra URL
    description: Base URL of the running Kestra instance whose flows and triggers
      are checked.
    defaults: "http://localhost:8080"

  - id: candidate_url
    type: STRING
    displayName: Candidate Kestra URL
    description: Base URL of the second Kestra instance running the candidate image.
    defaults: "http://localhost:8080"

  - id: namespace
    type: STRING
    displayName: Flow namespace
    description: Exact namespace exported and checked.
    defaults: company.team

  - id: tenant_id
    type: STRING
    displayName: Tenant id
    defaults: main

  - id: kestra_user
    type: STRING
    displayName: Kestra API username
    defaults: admin@kestra.io

  - id: kestra_password
    type: SECRET
    displayName: Kestra API password
    defaults: "{{ secret('KESTRA_PASSWORD') }}"

  - id: record_namespace
    type: STRING
    displayName: KV record namespace
    description: Namespace where quarantine records are written.
    defaults: company.team

  - id: auto_remediate
    type: BOOL
    displayName: Quarantine candidate-invalid flows
    description: Permit trigger disabling after approval. Keep false for an audit-only run.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Keep true to report regressions without changing trigger state.
    defaults: true

tasks:
  - id: inventory
    type: io.kestra.plugin.kestra.flows.List
    description: Read every deployed flow with the Kestra plugin so trigger ids and
      task types stay tied to live state.
    kestraUrl: "{{ inputs.kestra_url }}"
    tenantId: "{{ inputs.tenant_id }}"
    namespace: "{{ inputs.namespace }}"
    auth:
      username: "{{ inputs.kestra_user }}"
      password: "{{ inputs.kestra_password }}"

  - id: export
    type: io.kestra.plugin.kestra.flows.Export
    description: Export the complete namespace to Kestra internal storage as the
      source snapshot used by the audit.
    kestraUrl: "{{ inputs.kestra_url }}"
    tenantId: "{{ inputs.tenant_id }}"
    namespace: "{{ inputs.namespace }}"
    auth:
      username: "{{ inputs.kestra_user }}"
      password: "{{ inputs.kestra_password }}"

  - id: unzip
    type: io.kestra.plugin.compress.ArchiveDecompress
    description: Decompress the exported source snapshot so each validation uses the
      exact deployed YAML.
    algorithm: ZIP
    from: "{{ outputs.export.flowsZip }}"

  - id: scan
    type: io.kestra.plugin.core.flow.Loop
    description: Validate each live flow against both versions and emit one regression row.
    concurrencyLimit: 1
    values: "{{ outputs.unzip.files | keys }}"
    tasks:
      - id: validate_current
        type: io.kestra.plugin.core.http.Request
        description: Validate the live flow against the current Kestra API.
        uri: "{{ inputs.kestra_url }}/api/v1/{{ inputs.tenant_id }}/flows/validate"
        method: POST
        contentType: application/x-yaml
        body: "{{ read(outputs.unzip.files[item.value]) }}"
        options:
          auth:
            type: BASIC
            username: "{{ inputs.kestra_user }}"
            password: "{{ inputs.kestra_password }}"

      - id: validate_candidate
        type: io.kestra.plugin.core.http.Request
        description: Validate the same live flow against the candidate Kestra API.
        uri: "{{ inputs.candidate_url }}/api/v1/{{ inputs.tenant_id }}/flows/validate"
        method: POST
        contentType: application/x-yaml
        body: "{{ read(outputs.unzip.files[item.value]) }}"
        options:
          auth:
            type: BASIC
            username: "{{ inputs.kestra_user }}"
            password: "{{ inputs.kestra_password }}"

      - id: facts
        type: io.kestra.plugin.core.output.OutputValues
        values:
          current_constraints: "{{ outputs.validate_current.body | jq('.[0].constraints //
            \"\"') | first }}"
          candidate_constraints: "{{ outputs.validate_candidate.body |
            jq('.[0].constraints // \"\"') | first }}"
          current_deprecations: "{{ outputs.validate_current.body |
            jq('.[0].deprecationPaths // []') | first | toJson }}"
          candidate_deprecations: "{{ outputs.validate_candidate.body |
            jq('.[0].deprecationPaths // []') | first | toJson }}"
          current_warnings: "{{ outputs.validate_current.body | jq('.[0].warnings // []')
            | first | toJson }}"
          candidate_warnings: "{{ outputs.validate_candidate.body | jq('.[0].warnings //
            []') | first | toJson }}"
          candidate_infos: "{{ outputs.validate_candidate.body | jq('.[0].infos // []') |
            first | toJson }}"

      - id: classify
        type: io.kestra.plugin.core.output.OutputValues
        values:
          class: "{{ {'current_constraints': outputs.facts.values.current_constraints,
            'candidate_constraints': outputs.facts.values.candidate_constraints,
            'current_deprecations': outputs.facts.values.current_deprecations,
            'candidate_deprecations':
            outputs.facts.values.candidate_deprecations, 'current_warnings':
            outputs.facts.values.current_warnings, 'candidate_warnings':
            outputs.facts.values.candidate_warnings, 'candidate_infos':
            outputs.facts.values.candidate_infos} | toJson | jq('if
            ((.candidate_constraints | length) > 0 and .candidate_constraints !=
            .current_constraints) or ((.candidate_infos | fromjson) |
            any(contains(\"replaced\"))) then \"INVALID\" elif
            (((.candidate_deprecations | fromjson) - (.current_deprecations |
            fromjson)) | length) > 0 then \"NEW_DEPRECATION\" elif
            (((.candidate_warnings | fromjson) - (.current_warnings | fromjson))
            | length) > 0 then \"NEW_WARNING\" else \"UNCHANGED\" end') | first
            }}"
          new_deprecations: "{{ {'current': outputs.facts.values.current_deprecations,
            'candidate': outputs.facts.values.candidate_deprecations} | toJson |
            jq('(.candidate | fromjson) - (.current | fromjson)') | first |
            toJson }}"
          new_warnings: "{{ {'current': outputs.facts.values.current_warnings,
            'candidate': outputs.facts.values.candidate_warnings} | toJson |
            jq('(.candidate | fromjson) - (.current | fromjson)') | first |
            toJson }}"

      - id: row
        type: io.kestra.plugin.core.log.Log
        message: "{{ inputs.namespace }}/{{ item.value | replace({(inputs.namespace ~
          '-'): ''}) | replace({'.yml': ''}) }} | {{
          outputs.classify.values.class }} | current_constraints={{
          outputs.facts.values.current_constraints | default('none') }} |
          candidate_constraints={{ outputs.facts.values.candidate_constraints |
          default('none') }} | new_deprecations={{
          outputs.classify.values.new_deprecations }} | new_warnings={{
          outputs.classify.values.new_warnings }}"

      - id: record_invalid
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.classify.values.class == 'INVALID' }}"
        then:
          - id: save_finding
            type: io.kestra.plugin.core.kv.Set
            description: Record the candidate-invalid flow and its trigger ids before any
              approval or write.
            namespace: "{{ inputs.record_namespace }}"
            key: "kestra-compat-{{ execution.id }}-invalid-{{ inputs.namespace |
              replace({'.': '-'}) }}-{{ item.value | replace({(inputs.namespace
              ~ '-'): ''}) | replace({'/': '-'}) | replace({':': '-'}) |
              replace({'.': '-'}) | replace({' ': '-'}) }}"
            value: "{{ {'namespace': inputs.namespace, 'flow_id': item.value |
              replace({(inputs.namespace ~ '-'): ''}) | replace({'.yml': ''}),
              'kestra_url': inputs.kestra_url, 'tenant_id': inputs.tenant_id,
              'triggers': (outputs.inventory.flows | toJson | jq('[.[] |
              select(.id == \"' ~ (item.value | replace({(inputs.namespace ~
              '-'): ''}) | replace({'.yml': ''})) ~ '\") | .triggers[]]') |
              first | toJson), 'reason':
              outputs.facts.values.candidate_constraints | default('candidate
              plugin type is replaced or unavailable')} | toJson }}"
            kvType: STRING
            overwrite: true
        else:
          - id: no_invalid_record
            type: io.kestra.plugin.core.log.Log
            message: "{{ item.value | replace({(inputs.namespace ~ '-'): ''}) |
              replace({'.yml': ''}) }} is {{ outputs.classify.values.class }};
              no quarantine record was created."

  - id: finding_keys
    type: io.kestra.plugin.core.kv.GetKeys
    description: Read the findings written by the scan so approval sees the exact snapshot.
    namespace: "{{ inputs.record_namespace }}"
    prefix: "kestra-compat-{{ execution.id }}-invalid-"

  - id: remediation_gate
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.auto_remediate and not inputs.dry_run and
      (outputs.finding_keys.keys | length) > 0 }}"
    then:
      - id: approval
        type: io.kestra.plugin.core.flow.Pause
        description: Approve disabling triggers for flows that became invalid on the
          candidate.
        pauseDuration: PT1H
        behavior: RESUME
        onResume:
          - id: approved
            type: BOOL
            required: false
            description: Set true to quarantine the recorded candidate-invalid triggers.
          - id: note
            type: STRING
            required: false
            description: Optional reason for the quarantine record.

      - id: approved_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.approval.onResume.approved ?? false }}"
        then:
          - id: quarantine_flows
            type: io.kestra.plugin.core.flow.Loop
            description: Reconcile every recorded flow and disable its still-enabled
              triggers serially.
            concurrencyLimit: 1
            values: "{{ outputs.finding_keys.keys }}"
            tasks:
              - id: read_finding
                type: io.kestra.plugin.core.kv.Get
                namespace: "{{ inputs.record_namespace }}"
                key: "{{ item.value }}"
                errorOnMissing: true

              - id: recheck_source
                type: io.kestra.plugin.core.http.Request
                description: Read the flow source again after the approval so a fix made during
                  the Pause is seen.
                uri: "{{ inputs.kestra_url }}/api/v1/{{ inputs.tenant_id }}/flows/{{
                  fromJson(outputs.read_finding.value).namespace }}/{{
                  fromJson(outputs.read_finding.value).flow_id }}?source=true"
                method: GET
                options:
                  auth:
                    type: BASIC
                    username: "{{ inputs.kestra_user }}"
                    password: "{{ inputs.kestra_password }}"

              - id: recheck_candidate
                type: io.kestra.plugin.core.http.Request
                description: Validate the current source against the candidate again immediately
                  before any write.
                uri: "{{ inputs.candidate_url }}/api/v1/{{ inputs.tenant_id }}/flows/validate"
                method: POST
                contentType: application/x-yaml
                body: "{{ outputs.recheck_source.body | jq('.source') | first }}"
                options:
                  auth:
                    type: BASIC
                    username: "{{ inputs.kestra_user }}"
                    password: "{{ inputs.kestra_password }}"

              - id: recheck_gate
                type: io.kestra.plugin.core.flow.If
                description: Disable triggers only for a flow that still has a candidate
                  constraint or a replacement notice.
                condition: "{{ outputs.recheck_candidate.body | jq('((.[0].constraints // \"\")
                  | length) > 0 or ((.[0].infos // []) |
                  any(contains(\"replaced\")))') | first }}"
                then:
                  - id: prewrite_inventory
                    type: io.kestra.plugin.core.http.Request
                    description: Read runtime trigger state because flows.List exposes source
                      triggers but not the persisted disabled flag.
                    uri: "{{ fromJson(outputs.read_finding.value).kestra_url }}/api/v1/{{
                      fromJson(outputs.read_finding.value).tenant_id
                      }}/triggers/{{
                      fromJson(outputs.read_finding.value).namespace }}/{{
                      fromJson(outputs.read_finding.value).flow_id }}"
                    method: GET
                    options:
                      auth:
                        type: BASIC
                        username: "{{ inputs.kestra_user }}"
                        password: "{{ inputs.kestra_password }}"

                  - id: prewrite_gate
                    type: io.kestra.plugin.core.flow.If
                    description: Write only when a trigger is still enabled in the runtime state and
                      in the flow source.
                    condition: "{{ outputs.prewrite_inventory.body | jq('[.results[] |
                      select(.disabled != true and .sourceDisabled != true)] |
                      length > 0') | first }}"
                    then:
                      - id: save_original
                        type: io.kestra.plugin.core.kv.Set
                        description: Save the trigger ids that were enabled so the quarantine can be
                          reversed.
                        namespace: "{{ inputs.record_namespace }}"
                        key: "kestra-compat-{{ execution.id }}-disabled-{{
                          fromJson(outputs.read_finding.value).namespace |
                          replace({'.': '-'}) }}-{{
                          fromJson(outputs.read_finding.value).flow_id }}"
                        value: "{{ {'namespace': fromJson(outputs.read_finding.value).namespace,
                          'flow_id':
                          fromJson(outputs.read_finding.value).flow_id,
                          'enabled_before': (outputs.prewrite_inventory.body |
                          jq('[.results[] | select(.disabled != true and
                          .sourceDisabled != true) | .triggerId]') | first)} |
                          toJson }}"
                        kvType: STRING
                        overwrite: true

                      - id: disable_trigger
                        type: io.kestra.plugin.kestra.triggers.Toggle
                        kestraUrl: "{{ fromJson(outputs.read_finding.value).kestra_url }}"
                        tenantId: "{{ fromJson(outputs.read_finding.value).tenant_id }}"
                        namespace: "{{ fromJson(outputs.read_finding.value).namespace }}"
                        flowId: "{{ fromJson(outputs.read_finding.value).flow_id }}"
                        enabled: false
                        auth:
                          username: "{{ inputs.kestra_user }}"
                          password: "{{ inputs.kestra_password }}"

                      - id: settle_toggle
                        type: io.kestra.plugin.core.flow.Sleep
                        description: Allow Kestra's asynchronous trigger disable operation to settle
                          before verification.
                        duration: PT3S

                      - id: verify_inventory
                        type: io.kestra.plugin.core.http.Request
                        description: Re-read runtime trigger state after Toggle.
                        uri: "{{ fromJson(outputs.read_finding.value).kestra_url }}/api/v1/{{
                          fromJson(outputs.read_finding.value).tenant_id
                          }}/triggers/{{
                          fromJson(outputs.read_finding.value).namespace }}/{{
                          fromJson(outputs.read_finding.value).flow_id }}"
                        method: GET
                        options:
                          auth:
                            type: BASIC
                            username: "{{ inputs.kestra_user }}"
                            password: "{{ inputs.kestra_password }}"

                      - id: verify_gate
                        type: io.kestra.plugin.core.flow.If
                        condition: "{{ outputs.verify_inventory.body | jq('[.results[] |
                          select(.disabled != true and .sourceDisabled != true)]
                          | length == 0') | first }}"
                        then:
                          - id: quarantined
                            type: io.kestra.plugin.core.log.Log
                            message: "Disabled and verified every trigger of {{
                              fromJson(outputs.read_finding.value).namespace
                              }}/{{ fromJson(outputs.read_finding.value).flow_id
                              }}."
                        else:
                          - id: verify_failed
                            type: io.kestra.plugin.core.execution.Fail
                            errorMessage: "A trigger of {{ fromJson(outputs.read_finding.value).flow_id }}
                              did not remain disabled after Toggle."
                    else:
                      - id: stale_or_protected
                        type: io.kestra.plugin.core.log.Log
                        message: "Skipped {{ fromJson(outputs.read_finding.value).flow_id }} because
                          every trigger was already disabled or changed after
                          approval."
                else:
                  - id: fixed_during_pause
                    type: io.kestra.plugin.core.log.Log
                    message: "{{ fromJson(outputs.read_finding.value).namespace }}/{{
                      fromJson(outputs.read_finding.value).flow_id }} is no
                      longer invalid on the candidate after the approval. No
                      trigger was changed."
        else:
          - id: declined
            type: io.kestra.plugin.core.log.Log
            message: "Approval was not granted. No candidate-invalid trigger was changed."
    else:
      - id: safe_mode
        type: io.kestra.plugin.core.log.Log
        message: "Compatibility report complete. No trigger changes were permitted
          (auto_remediate={{ inputs.auto_remediate }}, dry_run={{ inputs.dry_run
          }}, candidate-invalid flows={{ outputs.finding_keys.keys | length
          }})."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Kestra upgrade compatibility gate failed for ' ~ inputs.namespace ~ ' against ' ~ inputs.candidate_url ~ ' (execution ' ~ execution.id ~ '). Review validation, quarantine and verification logs.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Disabled by default and forced to audit-only inputs.
    cron: "30 6 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true