data_pipeline_observability v1.0.0

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.