tutorial

Monitoring OpenLineage Data Lineage Pipelines with Vigilmon

OpenLineage is the open standard for data lineage metadata collection — but when lineage events stop flowing, your data catalog goes blind, broken pipelines stay invisible, and compliance teams lose the audit trail they depend on. Here's how to monitor OpenLineage event emission, backend ingestion, lineage coverage, and transport health with Vigilmon.

OpenLineage is an open standard and framework for collecting data lineage metadata across pipelines, jobs, and datasets. It defines a specification for job-level start, complete, and fail events — and integrates with Apache Spark, Airflow, dbt, Flink, and dozens of other data tools via facets. When your OpenLineage-instrumented pipelines run, they emit lineage events to a backend (Marquez, Atlan, OpenMetadata, or a custom HTTP endpoint). When event emission stops — due to a broken transport, a job that skips lineage instrumentation, or a backend that stops accepting events — your data catalog silently goes stale, data governance teams lose visibility, and compliance audit trails become incomplete. Vigilmon gives you external monitoring for OpenLineage transport health, event emission rates, backend ingestion, and lineage coverage across your data platform.

What You'll Set Up

  • OpenLineage backend HTTP endpoint availability
  • Event emission rate monitoring from instrumented pipelines
  • Lineage backend ingestion lag detection
  • Transport (HTTP/Kafka) health checks
  • Coverage monitoring for critical datasets

Prerequisites

  • OpenLineage-instrumented pipelines (Spark, Airflow, dbt, or custom)
  • OpenLineage backend (Marquez, Atlan, or custom HTTP endpoint)
  • Access to the backend's REST API or event queue
  • A free Vigilmon account

Step 1: Monitor the OpenLineage Backend HTTP Endpoint

OpenLineage events are delivered via HTTP POST to a backend endpoint (default path: /api/v1/lineage). If the backend goes down — OOM kill, failed deployment, or disk full on the metadata database — all instrumented pipelines continue running but events are silently dropped or queued until the transport retries are exhausted. Monitor the backend endpoint first:

  1. In Vigilmon, click Add Monitor → HTTP(S).
  2. Set the URL to your backend health endpoint (e.g., https://lineage.internal/api/v1/namespaces).
  3. Set Check interval to 1 minute.
  4. Set Expected status to 200.

For Marquez specifically:

# Verify Marquez is responding and listing namespaces
curl -s https://lineage.internal/api/v1/namespaces | python3 -c "
import sys, json
data = json.load(sys.stdin)
namespaces = data.get('namespaces', [])
print(f'Marquez OK: {len(namespaces)} namespaces')
sys.exit(0 if namespaces is not None else 1)
"

For a custom OpenLineage HTTP backend, add a heartbeat endpoint to your ingestion service:

# In your FastAPI / Flask lineage backend
from fastapi import FastAPI
from datetime import datetime

app = FastAPI()

last_event_received = None

@app.post("/api/v1/lineage")
async def receive_lineage_event(event: dict):
    global last_event_received
    last_event_received = datetime.utcnow()
    # Process and store the event...
    return {"status": "ok"}

@app.get("/health")
async def health():
    return {
        "status": "ok",
        "last_event_received": last_event_received.isoformat() if last_event_received else None,
    }

Point Vigilmon's HTTP monitor at /health. A 200 from the health endpoint proves the backend process is alive and ready to receive events.

Step 2: Monitor Event Emission Rate from Pipelines

OpenLineage-instrumented pipelines should emit a predictable number of events per day based on your pipeline schedule. If a platform upgrade silently disables the OpenLineage integration, a Spark job no longer has the openlineage-spark JAR on its classpath, or an Airflow integration breaks after an operator upgrade, events stop arriving without any visible error in the pipeline itself. Monitor emission rate to catch integration failures:

#!/bin/bash
# /usr/local/bin/openlineage-emission-check.sh

HEARTBEAT_URL="https://vigilmon.online/heartbeat/abc123"
MARQUEZ_URL="https://lineage.internal"
NAMESPACE="production"
MIN_EVENTS_PER_HOUR=50    # Expect at least 50 lineage events per hour during business hours
LOOKBACK_MINUTES=60
TIMEOUT=15

# Query Marquez for jobs updated in the last hour
RECENT_JOBS=$(curl -s \
  --max-time "$TIMEOUT" \
  "${MARQUEZ_URL}/api/v1/namespaces/${NAMESPACE}/jobs?limit=200&offset=0" | \
  python3 -c "
import sys, json
from datetime import datetime, timezone, timedelta

data = json.load(sys.stdin)
jobs = data.get('jobs', [])
cutoff = datetime.now(timezone.utc) - timedelta(minutes=60)
recent = [j for j in jobs if j.get('updatedAt') and
          datetime.fromisoformat(j['updatedAt'].replace('Z', '+00:00')) > cutoff]
print(len(recent))
" 2>/dev/null || echo 0)

RECENT_JOBS="${RECENT_JOBS:-0}"

HOUR=$(date +%H)
# Only enforce minimum during business hours (8am-8pm)
if [ "$HOUR" -ge 8 ] && [ "$HOUR" -le 20 ]; then
  if [ "$RECENT_JOBS" -ge "$MIN_EVENTS_PER_HOUR" ]; then
    echo "Lineage emission OK: ${RECENT_JOBS} jobs updated in last hour"
    curl -s "$HEARTBEAT_URL"
  else
    echo "Lineage emission low: only ${RECENT_JOBS} jobs updated in last hour (min: ${MIN_EVENTS_PER_HOUR})"
    exit 1
  fi
else
  # Off-hours: just ping to confirm backend reachability
  curl -s "$HEARTBEAT_URL"
fi

Set the Vigilmon heartbeat to 70 minutes. A drop to zero events during business hours — when dozens of Airflow DAGs and Spark jobs run continuously — almost certainly indicates an integration breakage rather than a quiet period.

Step 3: Monitor Lineage Event Ingestion Lag

OpenLineage backends process incoming events and store them in a metadata graph. When the backend falls behind under load — a high-volume Spark job emitting thousands of facets, a slow database write path, or an under-resourced Kafka consumer — the lineage graph shows events from hours ago while pipelines are actively running. Monitor ingestion lag to catch backend processing delays before they affect data governance workflows:

#!/bin/bash
# /usr/local/bin/openlineage-lag-check.sh

HEARTBEAT_URL="https://vigilmon.online/heartbeat/def456"
MARQUEZ_URL="https://lineage.internal"
NAMESPACE="production"
MAX_LAG_MINUTES=10   # Alert if most recent lineage event is older than 10 minutes
TIMEOUT=15

# Get the most recently updated job run in the namespace
LATEST_UPDATE=$(curl -s \
  --max-time "$TIMEOUT" \
  "${MARQUEZ_URL}/api/v1/namespaces/${NAMESPACE}/runs?limit=10" | \
  python3 -c "
import sys, json
data = json.load(sys.stdin)
runs = data.get('runs', [])
if not runs:
    print('')
    sys.exit(0)
# Find most recent updatedAt
latest = max(runs, key=lambda r: r.get('updatedAt', ''), default=None)
print(latest['updatedAt'] if latest else '')
" 2>/dev/null)

if [ -z "$LATEST_UPDATE" ]; then
  echo "No recent runs found in namespace $NAMESPACE"
  exit 1
fi

LAG_MINUTES=$(python3 -c "
from datetime import datetime, timezone
latest = datetime.fromisoformat('$LATEST_UPDATE'.replace('Z', '+00:00'))
lag = (datetime.now(timezone.utc) - latest).total_seconds() / 60
print(int(lag))
" 2>/dev/null || echo 999)

if [ "$LAG_MINUTES" -le "$MAX_LAG_MINUTES" ]; then
  echo "Lineage lag OK: most recent event ${LAG_MINUTES} min ago"
  curl -s "$HEARTBEAT_URL"
else
  echo "Lineage lag high: most recent event ${LAG_MINUTES} min ago (max: ${MAX_LAG_MINUTES} min)"
  exit 1
fi

For Kafka-based OpenLineage transports, monitor consumer lag directly:

#!/bin/bash
HEARTBEAT_URL="https://vigilmon.online/heartbeat/def456"
KAFKA_BOOTSTRAP="kafka.internal:9092"
CONSUMER_GROUP="openlineage-backend-consumer"
TOPIC="openlineage.events"
MAX_LAG=1000   # Alert if consumer is more than 1000 messages behind

CONSUMER_LAG=$(kafka-consumer-groups.sh \
  --bootstrap-server "$KAFKA_BOOTSTRAP" \
  --describe \
  --group "$CONSUMER_GROUP" 2>/dev/null | \
  awk -v topic="$TOPIC" '$1==topic {sum+=$6} END {print sum+0}')

if [ "${CONSUMER_LAG:-0}" -le "$MAX_LAG" ]; then
  echo "Kafka consumer lag OK: ${CONSUMER_LAG} messages behind"
  curl -s "$HEARTBEAT_URL"
else
  echo "Kafka consumer lag high: ${CONSUMER_LAG} messages (max: ${MAX_LAG})"
  exit 1
fi

Set the Vigilmon heartbeat to 15 minutes. Persistent ingestion lag means your lineage backend is processing older events while your data catalog shows an increasingly stale picture of your pipelines.

Step 4: Verify Lineage Coverage for Critical Datasets

Not all pipelines in your platform may have OpenLineage instrumentation — older Spark jobs, bespoke Python scripts, or recently onboarded teams may produce datasets that are invisible to the lineage graph. Verify that your most critical datasets (feeding production dashboards, regulatory reports, or ML models) have active lineage coverage:

#!/bin/bash
# /usr/local/bin/openlineage-coverage-check.sh

HEARTBEAT_URL="https://vigilmon.online/heartbeat/ghi789"
MARQUEZ_URL="https://lineage.internal"
NAMESPACE="production"
TIMEOUT=15
MAX_DATASET_AGE_HOURS=26   # Allow up to 26 hours for daily datasets

# Critical datasets that must have active lineage
CRITICAL_DATASETS=(
  "postgres://prod.db/analytics.user_features"
  "s3://data-lake/gold/revenue_daily.parquet"
  "bigquery://project/dataset.model_training_features"
)

FAILED=0

for DATASET in "${CRITICAL_DATASETS[@]}"; do
  ENCODED=$(python3 -c "import urllib.parse; print(urllib.parse.quote('$DATASET', safe=''))")

  DATASET_INFO=$(curl -s \
    --max-time "$TIMEOUT" \
    "${MARQUEZ_URL}/api/v1/namespaces/${NAMESPACE}/datasets/${ENCODED}")

  LAST_MODIFIED=$(echo "$DATASET_INFO" | python3 -c "
import sys, json
data = json.load(sys.stdin)
print(data.get('updatedAt', ''))
" 2>/dev/null)

  if [ -z "$LAST_MODIFIED" ]; then
    echo "MISSING coverage for critical dataset: $DATASET"
    FAILED=1
    continue
  fi

  AGE_HOURS=$(python3 -c "
from datetime import datetime, timezone
lm = datetime.fromisoformat('$LAST_MODIFIED'.replace('Z', '+00:00'))
age = (datetime.now(timezone.utc) - lm).total_seconds() / 3600
print(int(age))
" 2>/dev/null || echo 999)

  if [ "$AGE_HOURS" -le "$MAX_DATASET_AGE_HOURS" ]; then
    echo "Coverage OK: $DATASET (${AGE_HOURS}h ago)"
  else
    echo "Coverage STALE: $DATASET last seen ${AGE_HOURS}h ago (max: ${MAX_DATASET_AGE_HOURS}h)"
    FAILED=1
  fi
done

if [ "$FAILED" -eq 0 ]; then
  curl -s "$HEARTBEAT_URL"
else
  exit 1
fi

Run this every 30 minutes with a Vigilmon heartbeat at 35 minutes. Coverage gaps on critical datasets indicate either an instrumentation regression or a pipeline that moved to a new framework without carrying lineage collection forward.

Step 5: Monitor the OpenLineage HTTP Transport

When OpenLineage events are emitted via HTTP (not Kafka), the transport is a synchronous call from within the pipeline process. If the backend is unreachable, the default behavior in most integrations is to log a warning and silently drop the event — the pipeline job succeeds but lineage is lost. Add transport-level monitoring to detect silent drops:

# custom_openlineage_transport.py
# Wrap the default HTTP transport to track delivery failures

import requests
import logging
from openlineage.client.transport import HttpTransport, HttpConfig
from datetime import datetime

logger = logging.getLogger(__name__)

VIGILMON_HEARTBEAT = "https://vigilmon.online/heartbeat/jkl012"
FAILURE_COUNT_FILE = "/var/log/openlineage-transport-failures.txt"

class MonitoredHttpTransport(HttpTransport):
    def emit(self, event):
        try:
            super().emit(event)
            # Reset failure count on success
            with open(FAILURE_COUNT_FILE, "w") as f:
                f.write("0")
        except Exception as e:
            # Increment failure count
            try:
                with open(FAILURE_COUNT_FILE, "r") as f:
                    count = int(f.read().strip() or "0")
            except FileNotFoundError:
                count = 0
            count += 1
            with open(FAILURE_COUNT_FILE, "w") as f:
                f.write(str(count))
            logger.error(f"OpenLineage transport failed (consecutive: {count}): {e}")
            if count >= 3:
                logger.critical(f"OpenLineage backend unreachable for {count} consecutive events!")
            raise

Check delivery failures from a separate cron:

#!/bin/bash
HEARTBEAT_URL="https://vigilmon.online/heartbeat/jkl012"
FAILURE_FILE="/var/log/openlineage-transport-failures.txt"
MAX_CONSECUTIVE_FAILURES=3

FAILURES=$(cat "$FAILURE_FILE" 2>/dev/null || echo "0")

if [ "${FAILURES:-0}" -le "$MAX_CONSECUTIVE_FAILURES" ]; then
  echo "Transport OK: ${FAILURES} consecutive delivery failures"
  curl -s "$HEARTBEAT_URL"
else
  echo "Transport failing: ${FAILURES} consecutive delivery failures"
  exit 1
fi

Set the heartbeat to 5 minutes. Consecutive delivery failures that aren't being retried indicate the backend is down or the network path from the pipeline host to the backend is broken.

Step 6: Set Up Alert Channels

Configure Vigilmon to notify the data platform team:

  1. Go to Alert Channels in Vigilmon.
  2. Add a Slack webhook to #data-platform-alerts — lineage gaps are data governance incidents, not just ops.
  3. Add PagerDuty for the backend availability and transport health monitors — a down lineage backend affects all downstream catalog users.
  4. Add email notifications for coverage and lag monitors — these degrade silently over hours rather than failing instantaneously.
  5. Set Consecutive failures before alert to 2 for the emission rate monitor — a brief gap between pipeline batches is normal.

Add maintenance windows when upgrading the lineage backend or the OpenLineage integration libraries:

curl -X POST https://vigilmon.online/api/maintenance \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "monitor_id": "OPENLINEAGE_BACKEND_MONITOR_ID",
    "duration_minutes": 30,
    "reason": "Marquez backend upgrade to 0.45.0"
  }'

Summary

| Monitor | Target | What It Catches | |---|---|---| | HTTP monitor (backend) | /api/v1/namespaces or /health | Backend down, failed deployment, OOM kill | | Cron heartbeat (emission rate) | Marquez job update count | Broken Spark/Airflow integration, missing JAR | | Cron heartbeat (ingestion lag) | Most recent run updatedAt | Backend processing backlog, Kafka lag | | Cron heartbeat (coverage) | Critical dataset lineage age | Missing instrumentation, pipeline migration | | Cron heartbeat (transport) | Consecutive delivery failures | Network partition, backend temporarily unreachable |

OpenLineage's silent failure mode — pipelines succeed, lineage is dropped, the catalog shows a stale graph — is the most dangerous aspect of lineage infrastructure. Vigilmon's external monitoring closes that visibility gap: you know when lineage events stop flowing, when coverage degrades, and when the backend falls behind, all before your governance team discovers an audit trail hole.

Start monitoring for free at vigilmon.online.

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