Monitoring and Observability for Data Pipelines

SLI (measured):    pipeline completion time, data freshness, error rate, DLQ rate
SLO (internal target): "Silver orders completes within 90 min of scheduled start"
                        "Gold daily_revenue is no older than 2 hours"
                        "Error rate < 1% over any 7-day rolling window"
SLA (external promise): "Finance dashboards have yesterday's data by 08:00 ET"
                         "Any data correction is available within 4 hours"

ERROR BUDGET: SLO "99% of runs complete within 90 min", 6-hourly pipeline:
  Monthly runs: 30 × 4 = 120.  Allowed misses: 120 × 1% ≈ 1.2 runs/month.
  When the budget is exhausted: stop new features, focus on reliability.

Pipeline                SLI                 SLO             SLA
silver_orders_daily     completion_time      < 60 min        —
gold_daily_revenue      data_freshness       < 2h            data by 08:00 ET
ml_feature_store        completion_time      < 30 min        complete by 06:00 ET

CREATE TABLE monitoring.pipeline_slo_tracking (
    run_id UUID NOT NULL, pipeline_name VARCHAR(100) NOT NULL,
    scheduled_start TIMESTAMPTZ NOT NULL, actual_start TIMESTAMPTZ, actual_end TIMESTAMPTZ,
    slo_target_min INT NOT NULL, actual_duration_min DECIMAL(8,2), met_slo BOOLEAN,
    sla_deadline TIMESTAMPTZ, met_sla BOOLEAN, status VARCHAR(20) NOT NULL,
    rows_processed BIGINT, rows_rejected BIGINT, recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Daily SLO report
SELECT pipeline_name, COUNT(*) total_runs,
    ROUND(SUM(CASE WHEN met_slo THEN 1 ELSE 0 END)::NUMERIC / COUNT(*) * 100, 1) slo_met_pct,
    ROUND(PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY actual_duration_min), 1) p95_duration_min
FROM monitoring.pipeline_slo_tracking
WHERE scheduled_start >= CURRENT_DATE - 30 GROUP BY pipeline_name ORDER BY slo_met_pct ASC;

-- SLA breach history (the ones that matter most)
SELECT pipeline_name, scheduled_start, sla_deadline, actual_end - sla_deadline breach_duration
FROM monitoring.pipeline_slo_tracking
WHERE met_sla = FALSE AND sla_deadline IS NOT NULL AND scheduled_start >= CURRENT_DATE - 30
ORDER BY scheduled_start DESC;

pipeline_name       total_runs  slo_met_pct  p95_duration_min
silver_orders_daily 30          96.7         64.2
gold_daily_revenue  30          100.0        18.5   ← healthiest pipeline this month
ml_feature_store    30          83.3         41.8   ← worth investigating

import json, logging
from datetime import datetime, timezone
from typing import Any

class PipelineLogger:
    def __init__(self, pipeline_name: str, run_id: str):
        self.pipeline_name, self.run_id = pipeline_name, run_id

    def _emit(self, level: str, event: str, **kwargs: Any) -> None:
        entry = {'timestamp': datetime.now(timezone.utc).isoformat(), 'level': level,
                 'event': event, 'pipeline': self.pipeline_name, 'run_id': self.run_id, **kwargs}
        print(json.dumps(entry), flush=True)   # stdout → log aggregator

    def info(self, event: str, **kwargs):    self._emit('INFO', event, **kwargs)
    def warning(self, event: str, **kwargs): self._emit('WARNING', event, **kwargs)
    def error(self, event: str, **kwargs):   self._emit('ERROR', event, **kwargs)

def run_silver_pipeline(run_date: str) -> dict:
    run_id = str(uuid4())
    log = PipelineLogger('silver_orders', run_id)
    log.info('pipeline_started', run_date=run_date, trigger='scheduled')

    try:
        rows = extract_from_bronze(run_date)
        log.info('extract_complete', stage='extract', rows_extracted=len(rows), source='bronze.orders')

        valid, rejected = validate_rows(rows)
        if rejected:
            log.warning('validation_rejections', stage='validate', rejected_count=len(rejected),
                        rejection_rate=round(len(rejected) / len(rows), 4))
            write_to_dlq(rejected, run_id)

        rows_written = load_to_silver(valid, run_date)
        log.info('load_complete', stage='load', rows_written=rows_written, target='silver.orders')
        log.info('pipeline_complete', status='success', rows_written=rows_written)
        return {'status': 'success', 'rows_written': rows_written}

    except Exception as exc:
        log.error('pipeline_failed', error_type=type(exc).__name__, error_message=str(exc))
        raise

{"timestamp": "2026-03-17T06:14:32.847Z", "level": "INFO", "event": "extract_complete",
 "pipeline": "silver_orders", "run_id": "d7c7a7b8-...", "stage": "extract",
 "rows_extracted": 48234, "source": "bronze.orders"}

-- CloudWatch Insights: runs with > 5% rejection rate, last 7 days
fields @timestamp, pipeline, run_id, rejection_rate
| filter event = "pipeline_complete" and rejection_rate > 0.05
| sort @timestamp desc | limit 20

from uuid import uuid4

def generate_run_context(**context):
    """Generate correlation ID and push to XCom for all downstream tasks."""
    context['ti'].xcom_push(key='correlation_id', value=str(uuid4()))

def run_bronze_extraction(**context):
    correlation_id = context['ti'].xcom_pull(task_ids='generate_run_context', key='correlation_id')
    log = PipelineLogger('bronze_orders', run_id=correlation_id)
    log.info('extraction_started', stage='bronze')

def run_silver_transform(**context):
    correlation_id = context['ti'].xcom_pull(task_ids='generate_run_context', key='correlation_id')
    log = PipelineLogger('silver_orders', run_id=correlation_id)
    log.info('transform_started', stage='silver')   # same correlation_id — linkable to Bronze

-- search ONE id, see the whole run across Bronze → Silver → Gold:
fields @timestamp, event, stage, rows_extracted, rows_rejected, error_message
| filter run_id = "d7c7a7b8-3e1a-4a2c-9b4d-..."
| sort @timestamp asc
-- incident investigation: minutes, not hours of cross-log searching

BAD:  Title: silver_orders FAILED
      Body:  Pipeline silver_orders failed at 06:14:32.
      → engineer at 2 AM has no idea what to do next

GOOD: Title: [P2] silver_orders — FAILED — 2026-03-17 06:14 ET
      Failed at: validation stage (step 2 of 4)
      Error:     48,234 rows rejected — unrecognised status 'scheduled'
      Impact:    Gold daily_revenue build blocked. Finance dashboard will be stale.
      SLA:       Gold must be ready by 08:00 ET (1h 45m remaining)
      DLQ:       48,234 rows → pipeline/dlq_reprocess.py
      Runbook:   https://runbooks.freshcart.internal/silver-orders-failure

def format_alert_message(pipeline: str, run_date: str, stage: str, error: str,
                          impact: str, sla_time: str | None, run_id: str, runbook_url: str) -> str:
    time_to_sla = compute_time_to_sla(sla_time) if sla_time else None
    return f"""
Pipeline:  {pipeline}
Failed at: {stage}
Error:     {error}
Impact:    {impact}
{f"SLA:    {sla_time} ({time_to_sla} remaining)" if sla_time else ""}
Run ID:    {run_id}
Runbook:   {runbook_url}
""".strip()

def on_failure_callback(context):
    dag_id, task_id = context['dag'].dag_id, context['task_instance'].task_id
    priority = determine_alert_priority(dag_id, task_id, context)
    message = format_alert_message(
        pipeline=f'{dag_id}.{task_id}', run_date=context['ds'], stage=task_id,
        error=str(context.get('exception', 'unknown error')),
        impact=get_downstream_impact(dag_id, task_id), sla_time=get_sla_for_pipeline(dag_id),
        run_id=context['run_id'], runbook_url=f'https://runbooks.freshcart.internal/{dag_id}',
    )
    if priority == 'P1':
        send_pagerduty_alert(message, severity='critical')
        send_slack_alert('#incidents', message)
    elif priority == 'P2':
        send_slack_alert('#data-alerts', message)
    else:
        send_slack_alert('#data-warnings', message)

CREATE TABLE monitoring.pipeline_runs (
    run_id UUID NOT NULL PRIMARY KEY, pipeline_name VARCHAR(100) NOT NULL, dag_id VARCHAR(100),
    run_date DATE NOT NULL, scheduled_at TIMESTAMPTZ NOT NULL, started_at TIMESTAMPTZ, completed_at TIMESTAMPTZ,
    status VARCHAR(20) NOT NULL,   -- running/success/failed/skipped
    rows_extracted BIGINT, rows_rejected BIGINT, rows_written BIGINT, duration_sec DECIMAL(10,2),
    slo_target_sec INT, met_slo BOOLEAN, sla_deadline TIMESTAMPTZ, met_sla BOOLEAN, error_message TEXT
);

WITH latest_runs AS (
    SELECT DISTINCT ON (pipeline_name) pipeline_name, status, completed_at, met_sla, error_message
    FROM monitoring.pipeline_runs WHERE run_date = CURRENT_DATE
    ORDER BY pipeline_name, started_at DESC
)
SELECT pipeline_name,
    CASE WHEN status = 'success' AND met_sla THEN '✅ OK'
         WHEN status = 'success' AND NOT COALESCE(met_sla, TRUE) THEN '⚠️ SLA MISSED'
         WHEN status = 'running' THEN '🔄 RUNNING'
         WHEN status = 'failed' THEN '🔴 FAILED' ELSE '⏳ PENDING' END AS health_indicator,
    error_message
FROM latest_runs ORDER BY CASE status WHEN 'failed' THEN 0 WHEN 'running' THEN 1 ELSE 2 END;

pipeline_name          health_indicator   error_message
silver_orders_daily    🔴 FAILED          accepted_values: status 'scheduled' not in list
gold_daily_revenue     ✅ OK              (null)
ml_feature_store       ✅ OK              (null)

SELECT run_date, pipeline_name, COUNT(*) runs,
    ROUND(SUM(CASE WHEN met_slo THEN 1 ELSE 0 END)::NUMERIC / NULLIF(COUNT(*), 0) * 100, 1) slo_pct
FROM monitoring.pipeline_runs
WHERE run_date >= CURRENT_DATE - 7 AND status IN ('success', 'failed')
GROUP BY 1, 2 ORDER BY 1 DESC, 2;

SELECT pipeline_name, run_date, status, error_message,
    EXTRACT(EPOCH FROM (NOW() - started_at)) / 3600 hours_since_start
FROM monitoring.pipeline_runs
WHERE status IN ('failed', 'running') AND run_date >= CURRENT_DATE - 2
ORDER BY started_at;

SELECT pipeline_name, error_type, COUNT(*) pending_count,
    EXTRACT(EPOCH FROM (NOW() - MIN(arrived_at))) / 3600 hours_pending
FROM pipeline.dead_letter_queue WHERE status = 'pending'
GROUP BY pipeline_name, error_type ORDER BY pending_count DESC;

-- alert: same error type, > 1000 pending, older than 2 hours
SELECT pipeline_name, error_type, COUNT(*) depth
FROM pipeline.dead_letter_queue
WHERE status = 'pending' AND arrived_at < NOW() - INTERVAL '2 hours'
GROUP BY pipeline_name, error_type HAVING COUNT(*) > 1000 ORDER BY depth DESC;

pipeline_name    error_type              depth
silver_orders    accepted_values_status  48234   ← the exact incident from this
                                                    module's Real World section

def reprocess_dlq(pipeline_name: str, error_type: str, run_date: str, dry_run: bool = False) -> dict:
    records = fetch_pending_dlq_records(pipeline_name=pipeline_name, error_type=error_type, run_date=run_date)
    if not records:
        return {'status': 'no_records', 'count': 0}
    if dry_run:
        return {'status': 'dry_run', 'would_reprocess': len(records)}

    processed, failed = 0, 0
    for record in records:
        try:
            result = reprocess_single_record(record, pipeline_name)
            mark_dlq_resolved(record['dlq_id'], note=f'Reprocessed successfully. Row: {result}')
            processed += 1
        except Exception as exc:
            mark_dlq_failed(record['dlq_id'], note=str(exc))
            failed += 1
    return {'status': 'complete', 'processed': processed, 'failed': failed}

def check_dlq_health(**context):
    stats = query_dlq_stats(pipeline_name='silver_orders', run_date=context['ds'])
    if stats.pending_records > 10_000:
        raise ValueError(f"DLQ depth critical: {stats.pending_records} pending. "
                          f"Top error: {stats.top_error_type} ({stats.top_error_count} records).")
    elif stats.pending_records > 1_000:
        send_slack_warning(f"DLQ depth elevated: {stats.pending_records} pending. "
                            f"Top error: {stats.top_error_type}.")

COUNTER (always increasing):   pipeline.runs.total{status="success"}
                                pipeline.rows.rejected{pipeline="silver_orders"}
GAUGE (current value):         pipeline.dlq.depth{pipeline="silver_orders"}
                                pipeline.data_freshness_sec{table="silver.orders"}
HISTOGRAM (distribution):      pipeline.run_duration_seconds{pipeline="silver_orders"}

from datadog import DogStatsd
statsd = DogStatsd(host='localhost', port=8125)

def emit_pipeline_metrics(pipeline_name: str, status: str, duration_sec: float,
                           rows_extracted: int, rows_rejected: int) -> None:
    tags = [f'pipeline:{pipeline_name}']
    statsd.increment('pipeline.runs.total', tags=tags + [f'status:{status}'])
    statsd.histogram('pipeline.run_duration_seconds', duration_sec, tags=tags)
    if rows_extracted > 0:
        statsd.gauge('pipeline.rejection_rate', rows_rejected / rows_extracted, tags=tags)

import boto3
cloudwatch = boto3.client('cloudwatch')

def emit_to_cloudwatch(pipeline_name: str, rows_rejected: int, run_date: str) -> None:
    cloudwatch.put_metric_data(
        Namespace='FreshCart/DataPipelines',
        MetricData=[{'MetricName': 'RowsRejected',
                      'Dimensions': [{'Name': 'PipelineName', 'Value': pipeline_name}],
                      'Value': rows_rejected, 'Unit': 'Count'}],
    )
    # Alarm rule: RowsRejected > 10000 for ANY pipeline, 1 evaluation period

GRAFANA DASHBOARD PANELS:
1. Pipeline status grid (colored tile per pipeline)   5. DLQ depth (time series)
2. SLO compliance (30-day trend line)                 6. Warehouse credit usage
3. Daily row counts (stacked bar)                     7. Error rate vs SLO line
4. Run duration P95 vs SLO target                     8. Recent failures (table + log links)

The data engineering team has grown to 8 people, serving finance, operations, and product. The pipeline occasionally fails at night or on weekends and nobody finds out until Monday. The team needs a sustainable on-call process that doesn’t burn engineers out.

Runbooks — so the on-call engineer isn’t starting from zero

## silver_orders Runbook
Runs daily at 06:00 ET. SLA: complete by 07:30 ET. Owner: data-platform@freshcart.com

**Failure 1: accepted_values test fails for 'status' column**
Cause: Orders team added a new status value.
Fix:
  1. python dlq_reprocess.py --dry-run --pipeline silver_orders --date {DATE}
  2. Add new status to VALID_STATUSES in pipeline/validate.py
  3. dbt run -s silver_orders && dbt test -s silver_orders
  4. python dlq_reprocess.py --pipeline silver_orders --date {DATE}
Time to fix: 30 minutes.

**Failure 2: source freshness check fails (Bronze > 6 hours old)**
Fix: Check silver_ingestion Airflow DAG, trigger a manual run.
Escalate to: Marcus if ingestion issue persists > 2 hours.

Rotation, and the guardrails that prevent burnout

Week-long rotation, PagerDuty schedule: each engineer on-call once every 8 weeks.
Responsibilities: P1 within 5 min (any hour), P2 within 1 hour (business hours),
                  post-mortem for any P1 or repeated P2.

GUARDRAILS:
  Max 2 P1 pages per night — otherwise the process itself is broken.
  P3/P4 pages that wake someone → the threshold is wrong, fix it, don't just endure it.
  On-call engineer has zero feature work that week (protection time).

A real post-mortem, and the calibration loop that follows it

## Incident: silver_orders missed SLA — 2026-03-17
Duration: 06:00–09:15 ET (SLA breached at 07:30). Severity: P1.

Timeline:
  06:14 — pipeline failed: accepted_values error on status='scheduled'
  06:15 — P2 alert fired (should have escalated to P1 sooner)
  07:32 — on-call acknowledged after SLA breach escalation
  08:12 — fix deployed, DLQ reprocessed → 09:15 Gold rebuilt

Root cause: orders team deployed a new status enum without notifying data team.
Action items:
  [ ] Add data contract CI check for enum changes (owner: Emily, by 2026-03-31)
  [ ] Escalate silver_orders failures to P1 if SLA is within 1 hour

Monthly alert calibration review:
Target: 1-2 P1/P2 alerts per on-call week.
20+ alerts/week  → alert fatigue, raise thresholds.
0 alerts for 4 weeks, but incidents found later → too quiet, lower thresholds.
False positive rate (alerts needing no action / total alerts): target < 20%.