Detect dbt Model Runtime Regressions and Alert Slack

id: dbt-model-runtime-regression-alert
namespace: company.team
description: |
  Run dbt, keep a history of how long every model takes, and alert Slack when a model
  becomes much slower than its recent median. A slow model does not fail, so nothing
  else in the pipeline catches it.

inputs:
  - id: threshold_pct
    type: INT
    displayName: Regression threshold (%)
    description: Alert when a model runs this much slower than its baseline median.
    defaults: 50

  - id: baseline_runs
    type: INT
    displayName: Baseline runs
    description: Number of previous successful runs per model that make up the baseline.
    defaults: 7

  - id: min_duration_seconds
    type: FLOAT
    displayName: Minimum duration (seconds)
    description: Ignore models faster than this in the current run, so sub-second
      noise does not alert.
    defaults: 5.0

  - id: history_db
    type: STRING
    displayName: History database
    description: DuckDB database that stores the run history. Use md:<database> to
      keep it in MotherDuck.
    defaults: /app/storage/dbt_runtime_history.duckdb

tasks:
  - id: dbt
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
      - id: clone_repository
        type: io.kestra.plugin.git.Clone
        description: Check out the dbt project. Point this at your own repository.
        url: https://github.com/kestra-io/dbt-demo
        branch: main

      - id: dbt_build
        type: io.kestra.plugin.dbt.cli.DbtCLI
        description: Build the project. The task uploads target/run_results.json as an
          output file.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: ghcr.io/kestra-io/dbt-duckdb:latest
        profiles: |
          jaffle_shop:
            outputs:
              dev:
                type: duckdb
                path: ":memory:"
                threads: 4
            target: dev
        commands:
          - dbt build

  - id: record_run
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Append the execution time of every successful model in this run to
      the history table.
    url: "jdbc:duckdb:{{ inputs.history_db }}"
    inputFiles:
      run_results.json: "{{ outputs.dbt_build.outputFiles['run_results.json'] }}"
    sql: |
      CREATE TABLE IF NOT EXISTS dbt_model_runs (
        execution_id   VARCHAR,
        invocation_id  VARCHAR,
        recorded_at    TIMESTAMP,
        unique_id      VARCHAR,
        execution_time DOUBLE
      );

      INSERT INTO dbt_model_runs
      SELECT
        '{{ execution.id }}',
        metadata.invocation_id,
        now(),
        r.unique_id,
        r.execution_time
      FROM (
        SELECT metadata, unnest(results) AS r
        FROM read_json_auto('{{ workingDir }}/run_results.json', maximum_object_size = 268435456)
      )
      WHERE starts_with(r.unique_id, 'model.')
        AND r.status = 'success';

  - id: find_regressions
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Compare this run with the median of each model's previous runs and
      keep the models over the threshold.
    url: "jdbc:duckdb:{{ inputs.history_db }}"
    fetchType: FETCH
    sql: |
      WITH ranked AS (
        SELECT
          unique_id,
          execution_id,
          execution_time,
          row_number() OVER (PARTITION BY unique_id ORDER BY recorded_at DESC) AS rn
        FROM dbt_model_runs
      ),
      latest AS (
        SELECT unique_id, execution_time
        FROM ranked
        WHERE rn = 1 AND execution_id = '{{ execution.id }}'
      ),
      baseline AS (
        SELECT unique_id, median(execution_time) AS median_s, count(*) AS runs
        FROM ranked
        WHERE rn BETWEEN 2 AND {{ inputs.baseline_runs }} + 1
        GROUP BY unique_id
      )
      SELECT
        l.unique_id                                        AS model,
        round(b.median_s, 2)                               AS baseline_s,
        round(l.execution_time, 2)                         AS latest_s,
        CAST(round(100 * (l.execution_time / b.median_s - 1)) AS INTEGER) AS pct_slower
      FROM latest l
      JOIN baseline b USING (unique_id)
      WHERE b.runs >= 3
        AND l.execution_time >= {{ inputs.min_duration_seconds }}
        AND l.execution_time > b.median_s * (1 + {{ inputs.threshold_pct }} / 100.0)
      ORDER BY pct_slower DESC;

  - id: alert_on_regressions
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.find_regressions.size > 0 }}"
    then:
      - id: notify_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          dbt runtime regression in {{ flow.namespace }}.{{ flow.id }} (execution {{ execution.id }}).
          {{ outputs.find_regressions.size }} model(s) ran more than {{ inputs.threshold_pct }}% slower than their median over the last {{ inputs.baseline_runs }} runs:
          {% for row in outputs.find_regressions.rows %}- {{ row.model }}: {{ row.baseline_s }}s -> {{ row.latest_s }}s (+{{ row.pct_slower }}%)
          {% endfor %}
    else:
      - id: no_regressions
        type: io.kestra.plugin.core.log.Log
        message: No model ran more than {{ inputs.threshold_pct }}% slower than its
          baseline.

triggers:
  - id: nightly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run after the nightly load. Adjust or disable as needed.
    cron: "0 3 * * *"
    disabled: true

  - id: on_merge
    type: io.kestra.plugin.core.trigger.Webhook
    description: Call this webhook from CI after a merge to the main branch, so a
      slow model is caught on the change that introduced it. Replace the key
      with your own value.
    key: replace-with-a-random-key