Monitoring Great Expectations Data Pipelines with Vigilmon

Great Expectations is the gold standard for data quality validation. You define expectations about your data — column types, value ranges, null rates — and GX checks them every time your pipeline runs. When data violates those expectations, GX fails the checkpoint.

But here's the catch: GX can only tell you about bad data. It can't tell you when the validation job itself fails to run.

Your nightly validation suite might crash silently, your orchestrator might skip the step, or a misconfigured connection might cause GX to exit without running a single suite. Meanwhile, your downstream dashboards happily display stale data and nobody knows.

This tutorial shows you how to close that gap: build a health endpoint for your GX integration, add heartbeat monitoring for every validation checkpoint, and route alerts to Slack so your data team knows within minutes when something breaks.

What can go wrong with Great Expectations in production

Before jumping to solutions, it's worth naming the failure modes that traditional monitoring misses:

Silent checkpoint failures — a checkpoint runs but throws an unhandled exception. Your orchestrator marks the task as failed, but nobody is paged. The next step in your pipeline runs anyway because it doesn't depend on the checkpoint exit code.

Expectation suite drift — someone edits an expectation suite and removes a critical constraint. The checkpoint now passes everything — including data that should fail. You don't get an alert because GX reports success.

Data source connection drops — the datasource config points to a connection string that changed after a migration. GX initialization fails before it validates anything. No failures are recorded in the Data Docs because no validation ran.

Orphaned runs — your Airflow/Prefect/Dagster task is scheduled but the worker is overloaded. The task sits in the queue past its window. No validation ran, but no failure was recorded either.

All of these require external monitoring of the validation process, not just the validation results.

Step 1: Wrap your GX checkpoint in a monitoring-aware runner

Start by building a thin wrapper around your checkpoint execution that captures success, failure, and metadata:

# gx_runner.py
import great_expectations as gx
import httpx
import logging
import os
import time
from datetime import datetime

logger = logging.getLogger(__name__)


def run_checkpoint_with_heartbeat(
    context_root_dir: str,
    checkpoint_name: str,
    heartbeat_url: str | None = None,
) -> dict:
    """
    Run a GX checkpoint and optionally ping a heartbeat URL on success.
    Returns a summary dict suitable for logging or alerting.
    """
    context = gx.get_context(context_root_dir=context_root_dir)

    start = time.time()
    result = context.run_checkpoint(checkpoint_name=checkpoint_name)
    elapsed = round(time.time() - start, 2)

    run_results = result.run_results
    suite_results = list(run_results.values())

    total_expectations = sum(
        r.results.__len__() for r in suite_results
        if hasattr(r, "results")
    )
    failed_expectations = sum(
        1 for r in suite_results
        if hasattr(r, "results")
        for er in r.results
        if not er.success
    )

    summary = {
        "checkpoint": checkpoint_name,
        "success": result.success,
        "suites_run": len(suite_results),
        "total_expectations": total_expectations,
        "failed_expectations": failed_expectations,
        "elapsed_seconds": elapsed,
        "timestamp": datetime.utcnow().isoformat(),
    }

    if result.success and heartbeat_url:
        try:
            with httpx.Client(timeout=10) as client:
                client.get(heartbeat_url)
            logger.info(f"Heartbeat pinged for checkpoint '{checkpoint_name}'")
        except Exception as e:
            logger.warning(f"Heartbeat ping failed: {e}")

    return summary

The critical design decision: only ping the heartbeat on success. The absence of a ping is the alert — Vigilmon opens an incident when your heartbeat goes missing within its expected window. You get alerted even if your pipeline crashes before GX runs a single expectation.

Step 2: Build a health endpoint for your GX service

If you're running GX validations as part of a long-running service (a FastAPI microservice, a Flask app, or a custom worker), add a health endpoint that reports the state of the most recent validation runs:

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

app = FastAPI()

LAST_RUN_LOG = Path(os.environ.get("GX_RUN_LOG", "/tmp/gx_last_run.json"))


@app.get("/health")
def health_check():
    """
    Returns the health of the GX validation service.
    Returns 503 if the last run was more than 26 hours ago or failed.
    """
    if not LAST_RUN_LOG.exists():
        return JSONResponse(
            status_code=503,
            content={"status": "no_run_recorded", "detail": "No validation run found"},
        )

    try:
        run_data = json.loads(LAST_RUN_LOG.read_text())
    except Exception as e:
        return JSONResponse(
            status_code=503,
            content={"status": "error", "detail": f"Cannot read run log: {e}"},
        )

    from datetime import datetime, timezone, timedelta

    last_run_time = datetime.fromisoformat(run_data["timestamp"])
    age_hours = (datetime.utcnow() - last_run_time).total_seconds() / 3600

    if age_hours > 26:
        return JSONResponse(
            status_code=503,
            content={
                "status": "stale",
                "last_run": run_data["timestamp"],
                "age_hours": round(age_hours, 1),
                "detail": "No validation run in the last 26 hours",
            },
        )

    if not run_data.get("success"):
        return JSONResponse(
            status_code=503,
            content={
                "status": "failed",
                "last_run": run_data["timestamp"],
                "failed_expectations": run_data.get("failed_expectations", "unknown"),
                "detail": "Last validation run had failures",
            },
        )

    return JSONResponse(
        status_code=200,
        content={
            "status": "ok",
            "last_run": run_data["timestamp"],
            "checkpoint": run_data["checkpoint"],
            "total_expectations": run_data.get("total_expectations"),
            "elapsed_seconds": run_data.get("elapsed_seconds"),
        },
    )

Persist each run result after it completes:

# In your runner script
import json
from pathlib import Path

summary = run_checkpoint_with_heartbeat(
    context_root_dir="/opt/gx",
    checkpoint_name="nightly_orders_checkpoint",
    heartbeat_url=os.environ.get("ORDERS_HEARTBEAT_URL"),
)

Path("/tmp/gx_last_run.json").write_text(json.dumps(summary))

Step 3: Set up Vigilmon monitoring

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

HTTP monitor for your validation service

1. Sign up at vigilmon.online — free tier, no credit card
2. Click New Monitor → HTTP
3. Enter https://your-gx-service.internal/health
4. Set interval to 10 minutes
5. Save

When the last validation run is stale or failed, your health endpoint returns HTTP 503 — Vigilmon detects this and opens an incident.

Heartbeat monitor for each checkpoint

For each GX checkpoint that must run on a schedule:

1. Click New Monitor → Heartbeat
2. Name it (e.g. "nightly_orders_checkpoint")
3. Set the expected interval — for a nightly job use 25 hours (gives 1-hour buffer)
4. Copy the unique ping URL
5. Add to your environment:

# .env
ORDERS_HEARTBEAT_URL=https://vigilmon.online/api/heartbeat/your-unique-token
CUSTOMERS_HEARTBEAT_URL=https://vigilmon.online/api/heartbeat/another-token
INVENTORY_HEARTBEAT_URL=https://vigilmon.online/api/heartbeat/third-token

Now update your runner calls:

# run_validations.py
import os
from gx_runner import run_checkpoint_with_heartbeat
import json
from pathlib import Path

CHECKPOINTS = [
    ("nightly_orders_checkpoint", os.environ.get("ORDERS_HEARTBEAT_URL")),
    ("nightly_customers_checkpoint", os.environ.get("CUSTOMERS_HEARTBEAT_URL")),
    ("nightly_inventory_checkpoint", os.environ.get("INVENTORY_HEARTBEAT_URL")),
]

for checkpoint_name, heartbeat_url in CHECKPOINTS:
    summary = run_checkpoint_with_heartbeat(
        context_root_dir="/opt/gx",
        checkpoint_name=checkpoint_name,
        heartbeat_url=heartbeat_url,
    )
    # Persist per-checkpoint result for the health endpoint
    log_path = Path(f"/tmp/gx_{checkpoint_name}_last_run.json")
    log_path.write_text(json.dumps(summary))
    print(f"{checkpoint_name}: {'OK' if summary['success'] else 'FAILED'} "
          f"({summary['failed_expectations']} failures in {summary['elapsed_seconds']}s)")

Step 4: Alert routing to Slack

Configure alert channels in Vigilmon so your data engineering team gets notified immediately:

1. Create a Slack incoming webhook in your workspace
2. In Vigilmon go to Notifications → New Channel → Slack
3. Paste the webhook URL
4. Name it "Data Engineering Alerts"
5. Enable it on all your GX monitors

Your Slack channel will receive:

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

🔴 DOWN: gx-service.internal/health
Status: 503 Service Unavailable — Last validation run had failures

And when things recover:

✅ RECOVERED: nightly_orders_checkpoint
✅ UP: gx-service.internal/health

Step 5: Integrate with your orchestrator

If you're using Airflow, Prefect, or Dagster, wrap the GX checkpoint call within the task and handle the heartbeat there:

# Airflow example
from airflow.decorators import task
from gx_runner import run_checkpoint_with_heartbeat
import os


@task
def validate_orders():
    summary = run_checkpoint_with_heartbeat(
        context_root_dir="/opt/gx",
        checkpoint_name="nightly_orders_checkpoint",
        heartbeat_url=os.environ.get("ORDERS_HEARTBEAT_URL"),
    )
    if not summary["success"]:
        raise ValueError(
            f"GX checkpoint failed: {summary['failed_expectations']} expectations failed"
        )
    return summary

The task raises on failure, which causes the Airflow task to fail (so your orchestrator alerts fire too). But crucially, the Vigilmon heartbeat only pings on success — so if the task is never scheduled, crashes before GX runs, or is stuck in a queue, you still get paged by Vigilmon independently of your orchestrator.

Step 6: Public status page for data stakeholders

When your data validation is broken, business stakeholders need to know — not just engineers.

In Vigilmon:

1. Go to Status Pages → New Status Page
2. Name it "Data Pipeline Health"
3. Add your GX checkpoint heartbeat monitors and the health endpoint monitor
4. Save and share the URL with your data consumers

The status page shows uptime history and current status without requiring Vigilmon login. When a validation is failing, stakeholders see it immediately rather than asking the data team "is the reporting data fresh?"

Monitoring matrix

| Monitor type | What it detects | |---|---| | HTTP monitor on /health | Service down, last run stale, last run had failures | | Heartbeat per checkpoint | Checkpoint skipped, orchestrator missed schedule, job crashed before completing | | Slack alerts | Instant notification to data engineering team | | Public status page | Visibility for data consumers and business stakeholders |

Summary

Great Expectations tells you when your data is bad. Vigilmon tells you when your data validation is broken. Together they cover both failure modes:

• GX expectation failures: caught by the validation runner and surfaced via the health endpoint
• Silent pipeline failures: caught by heartbeat monitors that alert when a ping goes missing

The setup takes under an hour and the monitoring runs free on Vigilmon's free tier.

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