Pipeline Monitoring and Observability

Pipeline Monitoring and Observability: Ensuring Reliability at Scale

Monitoring and observability are the eyes and ears of production data pipelines.

The Three Pillars of Observability

Core Pillars:

1. Metrics — numerical measurements over time (throughput, latency, error rate)
2. Logs — structured event records with context (what happened, when, why)
3. Traces — distributed execution paths across services (where time is spent)

For Data Pipelines:

• Metrics are the primary monitoring tool
• Logs provide debugging context
• Traces are essential for multi-service pipelines

Key Insight: Monitoring tells you what is happening; observability tells you why. A well-instrumented pipeline provides real-time visibility into health, performance, data quality, and business metrics.

Three Pillars of Observability

SLI / SLO / Error Budget

Architecture Diagram

Key Concepts

Production Code

Instrumented Pipeline with Prometheus Metrics

from prometheus_client import Counter, Histogram, Gauge, start_http_server
from datetime import datetime
import time
import logging
import json
from typing import Dict, Optional
from dataclasses import dataclass, asdict

logger = logging.getLogger(__name__)

# ------------------------------------------------------
# PROMETHEUS METRICS DEFINITIONS
# ------------------------------------------------------
RECORDS_PROCESSED = Counter(
    "pipeline_records_processed_total",
    "Total records processed",
    ["pipeline", "stage", "status"],
)

RECORDS_FAILED = Counter(
    "pipeline_records_failed_total",
    "Total records that failed processing",
    ["pipeline", "stage", "error_type"],
)

PIPELINE_DURATION = Histogram(
    "pipeline_stage_duration_seconds",
    "Duration of pipeline stage in seconds",
    ["pipeline", "stage"],
    buckets=[1, 5, 10, 30, 60, 120, 300, 600],
)

QUEUE_DEPTH = Gauge(
    "pipeline_queue_depth",
    "Current number of records waiting to be processed",
    ["pipeline", "stage"],
)

DATA_QUALITY_SCORE = Gauge(
    "pipeline_data_quality_score",
    "Data quality score (0-1)",
    ["pipeline", "dimension"],
)

FRESHNESS_SECONDS = Gauge(
    "pipeline_freshness_seconds",
    "Seconds since last successful pipeline run",
    ["pipeline"],
)


@dataclass
class PipelineMetrics:
    """Context manager for collecting pipeline metrics."""
    pipeline_name: str
    stage_name: str

    def __enter__(self):
        self.start_time = time.time()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        duration = time.time() - self.start_time
        PIPELINE_DURATION.labels(
            pipeline=self.pipeline_name,
            stage=self.stage_name,
        ).observe(duration)

        if exc_type is not None:
            RECORDS_FAILED.labels(
                pipeline=self.pipeline_name,
                stage=self.stage_name,
                error_type=exc_type.__name__,
            ).inc()
        else:
            RECORDS_PROCESSED.labels(
                pipeline=self.pipeline_name,
                stage=self.stage_name,
                status="success",
            ).inc()

    def record_batch(self, count: int, status: str = "success"):
        """Record a batch of processed records."""
        RECORDS_PROCESSED.labels(
            pipeline=self.pipeline_name,
            stage=self.stage_name,
            status=status,
        ).inc(count)


class InstrumentedPipeline:
    """Data pipeline with full Prometheus instrumentation."""

    def __init__(self, pipeline_name: str):
        self.pipeline_name = pipeline_name

    def extract(self, source: str) -> list:
        """Extract data from source with metrics."""
        with PipelineMetrics(self.pipeline_name, "extract"):
            logger.info(json.dumps({
                "event": "extract_started",
                "pipeline": self.pipeline_name,
                "source": source,
                "timestamp": datetime.utcnow().isoformat(),
            }))
            # Simulate extraction
            data = [{"id": i, "value": f"record_{i}"} for i in range(1000)]
            RECORDS_PROCESSED.labels(
                pipeline=self.pipeline_name,
                stage="extract",
                status="success",
            ).inc(len(data))
            return data

    def transform(self, data: list) -> list:
        """Transform data with metrics."""
        with PipelineMetrics(self.pipeline_name, "transform"):
            # Simulate transformation
            transformed = [{"id": d["id"], "value": d["value"].upper()} for d in data]
            RECORDS_PROCESSED.labels(
                pipeline=self.pipeline_name,
                stage="transform",
                status="success",
            ).inc(len(transformed))
            return transformed

    def validate(self, data: list) -> list:
        """Validate data quality with metrics."""
        with PipelineMetrics(self.pipeline_name, "validate"):
            valid = [d for d in data if d["value"] is not None]
            invalid_count = len(data) - len(valid)

            RECORDS_PROCESSED.labels(
                pipeline=self.pipeline_name,
                stage="validate",
                status="success",
            ).inc(len(valid))

            if invalid_count > 0:
                RECORDS_FAILED.labels(
                    pipeline=self.pipeline_name,
                    stage="validate",
                    error_type="validation_failure",
                ).inc(invalid_count)

            # Record quality metrics
            quality_score = len(valid) / len(data) if data else 1.0
            DATA_QUALITY_SCORE.labels(
                pipeline=self.pipeline_name,
                dimension="completeness",
            ).set(quality_score)

            return valid

    def load(self, data: list, target: str) -> None:
        """Load data with metrics."""
        with PipelineMetrics(self.pipeline_name, "load"):
            logger.info(json.dumps({
                "event": "load_completed",
                "pipeline": self.pipeline_name,
                "target": target,
                "record_count": len(data),
                "timestamp": datetime.utcnow().isoformat(),
            }))
            RECORDS_PROCESSED.labels(
                pipeline=self.pipeline_name,
                stage="load",
                status="success",
            ).inc(len(data))

    def run(self, source: str, target: str) -> Dict:
        """Run the full pipeline with metrics."""
        start_time = time.time()
        try:
            data = self.extract(source)
            transformed = self.transform(data)
            validated = self.validate(transformed)
            self.load(validated, target)

            duration = time.time() - start_time
            FRESHNESS_SECONDS.labels(pipeline=self.pipeline_name).set(0)

            return {
                "status": "success",
                "records": len(validated),
                "duration": duration,
            }
        except Exception as e:
            logger.error(json.dumps({
                "event": "pipeline_failed",
                "pipeline": self.pipeline_name,
                "error": str(e),
                "timestamp": datetime.utcnow().isoformat(),
            }))
            raise


# Start Prometheus metrics server
# start_http_server(8000)

# Run instrumented pipeline
pipeline = InstrumentedPipeline("daily_orders_etl")
result = pipeline.run(source="s3://raw/orders", target="s3://curated/orders")
logger.info(json.dumps({"event": "pipeline_result", **result}))

Structured Logging and Alert Configuration

import logging
import json
from datetime import datetime
from typing import Dict, Any
from pythonjsonlogger import jsonlogger


class StructuredFormatter(jsonlogger.JsonFormatter):
    """Custom JSON formatter for structured logging."""

    def add_fields(self, log_record: Dict[str, Any], record: logging.LogRecord, message_dict: Dict):
        super().add_fields(log_record, record, message_dict)
        log_record["timestamp"] = datetime.utcnow().isoformat()
        log_record["level"] = record.levelname
        log_record["logger"] = record.name
        log_record["module"] = record.module
        log_record["function"] = record.funcName
        log_record["line"] = record.lineno


def setup_logging():
    """Configure structured logging for pipeline observability."""
    logger = logging.getLogger("pipeline")
    logger.setLevel(logging.INFO)

    handler = logging.StreamHandler()
    handler.setFormatter(StructuredFormatter())
    logger.addHandler(handler)

    return logger


# Prometheus Alertmanager configuration (YAML)
ALERT_MANAGER_CONFIG = """
global:
  resolve_timeout: 5m

route:
  group_by: ['alertname', 'pipeline']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 4h
  receiver: 'slack-notifications'
  routes:
    - match:
        severity: critical
      receiver: 'pagerduty-critical'
    - match:
        severity: warning
      receiver: 'slack-warnings'

receivers:
  - name: 'slack-notifications'
    slack_configs:
      - api_url: '${SLACK_WEBHOOK_URL}'
        channel: '#data-alerts'
        title: '{{ .GroupLabels.alertname }}'
        text: '{{ .CommonAnnotations.summary }}'

  - name: 'pagerduty-critical'
    pagerduty_configs:
      - service_key: '${PAGERDuty_SERVICE_KEY}'
        severity: critical

  - name: 'slack-warnings'
    slack_configs:
      - api_url: '${SLACK_WEBHOOK_URL}'
        channel: '#data-warnings'
        title: '{{ .GroupLabels.alertname }}'
        text: '{{ .CommonAnnotations.summary }}'

inhibit_rules:
  - source_match:
      severity: 'critical'
    target_match:
      severity: 'warning'
    equal: ['alertname', 'pipeline']
"""

# Prometheus alert rules (YAML)
ALERT_RULES = """
groups:
  - name: pipeline_alerts
    rules:
      - alert: PipelineFailureRateHigh
        expr: |
          rate(pipeline_records_failed_total[5m])
          / rate(pipeline_records_processed_total[5m]) > 0.05
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Pipeline {{ $labels.pipeline }} failure rate > 5%"
          description: "Failure rate is {{ $value | humanizePercentage }}"

      - alert: PipelineFreshnessBreached
        expr: |
          pipeline_freshness_seconds > 7200
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Pipeline {{ $labels.pipeline }} freshness > 2 hours"
          description: "Last successful run was {{ $value | humanizeDuration }} ago"

      - alert: PipelineThroughputDegraded
        expr: |
          rate(pipeline_records_processed_total[10m])
          < 0.5 * avg_over_time(
            rate(pipeline_records_processed_total[10m])[7d:10m]
          )
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Pipeline {{ $labels.pipeline }} throughput degraded > 50%"

      - alert: DataQualityScoreLow
        expr: |
          pipeline_data_quality_score < 0.95
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Pipeline {{ $labels.pipeline }} quality score < 95%"
          description: "{{ $labels.dimension }} score is {{ $value }}"

      - alert: QueueDepthHigh
        expr: |
          pipeline_queue_depth > 10000
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Pipeline {{ $labels.pipeline }} queue depth > 10K records"
"""

Best Practices

1. Define SLIs and SLOs for every production pipeline. Track availability, latency, throughput, and freshness.
2. Use SLO-based alerts instead of threshold-based alerts to reduce noise and focus on outcomes.
3. Instrument every pipeline stage with Prometheus counters, histograms, and gauges for throughput, latency, and errors.
4. Use structured logging (JSON format) for all pipeline events. Include pipeline name, stage, and timestamp in every log line.
5. Build Grafana dashboards for: pipeline health overview, per-stage performance, data quality metrics, and SLA compliance.
6. Configure alert routing with Alertmanager: critical alerts to PagerDuty, warnings to Slack, info to email.
7. Create runbooks for each alert type with diagnostic commands, common causes, and escalation paths.
8. Implement distributed tracing with OpenTelemetry for multi-service pipelines to identify bottlenecks.
9. Review metrics weekly: analyze trends, adjust thresholds, and add new checks based on observed patterns.
10. Conduct blameless postmortems after incidents. Focus on systemic improvements, not individual blame.

Observability Stack Comparison

SLO Target Reference

See Also

• 028 - Error Handling, Retries, and Dead Letter Queues - Error handling and alerting
• 025 - Data Quality: Validation Frameworks - Quality monitoring metrics
• 026 - Data Pipeline Testing - Testing monitoring logic
• 017 - Apache Airflow: DAGs, Operators, and Scheduling - Monitoring Airflow pipelines
• 030 - Capstone Project: Real-Time Streaming Pipeline - Production monitoring implementation