Release Kestra Flows from Git with Smoke Tests and Automatic Rollback

id: kestra-gitops-deploy-smoke-test-rollback
namespace: company.platform

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: repository_url
    type: URI
    defaults: https://github.com/your-org/kestra-flows
    description: Git repository that holds the flow definitions to release.
  - id: branch
    type: STRING
    defaults: main
    description: Branch to release from.
  - id: git_directory
    type: STRING
    defaults: flows
    description: Directory of the repository that contains the flow YAML files.
  - id: target_namespace
    type: STRING
    defaults: company.payments
    description: Namespace the flows are deployed to. It must not be the namespace
      of this deployer flow.
  - id: smoke_test_flows
    type: ARRAY
    itemType: STRING
    defaults:
      - payments-smoke-test
    description: Flows in the target namespace that must end in SUCCESS after the
      release. They run one by one, after the deploy and again after a rollback.
  - id: delete_missing
    type: BOOL
    defaults: false
    description: Also delete flows of the target namespace that are not in Git. A
      release that deletes flows always waits for approval.
  - id: require_approval
    type: BOOL
    defaults: false
    description: Wait for a manual approval before every release, not only releases
      that delete flows.
  - id: kestra_url
    type: URI
    defaults: http://localhost:8080
    description: Base URL of the Kestra API that receives the release.

variables:
  tenant: main

tasks:
  - id: check_target
    type: io.kestra.plugin.core.execution.Assert
    description: Refuse to release into this flow's own namespace or into one of its
      parents, where a sync could change or delete the deployer itself.
    conditions:
      - "{{ inputs.target_namespace != flow.namespace and not (flow.namespace |
        startsWith(inputs.target_namespace ~ '.')) }}"
    errorMessage: "target_namespace {{ inputs.target_namespace }} must not contain
      the deployer flow ({{ flow.namespace }}.{{ flow.id }})"

  - id: snapshot_ids
    type: io.kestra.plugin.kestra.flows.List
    description: Records which flows exist before the release, so a rollback can
      remove flows the release added.
    kestraUrl: "{{ inputs.kestra_url }}"
    tenantId: "{{ vars.tenant }}"
    namespace: "{{ inputs.target_namespace }}"
    auth:
      username: "{{ secret('KESTRA_USERNAME') }}"
      password: "{{ secret('KESTRA_PASSWORD') }}"

  - id: snapshot
    type: io.kestra.plugin.kestra.flows.Export
    description: Exports the current flow sources of the target namespace as the
      rollback point.
    kestraUrl: "{{ inputs.kestra_url }}"
    tenantId: "{{ vars.tenant }}"
    namespace: "{{ inputs.target_namespace }}"
    auth:
      username: "{{ secret('KESTRA_USERNAME') }}"
      password: "{{ secret('KESTRA_PASSWORD') }}"

  - id: preview
    type: io.kestra.plugin.git.SyncFlows
    description: Dry run. Clones the branch and writes the per-flow diff without
      changing Kestra.
    url: "{{ inputs.repository_url }}"
    branch: "{{ inputs.branch }}"
    gitDirectory: "{{ inputs.git_directory }}"
    username: "{{ secret('GIT_USERNAME') }}"
    password: "{{ secret('GIT_ACCESS_TOKEN') }}"
    targetNamespace: "{{ inputs.target_namespace }}"
    delete: "{{ inputs.delete_missing }}"
    dryRun: true
    kestraUrl: "{{ inputs.kestra_url }}"
    auth:
      username: "{{ secret('KESTRA_USERNAME') }}"
      password: "{{ secret('KESTRA_PASSWORD') }}"
    timeout: PT5M
    retry:
      type: constant
      interval: PT30S
      maxAttempts: 3

  - id: plan
    type: io.kestra.plugin.core.output.OutputValues
    description: Counts the changes in the dry-run diff, one line per flow.
    values:
      added: "{{ (read(outputs.preview.flows) | split('\"ADDED\"') | length) - 1 }}"
      updated: "{{ (read(outputs.preview.flows) | split('\"UPDATED\"') | length) - 1 }}"
      deleted: "{{ (read(outputs.preview.flows) | split('\"DELETED\"') | length) - 1 }}"

  - id: release
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.plan.values.added | number) +
      (outputs.plan.values.updated | number) + (outputs.plan.values.deleted |
      number) > 0 }}"
    then:
      - id: approval
        type: io.kestra.plugin.core.flow.Pause
        description: Waits up to 4 hours for an operator to resume the execution.
          Without approval the release is cancelled and nothing is deployed.
        runIf: "{{ inputs.require_approval or (outputs.plan.values.deleted | number) > 0
          }}"
        pauseDuration: PT4H
        behavior: CANCEL

      - id: deploy_and_verify
        type: io.kestra.plugin.core.flow.Sequential
        tasks:
          - id: deploy
            type: io.kestra.plugin.git.SyncFlows
            url: "{{ inputs.repository_url }}"
            branch: "{{ inputs.branch }}"
            gitDirectory: "{{ inputs.git_directory }}"
            username: "{{ secret('GIT_USERNAME') }}"
            password: "{{ secret('GIT_ACCESS_TOKEN') }}"
            targetNamespace: "{{ inputs.target_namespace }}"
            delete: "{{ inputs.delete_missing }}"
            dryRun: false
            kestraUrl: "{{ inputs.kestra_url }}"
            auth:
              username: "{{ secret('KESTRA_USERNAME') }}"
              password: "{{ secret('KESTRA_PASSWORD') }}"
            timeout: PT5M

          - id: smoke_tests
            type: io.kestra.plugin.core.flow.Loop
            values: "{{ inputs.smoke_test_flows }}"
            tasks:
              - id: smoke_test
                type: io.kestra.plugin.core.flow.Subflow
                namespace: "{{ inputs.target_namespace }}"
                flowId: "{{ item.value }}"
                wait: true
                transmitFailed: true
        errors:
          - id: restore_snapshot
            type: io.kestra.plugin.core.http.Request
            description: Re-imports the pre-release sources. Import is idempotent, so the
              bounded retry is safe.
            runIf: "{{ outputs.snapshot_ids.flows | length > 0 }}"
            uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant
              }}/flows/import?failOnError=true"
            method: POST
            contentType: multipart/form-data
            formData:
              fileUpload:
                name: rollback.zip
                content: "{{ outputs.snapshot.flowsZip }}"
            options:
              auth:
                type: BASIC
                username: "{{ secret('KESTRA_USERNAME') }}"
                password: "{{ secret('KESTRA_PASSWORD') }}"
            retry:
              type: constant
              interval: PT10S
              maxAttempts: 3

          - id: flows_after_release
            type: io.kestra.plugin.kestra.flows.List
            kestraUrl: "{{ inputs.kestra_url }}"
            tenantId: "{{ vars.tenant }}"
            namespace: "{{ inputs.target_namespace }}"
            auth:
              username: "{{ secret('KESTRA_USERNAME') }}"
              password: "{{ secret('KESTRA_PASSWORD') }}"

          - id: rollback_plan
            type: io.kestra.plugin.core.output.OutputValues
            description: Flows the release added, and the smoke tests that already existed
              before it. Computed here because loop iterations run in an
              isolated context.
            values:
              added_flows: "{{ {'after': outputs.flows_after_release.flows, 'before':
                outputs.snapshot_ids.flows} | jq('[.after[].id] -
                [.before[].id]') | first | toJson }}"
              existing_smoke_tests: "{{ {'tests': inputs.smoke_test_flows, 'before':
                outputs.snapshot_ids.flows} | jq('[.before[].id] as $ids |
                [.tests[] | select(. as $t | $ids | any(. == $t))]') | first |
                toJson }}"

          - id: remove_added_flows
            type: io.kestra.plugin.core.flow.Loop
            description: Deletes only the flows that did not exist before the release.
            runIf: "{{ outputs.rollback_plan.values.added_flows != '[]' }}"
            values: "{{ outputs.rollback_plan.values.added_flows }}"
            tasks:
              - id: remove_added_flow
                type: io.kestra.plugin.core.http.Request
                uri: "{{ inputs.kestra_url }}/api/v1/{{ vars.tenant }}/flows/{{
                  inputs.target_namespace }}/{{ item.value }}"
                method: DELETE
                options:
                  auth:
                    type: BASIC
                    username: "{{ secret('KESTRA_USERNAME') }}"
                    password: "{{ secret('KESTRA_PASSWORD') }}"

          - id: flows_after_rollback
            type: io.kestra.plugin.kestra.flows.List
            kestraUrl: "{{ inputs.kestra_url }}"
            tenantId: "{{ vars.tenant }}"
            namespace: "{{ inputs.target_namespace }}"
            auth:
              username: "{{ secret('KESTRA_USERNAME') }}"
              password: "{{ secret('KESTRA_PASSWORD') }}"

          - id: check_rollback
            type: io.kestra.plugin.core.execution.Assert
            description: The namespace must contain exactly the flows it had before the
              release.
            conditions:
              - "{{ (outputs.flows_after_rollback.flows | jq('[.[].id] | sort')
                | first) == (outputs.snapshot_ids.flows | jq('[.[].id] | sort')
                | first) }}"
            errorMessage: "Rollback incomplete: {{ outputs.flows_after_rollback.flows |
              jq('[.[].id] | sort') | first }} differs from the snapshot {{
              outputs.snapshot_ids.flows | jq('[.[].id] | sort') | first }}"

          - id: smoke_tests_after_rollback
            type: io.kestra.plugin.core.flow.Loop
            description: Proves the restored release works. Smoke-test flows that did not
              exist before the release are skipped.
            runIf: "{{ outputs.rollback_plan.values.existing_smoke_tests != '[]' }}"
            values: "{{ outputs.rollback_plan.values.existing_smoke_tests }}"
            tasks:
              - id: smoke_test_after_rollback
                type: io.kestra.plugin.core.flow.Subflow
                namespace: "{{ inputs.target_namespace }}"
                flowId: "{{ item.value }}"
                wait: true
                transmitFailed: true

      - id: summary
        type: io.kestra.plugin.core.log.Log
        message: "Released {{ inputs.branch }} to {{ inputs.target_namespace }}: {{
          outputs.plan.values.added }} added, {{ outputs.plan.values.updated }}
          updated, {{ outputs.plan.values.deleted }} deleted, smoke tests
          passed."

    else:
      - id: nothing_to_release
        type: io.kestra.plugin.core.log.Log
        message: "{{ inputs.target_namespace }} already matches {{ inputs.branch }},
          nothing to release."

triggers:
  - id: release_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('DEPLOY_WEBHOOK_KEY') }}"