---
name: "data_pipeline_observability"
description: "Adds observability to data pipelines: freshness checks, volume anomaly detection, schema drift alerts, lineage-aware alert routing, and a runbook for silent failures. Triggers when an agent needs to monitor a data pipeline, add data-quality checks, or debug a silently broken data flow."
---

# Data Pipeline Observability

## Purpose

Catch pipeline failures before consumers notice. A pipeline that errors loudly is easy to fix. A pipeline that succeeds while writing stale, duplicated, or schema-shifted data is the one that quietly ruins dashboards and models. This skill sets up the checks that catch the quiet failures: freshness, volume, schema, and key integrity, with alerts routed through lineage so the right owner gets paged.

## Workflow

1. **Tier the tables.** Tier 1 tables feed dashboards, models, or money. Tier 2 feed internal analysis. Tier 3 are convenience. Only Tier 1/2 get all checks; Tier 3 gets freshness plus volume.
2. **Add the cheap checks first.** Freshness, row-count volume, not-null/unique on key columns, referential integrity on foreign keys. These catch the majority of real incidents before anything fancier.
3. **Pin schemas from data contracts.** Snapshot each table's expected columns and types. Compare on every run: additions are INFO, removals and type changes are BREAKING.
4. **Baseline before alerting.** Collect at least a couple weeks of history before trusting adaptive thresholds. Simple explainable rules beat ML for v1.
5. **Wire lineage to alerts.** Map each alert to the producing team and the at-risk consumers. Every alert links to the metric history and the affected table.
6. **Tune on a schedule.** Weekly: review fired alerts, re-baseline thresholds, escalate warnings that fire three times. Monthly: check alert precision (aim for under 30% noise). Monitor failures alert like pipeline failures.

## Output Contract

The bundled monitor (`bin/pipeline_monitor.py`) emits one JSON report per run:

```json
{
  "ran_at": "2026-10-01T21:37:00Z",
  "tables": {
    "orders": {
      "checks": {
        "freshness": {"status": "OK", "lag_hours": 1.2, "detail": "..."},
        "volume": {"status": "WARN", "z_score": 4.1, "detail": "..."},
        "schema": {"status": "OK", "detail": "matches pinned schema v3"},
        "keys": {"status": "ERROR", "detail": "order_id has 14 duplicates"}
      },
      "worst": "ERROR"
    }
  },
  "alerts": [
    {"severity": "ERROR", "table": "orders", "check": "keys",
     "owner": "payments", "consumers": ["rev_dashboard"], "detail": "..."}
  ]
}
```

Status levels: `OK`, `INFO` (compatible change, note only), `WARN` (review soon), `ERROR` (page the owner). Exit code 0 when worst is OK/INFO/WARN, 1 when any ERROR, 2 on config or database failure. Every run writes metrics to the history table `(table_name, measured_at, metric, value)` so the next run has a baseline.

## Operating Rules

1. Freshness windows come from the table's actual cadence: warn after the load is late, error after it is badly late. Never guess a window, read it from the orchestrator schedule or the contract.
2. Volume thresholds are adaptive (z-score against trailing 30-day mean), not static counts. Business cycles make static counts noisy.
3. Schema removals and type changes are never auto-accepted. Quarantine or fail the pipeline until someone decides.
4. An alert without lineage context is a page nobody can act on. Include producer, consumers, run id, and metric history link.
5. The monitor is part of the pipeline, not a sidecar. If the monitor job itself fails or stops writing metrics, that pages exactly like a pipeline failure.
6. Every factual claim in a skill must be verified via web search or a primary source. Never invent numbers, dates, quotes, or events. Unverifiable = cut it or frame as opinion. No confident unverified claims, ever.

## Quickstart

```bash
# 1. Pin a schema from the live table
python3 bin/pipeline_monitor.py --pin-schema --database warehouse.sqlite \
  --table orders --schema-file schemas/orders.schema.json

# 2. Seed metric history, then check
python3 bin/pipeline_monitor.py --config checks.json --report report.json
cat report.json
```

Minimal `checks.json`:

```json
{
  "database": "warehouse.sqlite",
  "metrics_table": "pipeline_metrics",
  "owners": {"orders": "payments"},
  "consumers": {"orders": ["rev_dashboard"]},
  "tables": [
    {"name": "orders", "tier": 1, "timestamp_column": "loaded_at",
     "warn_after_hours": 2, "error_after_hours": 4,
     "key_columns": ["order_id"], "schema_file": "schemas/orders.schema.json"}
  ]
}
```

For the silent-failure runbook (dashboard looks fine but numbers are wrong, jobs succeed on zero rows, upstream formats changed without warning), read `references/silent-failure-runbook.md`.