Pipeline Observability: Prometheus Metrics, Grafana Dashboards, and Alerting

You’ve built a pipeline that downloads, transforms, validates, and curates AIS data. But how do you know if it’s working? Observability makes your pipeline’s behavior visible, measurable, and actionable—so you can detect failures before users complain.

Why Observability Matters

Data pipelines fail in many ways:

• Silent failures: Data source changes schema, validation passes but data is wrong
• Performance degradation: Pipeline slows down over time due to data growth
• Partial failures: One partition fails but others succeed
• Resource exhaustion: Running out of memory or disk space
• Data quality drift: Gradual degradation in data quality

Without observability, you discover these problems when users complain. With observability, you detect and fix them proactively.

Three Pillars of Observability

Pillar	What It Captures	When to Use
Logs	Detailed event records	Debugging specific failures
Metrics	Aggregate statistics	Trend analysis, alerting, dashboards
Traces	Request flow through system	Finding bottlenecks, understanding dependencies

We’ll focus on logs (Dagster built-in) and metrics (Prometheus + Grafana).

Dagster’s Built-in Observability

Dagster provides excellent observability out of the box.

Run History

Every asset materialization is tracked in Dagster’s PostgreSQL database:

• Run status: Success, failure, in progress
• Duration: How long each run took
• Logs: Detailed execution logs
• Metadata: Custom metadata returned from assets

Access via Dagit UI at /runs with filtering by asset name, date range, status, and partition.

Asset Status

The asset graph view shows:

• Materialization status: Green (success), red (failure), gray (not materialized)
• Last updated: When each asset was last materialized
• Staleness: Assets that depend on updated upstream assets

Structured Logging

Every asset execution generates logs:

@asset(...)
def dma_ais_silver(context, ...):
    context.log.info(f"Converting {partition_date} to Parquet")
    context.log.info(f"Processed {row_count:,} rows in {duration:.2f}s")
    context.log.warning(f"Found {invalid_rows} invalid MMSI values")

Custom Prometheus Metrics

Dagster’s monitoring is great for debugging, but for trend analysis and alerting, you need time-series metrics.

Metrics to Track

Volume Metrics:

• ais_rows_ingested_total - Total rows ingested (counter)
• ais_rows_validated_total - Rows that passed validation (counter)
• ais_unique_vessels_total - Unique vessels seen (counter)
• ais_storage_bytes - Storage used by layer (gauge)

Performance Metrics:

• ais_asset_duration_seconds - Asset execution time (histogram)
• ais_download_duration_seconds - Download time (histogram)
• ais_query_duration_seconds - DuckDB query time (histogram)

Quality Metrics:

• ais_validation_success_rate - Validation pass rate (gauge)
• ais_validation_failures_total - Failed expectations (counter)

Error Metrics:

• ais_asset_failures_total - Asset failures by name (counter)
• ais_download_errors_total - Download failures by source (counter)

Implementing Metrics

# src/monitoring/metrics.py
from prometheus_client import Counter, Gauge, Histogram, CollectorRegistry

registry = CollectorRegistry()

# Volume Metrics
rows_ingested = Counter(
    'ais_rows_ingested_total',
    'Total AIS position reports ingested',
    ['source', 'partition_date'],
    registry=registry
)

unique_vessels = Gauge(
    'ais_unique_vessels',
    'Number of unique vessels seen in partition',
    ['source', 'partition_date'],
    registry=registry
)

storage_bytes = Gauge(
    'ais_storage_bytes',
    'Storage used by data layer',
    ['layer', 'source'],
    registry=registry
)

# Performance Metrics
asset_duration = Histogram(
    'ais_asset_duration_seconds',
    'Asset execution duration',
    ['asset_name', 'partition_date'],
    buckets=[1, 5, 10, 30, 60, 120, 300, 600],
    registry=registry
)

# Quality Metrics
validation_success_rate = Gauge(
    'ais_validation_success_rate',
    'Validation success rate (0-100)',
    ['source', 'partition_date'],
    registry=registry
)

# Error Metrics
asset_failures = Counter(
    'ais_asset_failures_total',
    'Asset execution failures',
    ['asset_name', 'error_type'],
    registry=registry
)

Instrumenting Assets

Update assets to emit metrics:

from monitoring.metrics import rows_ingested, asset_duration, asset_failures
from datetime import datetime

@asset(...)
def dma_ais_silver(context, dma_ais_bronze):
    partition_date = context.partition_key
    start_time = datetime.now()

    try:
        # ... DuckDB conversion logic ...

        # Emit metrics
        rows_ingested.labels(source='dma', partition_date=partition_date).inc(total_rows)

        duration = (datetime.now() - start_time).total_seconds()
        asset_duration.labels(
            asset_name='dma_ais_silver',
            partition_date=partition_date
        ).observe(duration)

        context.log.info(f"Metrics emitted: {total_rows:,} rows, {duration:.2f}s")

    except Exception as e:
        asset_failures.labels(
            asset_name='dma_ais_silver',
            error_type=type(e).__name__
        ).inc()
        raise

Metrics HTTP Endpoint

Prometheus scrapes metrics via HTTP:

# src/monitoring/metrics_server.py
from prometheus_client import make_wsgi_app
from wsgiref.simple_server import make_server
from monitoring.metrics import registry

def start_metrics_server(port=9091):
    """Start HTTP server for Prometheus metrics scraping."""
    app = make_wsgi_app(registry)
    httpd = make_server('', port, app)
    print(f"Metrics server listening on port {port}")
    httpd.serve_forever()

Test the endpoint:

curl http://localhost:9091/metrics

# Output:
# HELP ais_rows_ingested_total Total AIS position reports ingested
# TYPE ais_rows_ingested_total counter
ais_rows_ingested_total{source="dma",partition_date="2024-01-15"} 12458392.0

Configuring Prometheus

Update Prometheus to scrape your metrics:

# prometheus-config.yaml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: 'ais-pipeline'
    static_configs:
      - targets: ['dagster-daemon:9091']
    metric_relabel_configs:
      - source_labels: [__name__]
        target_label: cluster
        replacement: 'ais-pipeline'

alerting:
  alertmanagers:
    - static_configs:
        - targets: ['alertmanager:9093']

rule_files:
  - '/etc/prometheus/alerts.yml'

Verify scraping in Prometheus UI under Status → Targets. The ais-pipeline job should show UP status.

Grafana Dashboards

Prometheus collects metrics; Grafana visualizes them.

Dashboard Layout

Row 1: Overview

• Total rows ingested today (stat panel)
• Unique vessels today (stat panel)
• Validation success rate (gauge)

Row 2: Volume Trends

• Rows ingested over time (graph by source)
• Storage growth (graph by layer)

Row 3: Performance

• Asset duration heatmap
• Download duration (graph)
• Query duration distribution (histogram)

Row 4: Errors and Quality

• Failure rate (graph)
• Top validation failures (table)

Example Panel Configurations

Total Rows Ingested:

{
  "title": "Total Rows Ingested Today",
  "type": "stat",
  "targets": [{
    "expr": "sum(increase(ais_rows_ingested_total[24h]))",
    "legendFormat": "Total Rows"
  }],
  "fieldConfig": {
    "defaults": {
      "thresholds": {
        "steps": [
          {"value": 0, "color": "red"},
          {"value": 1000000, "color": "yellow"},
          {"value": 10000000, "color": "green"}
        ]
      }
    }
  }
}

Validation Success Rate:

{
  "title": "Validation Success Rate",
  "type": "gauge",
  "targets": [{
    "expr": "avg(ais_validation_success_rate)",
    "legendFormat": "Success %"
  }],
  "fieldConfig": {
    "defaults": {
      "min": 0, "max": 100,
      "thresholds": {
        "steps": [
          {"value": 0, "color": "red"},
          {"value": 95, "color": "yellow"},
          {"value": 99, "color": "green"}
        ]
      }
    }
  }
}

Alerting

Metrics and dashboards are great, but you can’t watch them 24/7. Alerting notifies you when things go wrong.

Alert Rules

groups:
  - name: ais_pipeline
    interval: 30s
    rules:
      # No data ingested in 2 hours
      - alert: AISNoDataIngested
        expr: rate(ais_rows_ingested_total[2h]) == 0
        for: 15m
        labels:
          severity: critical
        annotations:
          summary: "No AIS data ingested in 2 hours"
          description: "Pipeline may be stuck or data source unavailable"

      # Validation quality drop
      - alert: AISValidationQualityDrop
        expr: avg(ais_validation_success_rate) < 95
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Data quality below threshold"
          description: "Validation success rate: {{ $value }}%"

      # Asset failures
      - alert: AISAssetFailuresIncreased
        expr: rate(ais_asset_failures_total[1h]) > 0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Asset failures detected"
          description: "{{ $labels.asset_name }} failed"

      # Storage nearly full
      - alert: AISStorageNearFull
        expr: sum(ais_storage_bytes) / 1e9 > 800
        for: 30m
        labels:
          severity: critical
        annotations:
          summary: "Storage nearly full"
          description: "Using {{ $value }}GB of capacity"

      # Pipeline running slow
      - alert: AISPipelineSlow
        expr: |
          (sum(rate(ais_asset_duration_seconds_sum[1h]))
          / sum(rate(ais_asset_duration_seconds_count[1h]))) > 300
        for: 30m
        labels:
          severity: warning
        annotations:
          summary: "Pipeline running slow"
          description: "Average asset duration: {{ $value }}s"

Alert Notifications

Configure Alertmanager to send notifications:

# alertmanager.yml
global:
  resolve_timeout: 5m
  slack_api_url: 'YOUR_SLACK_WEBHOOK_URL'

route:
  receiver: 'slack-notifications'
  group_by: ['alertname', 'severity']
  routes:
    - match:
        severity: critical
      receiver: 'slack-critical'
      repeat_interval: 1h
    - match:
        severity: warning
      receiver: 'slack-warnings'
      repeat_interval: 4h

receivers:
  - name: 'slack-critical'
    slack_configs:
      - channel: '#pipeline-critical'
        title: 'CRITICAL: AIS Pipeline'
        send_resolved: true

  - name: 'slack-warnings'
    slack_configs:
      - channel: '#pipeline-alerts'
        title: 'Warning: AIS Pipeline'

Storage Monitoring

Track storage usage to avoid running out of space:

# src/monitoring/storage_monitor.py
from minio import Minio
from monitoring.metrics import storage_bytes

def monitor_storage(minio_client: Minio, bucket: str = "ais-lake"):
    """Monitor storage usage by layer and emit metrics."""
    layers = {
        'bronze': 'bronze/',
        'silver': 'silver/ais/',
        'gold_tracks': 'gold/vessel_tracks/',
        'gold_density': 'gold/port_density/',
        'gold_stats': 'gold/daily_stats/',
    }

    for layer, prefix in layers.items():
        total_bytes = 0
        objects = minio_client.list_objects(bucket, prefix=prefix, recursive=True)
        for obj in objects:
            total_bytes += obj.size

        storage_bytes.labels(layer=layer, source='dma').set(total_bytes)
        print(f"{layer}: {total_bytes / 1e9:.2f} GB")

Run this periodically via a CronJob or alongside your pipeline.

What You’ve Built

Your pipeline now has comprehensive observability:

• Dagster monitoring: Built-in run history, logs, and asset status
• Custom metrics: Volume, performance, quality, and error tracking
• Visualization: Real-time Grafana dashboards
• Alerting: Proactive notifications for failures and quality drops
• Storage tracking: Monitoring disk usage to prevent exhaustion

What’s Next

With full observability in place, Chapter 10 covers deploying the complete pipeline to Kubernetes using Docker images, Helm charts, and proper secrets management.

---

Building data pipelines that need production-grade monitoring? We help organizations implement observability strategies that catch issues before they impact business. Get in touch to discuss your project.