Deployment request form with dependent inputs and KV-driven dropdowns

id: kv-driven-dynamic-deployment-request-form
namespace: company.team

description: |
  Deployment request form whose inputs appear step by step (`dependsOn`) and take their options from the KV store
  (`expression`). The flow validates the request again server-side, records it in KV, waits for an approval when the
  target is production or a service is not in the catalog, and posts the approved request to the deployer webhook.

labels:
  team: platform
  purpose: deployment-request

inputs:
  - id: action
    type: SELECT
    displayName: What do you want to do?
    description: "`request` submits a deployment request. `seed_catalog` writes the
      catalog below to the KV store (run it once before the first request)."
    isRadio: true
    defaults: request
    values:
      - label: Request a deployment
        value: request
      - label: Seed or replace the catalog in KV
        value: seed_catalog

  - id: environment
    type: SELECT
    displayName: Target environment
    description: Production requests always wait for an approval. Development
      requests are rolling deployments and skip the strategy question.
    isRadio: true
    defaults: dev
    values:
      - label: Development
        value: dev
      - label: Staging
        value: staging
      - label: Production
        value: production
    dependsOn:
      inputs: [ action ]
      condition: "{{ inputs.action == 'request' }}"

  - id: region
    type: SELECT
    displayName: Region
    description: Regions allowed for the chosen environment, read from the KV entry
      `deploy_form_regions` (one list per environment). Appears once an
      environment is picked.
    expression: "{{ kv(vars.regions_key)[inputs.environment] }}"
    dependsOn:
      inputs: [ action, environment ]
      condition: "{{ inputs.action == 'request' }}"

  - id: services
    type: MULTISELECT
    displayName: Services to deploy
    description: Pick one or more services from the KV catalog
      `deploy_form_services`. A service that is not listed yet can be typed in;
      the request is then routed to an approver whatever the environment.
    expression: "{{ kv(vars.services_key) }}"
    allowCustomValue: true
    dependsOn:
      inputs: [ action, region ]
      condition: "{{ inputs.action == 'request' }}"

  - id: version
    type: STRING
    displayName: Version tag
    description: Semantic version of the build, with an optional `v` prefix and an
      optional pre-release suffix (`v2.3.0-rc1`). The suggested value is a hint,
      not a default; replace it.
    prefill: v1.0.0
    validator: "^v?\\d+\\.\\d+\\.\\d+(-[0-9A-Za-z.]+)?$"
    dependsOn:
      inputs: [ action, services ]
      condition: "{{ inputs.action == 'request' }}"

  - id: strategy
    type: SELECT
    displayName: Rollout strategy
    description: Only asked for staging and production. Development always deploys
      with a rolling restart.
    isRadio: true
    defaults: rolling
    values: [ rolling, blue_green, canary ]
    dependsOn:
      inputs: [ action, environment ]
      condition: "{{ inputs.action == 'request' and inputs.environment != 'dev' }}"

  - id: canary_percent
    type: INT
    displayName: Canary traffic share (%)
    description: Share of traffic sent to the new version first. Only asked for a
      canary rollout; production is capped at 25 by a flow check.
    defaults: 10
    min: 1
    max: 50
    dependsOn:
      inputs: [ action, strategy ]
      condition: "{{ inputs.action == 'request' and inputs.strategy == 'canary' }}"

  - id: notify_channels
    type: ARRAY
    itemType: STRING
    displayName: Channels to notify
    description: Chat channels written into the request record for the deployer to
      notify when the rollout ends. A JSON array when submitted through the API.
    defaults: [ "#deployments" ]
    dependsOn:
      inputs: [ action ]
      condition: "{{ inputs.action == 'request' }}"

  - id: use_private_registry
    type: BOOL
    displayName: Images come from the private registry
    description: Staging and production builds may live in a private registry. When
      true, a pull token is asked for and forwarded to the deployer in the
      `Authorization` header.
    defaults: false
    dependsOn:
      inputs: [ action, environment ]
      condition: "{{ inputs.action == 'request' and inputs.environment != 'dev' }}"

  - id: registry_token
    type: SECRET
    displayName: Registry pull token
    description: Encrypted at rest and masked in the UI and the logs. Needs
      `kestra.encryption.secret-key` in the server configuration; without it the
      API answers 422 as soon as a value is given. Letters, digits, `_ . -`, at
      least 16 characters.
    validator: "^[A-Za-z0-9_.-]{16,}$"
    dependsOn:
      inputs: [ action, use_private_registry ]
      condition: "{{ inputs.action == 'request' and inputs.use_private_registry ==
        true }}"

  - id: catalog
    type: JSON
    displayName: Catalog to write to KV
    description: Only used by `seed_catalog`. `regions` maps every environment to
      its allowed regions; `services` is the list offered in the form. Replace
      the sample with your own, then run `seed_catalog` once.
    defaults: |
      {
        "regions": {
          "dev": ["local"],
          "staging": ["eu-west-1"],
          "production": ["eu-west-1", "us-east-1", "ap-southeast-1"]
        },
        "services": ["api", "web", "worker", "scheduler"]
      }
    dependsOn:
      inputs: [ action ]
      condition: "{{ inputs.action == 'seed_catalog' }}"

variables:
  regions_key: deploy_form_regions
  services_key: deploy_form_services
  record_prefix: deploy_request_
  record_ttl: P30D
  production_canary_max: 25
  deployer_webhook: https://httpbin.org/post

checks:
  - when: "{{ inputs.action != 'request' or ((inputs.services ?? []) | length) > 0
      }}"
    message: Pick at least one service. An empty selection passes the MULTISELECT
      validation, so it is refused here.
    style: ERROR
    behavior: BLOCK_EXECUTION
  - when: "{{ inputs.action != 'request' or inputs.environment != 'production' or
      (inputs.version | split('-') | length) == 1 }}"
    message: A pre-release version (with a `-suffix`) cannot be deployed to
      production. Pick a final tag or another environment.
    style: ERROR
    behavior: BLOCK_EXECUTION
  - when: "{{ inputs.action != 'request' or inputs.environment != 'production' or
      (inputs.strategy ?? 'rolling') != 'canary' or inputs.canary_percent <=
      vars.production_canary_max }}"
    message: A production canary cannot take more than 25% of the traffic. Lower
      `canary_percent` or use blue_green.
    style: ERROR
    behavior: BLOCK_EXECUTION
  - when: "{{ inputs.action != 'seed_catalog' or (inputs.catalog | jq('(.regions |
      type == \"object\" and length > 0) and (.services | type == \"array\" and
      length > 0) and ([.services[] | type == \"string\"] | all)') | first) }}"
    message: "`catalog` must be an object with a non-empty `regions` map
      (environment to list of regions) and a non-empty `services` list of
      strings."
    style: ERROR
    behavior: BLOCK_EXECUTION

concurrency:
  limit: 3
  behavior: QUEUE

sla:
  - id: max_runtime
    type: MAX_DURATION
    duration: PT5H
    behavior: FAIL
    labels:
      deploy_form_sla: breached

tasks:
  - id: seed
    type: io.kestra.plugin.core.flow.If
    description: Writes the two KV entries the form reads, as JSON structures so the
      input expressions can index them.
    condition: "{{ inputs.action == 'seed_catalog' }}"
    then:
      - id: seed_regions
        type: io.kestra.plugin.core.kv.Set
        key: "{{ vars.regions_key }}"
        kvType: JSON
        kvDescription: Allowed regions per environment, read by the `region` input of {{
          flow.id }}
        value: "{{ inputs.catalog.regions | toJson }}"
      - id: seed_services
        type: io.kestra.plugin.core.kv.Set
        key: "{{ vars.services_key }}"
        kvType: JSON
        kvDescription: Services offered by the `services` input of {{ flow.id }}
        value: "{{ inputs.catalog.services | toJson }}"
      - id: seed_done
        type: io.kestra.plugin.core.output.OutputValues
        values:
          status: catalog_seeded
          summary: "Catalog written: {{ inputs.catalog.regions | keys | join(', ') }} ({{
            inputs.catalog.services | length }} services)."
      - id: log_seed
        type: io.kestra.plugin.core.log.Log
        message: "{{ outputs.seed_done.values.summary }}"

  - id: request
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.action == 'request' }}"
    then:
      - id: load_services
        type: io.kestra.plugin.core.kv.Get
        description: The catalog the form offered. `region` needs no second check, the
          API refuses a value outside the rendered list (422) even when the
          caller never saw the form.
        key: "{{ vars.services_key }}"
        errorOnMissing: true

      - id: review
        type: io.kestra.plugin.core.output.OutputValues
        description: |
          Services typed in through `allowCustomValue` are not in the catalog; they make the request need an approval.
          Hidden inputs are rendered with their default (`strategy` is `rolling` for dev) or `null` (no default), hence the `??`.
        values:
          custom_services: "{% set custom = [] %}{% for s in inputs.services %}{% if not
            (outputs.load_services.value contains s) %}{% set custom = custom +
            [s] %}{% endif %}{% endfor %}{{ custom | toJson }}"
          requested_by: "{{ labels.system.username ?? labels.system.from ?? 'api' }}"

      - id: build_record
        type: io.kestra.plugin.core.output.OutputValues
        values:
          needs_approval: "{{ inputs.environment == 'production' or
            (fromJson(outputs.review.values.custom_services) | length) > 0 }}"
          record: |
            {{ {
              'request_id': execution.id,
              'requested_at': execution.startDate,
              'requested_by': outputs.review.values.requested_by,
              'environment': inputs.environment,
              'region': inputs.region,
              'services': inputs.services,
              'custom_services': fromJson(outputs.review.values.custom_services),
              'version': inputs.version,
              'strategy': inputs.strategy ?? 'rolling',
              'canary_percent': (inputs.strategy ?? 'rolling') == 'canary' ? inputs.canary_percent : null,
              'notify_channels': inputs.notify_channels,
              'private_registry': inputs.use_private_registry ?? false,
              'status': 'pending'
            } | toJson }}

      - id: label_request
        type: io.kestra.plugin.core.execution.Labels
        description: Filter the Executions page by environment, region or approval state.
        labels:
          deploy_environment: "{{ inputs.environment }}"
          deploy_region: "{{ inputs.region }}"
          deploy_version: "{{ inputs.version }}"
          deploy_approval: "{{ outputs.build_record.values.needs_approval == 'true' ?
            'pending' : 'auto' }}"

      - id: record_pending
        type: io.kestra.plugin.core.kv.Set
        description: One KV entry per request, expiring after 30 days. The status is
          updated as the request moves on.
        key: "{{ vars.record_prefix }}{{ execution.id }}"
        kvType: JSON
        ttl: "{{ vars.record_ttl }}"
        value: "{{ outputs.build_record.values.record }}"

      - id: approval
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.build_record.values.needs_approval == 'true' }}"
        then:
          - id: notify_approver
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Best effort. The approver can also find the paused execution in the
              UI; a missing webhook must not block the request.
            allowFailure: true
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: ":clipboard: Deployment request {{ execution.id }} by {{
              outputs.review.values.requested_by }}: {{ inputs.services |
              join(', ') }} {{ inputs.version }} to {{ inputs.environment }}/{{
              inputs.region }} ({{ inputs.strategy ?? 'rolling' }}){{
              (fromJson(outputs.review.values.custom_services) | length) > 0 ?
              '. Not in the catalog: ' ~
              (fromJson(outputs.review.values.custom_services) | join(', ')) :
              '' }}. Approve within 4 hours: {{ kestra.url ??
              'http://localhost:8080' }}/ui/main/executions/{{ flow.namespace
              }}/{{ flow.id }}/{{ execution.id }}"

          - id: wait_for_approval
            type: io.kestra.plugin.core.flow.Pause
            description: "Four hours without a decision fails the run (`behavior: FAIL`), so
              a forgotten request never reaches the deployer. Resume from the UI
              or with POST /api/v1/main/executions/{id}/actions/resume
              (multipart `approve`, `comment`)."
            pauseDuration: PT4H
            behavior: FAIL
            onResume:
              - id: approve
                type: BOOL
                displayName: Approve this deployment request?
                defaults: false
              - id: comment
                type: STRING
                displayName: Comment for the requester
                defaults: ""

          - id: decision
            type: io.kestra.plugin.core.flow.If
            condition: "{{ not (outputs.wait_for_approval.onResume.approve ?? false) }}"
            then:
              - id: record_rejected
                type: io.kestra.plugin.core.kv.Set
                key: "{{ vars.record_prefix }}{{ execution.id }}"
                kvType: JSON
                ttl: "{{ vars.record_ttl }}"
                value: "{{ fromJson(outputs.build_record.values.record) | merge({'status':
                  'rejected', 'decision_comment':
                  outputs.wait_for_approval.onResume.comment ?? ''}) | toJson
                  }}"
              - id: label_rejected
                type: io.kestra.plugin.core.execution.Labels
                labels:
                  deploy_approval: rejected
              - id: rejected
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "Deployment request rejected{{
                  (outputs.wait_for_approval.onResume.comment ?? '') != '' ? ':
                  ' ~ outputs.wait_for_approval.onResume.comment : '' }}.
                  Nothing was sent to the deployer."

      - id: submit
        type: io.kestra.plugin.core.http.Request
        description: |
          Hands the request to the deployer (a CI job, Argo CD, a GitHub `repository_dispatch`). The sample URL echoes the
          body back. Three attempts; a 4xx or 5xx after the last one fails the run and the `errors` branch marks the record.
        uri: "{{ vars.deployer_webhook }}"
        method: POST
        contentType: application/json
        headers:
          X-Deploy-Request-Id: "{{ execution.id }}"
          Authorization: "Bearer {{ inputs.registry_token ?? 'none' }}"
        body: "{{ fromJson(outputs.build_record.values.record) | merge({'status':
          'approved', 'approved_by': outputs.build_record.values.needs_approval
          == 'true' ? 'manual' : 'policy'}) | toJson }}"
        timeout: PT1M
        retry: { type: exponential, interval: PT5S, maxInterval: PT1M, maxAttempts: 3 }

      - id: record_submitted
        type: io.kestra.plugin.core.kv.Set
        key: "{{ vars.record_prefix }}{{ execution.id }}"
        kvType: JSON
        ttl: "{{ vars.record_ttl }}"
        value: "{{ fromJson(outputs.build_record.values.record) | merge({'status':
          'submitted', 'deployer_http_code': outputs.submit.code, 'approved_by':
          outputs.build_record.values.needs_approval == 'true' ? 'manual' :
          'policy', 'decision_comment':
          outputs.wait_for_approval.onResume.comment ?? ''}) | toJson }}"

      - id: label_submitted
        type: io.kestra.plugin.core.execution.Labels
        labels:
          deploy_approval: "{{ outputs.build_record.values.needs_approval == 'true' ?
            'approved' : 'auto' }}"

      - id: finish
        type: io.kestra.plugin.core.output.OutputValues
        values:
          status: submitted
          summary: "Request {{ execution.id }} submitted: {{ inputs.services | join(', ')
            }} {{ inputs.version }} to {{ inputs.environment }}/{{ inputs.region
            }}, deployer answered HTTP {{ outputs.submit.code }}."

      - id: log_finish
        type: io.kestra.plugin.core.log.Log
        message: "{{ outputs.finish.values.summary }}"

errors:
  - id: failure_details
    type: io.kestra.plugin.core.output.OutputValues
    description: errorLogs() also returns lines logged by this branch, so it is read
      once, before anything is logged.
    values:
      approval_expired: "{{ outputs.wait_for_approval is defined and
        outputs.wait_for_approval.onResume is not defined }}"
      task: "{{ ((errorLogs() | length) > 0 and (errorLogs()[0].taskId ?? '') != '') ?
        errorLogs()[0].taskId : (outputs.wait_for_approval is defined and
        outputs.wait_for_approval.onResume is not defined ? 'wait_for_approval'
        : 'none') }}"
      message: "{{ (errorLogs() | length) > 0 ? errorLogs()[0].message :
        (outputs.wait_for_approval is defined and
        outputs.wait_for_approval.onResume is not defined ? 'nobody approved or
        rejected the request within 4 hours, nothing was sent to the deployer' :
        'the execution failed without a task error log') }}"

  - id: label_expired
    type: io.kestra.plugin.core.execution.Labels
    description: A Pause that expires leaves no error log and no onResume values;
      the label tells an expired request apart from a technical failure.
    runIf: "{{ outputs.failure_details.values.approval_expired == 'true' }}"
    labels:
      deploy_approval: expired

  - id: log_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: "Deployment request {{ execution.id }} failed in {{
      outputs.failure_details.values.task }}: {{
      outputs.failure_details.values.message }}"

  - id: record_failed
    type: io.kestra.plugin.core.kv.Set
    description: The request record (or a minimal one when the failure came before
      it was built) keeps the reason for 30 days. A rejection already wrote its
      record and is reported by `afterExecution`, not as a failure.
    runIf: "{{ outputs.record_rejected is not defined }}"
    key: "{{ vars.record_prefix }}{{ execution.id }}"
    kvType: JSON
    ttl: "{{ vars.record_ttl }}"
    value: "{{ (outputs.build_record.values.record is defined ?
      fromJson(outputs.build_record.values.record) : {'request_id':
      execution.id, 'action': inputs.action, 'environment': inputs.environment
      ?? null, 'services': inputs.services ?? []}) | merge({'status':
      outputs.failure_details.values.approval_expired == 'true' ? 'expired' :
      'failed', 'failed_task': outputs.failure_details.values.task, 'error':
      outputs.failure_details.values.message}) | toJson }}"

  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    runIf: "{{ outputs.record_rejected is not defined }}"
    allowFailure: true
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: ":rotating_light: Deployment request {{ execution.id }} ({{
      inputs.action }}, {{ inputs.environment ?? 'n/a' }}) failed in `{{
      outputs.failure_details.values.task }}`: {{
      outputs.failure_details.values.message }} {{ kestra.url ??
      'http://localhost:8080' }}/ui/main/executions/{{ flow.namespace }}/{{
      flow.id }}/{{ execution.id }}"

finally:
  - id: log_record_key
    type: io.kestra.plugin.core.log.Log
    description: Runs after a success, a rejection or a failure, so the log always
      says where the request record is.
    message: "{{ inputs.action == 'request' ? 'Request record: KV key ' ~
      vars.record_prefix ~ execution.id ~ ' in ' ~ flow.namespace ~ ' (expires
      after ' ~ vars.record_ttl ~ ').' : 'Catalog keys: ' ~ vars.regions_key ~ '
      and ' ~ vars.services_key ~ ' in ' ~ flow.namespace ~ '.' }}"

afterExecution:
  - id: notify_outcome
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Reports a submitted or a rejected request. Failures were already
      reported by `errors`, and an SLA breach kills this task along with the
      run, so it is not handled here.
    runIf: "{{ inputs.action == 'request' and (execution.state != 'FAILED' or
      (labels.deploy_approval ?? '') == 'rejected') }}"
    allowFailure: true
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: "Deployment request {{ execution.id }} ended {{ execution.state
      }}{{ (labels.deploy_approval ?? '') == 'rejected' ? ' (rejected by the
      approver: ' ~ (outputs.wait_for_approval.onResume.comment ?? 'no comment')
      ~ ')' : '' }}: {{ inputs.services | join(', ') }} {{ inputs.version }} to
      {{ inputs.environment }}/{{ inputs.region }}. {{ kestra.url ??
      'http://localhost:8080' }}/ui/main/executions/{{ flow.namespace }}/{{
      flow.id }}/{{ execution.id }}"

outputs:
  - id: request_id
    type: STRING
    value: "{{ execution.id }}"
  - id: status
    type: STRING
    value: "{{ outputs.finish.values.status ?? outputs.seed_done.values.status ??
      (outputs.record_rejected is defined ? 'rejected' :
      (outputs.failure_details is defined ?
      (outputs.failure_details.values.approval_expired == 'true' ? 'expired' :
      'failed') : 'not_submitted')) }}"
  - id: record
    type: JSON
    description: The request as built, with the final status; the KV record holds
      the same plus the deployer's answer or the error.
    value: "{{ outputs.build_record.values.record is defined ?
      (fromJson(outputs.build_record.values.record) | merge({'status':
      outputs.finish.values.status ?? (outputs.record_rejected is defined ?
      'rejected' : (outputs.failure_details is defined ?
      (outputs.failure_details.values.approval_expired == 'true' ? 'expired' :
      'failed') : 'not_submitted'))})) : {} }}"