Databricks native dbt build with AllowFailure quarantine and Slack alerts

id: databricks-native-dbt-allow-failure
namespace: company.team
description: |
  Lands a CSV in a Databricks Unity Catalog Volume, runs it through a native dbt build
  job with a Write-Audit-Publish quality gate, and alerts Slack if any staging test fails.

inputs:
  - id: csv_content
    type: STRING
    description: Raw CSV text to upload.
    defaults: |
      order_id,customer_id,order_date,status,amount
      1,101,2026-10-01,delivered,250.00
      2,102,2026-10-02,shipped,180.50
      3,103,2026-10-03,pending,99.99
  - id: csv_file_name
    type: STRING
    description: Destination filename.
    defaults: new_orders.csv
  - id: volume_path
    type: STRING
    description: Full Unity Catalog volume path.
    defaults: /Volumes/workspace/staging/raw_landing_volume
  - id: dbt_catalog
    type: STRING
    description: Target catalog for dbt assets.
    defaults: workspace
  - id: dbt_schema
    type: STRING
    description: Target schema for test records.
    defaults: staging
  - id: dbt_job_id
    type: STRING
    description: Databricks Job ID.
    defaults: "000000000000000"

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "0 * * * *"

tasks:
  - id: upload_to_uc_volume
    type: io.kestra.plugin.core.http.Request
    uri: "https://{{ secret('DATABRICKS_HOST') }}/api/2.0/fs/files{{
      inputs.volume_path }}/{{ inputs.csv_file_name }}?overwrite=true"
    method: PUT
    contentType: text/csv
    headers:
      Authorization: "Bearer {{ secret('DATABRICKS_TOKEN') }}"
    body: "{{ inputs.csv_content }}"

  - id: allow_dbt_failure_trap
    type: io.kestra.plugin.core.flow.AllowFailure
    tasks:
      - id: run_governed_dbt_build
        type: io.kestra.plugin.databricks.job.SubmitRun
        host: "{{ secret('DATABRICKS_HOST') }}"
        authentication:
          token: "{{ secret('DATABRICKS_TOKEN') }}"
        waitForCompletion: PT30M
        runTasks:
          - taskKey: invoke_dbt_job
            runJobTask:
              jobId: "{{ inputs.dbt_job_id }}"

  - id: fetch_dbt_test_telemetry
    type: io.kestra.plugin.databricks.sql.Query
    accessToken: "{{ secret('DATABRICKS_TOKEN') }}"
    host: "{{ secret('DATABRICKS_HOST') }}"
    httpPath: "{{ secret('DATABRICKS_HTTP_PATH') }}"
    sql: |
      WITH latest AS (
        SELECT invocation_id
        FROM {{ inputs.dbt_catalog }}.{{ inputs.dbt_schema }}.dbt_test_results
        WHERE run_started_at >= TIMESTAMP'{{ execution.startDate | date(format="yyyy-MM-dd HH:mm:ss", timeZone="UTC") }}'
        ORDER BY run_started_at DESC
        LIMIT 1
      )
      SELECT
        COUNT(*) AS total_tests,
        COUNT_IF(status IN ('fail','error')) AS failure_count,
        COALESCE(
          ARRAY_JOIN(COLLECT_LIST(CASE WHEN status IN ('fail','error') THEN test_name END), '; '),
          'none'
        ) AS failed_tests
      FROM {{ inputs.dbt_catalog }}.{{ inputs.dbt_schema }}.dbt_test_results
      WHERE invocation_id IN (SELECT invocation_id FROM latest);

  - id: results_to_json
    type: io.kestra.plugin.serdes.json.IonToJson
    from: "{{ outputs.fetch_dbt_test_telemetry.uri }}"
    newLine: false

  - id: evaluate_test_results
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(read(outputs.results_to_json.uri)).failure_count > 0 }}"
    then:
      - id: send_quarantine_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":rotating_light: *dbt quarantine triggered*\n*Failed tests:* {{ fromJson(read(outputs.results_to_json.uri)).failed_tests }}\n*Impact:* this run did NOT update production tables - upstream models were skipped by dbt. Delta tables are safe."
          }
    else:
      - id: check_zero_tests
        type: io.kestra.plugin.core.flow.If
        condition: "{{ fromJson(read(outputs.results_to_json.uri)).total_tests == 0 }}"
        then:
          - id: warn_missing_telemetry
            type: io.kestra.plugin.core.log.Log
            message: "No test rows found for this execution - check that the on-run-end hook
              is deployed in the dbt project."
        else:
          - id: log_success
            type: io.kestra.plugin.core.log.Log
            message: "All {{ fromJson(read(outputs.results_to_json.uri)).total_tests }} dbt
              tests passed. Production tables updated."

outputs:
  - id: total_tests
    type: INT
    value: "{{ fromJson(read(outputs.results_to_json.uri)).total_tests }}"