Automate dbt Anomaly Detection and Slack Triage with Elementary

id: dbt-elementary-data-observability-and-slack-triage
namespace: company.analytics

description: |
  Automated dbt observability pipeline that executes dbt models and Elementary anomaly detection tests,
  generates a standalone HTML health report, and triages schema and volume anomalies to Slack.

inputs:
  - id: git_repository
    type: URI
    displayName: Git Repository URL
    description: Git repository containing the dbt project and packages.yml with Elementary.
    defaults: https://github.com/elementary-data/dbt-project-demo

  - id: git_branch
    type: STRING
    displayName: Git Branch
    description: Target Git branch to check out and test.
    defaults: main

  - id: dbt_target
    type: STRING
    displayName: dbt Profile Target
    description: Target database profile declared in profiles.yml (e.g., dev, prod,
      staging).
    defaults: prod

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    description: Destination Slack channel for data quality anomaly triage alerts.
    defaults: "#data-ops-alerts"

triggers:
  - id: daily_morning_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Automated morning data observability audit scheduled before
      business dashboard refresh. Shipped disabled by default.
    cron: "0 6 * * 1-5"
    disabled: true

  - id: upstream_warehouse_sync
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook trigger to run observability tests following
      upstream Fivetran or Airbyte loads.
    key: "{{ secret('WEBHOOK_KEY') }}"

tasks:
  - id: dbt_and_elementary_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: Isolated shared workspace for cloning repository, compiling dbt,
      and running Elementary CLI.
    tasks:
      - id: clone_repository
        type: io.kestra.plugin.git.Clone
        description: Pull dbt project models, configurations, and packages from version
          control.
        url: "{{ inputs.git_repository }}"
        branch: "{{ inputs.git_branch }}"

      - id: run_dbt_and_elementary
        type: io.kestra.plugin.scripts.shell.Commands
        description: Execute dbt dependencies, run Elementary anomaly tests, and compile
          the observability report.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.11-slim
        env:
          DBT_PROFILES_DIR: "{{ workingDir }}"
          POSTGRES_HOST: "{{ secret('POSTGRES_HOST') }}"
          POSTGRES_PORT: "{{ secret('POSTGRES_PORT') ?? '5432' }}"
          POSTGRES_DATABASE: "{{ secret('POSTGRES_DB') }}"
          POSTGRES_USER: "{{ secret('POSTGRES_USER') }}"
          POSTGRES_PASSWORD: "{{ secret('POSTGRES_PASSWORD') }}"
          POSTGRES_SCHEMA: "{{ secret('POSTGRES_SCHEMA') ?? 'analytics' }}"
        commands:
          - pip install --no-cache-dir "dbt-postgres>=1.8.0,<1.9.0"
            "elementary-data>=0.14.0"
          - |
            cat << 'EOF' > profiles.yml
            defaults:
              target: "{{ inputs.dbt_target }}"
              outputs:
                prod:
                  type: postgres
                  host: "{{ secret('POSTGRES_HOST') }}"
                  port: {{ secret('POSTGRES_PORT') ?? 5432 }}
                  user: "{{ secret('POSTGRES_USER') }}"
                  password: "{{ secret('POSTGRES_PASSWORD') }}"
                  dbname: "{{ secret('POSTGRES_DB') }}"
                  schema: "{{ secret('POSTGRES_SCHEMA') ?? 'analytics' }}"
                  threads: 4
                dev:
                  type: postgres
                  host: "{{ secret('POSTGRES_HOST') }}"
                  port: {{ secret('POSTGRES_PORT') ?? 5432 }}
                  user: "{{ secret('POSTGRES_USER') }}"
                  password: "{{ secret('POSTGRES_PASSWORD') }}"
                  dbname: "{{ secret('POSTGRES_DB') }}"
                  schema: "{{ secret('POSTGRES_SCHEMA') ?? 'analytics' }}_dev"
                  threads: 2
            EOF
          - dbt deps
          - dbt test --select tag:elementary || true
          - edr report --file-path elementary_report.html || true

      - id: parse_elementary_results
        type: io.kestra.plugin.scripts.python.Script
        description: Inspect dbt run results and Elementary test metadata to extract
          anomaly counts and status.
        containerImage: python:3.11-slim
        script: |
          import json
          import os
          import sys

          run_results_path = os.path.join("target", "run_results.json")
          has_anomalies = False
          failed_tests = []
          total_tested = 0

          if os.path.exists(run_results_path):
              with open(run_results_path, "r", encoding="utf-8") as f:
                  data = json.load(f)
                  results = data.get("results", [])
                  total_tested = len(results)
                  for r in results:
                      status = r.get("status", "").lower()
                      if status in ["fail", "error"]:
                          test_name = r.get("unique_id", "unknown_test")
                          message = r.get("message", "No message")
                          failed_tests.append({"test": test_name, "message": message})
                          has_anomalies = True

          outputs = {
              "has_anomalies": has_anomalies,
              "anomaly_count": len(failed_tests),
              "total_tests": total_tested,
              "failed_tests": failed_tests,
              "summary": f"{len(failed_tests)} anomalies detected out of {total_tested} tests executed."
          }

          print(f"Observability check completed: {outputs['summary']}")
          print("::" + json.dumps({"outputs": outputs}) + "::")

  - id: evaluate_anomalies
    type: io.kestra.plugin.core.flow.If
    description: Branch execution based on whether Elementary detected data quality
      anomalies.
    condition: "{{ outputs.parse_elementary_results.vars.has_anomalies }}"
    then:
      - id: notify_slack_anomalies
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatch high-priority Slack Block Kit alert detailing failed dbt
          models and anomaly metrics.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "blocks": [
              {
                "type": "header",
                "text": {
                  "type": "plain_text",
                  "text": "🚨 dbt Data Observability Alert: Anomalies Detected"
                }
              },
              {
                "type": "section",
                "fields": [
                  {
                    "type": "mrkdwn",
                    "text": "*Execution ID:*\n<{{ (kestra.url ?? 'http://localhost:8080/') ~ 'ui/' ~ execution.namespace ~ '/executions/' ~ execution.id }}|{{ execution.id }}>"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Anomalies Found:*\n{{ outputs.parse_elementary_results.vars.anomaly_count }} of {{ outputs.parse_elementary_results.vars.total_tests }} tests"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Target Profile:*\n`{{ inputs.dbt_target }}`"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Repository:*\n`{{ inputs.git_repository }}`"
                  }
                ]
              },
              {
                "type": "section",
                "text": {
                  "type": "mrkdwn",
                  "text": "*Failed Anomaly Tests:*\n```\n{% for item in outputs.parse_elementary_results.vars.failed_tests %}- {{ item.test }}: {{ item.message }}\n{% endfor %}```"
                }
              },
              {
                "type": "context",
                "elements": [
                  {
                    "type": "mrkdwn",
                    "text": "Generated by *Kestra dbt Elementary Observability Pipeline* • Inspect `elementary_report.html` in execution outputs."
                  }
                ]
              }
            ]
          }
    else:
      - id: log_clean_run
        type: io.kestra.plugin.core.log.Log
        description: Record passing audit status when all data quality and volume checks
          pass.
        message: "✅ All dbt models and Elementary anomaly detection tests passed with
          zero failures."

errors:
  - id: alert_pipeline_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Critical alert dispatched if the runner crashes or database
      connectivity drops.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": "🔥 *Fatal Crash in dbt Elementary Pipeline* for execution `<{{ (kestra.url ?? 'http://localhost:8080/') ~ 'ui/' ~ execution.namespace ~ '/executions/' ~ execution.id }}|{{ execution.id }}>`. The runner aborted before completing validation."
      }