tutorial

Monitoring dbt Core Jobs with Vigilmon

dbt Core transforms your data but won't alert you when runs silently fail or never start. Add heartbeat monitoring, run health endpoints, and instant alerts so your data team knows the moment a dbt job breaks down.

Monitoring dbt Core Jobs with Vigilmon

dbt Core is the backbone of modern data transformation pipelines. You write models in SQL, dbt resolves the dependency graph and executes them in the right order, and your transformed data lands in your warehouse ready for analytics. When dbt runs are healthy, your data team ships fast.

But dbt Core has no built-in alerting. A failed model raises an exit code, but if your scheduler doesn't catch that exit code, nobody knows. A dbt run that hangs indefinitely ties up your orchestrator and blocks downstream models. A model that was accidentally excluded from the run produces stale data silently.

This tutorial shows you how to wrap dbt Core with production-grade monitoring: a health endpoint that surfaces run status, heartbeat monitors in Vigilmon for every critical dbt job, and Slack alerts that fire the moment a run goes wrong.

The monitoring gaps in dbt Core

dbt Core is a CLI tool — it runs, writes artifacts, exits. What it doesn't do:

Proactive alerting — dbt logs failures to the terminal and writes run_results.json, but it doesn't push alerts anywhere. If your CI/CD job fails silently, the failure is invisible until someone checks the logs.

Scheduler visibility — dbt Core doesn't schedule itself. If your orchestrator (cron, Airflow, Prefect, Dagster) fails to trigger a run, dbt never gets invoked. You won't see a failure in run_results.json because there's no run_results.json at all.

Partial run detection — if you run dbt run --select some_model and accidentally exclude a critical model, the run succeeds but the excluded model's data goes stale. No error, no alert.

Long-running detection — a dbt run that takes 3× its normal time isn't failing yet, but it's a leading indicator. Without monitoring run duration trends, you find out about the problem when the run finally fails hours later.

These gaps are solved by external monitoring of the run process, not just the run results.

Step 1: Write a dbt run wrapper with heartbeat support

Build a Python wrapper that invokes dbt, parses the results, and pings a Vigilmon heartbeat on success:

# dbt_runner.py
import subprocess
import json
import httpx
import logging
import os
import time
from datetime import datetime
from pathlib import Path

logger = logging.getLogger(__name__)


def run_dbt(
    command: list[str],
    project_dir: str,
    profiles_dir: str | None = None,
    heartbeat_url: str | None = None,
    run_results_path: str | None = None,
) -> dict:
    """
    Execute a dbt command and optionally ping a heartbeat URL on success.

    Args:
        command: dbt command as list, e.g. ["dbt", "run", "--select", "tag:daily"]
        project_dir: path to dbt project directory
        profiles_dir: path to profiles.yml directory (defaults to ~/.dbt)
        heartbeat_url: Vigilmon heartbeat URL to ping on clean success
        run_results_path: path to dbt run_results.json (auto-detected if None)

    Returns:
        dict with run summary
    """
    if profiles_dir:
        command = command + ["--profiles-dir", profiles_dir]

    start = time.time()
    result = subprocess.run(
        command,
        cwd=project_dir,
        capture_output=True,
        text=True,
    )
    elapsed = round(time.time() - start, 2)

    exit_code = result.returncode
    stdout = result.stdout
    stderr = result.stderr

    # Parse dbt run results artifact if available
    if run_results_path is None:
        run_results_path = os.path.join(project_dir, "target", "run_results.json")

    run_results = None
    model_summary = {"total": 0, "success": 0, "error": 0, "skipped": 0, "warn": 0}

    if os.path.exists(run_results_path):
        try:
            with open(run_results_path) as f:
                run_results = json.load(f)

            for r in run_results.get("results", []):
                status = r.get("status", "").lower()
                model_summary["total"] += 1
                if status in ("success", "pass"):
                    model_summary["success"] += 1
                elif status == "error":
                    model_summary["error"] += 1
                elif status == "skip":
                    model_summary["skipped"] += 1
                elif status == "warn":
                    model_summary["warn"] += 1
        except Exception as e:
            logger.warning(f"Could not parse run_results.json: {e}")

    success = exit_code == 0 and model_summary.get("error", 0) == 0

    summary = {
        "command": " ".join(command),
        "exit_code": exit_code,
        "success": success,
        "elapsed_seconds": elapsed,
        "models": model_summary,
        "timestamp": datetime.utcnow().isoformat(),
    }

    if success and heartbeat_url:
        try:
            with httpx.Client(timeout=10) as client:
                client.get(heartbeat_url)
            logger.info(f"Heartbeat pinged for dbt run")
        except Exception as e:
            logger.warning(f"Heartbeat ping failed: {e}")
    elif not success:
        logger.error(
            f"dbt run failed (exit={exit_code}, "
            f"errors={model_summary.get('error', 0)}). Skipping heartbeat."
        )
        if stderr:
            logger.error(f"stderr: {stderr[:500]}")

    return summary

The heartbeat is only pinged when:

  1. dbt exits with code 0 (no command-level failure)
  2. No models have error status in run_results.json

If either condition fails — or if the script is never invoked — Vigilmon sees a missing heartbeat and opens an incident.

Step 2: Add a run results health endpoint

Expose dbt run health through an HTTP endpoint for your ops team and monitoring:

# app.py
from fastapi import FastAPI
from fastapi.responses import JSONResponse
import json
import os
from pathlib import Path
from datetime import datetime

app = FastAPI()

RUN_LOG_DIR = Path(os.environ.get("DBT_RUN_LOG_DIR", "/tmp/dbt_runs"))


@app.get("/health/dbt")
def dbt_health():
    """
    Returns health status of dbt runs.
    Returns 503 if any tracked run is stale or failed.
    """
    if not RUN_LOG_DIR.exists():
        return JSONResponse(
            status_code=503,
            content={"status": "no_runs_recorded"},
        )

    run_files = list(RUN_LOG_DIR.glob("*.json"))
    if not run_files:
        return JSONResponse(
            status_code=503,
            content={"status": "no_runs_recorded"},
        )

    runs = []
    any_degraded = False

    for run_file in run_files:
        try:
            data = json.loads(run_file.read_text())
            last_ts = datetime.fromisoformat(data["timestamp"])
            age_hours = (datetime.utcnow() - last_ts).total_seconds() / 3600

            stale = age_hours > 26
            failed = not data.get("success", False)
            degraded = stale or failed

            if degraded:
                any_degraded = True

            runs.append({
                "job": run_file.stem,
                "command": data.get("command"),
                "success": data.get("success"),
                "elapsed_seconds": data.get("elapsed_seconds"),
                "models": data.get("models"),
                "age_hours": round(age_hours, 1),
                "stale": stale,
                "failed": failed,
            })
        except Exception:
            pass

    if any_degraded:
        return JSONResponse(
            status_code=503,
            content={"status": "degraded", "runs": runs},
        )

    return JSONResponse(
        status_code=200,
        content={"status": "ok", "runs": runs},
    )

Step 3: Set up your dbt jobs

Structure your dbt runs per scheduling tier and monitor each independently:

# run_dbt_jobs.py
import os
import json
import sys
from pathlib import Path
from dbt_runner import run_dbt

RUN_LOG_DIR = Path("/tmp/dbt_runs")
RUN_LOG_DIR.mkdir(exist_ok=True)

DBT_PROJECT_DIR = os.environ.get("DBT_PROJECT_DIR", "/opt/dbt")

JOBS = [
    {
        "name": "daily_core_models",
        "command": ["dbt", "run", "--select", "tag:daily"],
        "heartbeat_url": os.environ.get("DBT_DAILY_HEARTBEAT_URL"),
    },
    {
        "name": "daily_tests",
        "command": ["dbt", "test", "--select", "tag:daily"],
        "heartbeat_url": os.environ.get("DBT_TESTS_HEARTBEAT_URL"),
    },
    {
        "name": "daily_snapshots",
        "command": ["dbt", "snapshot"],
        "heartbeat_url": os.environ.get("DBT_SNAPSHOTS_HEARTBEAT_URL"),
    },
]

exit_codes = []
for job in JOBS:
    print(f"Running job: {job['name']}")
    summary = run_dbt(
        command=job["command"],
        project_dir=DBT_PROJECT_DIR,
        heartbeat_url=job["heartbeat_url"],
    )
    (RUN_LOG_DIR / f"{job['name']}.json").write_text(json.dumps(summary))

    status = "OK" if summary["success"] else f"FAILED ({summary['models']['error']} errors)"
    print(f"  {status} in {summary['elapsed_seconds']}s — "
          f"{summary['models']['success']}/{summary['models']['total']} models OK")

    exit_codes.append(0 if summary["success"] else 1)

sys.exit(max(exit_codes))

Step 4: Set up Vigilmon monitoring

Connect your dbt pipeline to Vigilmon.

HTTP monitor for run health

  1. Sign up at vigilmon.online — free tier, no credit card
  2. Click New Monitor → HTTP
  3. Enter your health endpoint URL (e.g. https://data-platform.internal/health/dbt)
  4. Set interval: 10 minutes
  5. Save

Heartbeat monitor per dbt job

For each dbt job:

  1. Click New Monitor → Heartbeat
  2. Name it (e.g. "daily_core_models")
  3. Set the expected interval — 25 hours for a daily job
  4. Copy the unique ping URL
# .env
DBT_PROJECT_DIR=/opt/dbt
DBT_DAILY_HEARTBEAT_URL=https://vigilmon.online/api/heartbeat/token-daily
DBT_TESTS_HEARTBEAT_URL=https://vigilmon.online/api/heartbeat/token-tests
DBT_SNAPSHOTS_HEARTBEAT_URL=https://vigilmon.online/api/heartbeat/token-snapshots

Step 5: Integrate with Airflow

If you orchestrate dbt with Airflow, use the BashOperator or the dbt-airflow provider:

from airflow.decorators import dag, task
from airflow.operators.bash import BashOperator
from datetime import datetime
import os


@dag(
    schedule="0 6 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["dbt", "data-quality"],
)
def dbt_daily_pipeline():

    dbt_run = BashOperator(
        task_id="dbt_run_daily",
        bash_command="""
            cd {{ var.value.dbt_project_dir }} && \
            python /opt/scripts/run_dbt_jobs.py
        """,
        env={
            "DBT_PROJECT_DIR": "{{ var.value.dbt_project_dir }}",
            "DBT_DAILY_HEARTBEAT_URL": os.environ.get("DBT_DAILY_HEARTBEAT_URL", ""),
            "DBT_TESTS_HEARTBEAT_URL": os.environ.get("DBT_TESTS_HEARTBEAT_URL", ""),
            "DBT_SNAPSHOTS_HEARTBEAT_URL": os.environ.get("DBT_SNAPSHOTS_HEARTBEAT_URL", ""),
        },
    )

    dbt_run


dbt_pipeline = dbt_daily_pipeline()

Airflow marks the task as failed when the script exits non-zero (which happens when any dbt job fails). The Vigilmon heartbeat provides an independent second signal — if Airflow itself is unhealthy, you still get paged.

Step 6: Alert routing

Configure Vigilmon to notify your data engineering team:

Slack alerts:

  1. Create an incoming webhook for #dbt-alerts
  2. In Vigilmon: Notifications → New Channel → Slack
  3. Paste the webhook URL
  4. Enable on all dbt job monitors

Alert format in Slack:

🔴 MISSED: daily_core_models heartbeat
Expected every: 25 hours
Last ping: 27 hours ago

🔴 DOWN: data-platform.internal/health/dbt
Status: 503 — daily_tests job failed: 3 model errors

Recovery notifications:

✅ RECOVERED: daily_core_models
✅ UP: data-platform.internal/health/dbt

PagerDuty escalation:

For business-critical dbt models (revenue calculations, customer metrics), escalate after 30 minutes of no heartbeat to your on-call engineer:

  1. In Vigilmon: Notifications → New Channel → Webhook
  2. Point at your PagerDuty events API
  3. Set escalation delay: 30 minutes
  4. Associate only with your most critical monitors

Step 7: Monitor dbt test failures separately

dbt tests are distinct from dbt runs — a model can build successfully but fail data quality tests. Monitor them separately:

# In your JOBS list, add dedicated test heartbeats:
{
    "name": "source_freshness_check",
    "command": ["dbt", "source", "freshness"],
    "heartbeat_url": os.environ.get("DBT_FRESHNESS_HEARTBEAT_URL"),
},

And a dedicated Vigilmon heartbeat monitor named "source_freshness_check" with a 25-hour interval. Source freshness failures are often the first signal of upstream data pipeline problems.

Step 8: Public status page

Give your data consumers visibility into dbt pipeline health:

  1. In Vigilmon: Status Pages → New Status Page
  2. Name it "Data Transformation Pipeline"
  3. Add your dbt job heartbeat monitors
  4. Share the URL with your BI team and analytics stakeholders

When dbt runs are failing and dashboards show stale data, stakeholders can check the status page instead of filing support tickets asking "why is the data wrong?"

Monitoring coverage summary

| Monitor | Type | Expected interval | What it catches | |---|---|---|---| | daily_core_models heartbeat | Heartbeat | 25 hours | Run not invoked, scheduler failure, model errors | | daily_tests heartbeat | Heartbeat | 25 hours | Test suite skipped, test failures | | daily_snapshots heartbeat | Heartbeat | 25 hours | Snapshot job failed | | source_freshness_check heartbeat | Heartbeat | 25 hours | Upstream data source freshness degradation | | /health/dbt endpoint | HTTP | 10 minutes | Cross-job health rollup, stale runs |

Advanced: Track model run duration as a signal

Add duration tracking to catch slow-running models before they fail completely:

# In dbt_runner.py, add duration warnings
DURATION_WARNING_THRESHOLD = float(
    os.environ.get("DBT_DURATION_WARNING_SECONDS", "1800")  # 30 minutes
)

if elapsed > DURATION_WARNING_THRESHOLD:
    logger.warning(
        f"dbt run took {elapsed}s — exceeds threshold of {DURATION_WARNING_THRESHOLD}s. "
        f"Check for warehouse resource contention or runaway queries."
    )
    # Don't fail the heartbeat for slow runs — they may still succeed
    # But log for your observability system to pick up

Summary

dbt Core runs your transformations. Vigilmon ensures those transformations actually run.

The combination gives you:

  • Run-level coverage: heartbeat per job catches any failure mode that prevents a clean run
  • Model-level coverage: run_results.json parsing catches individual model failures
  • Freshness coverage: source freshness checks catch upstream data quality issues
  • Stakeholder visibility: public status page surfaces pipeline health without paging engineers

The setup is a one-time 45-minute investment. Once it's in place, your on-call engineer gets paged within the heartbeat window of any dbt failure — not the next morning when someone notices the dashboard is stale.

Get started free at vigilmon.online — your first heartbeat monitor is live in under two minutes.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles