Monitoring Soda Core Data Quality Scans with Vigilmon

Soda Core is a powerful open-source data quality framework. You write SodaCL checks in YAML, Soda scans your data source, and failed checks surface in your pipeline logs. It's a clean, declarative approach to ensuring data quality.

But Soda Core only validates what it scans — it can't monitor itself. If a scan crashes before it reads a single row, or if your scheduler silently drops the job, Soda reports nothing. Your data quality process is broken, and nobody knows.

This tutorial closes that gap. You'll build a monitoring wrapper around Soda Core scans, expose a health endpoint for your scan service, set up heartbeat monitors in Vigilmon for every scan job, and route alerts to Slack when something breaks.

The failure modes Soda Core can't catch itself

Soda Core is excellent at what it does, but it has a monitoring blind spot: it can only report on scans it actually runs. A few scenarios that slip past it:

Scan process crash before first check — if Soda fails to connect to the data source, the scan exits with an error. No checks are evaluated, so no failures are recorded in your Soda Cloud account or local logs. Your alert on "failed checks" never fires.

Scheduler drift — your cron job or orchestrator task is deprioritized due to resource pressure. The scan window passes without a scan running. No notification goes out because there's nothing to notify about.

Configuration rot — someone renames a column in your warehouse. Soda tries to check a column that no longer exists and throws a ScanError. The scan is aborted, not failed — so check-based alerts don't trigger.

Connection string rotation — your data warehouse credentials rotate. The next Soda scan fails at the datasource connection step. Without external monitoring of the scan process, you only discover this when a stakeholder notices stale data.

All of these require monitoring the scan execution externally, not just the scan results.

Step 1: Install Soda Core and build a monitored scan runner

Install Soda Core for your data source (PostgreSQL in this example):

pip install soda-core-postgres httpx

Write your SodaCL checks file:

# checks/orders_checks.yml
checks for orders:
  - row_count > 0:
      name: Orders table is not empty
  - missing_count(customer_id) = 0:
      name: No missing customer IDs
  - duplicate_count(order_id) = 0:
      name: No duplicate order IDs
  - min(order_total) >= 0:
      name: No negative order totals
  - freshness(created_at) < 25h:
      name: Orders data is fresh

Write a wrapper that runs the scan and handles monitoring:

# soda_runner.py
import httpx
import logging
import os
import time
from datetime import datetime

from soda.scan import Scan

logger = logging.getLogger(__name__)


def run_soda_scan(
    data_source_name: str,
    checks_file: str,
    configuration_file: str,
    heartbeat_url: str | None = None,
) -> dict:
    """
    Execute a Soda scan and optionally ping a heartbeat URL on success.
    Returns a summary dict with scan results.
    """
    scan = Scan()
    scan.set_data_source_name(data_source_name)
    scan.add_configuration_yaml_file(file_path=configuration_file)
    scan.add_sodacl_yaml_file(file_path=checks_file)

    start = time.time()
    exit_code = scan.execute()
    elapsed = round(time.time() - start, 2)

    all_checks = scan.get_checks()
    passed = [c for c in all_checks if c.outcome and c.outcome.name == "pass"]
    failed = [c for c in all_checks if c.outcome and c.outcome.name == "fail"]
    warnings = [c for c in all_checks if c.outcome and c.outcome.name == "warn"]

    errors = scan.get_error_logs()
    has_errors = len(errors) > 0

    success = exit_code == 0 and not has_errors

    summary = {
        "data_source": data_source_name,
        "checks_file": checks_file,
        "success": success,
        "exit_code": exit_code,
        "total_checks": len(all_checks),
        "passed": len(passed),
        "failed": len(failed),
        "warnings": len(warnings),
        "errors": [str(e) for e in errors],
        "elapsed_seconds": elapsed,
        "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 scan '{checks_file}'")
        except Exception as e:
            logger.warning(f"Heartbeat ping failed: {e}")
    elif not success:
        logger.error(
            f"Scan failed: {len(failed)} checks failed, {len(errors)} errors. "
            f"Skipping heartbeat."
        )

    return summary

The pattern is the same as with any scheduled job: only ping on clean success. A scan with errors, a scan with failed checks, or a scan that never ran all produce the same observable result from Vigilmon's perspective — a missing heartbeat — and all trigger the same alert.

Step 2: Configure your Soda datasource

Write your Soda configuration file:

# soda_config.yml
data_sources:
  orders_warehouse:
    type: postgres
    host: ${POSTGRES_HOST}
    port: 5432
    username: ${POSTGRES_USER}
    password: ${POSTGRES_PASSWORD}
    database: ${POSTGRES_DB}
    schema: public

Soda resolves ${VAR} references from environment variables automatically, so credentials stay out of your config files.

Step 3: Build a health endpoint for your scan service

If you have a service that runs Soda scans, expose a health endpoint that surfaces the state of recent scans:

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

app = FastAPI()

SCAN_LOG_DIR = Path(os.environ.get("SCAN_LOG_DIR", "/tmp/soda_runs"))


@app.get("/health")
def health_check():
    if not SCAN_LOG_DIR.exists():
        return JSONResponse(
            status_code=503,
            content={"status": "no_runs_recorded"},
        )

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

    scans = []
    any_failed = False
    any_stale = False

    for scan_file in scan_files:
        try:
            data = json.loads(scan_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)

            if stale:
                any_stale = True
            if failed:
                any_failed = True

            scans.append({
                "scan": data["checks_file"],
                "success": data.get("success"),
                "failed_checks": data.get("failed"),
                "age_hours": round(age_hours, 1),
                "stale": stale,
            })
        except Exception:
            pass

    if any_failed or any_stale:
        return JSONResponse(
            status_code=503,
            content={
                "status": "degraded",
                "scans": scans,
            },
        )

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

Step 4: Set up Vigilmon monitoring

With your scan runner and health endpoint in place, connect them to Vigilmon.

HTTP monitor for scan health

1. Sign up at vigilmon.online — free tier, no credit card
2. Click New Monitor → HTTP
3. Enter your health endpoint URL
4. Set interval to 10 minutes
5. Save

Heartbeat monitor per scan

For each Soda scan that must run on a schedule:

1. Click New Monitor → Heartbeat
2. Name it (e.g. "orders_soda_scan")
3. Set the expected interval — for a daily scan use 25 hours
4. Copy the unique ping URL

# .env
ORDERS_SCAN_HEARTBEAT_URL=https://vigilmon.online/api/heartbeat/token-orders
CUSTOMERS_SCAN_HEARTBEAT_URL=https://vigilmon.online/api/heartbeat/token-customers
PRODUCTS_SCAN_HEARTBEAT_URL=https://vigilmon.online/api/heartbeat/token-products

Run your scans with heartbeat support:

# run_daily_scans.py
import os
import json
from pathlib import Path
from soda_runner import run_soda_scan

SCAN_LOG_DIR = Path("/tmp/soda_runs")
SCAN_LOG_DIR.mkdir(exist_ok=True)

SCANS = [
    {
        "data_source_name": "orders_warehouse",
        "checks_file": "checks/orders_checks.yml",
        "configuration_file": "soda_config.yml",
        "heartbeat_url": os.environ.get("ORDERS_SCAN_HEARTBEAT_URL"),
        "log_file": "orders.json",
    },
    {
        "data_source_name": "orders_warehouse",
        "checks_file": "checks/customers_checks.yml",
        "configuration_file": "soda_config.yml",
        "heartbeat_url": os.environ.get("CUSTOMERS_SCAN_HEARTBEAT_URL"),
        "log_file": "customers.json",
    },
]

for scan_config in SCANS:
    log_file = scan_config.pop("log_file")
    summary = run_soda_scan(**scan_config)
    (SCAN_LOG_DIR / log_file).write_text(json.dumps(summary))
    status = "OK" if summary["success"] else f"FAILED ({summary['failed']} checks)"
    print(f"{scan_config['checks_file']}: {status} in {summary['elapsed_seconds']}s")

Step 5: Alert routing and escalation

Configure Vigilmon notification channels for your data team:

Slack:

1. Create a Slack incoming webhook
2. In Vigilmon: Notifications → New Channel → Slack
3. Paste the webhook URL
4. Associate with all Soda scan monitors

Email escalation:

1. In Vigilmon: Notifications → New Channel → Email
2. Add your data engineering manager's email
3. Set up an escalation policy — if a heartbeat has been missing for 2+ hours, escalate to email in addition to Slack

Alert format in Slack:

🔴 MISSED: orders_soda_scan heartbeat
Expected every: 25 hours
Last ping: 28 hours ago
Action: check your scan scheduler and Soda logs

🔴 DOWN: scan-service.internal/health
Status: 503 — Last run had 3 failed checks on orders table

Step 6: Integrate with Airflow

If you run Soda scans via Airflow:

from airflow.decorators import dag, task
from datetime import datetime
import os


@dag(
    schedule="0 6 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
)
def data_quality_pipeline():

    @task
    def scan_orders():
        from soda_runner import run_soda_scan
        summary = run_soda_scan(
            data_source_name="orders_warehouse",
            checks_file="checks/orders_checks.yml",
            configuration_file="soda_config.yml",
            heartbeat_url=os.environ.get("ORDERS_SCAN_HEARTBEAT_URL"),
        )
        if not summary["success"]:
            raise ValueError(
                f"Soda scan failed: {summary['failed']} checks failed, "
                f"{len(summary['errors'])} errors"
            )
        return summary

    @task
    def scan_customers():
        from soda_runner import run_soda_scan
        summary = run_soda_scan(
            data_source_name="orders_warehouse",
            checks_file="checks/customers_checks.yml",
            configuration_file="soda_config.yml",
            heartbeat_url=os.environ.get("CUSTOMERS_SCAN_HEARTBEAT_URL"),
        )
        if not summary["success"]:
            raise ValueError(f"Customer scan failed: {summary['failed']} checks failed")
        return summary

    scan_orders()
    scan_customers()


data_quality_dag = data_quality_pipeline()

The task raising on failure ensures Airflow retries and marks the DAG run as failed (so Airflow-level alerts fire). The Vigilmon heartbeat is independent — if Airflow itself is unhealthy, the heartbeat still goes missing.

Step 7: Public status page

Share scan health with data consumers without exposing your infrastructure:

1. In Vigilmon: Status Pages → New Status Page
2. Name it "Data Quality Status"
3. Add all your Soda scan heartbeat monitors
4. Make it public and share the URL with your analytics and BI teams

Now when a scan is failing, your BI analysts know before they open a Jira ticket asking "why are the numbers wrong?"

Monitoring summary

| Layer | What it covers | |---|---| | SodaCL checks | Column-level data quality constraints | | Soda exit code | Scan-level errors and configuration failures | | Heartbeat monitors | Scheduler failures, crashed processes, missed windows | | HTTP health endpoint | Cross-scan health status for ops teams | | Slack/email alerts | Instant notification to data engineering team | | Public status page | Visibility for BI, analytics, and stakeholders |

Soda Core handles data quality. Vigilmon handles operational reliability. Together they ensure both the data and the process are working.

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