Flink SQL Pipeline Deploy Gate

id: flink-sql-pipeline-deploy-gate
namespace: company.team

description: |
  Compile check a Flink SQL streaming statement against the live catalog on the Flink SQL Gateway, then
  in remediate mode deploy it and read the Flink REST API back to confirm the resulting job is running.
  The dry run only runs EXPLAIN, so a renamed column, a missing table or an incompatible catalog object
  is caught before anything is deployed.

inputs:
  - id: gateway_url
    type: STRING
    displayName: Flink SQL Gateway URL
    description: Base URL of the Flink SQL Gateway, reachable from Kestra, for
      example http://flink-sql-gateway:8083.
    defaults: "http://flink-sql-gateway:8083"

  - id: rest_url
    type: STRING
    displayName: Flink JobManager REST URL
    description: Base URL of the Flink JobManager REST API, used to confirm the
      deployed job is running, for example http://flink-jobmanager:8081.
    defaults: "http://flink-jobmanager:8081"

  - id: statement
    type: STRING
    displayName: Streaming SQL statement
    description: The INSERT statement that launches the streaming job. Use fully
      qualified names (catalog.database.table) and no trailing semicolon.
    defaults: "INSERT INTO sink_table SELECT col_a, col_b FROM source_table"

  - id: runtime_mode
    type: SELECT
    displayName: Execution runtime mode
    description: streaming keeps the deployed job running. batch runs it as a
      bounded job that finishes.
    values:
      - streaming
      - batch
    defaults: streaming

  - id: job_name_contains
    type: STRING
    displayName: Deployed job name filter
    description: Optional substring the deployed job name must contain in the REST
      re-verify. Empty accepts any newly running job.
    defaults: ""

  - id: settle_seconds
    type: INT
    displayName: Settle seconds
    description: Seconds to wait after the deploy before reading the Flink REST API,
      so the job reaches a steady state.
    defaults: 8

  - id: statement_timeout
    type: INT
    displayName: Statement timeout seconds
    description: How long the SQL Gateway waits for the EXPLAIN or the INSERT to
      reach a terminal state.
    defaults: 120

  - id: auto_remediate
    type: BOOL
    displayName: Allow the deploy
    description: Must be true, with dry_run false and an approval, before the
      statement is deployed.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: When true the flow only runs EXPLAIN to report whether the
      statement compiles. Nothing is deployed.
    defaults: true

tasks:
  - id: detect
    type: io.kestra.plugin.core.http.Request
    description: Read the current jobs from the Flink REST API as the baseline
      before any deploy.
    uri: "{{ inputs.rest_url }}/jobs/overview"
    method: GET

  - id: baseline
    type: io.kestra.plugin.core.output.OutputValues
    description: Keep the ids and the count of jobs that are already running, so a
      new one can be told apart later.
    values:
      r: >-
        {{ {'jobs': fromJson(outputs.detect.body).jobs} | jq('
           { running_ids: [ .jobs[] | select(.state == "RUNNING") | .jid ],
             running_count: ([ .jobs[] | select(.state == "RUNNING") ] | length) }') | first | toJson }}

  - id: findings_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Name the statement the gate is about to compile, because a deploy
      changes a live cluster.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Flink SQL deploy gate on ' ~ inputs.gateway_url ~ '. ' ~ fromJson(outputs.baseline.values.r).running_count ~ ' jobs running now. dry_run ' ~ inputs.dry_run ~ ', auto_remediate ' ~ inputs.auto_remediate ~ '. Statement: ' ~ inputs.statement} | toJson }}

  - id: compat_gate
    type: io.kestra.plugin.flink.SubmitSql
    description: Compile the statement with EXPLAIN against the live catalog. This
      reads the catalog and deploys nothing.
    gatewayUrl: "{{ inputs.gateway_url }}"
    statement: "EXPLAIN PLAN FOR {{ inputs.statement }}"
    statementTimeout: "{{ inputs.statement_timeout }}"
    acceptableStates:
      - FINISHED
      - ERROR
    sessionConfig:
      configuration:
        execution.runtime-mode: "{{ inputs.runtime_mode }}"

  - id: gate_eval
    type: io.kestra.plugin.core.output.OutputValues
    description: A FINISHED EXPLAIN means the statement compiles. An ERROR means it
      does not.
    values:
      compatible: "{{ outputs.compat_gate.status == 'FINISHED' }}"
      status: "{{ outputs.compat_gate.status }}"

  - id: gate_log
    type: io.kestra.plugin.core.log.Log
    description: Report the compile result and whether a deploy can follow.
    message: >-
      Flink SQL compatibility gate on {{ inputs.gateway_url }}: EXPLAIN status
      {{ outputs.compat_gate.status }}, compatible {{
      outputs.gate_eval.values.compatible }}. dry_run {{ inputs.dry_run }},
      auto_remediate {{ inputs.auto_remediate }}.

  - id: incompatible_guard
    type: io.kestra.plugin.core.flow.If
    description: Stop the run when the statement does not compile, so an
      incompatible change can never deploy.
    condition: "{{ outputs.compat_gate.status != 'FINISHED' }}"
    then:
      - id: incompatible_fail
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: >-
          Flink SQL statement did not compile against the catalog on {{
          inputs.gateway_url }}. EXPLAIN status {{ outputs.compat_gate.status
          }}. Nothing was deployed. Fix the statement or the catalog and run
          again.

  - id: apply_gate
    type: io.kestra.plugin.core.flow.If
    description: A deploy runs only with auto_remediate true and dry_run false.
      Otherwise the run stops after the compile check.
    condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
    then:
      - id: approve_deploy
        type: io.kestra.plugin.core.flow.Pause
        description: Deploying a streaming job starts a long running job that writes to
          a sink, so a human approves it.
        pauseDuration: PT2H
        behavior: FAIL
        onResume:
          - id: approved
            type: BOOL
            defaults: false
            description: Set true to submit the statement and start the streaming job.
          - id: reason
            type: STRING
            defaults: "planned deploy"
            description: Recorded in the execution so the decision is auditable.

      - id: deploy
        type: io.kestra.plugin.core.flow.If
        condition: "{{ (outputs.approve_deploy.onResume is defined) and
          outputs.approve_deploy.onResume.approved }}"
        then:
          - id: submit
            type: io.kestra.plugin.flink.SubmitSql
            description: Submit the statement to the SQL Gateway, which starts the job on
              the cluster.
            gatewayUrl: "{{ inputs.gateway_url }}"
            statement: "{{ inputs.statement }}"
            statementTimeout: "{{ inputs.statement_timeout }}"
            acceptableStates:
              - RUNNING
              - FINISHED
            sessionConfig:
              configuration:
                execution.runtime-mode: "{{ inputs.runtime_mode }}"

          - id: settle
            type: io.kestra.plugin.core.flow.Sleep
            duration: "PT{{ inputs.settle_seconds }}S"

          - id: reverify
            type: io.kestra.plugin.core.http.Request
            description: Read the Flink REST API again to confirm the deployed job is
              running.
            uri: "{{ inputs.rest_url }}/jobs/overview"
            method: GET

          - id: verdict
            type: io.kestra.plugin.core.output.OutputValues
            description: The deploy counts as landed only when a job that was not running
              before is running now.
            values:
              v: >-
                {{ {'after': fromJson(outputs.reverify.body).jobs, 'base':
                fromJson(outputs.baseline.values.r).running_ids, 'st':
                outputs.submit.status, 'nc': inputs.job_name_contains} | jq('
                   .base as $base | .st as $st | .nc as $nc
                   | [ .after[] | select(.state == "RUNNING") ] as $run
                   | [ $run[] | . as $r | select(($base | index($r.jid)) == null) ] as $new
                   | (if $nc == "" then $new else [ $run[] | select(.name | contains($nc)) ] end) as $match
                   | { submit_status: $st, running_after: ($run | length), new_running: ($new | length),
                       new_ids: [ $new[] | .jid ], matched: ($match | length),
                       ok: (($st == "RUNNING" or $st == "FINISHED") and ($new | length > 0) and ($match | length > 0)) }') | first | toJson }}

          - id: verify
            type: io.kestra.plugin.core.flow.If
            description: Fail the run unless the deployed statement left a job running on
              the cluster.
            condition: "{{ not fromJson(outputs.verdict.values.v).ok }}"
            then:
              - id: deploy_not_running
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: >-
                  Flink SQL deploy did not reach a running job on {{
                  inputs.rest_url }}. Submit status {{
                  fromJson(outputs.verdict.values.v).submit_status }}, newly
                  running jobs {{ fromJson(outputs.verdict.values.v).new_running
                  }}, matched {{ fromJson(outputs.verdict.values.v).matched }}.
                  The statement compiled but the job is not running, so the
                  pipeline is not deployed.
            else:
              - id: deployed_alert
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                payload: |
                  {{ {'text': 'Flink SQL pipeline deployed on ' ~ inputs.rest_url ~ '. New running job(s): ' ~ fromJson(outputs.verdict.values.v).new_ids ~ '. Submit status ' ~ fromJson(outputs.verdict.values.v).submit_status ~ '. Approved with reason "' ~ outputs.approve_deploy.onResume.reason ~ '".'} | toJson }}

              - id: deployed_log
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Flink SQL pipeline deployed and verified on {{ inputs.rest_url
                  }}. Submit status {{
                  fromJson(outputs.verdict.values.v).submit_status }}, newly
                  running jobs {{ fromJson(outputs.verdict.values.v).new_ids }}
                  (running total {{
                  fromJson(outputs.verdict.values.v).running_after }}).
        else:
          - id: not_approved
            type: io.kestra.plugin.core.log.Log
            message: "Deploy on {{ inputs.gateway_url }} was not approved, so the statement
              was checked only and not deployed."
    else:
      - id: compile_only_log
        type: io.kestra.plugin.core.log.Log
        message: >-
          Statement compiles on {{ inputs.gateway_url }} and was not deployed.
          dry_run {{ inputs.dry_run }}, auto_remediate {{ inputs.auto_remediate
          }}. Set auto_remediate true and dry_run false, then approve the paused
          execution, to deploy it.

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Flink SQL deploy gate failed on ' ~ inputs.gateway_url ~ ' (execution ' ~ execution.id ~ '). The statement was not deployed or did not verify, so check it before deploying again.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and in safe mode. Enable it once the inputs match
      your cluster.
    cron: "15 3 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false