dbt Core Semantic Layer Metric Drift Gate

id: dbt-core-semantic-layer-metric-drift-gate
namespace: company.analytics

description: |
  Inspect dbt Core semantic manifests for breaking changes across metrics, entities,
  measures, and dimensions, evaluate numerical KPI drift against warehouse queries,
  persist audit telemetry in Kestra KV, and gate deployments to prevent downstream BI outages.

concurrency:
  limit: 1

inputs:
  - id: candidate_manifest
    type: STRING
    displayName: Candidate Semantic Manifest JSON
    description: Raw JSON content of semantic_manifest.json produced by dbt compile
      or dbt parse during CI/CD.
    defaults: |
      {
        "semantic_models": [
          {
            "name": "orders_source",
            "description": "Core orders transaction semantic model",
            "node_relation": {
              "relation_name": "analytics.fct_orders"
            },
            "entities": [
              {"name": "order_id", "type": "primary"},
              {"name": "customer_id", "type": "foreign"}
            ],
            "measures": [
              {"name": "order_total", "agg": "sum", "expr": "order_total_usd", "description": "Gross completed order transaction value in USD"},
              {"name": "order_count", "agg": "sum", "expr": "1", "description": "Total volume of completed customer orders"},
              {"name": "distinct_customers", "agg": "count_distinct", "expr": "customer_id", "description": "Count of transacting customer accounts"}
            ],
            "dimensions": [
              {"name": "order_date", "type": "time", "type_params": {"time_granularity": "day"}, "is_partition": true},
              {"name": "order_status", "type": "categorical"},
              {"name": "region", "type": "categorical"}
            ]
          }
        ],
        "metrics": [
          {
            "name": "revenue",
            "type": "simple",
            "type_params": {
              "measure": "order_total"
            },
            "description": "Total gross revenue generated by orders"
          },
          {
            "name": "daily_active_users",
            "type": "simple",
            "type_params": {
              "measure": "distinct_customers"
            },
            "description": "Daily active transacting users count"
          },
          {
            "name": "order_volume",
            "type": "simple",
            "type_params": {
              "measure": "order_count"
            },
            "description": "Total completed orders volume"
          },
          {
            "name": "average_order_value",
            "type": "ratio",
            "type_params": {
              "numerator": "revenue",
              "denominator": "order_volume"
            },
            "description": "Average transaction value calculated as revenue divided by order volume"
          }
        ]
      }

  - id: kv_baseline_key
    type: STRING
    displayName: Baseline Manifest KV Key
    defaults: dbt_semantic_manifest_prod
    description: Key in Kestra KV store holding the certified production semantic manifest.

  - id: kv_audit_report_key
    type: STRING
    displayName: Audit Report KV Key
    defaults: dbt_semantic_layer_audit_report
    description: Key in Kestra KV store where audit findings and KPI drift telemetry
      are persisted.

  - id: max_kpi_drift_pct
    type: FLOAT
    displayName: Max Allowed KPI Drift (%)
    defaults: 5.0
    description: Maximum allowable percentage drift between candidate warehouse
      verification metrics and baseline values before failing the gate.

  - id: fail_on_breaking_change
    type: BOOL
    displayName: Enforce Deployment Gate
    defaults: true
    description: When true, halts the deployment with an execution failure if
      breaking changes or severe KPI drift are detected.

  - id: alert_channel
    type: STRING
    displayName: Slack Alert Channel
    defaults: "#analytics-engineering"
    description: Slack channel destination for semantic layer gate notifications and
      failure alerts.

  - id: fault_injection_mode
    type: SELECT
    displayName: Fault Drill Mode
    values:
      - NONE
      - REMOVE_REVENUE_METRIC
      - MUTATE_AGGREGATION_SUM_TO_AVG
      - DROP_GRAIN_DIMENSION
      - SIMULATE_KPI_NUMERICAL_DRIFT
    defaults: NONE
    description: Optional fault drill mode to test gate rejection on breaking
      changes and KPI drift without modifying source dbt code.

tasks:
  - id: load_baseline_manifest
    type: io.kestra.plugin.core.kv.Get
    description: Fetch the previously certified production semantic manifest from
      the Kestra KV store.
    key: "{{ inputs.kv_baseline_key }}"
    errorOnMissing: false

  - id: analyze_semantic_manifest_diff
    type: io.kestra.plugin.scripts.python.Script
    description: Parse candidate and baseline semantic manifests, audit breaking
      changes across metrics and models, and evaluate warehouse KPI drift using
      DuckDB.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install duckdb kestra
    env:
      CANDIDATE_MANIFEST: "{{ inputs.candidate_manifest }}"
      BASELINE_MANIFEST_RAW: "{{ outputs.load_baseline_manifest.value ?? '' }}"
      FAULT_MODE: "{{ inputs.fault_injection_mode }}"
      MAX_KPI_DRIFT: "{{ inputs.max_kpi_drift_pct }}"
      EXECUTION_ID: "{{ execution.id }}"
    script: |
      import json
      import os
      import sys
      from datetime import datetime, timezone
      import duckdb

      try:
          from kestra import Kestra
      except ImportError:
          class Kestra:
              @staticmethod
              def outputs(data):
                  print("::" + json.dumps({"outputs": data}) + "::")

      candidate_raw = os.environ.get("CANDIDATE_MANIFEST", "{}").strip()
      baseline_raw = os.environ.get("BASELINE_MANIFEST_RAW", "").strip()
      fault_mode = os.environ.get("FAULT_MODE", "NONE")
      max_kpi_drift = float(os.environ.get("MAX_KPI_DRIFT", "5.0"))
      execution_id = os.environ.get("EXECUTION_ID", "local")

      # Strip Markdown wrappers if present
      if candidate_raw.startswith("```"):
          candidate_raw = candidate_raw.strip("`").lstrip("json").strip()

      candidate_manifest = json.loads(candidate_raw)

      # Fallback certified production baseline if KV store key is uninitialized
      if not baseline_raw or baseline_raw == "null":
          baseline_manifest = {
              "semantic_models": [
                  {
                      "name": "orders_source",
                      "description": "Core orders transaction semantic model",
                      "node_relation": {
                          "relation_name": "analytics.fct_orders"
                      },
                      "entities": [
                          {"name": "order_id", "type": "primary"},
                          {"name": "customer_id", "type": "foreign"}
                      ],
                      "measures": [
                          {"name": "order_total", "agg": "sum", "expr": "order_total_usd", "description": "Gross completed order transaction value in USD"},
                          {"name": "order_count", "agg": "sum", "expr": "1", "description": "Total volume of completed customer orders"},
                          {"name": "distinct_customers", "agg": "count_distinct", "expr": "customer_id", "description": "Count of transacting customer accounts"}
                      ],
                      "dimensions": [
                          {"name": "order_date", "type": "time", "type_params": {"time_granularity": "day"}, "is_partition": True},
                          {"name": "order_status", "type": "categorical"},
                          {"name": "region", "type": "categorical"}
                      ]
                  }
              ],
              "metrics": [
                  {
                      "name": "revenue",
                      "type": "simple",
                      "type_params": {"measure": "order_total"},
                      "description": "Total gross revenue generated by orders"
                  },
                  {
                      "name": "daily_active_users",
                      "type": "simple",
                      "type_params": {"measure": "distinct_customers"},
                      "description": "Daily active transacting users count"
                  },
                  {
                      "name": "order_volume",
                      "type": "simple",
                      "type_params": {"measure": "order_count"},
                      "description": "Total completed orders volume"
                  },
                  {
                      "name": "average_order_value",
                      "type": "ratio",
                      "type_params": {"numerator": "revenue", "denominator": "order_volume"},
                      "description": "Average transaction value calculated as revenue divided by order volume"
                  }
              ]
          }
      else:
          if baseline_raw.startswith("```"):
              baseline_raw = baseline_raw.strip("`").lstrip("json").strip()
          baseline_manifest = json.loads(baseline_raw)

      # Apply optional fault injection simulation
      import copy
      cand = copy.deepcopy(candidate_manifest)
      if fault_mode == "REMOVE_REVENUE_METRIC":
          cand["metrics"] = [m for m in cand.get("metrics", []) if m["name"] != "revenue"]
      elif fault_mode == "MUTATE_AGGREGATION_SUM_TO_AVG":
          for sm in cand.get("semantic_models", []):
              for m in sm.get("measures", []):
                  if m["name"] == "order_total":
                      m["agg"] = "average"
      elif fault_mode == "DROP_GRAIN_DIMENSION":
          for sm in cand.get("semantic_models", []):
              sm["dimensions"] = [d for d in sm.get("dimensions", []) if d["name"] != "order_date"]
              sm["entities"] = [e for e in sm.get("entities", []) if e["name"] != "order_id"]

      breaking_changes = []

      # 1. Audit Metrics
      base_metrics = {m["name"]: m for m in baseline_manifest.get("metrics", [])}
      cand_metrics = {m["name"]: m for m in cand.get("metrics", [])}

      for m_name, base_m in base_metrics.items():
          if m_name not in cand_metrics:
              breaking_changes.append({
                  "category": "REMOVED_METRIC",
                  "object": f"metric:{m_name}",
                  "message": f"Metric '{m_name}' deleted in candidate manifest. Downstream BI queries will fail."
              })
          else:
              cand_m = cand_metrics[m_name]
              if cand_m.get("type") != base_m.get("type"):
                  breaking_changes.append({
                      "category": "ALTERED_METRIC_TYPE",
                      "object": f"metric:{m_name}",
                      "message": f"Metric '{m_name}' type altered from '{base_m.get('type')}' to '{cand_m.get('type')}'."
                  })
              base_tp = base_m.get("type_params", {})
              cand_tp = cand_m.get("type_params", {})
              if base_tp.get("measure") and cand_tp.get("measure") and base_tp.get("measure") != cand_tp.get("measure"):
                  breaking_changes.append({
                      "category": "ALTERED_METRIC_MEASURE",
                      "object": f"metric:{m_name}",
                      "message": f"Metric '{m_name}' underlying measure reference altered from '{base_tp.get('measure')}' to '{cand_tp.get('measure')}'."
                  })

      # 2. Audit Semantic Models
      base_models = {sm["name"]: sm for sm in baseline_manifest.get("semantic_models", [])}
      cand_models = {sm["name"]: sm for sm in cand.get("semantic_models", [])}

      for sm_name, base_sm in base_models.items():
          if sm_name not in cand_models:
              breaking_changes.append({
                  "category": "REMOVED_SEMANTIC_MODEL",
                  "object": f"model:{sm_name}",
                  "message": f"Semantic model '{sm_name}' removed in candidate manifest."
              })
              continue

          cand_sm = cand_models[sm_name]

          # Entities inspection
          base_entities = {e["name"]: e for e in base_sm.get("entities", [])}
          cand_entities = {e["name"]: e for e in cand_sm.get("entities", [])}

          for e_name, base_e in base_entities.items():
              if e_name not in cand_entities:
                  breaking_changes.append({
                      "category": "DELETED_ENTITY",
                      "object": f"entity:{sm_name}.{e_name}",
                      "message": f"Entity '{e_name}' ({base_e.get('type')}) removed from model '{sm_name}'. Primary join resolution broken."
                  })
              elif base_e.get("type") != cand_entities[e_name].get("type"):
                  breaking_changes.append({
                      "category": "ALTERED_ENTITY_TYPE",
                      "object": f"entity:{sm_name}.{e_name}",
                      "message": f"Entity '{e_name}' type changed from '{base_e.get('type')}' to '{cand_entities[e_name].get('type')}'."
                  })

          # Measures inspection
          base_measures = {m["name"]: m for m in base_sm.get("measures", [])}
          cand_measures = {m["name"]: m for m in cand_sm.get("measures", [])}

          for m_name, base_m in base_measures.items():
              if m_name not in cand_measures:
                  breaking_changes.append({
                      "category": "REMOVED_MEASURE",
                      "object": f"measure:{sm_name}.{m_name}",
                      "message": f"Measure '{m_name}' removed from model '{sm_name}'."
                  })
              else:
                  cand_m = cand_measures[m_name]
                  if base_m.get("agg") != cand_m.get("agg"):
                      breaking_changes.append({
                          "category": "MODIFIED_AGGREGATION_TYPE",
                          "object": f"measure:{sm_name}.{m_name}",
                          "message": f"Measure '{m_name}' aggregation modified from '{base_m.get('agg')}' to '{cand_m.get('agg')}'. Calculation semantics changed."
                      })

          # Dimensions inspection
          base_dims = {d["name"]: d for d in base_sm.get("dimensions", [])}
          cand_dims = {d["name"]: d for d in cand_sm.get("dimensions", [])}

          for d_name, base_d in base_dims.items():
              if d_name not in cand_dims:
                  breaking_changes.append({
                      "category": "DELETED_DIMENSION",
                      "object": f"dimension:{sm_name}.{d_name}",
                      "message": f"Grain dimension '{d_name}' removed from model '{sm_name}'. Downstream slicing queries will fail."
                  })
              else:
                  cand_d = cand_dims[d_name]
                  if base_d.get("type") != cand_d.get("type"):
                      breaking_changes.append({
                          "category": "ALTERED_DIMENSION_TYPE",
                          "object": f"dimension:{sm_name}.{d_name}",
                          "message": f"Dimension '{d_name}' type altered from '{base_d.get('type')}' to '{cand_d.get('type')}'."
                      })
                  base_gran = (base_d.get("type_params") or {}).get("time_granularity")
                  cand_gran = (cand_d.get("type_params") or {}).get("time_granularity")
                  if base_gran and cand_gran and base_gran != cand_gran:
                      breaking_changes.append({
                          "category": "ALTERED_TIME_GRANULARITY",
                          "object": f"dimension:{sm_name}.{d_name}",
                          "message": f"Time dimension '{d_name}' granularity altered from '{base_gran}' to '{cand_gran}'."
                      })

      # 3. Warehouse Verification Queries & Numerical Metric Drift
      con = duckdb.connect(":memory:")
      con.execute("""
          CREATE SCHEMA analytics;
          CREATE TABLE analytics.fct_orders (
              order_id VARCHAR,
              customer_id VARCHAR,
              order_date DATE,
              order_total_usd DOUBLE,
              status VARCHAR
          );
          INSERT INTO analytics.fct_orders VALUES
              ('ord-001', 'cust-101', DATE '2026-10-01', 1250.00, 'completed'),
              ('ord-002', 'cust-102', DATE '2026-10-02', 450.50, 'completed'),
              ('ord-003', 'cust-103', DATE '2026-10-02', 890.00, 'completed'),
              ('ord-004', 'cust-101', DATE '2026-10-03', 210.00, 'completed'),
              ('ord-005', 'cust-104', DATE '2026-10-03', 1500.00, 'cancelled'),
              ('ord-006', 'cust-105', DATE '2026-10-04', 3200.00, 'completed'),
              ('ord-007', 'cust-106', DATE '2026-10-04', 780.25, 'completed'),
              ('ord-008', 'cust-102', DATE '2026-10-05', 640.00, 'completed');
      """)

      rev_res = con.execute("SELECT SUM(order_total_usd) FROM analytics.fct_orders WHERE status = 'completed'").fetchone()[0]
      dau_res = con.execute("SELECT COUNT(DISTINCT customer_id) FROM analytics.fct_orders WHERE status = 'completed'").fetchone()[0]
      vol_res = con.execute("SELECT COUNT(order_id) FROM analytics.fct_orders WHERE status = 'completed'").fetchone()[0]

      baseline_kpis = {
          "revenue": float(rev_res),
          "daily_active_users": float(dau_res),
          "order_volume": float(vol_res)
      }

      candidate_kpis = dict(baseline_kpis)
      if fault_mode == "SIMULATE_KPI_NUMERICAL_DRIFT":
          candidate_kpis["revenue"] = baseline_kpis["revenue"] * 0.755 # 24.5% drift drop

      kpi_drift_results = []
      has_severe_drift = False

      for kpi_name, base_val in baseline_kpis.items():
          cand_val = candidate_kpis.get(kpi_name, base_val)
          delta = abs(cand_val - base_val)
          drift_pct = round((delta / base_val) * 100.0, 2) if base_val > 0 else 0.0
          breach = drift_pct > max_kpi_drift
          if breach:
              has_severe_drift = True

          kpi_drift_results.append({
              "metric": kpi_name,
              "baseline_value": base_val,
              "candidate_value": cand_val,
              "drift_pct": drift_pct,
              "threshold_pct": max_kpi_drift,
              "breach": breach
          })

      has_breaking = len(breaking_changes) > 0
      is_valid = not (has_breaking or has_severe_drift)

      violations_summary_lines = []
      if breaking_changes:
          violations_summary_lines.append(f"Breaking Changes ({len(breaking_changes)}):")
          for bc in breaking_changes:
              violations_summary_lines.append(f"- [{bc['category']}] {bc['message']}")
      if has_severe_drift:
          violations_summary_lines.append("Warehouse KPI Drift Breaches:")
          for kd in kpi_drift_results:
              if kd["breach"]:
                  violations_summary_lines.append(f"- {kd['metric']}: {kd['drift_pct']}% drift exceeds {kd['threshold_pct']}% threshold (base: {kd['baseline_value']}, cand: {kd['candidate_value']})")

      violations_summary = "\n".join(violations_summary_lines) if violations_summary_lines else "None"

      audit_report = {
          "execution_id": execution_id,
          "timestamp": datetime.now(timezone.utc).isoformat(),
          "is_valid": is_valid,
          "has_breaking_changes": has_breaking,
          "has_severe_drift": has_severe_drift,
          "breaking_changes_count": len(breaking_changes),
          "breaking_changes": breaking_changes,
          "kpi_drift_results": kpi_drift_results,
          "violations_summary": violations_summary
      }

      Kestra.outputs({
          "is_valid": is_valid,
          "has_breaking_changes": has_breaking,
          "has_severe_drift": has_severe_drift,
          "breaking_changes_count": len(breaking_changes),
          "violations_summary": violations_summary,
          "promoted_manifest": json.dumps(cand),
          "audit_report": audit_report
      })

  - id: persist_audit_report
    type: io.kestra.plugin.core.kv.Set
    description: Record semantic diff and numerical drift audit report into Kestra
      KV store for telemetry and governance tracking.
    key: "{{ inputs.kv_audit_report_key }}"
    kvType: JSON
    value: "{{ outputs.analyze_semantic_manifest_diff.vars.audit_report }}"

  - id: evaluate_semantic_gate
    type: io.kestra.plugin.core.flow.If
    description: Branch execution based on whether breaking changes or severe metric
      drift were discovered.
    condition: "{{ (outputs.analyze_semantic_manifest_diff.vars.has_breaking_changes
      or outputs.analyze_semantic_manifest_diff.vars.has_severe_drift) and
      inputs.fail_on_breaking_change }}"
    then:
      - id: notify_slack_breach
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority alert to analytics engineering with
          breaking changes and metric drift summary.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.alert_channel }}",
            "username": "dbt-Semantic-Gate",
            "text": "🚨 *CRITICAL: dbt Semantic Layer Deployment Gated!*\n• *Status:* `FAILED_SEMANTIC_GATE`\n• *Breaking Changes Count:* `{{ outputs.analyze_semantic_manifest_diff.vars.breaking_changes_count }}`\n• *Severe KPI Drift:* `{{ outputs.analyze_semantic_manifest_diff.vars.has_severe_drift }}`\n• *Violations:*\n```\n{{ outputs.analyze_semantic_manifest_diff.vars.violations_summary }}\n```\n• *Execution:* `{{ execution.id }}`\nDeployment halted to prevent downstream dashboard and MetricFlow API outages."
          }

      - id: fail_pipeline_gate
        type: io.kestra.plugin.core.execution.Fail
        description: Halts execution to block downstream CI/CD deployment or production
          model promotion.
        errorMessage: "dbt Semantic Layer gate failed: detected {{
          outputs.analyze_semantic_manifest_diff.vars.breaking_changes_count }}
          breaking change(s) or KPI numerical drift exceeding {{
          inputs.max_kpi_drift_pct }}% SLA threshold."

    else:
      - id: notify_slack_success
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Posts deployment clearance notification to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.alert_channel }}",
            "username": "dbt-Semantic-Gate",
            "text": "✅ *dbt Semantic Layer Gate Passed*\nAll metrics, dimensions, and measures are backwards-compatible. Warehouse KPI verification drift is within the {{ inputs.max_kpi_drift_pct }}% threshold. Certified production manifest updated."
          }

      - id: promote_manifest_to_kv
        type: io.kestra.plugin.core.kv.Set
        description: Promotes the verified candidate semantic manifest to production
          baseline in KV store.
        key: "{{ inputs.kv_baseline_key }}"
        kvType: JSON
        value: "{{ outputs.analyze_semantic_manifest_diff.vars.promoted_manifest }}"

      - id: log_gate_passed
        type: io.kestra.plugin.core.log.Log
        description: Logs nominal gate clearance.
        message: "dbt Semantic Layer Gate PASSED for execution {{ execution.id }}.
          Production manifest baseline updated in KV key '{{
          inputs.kv_baseline_key }}'."

triggers:
  - id: on_dbt_ci_build
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP webhook endpoint for CI/CD runners (GitHub
      Actions, GitLab CI, dbt Cloud) to trigger semantic manifest gating.
    key: "{{ secret('DBT_GATE_WEBHOOK_KEY') }}"

  - id: scheduled_weekly_drift_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekly scheduled metric drift audit comparing warehouse KPIs
      against production baseline numbers.
    cron: "0 7 * * 1"
    timezone: UTC
    disabled: true

errors:
  - id: notify_execution_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Notifies analytics engineering if the gate flow encounters an
      unhandled runtime error.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.alert_channel }}",
        "username": "dbt-Semantic-Gate",
        "text": "❌ *dbt Semantic Gate Error*\nExecution `{{ execution.id }}` failed unexpectedly during semantic manifest audit. Check Kestra execution logs for error traces."
      }

Automate semantic contract governance for dbt Core and MetricFlow by validating candidate semantic_manifest.json artifacts against certified production baselines. Detect breaking mutations in metrics, dimensions, measures, and entities, execute warehouse verification queries to measure numerical KPI drift, and gate CI/CD deployments before bad metrics break executive dashboards.

How it works

Baseline Ingestion: The load_baseline_manifest task reads the certified production semantic manifest from the Kestra KV store (dbt_semantic_manifest_prod) with automatic fallback to initial certified schemas.
Semantic Diff Inspection: The analyze_semantic_manifest_diff task runs an isolated Python Docker container to compare candidate and baseline manifests. It flags removed metrics, modified calculation types, altered entity joins, mutated measure aggregations (such as sum to average), and deleted grain dimensions.
Warehouse KPI Verification: The Python audit task executes verification SQL queries against in-memory DuckDB warehouse tables representing key business metrics (revenue, active users, order volume) to detect numerical calculation drift against defined percentage thresholds.
Telemetry Persistence: The persist_audit_report task writes the complete structured audit manifest into Kestra KV (dbt_semantic_layer_audit_report) for historical observability.
Quality Gating: If breaking changes or KPI drift exceed thresholds, evaluate_semantic_gate alerts the #analytics-engineering Slack channel and terminates the run with io.kestra.plugin.core.execution.Fail. Otherwise, it updates the production baseline in KV and approves the promotion.

What you get

Automated detection of breaking dbt semantic layer changes in CI/CD before deployment.
Protection against silent aggregation mutations (such as changing a measure from sum to average).
Granular warehouse verification comparing numerical KPI drift against SLA thresholds.
Complete audit trail persisted in Kestra KV store with structured JSON telemetry.
Instant Slack alerts with actionable violation summaries for analytics engineers.

Who it's for

Analytics Engineers maintaining dbt Core Semantic Layer or MetricFlow models.
Data Platform Teams implementing automated CI/CD guardrails and deployment gates.
BI Engineers protecting downstream dashboards in Superset, Tableau, or Hex from semantic outages.

Why orchestrate this with Kestra

Containerized Auditing: Executes complex semantic graph parsing and DuckDB verification queries in isolated Docker containers without host dependencies.
Native KV State Management: Stores certified baseline manifests and audit telemetry directly in Kestra KV store without external databases.
Declarative Quality Gating: Combines native io.kestra.plugin.core.flow.If and io.kestra.plugin.core.execution.Fail tasks for deterministic pipeline control.
Multi-Trigger Support: Operates as an automated webhook gate for CI/CD runners or as a scheduled recurring weekly audit.

Prerequisites

Docker daemon accessible to Kestra workers.
Slack Incoming Webhook URL configured in Kestra secrets.
Webhook authentication key configured for CI/CD invocation.

Secrets

SLACK_WEBHOOK_URL: Slack Incoming Webhook URL for analytics team alerts.
DBT_GATE_WEBHOOK_KEY: Secret authentication key for the inbound webhook trigger.

Quick start

Store your Slack webhook URL in Kestra secrets under SLACK_WEBHOOK_URL.
Store your webhook authentication key under DBT_GATE_WEBHOOK_KEY.
Run a manual execution with fault_injection_mode: NONE to verify the nominal pass path and populate the initial KV baseline.
Execute a test run with fault_injection_mode: MUTATE_AGGREGATION_SUM_TO_AVG to verify that breaking measure aggregations are caught and gated.

How to extend

Warehouse Connection: Replace the in-memory DuckDB warehouse verification queries with native Snowflake (io.kestra.plugin.jdbc.snowflake.Query), BigQuery (io.kestra.plugin.gcp.bigquery.Query), or PostgreSQL (io.kestra.plugin.jdbc.postgresql.Query) tasks.
S3 / GCS Artifact Storage: Store historical semantic manifests in cloud object storage using io.kestra.plugin.aws.s3.Upload or io.kestra.plugin.gcp.gcs.Upload.
Human Approval Pause: Add an io.kestra.plugin.core.flow.Pause task allowing lead data stewards to manually approve minor metric drift before failing.

dbt Core Semantic Layer Metric Drift & Semantic Manifest Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

dbt Core Semantic Layer Metric Drift & Semantic Manifest Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

NocoDB Database Platform Health, Storage Engine, and Project Sentinel

Detect and Resume Stuck Paused NSQ Channels with Accumulating Backlog

Automated OpenLineage Metadata Extraction and Marquez Governance Pipeline

Hourly OpenSearch Error Spike Gate with Discord Alerting

Object Storage Integrity and Streaming Replica Parity Audit

Use OpenSearch as a Config Store with Put and Get Roundtrip

More Core Blueprints

Block breaking OpenAPI changes before they merge

Netskope Security Alert Volume Sentinel

Open-Source Issue Readiness & Contribution Planner

Newman (Postman CLI) Automated API Contract Regression Gate

Open Graph Card Generator

Nexus Repository Blobstore Storage Space Exhaustion Sentinel

New to Kestra?