graph TD
    A[Data Platform Observability] --> B[Metrics]
    A --> C[Logs]
    A --> D[Distributed Traces]
    A --> E[Data Quality]
    B --> B1[Throughput, Latency, Error Rate]
    C --> C1[Structured Pipeline Events]
    D --> D1[End-to-End Job Lineage]
    E --> E1[Freshness, Volume, Schema, Distribution]

# Prometheus-Metrics, die du in deinem Pipeline-Code instrumentierst
metrics:
  ingestion:
    - records_ingested_total{source, table, environment}
    - ingestion_latency_seconds{source, table}
    - ingestion_errors_total{source, table, error_type}
    - last_successful_run_timestamp{pipeline_id}

  transformation:
    - records_transformed_total{stage, table}
    - records_dropped_total{stage, table, reason}
    - transformation_duration_seconds{pipeline_id, stage}
    - schema_drift_events_total{table, field}

  serving:
    - query_latency_p99_seconds{dataset, query_type}
    - stale_data_seconds{table}  # now - max(updated_at)
    - query_error_rate{dataset}

# docker-compose.yml für lokalen Observability-Stack
version: "3.8"
services:
  prometheus:
    image: prom/prometheus:v2.47.0
    ports: ["9090:9090"]
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    command:
      - --storage.tsdb.retention.time=90d
      - --web.enable-lifecycle

  grafana:
    image: grafana/grafana:10.1.0
    ports: ["3000:3000"]
    environment:
      GF_AUTH_ANONYMOUS_ENABLED: "true"
      GF_AUTH_ANONYMOUS_ORG_ROLE: Viewer
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards

volumes:
  prometheus_data:
  grafana_data:

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - "alerts/*.yml"

scrape_configs:
  - job_name: spark_pipelines
    static_configs:
      - targets: ['spark-driver:4040']
    metrics_path: /metrics/prometheus

  - job_name: airflow
    static_configs:
      - targets: ['airflow-webserver:8080']
    metrics_path: /metrics

  - job_name: kafka
    static_configs:
      - targets: ['kafka-broker-1:9308', 'kafka-broker-2:9308']

# Prometheus-Alert für Daten-Freshness-SLO-Breach
groups:
  - name: data_freshness
    rules:
      - alert: OrdersTableStale
        expr: |
          (time() - data_platform_last_successful_ingestion_timestamp{table="orders"}) > 300
        for: 2m
        labels:
          severity: warning
          team: data-platform
        annotations:
          summary: "Orders table is stale ({{ $value | humanizeDuration }})"
          runbook: "https://wiki.internal/runbooks/stale-orders"

      - alert: SilverTransformSLOBreach
        expr: |
          data_platform_transformation_duration_seconds{pipeline="bronze_to_silver_orders"} > 1800
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Silver transform exceeding 30-min SLO"

# Structured-Log-Schema (JSON)
log_event:
  timestamp: "2024-01-15T08:32:11.421Z"
  level: INFO | WARN | ERROR
  pipeline_id: "bronze_to_silver_orders"
  run_id: "run_20240115_083200"
  stage: "read | transform | write | validate"
  table: "silver.orders"
  records_in: 142847
  records_out: 142831
  records_dropped: 16
  drop_reason: "schema_mismatch"
  duration_ms: 28431
  environment: "prod"
  spark_app_id: "app-20240115083200-0001"
  correlation_id: "req-abc123"

resource "aws_cloudwatch_log_group" "pipeline_logs" {
  for_each          = toset(["bronze-ingestion", "silver-transform", "gold-aggregate", "data-quality"])
  name              = "/data-platform/${var.environment}/${each.key}"
  retention_in_days = 90
  kms_key_id        = aws_kms_key.data_platform.arn

  tags = {
    Pipeline    = each.key
    Environment = var.environment
  }
}

# Log-Metric-Filter: Error-Count aus structured Logs extrahieren
resource "aws_cloudwatch_log_metric_filter" "pipeline_errors" {
  for_each       = aws_cloudwatch_log_group.pipeline_logs
  name           = "${each.key}-errors"
  pattern        = "{ $.level = "ERROR" }"
  log_group_name = each.value.name

  metric_transformation {
    name          = "PipelineErrors"
    namespace     = "DataPlatform/${var.environment}"
    value         = "1"
    default_value = "0"
    dimensions = {
      Pipeline = each.key
    }
  }
}

# OpenTelemetry-Collector-Config
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    timeout: 10s
    send_batch_size: 512
  resource:
    attributes:
      - key: environment
        value: prod
        action: upsert

exporters:
  jaeger:
    endpoint: jaeger-collector:14250
    tls:
      insecure: false
  prometheus:
    endpoint: "0.0.0.0:8889"

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch, resource]
      exporters: [jaeger]
    metrics:
      receivers: [otlp]
      processors: [batch]
      exporters: [prometheus]

# OTel-Java-Agent zum Spark-Submit adden
spark-submit   --conf "spark.driver.extraJavaOptions=-javaagent:/opt/otel-javaagent.jar"   --conf "spark.executor.extraJavaOptions=-javaagent:/opt/otel-javaagent.jar"   --conf "spark.driver.extraJavaOptions=-Dotel.service.name=etl-silver-transform"   --conf "spark.driver.extraJavaOptions=-Dotel.exporter.otlp.endpoint=http://otel-collector:4317"   --conf "spark.driver.extraJavaOptions=-Dotel.traces.sampler=parentbased_traceidratio"   --conf "spark.driver.extraJavaOptions=-Dotel.traces.sampler.arg=0.1"   my-etl-job.jar

graph LR
    A[Data Quality] --> B[Freshness
How recent is it?]
    A --> C[Volume
Are row counts normal?]
    A --> D[Schema
Did structure change?]
    A --> E[Distribution
Are values in expected range?]
    A --> F[Referential Integrity
Do FKs resolve?]
    A --> G[Uniqueness
Are there unexpected duplicates?]

# schema.yml
version: 2

models:
  - name: silver_orders
    description: "Cleaned and deduplicated orders"
    columns:
      - name: order_id
        tests:
          - not_null
          - unique
      - name: customer_id
        tests:
          - not_null
          - relationships:
              to: ref('silver_customers')
              field: customer_id
      - name: amount
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 100000
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled']
      - name: created_at
        tests:
          - not_null
          - dbt_utils.recency:
              datepart: hour
              field: created_at
              interval: 6

# SQL: Volume-Anomalien mit 7-Tage-Rolling-Stats detecten
WITH daily_volumes AS (
  SELECT
    DATE(created_at) AS dt,
    COUNT(*) AS row_count
  FROM silver.orders
  WHERE created_at >= CURRENT_DATE - INTERVAL '30 days'
  GROUP BY 1
),
stats AS (
  SELECT
    AVG(row_count) AS mean,
    STDDEV(row_count) AS stddev
  FROM daily_volumes
  WHERE dt < CURRENT_DATE  -- exclude today
)
SELECT
  dv.dt,
  dv.row_count,
  s.mean,
  s.stddev,
  ABS(dv.row_count - s.mean) / NULLIF(s.stddev, 0) AS z_score,
  CASE WHEN ABS(dv.row_count - s.mean) / NULLIF(s.stddev, 0) > 3
    THEN 'ANOMALY' ELSE 'OK' END AS status
FROM daily_volumes dv
CROSS JOIN stats s
WHERE dv.dt = CURRENT_DATE;

resource "aws_cloudwatch_metric_alarm" "pipeline_freshness" {
  alarm_name          = "orders-table-freshness-breach"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 2
  metric_name         = "DataStalenessSeconds"
  namespace           = "DataPlatform/prod"
  period              = 300
  statistic           = "Maximum"
  threshold           = 14400  # 4 Stunden
  alarm_description   = "Orders table data is older than 4 hours"
  treat_missing_data  = "breaching"

  alarm_actions = [aws_sns_topic.data_platform_alerts.arn]
  ok_actions    = [aws_sns_topic.data_platform_alerts.arn]

  dimensions = {
    Table = "orders"
  }
}

# Verify Pipeline hat exposed Prometheus-Metrics
curl -s http://spark-driver:4040/metrics/prometheus | grep records_processed

# Check Freshness einer Critical-Table
psql -c "SELECT MAX(updated_at), NOW() - MAX(updated_at) AS age FROM silver.orders"

# dbt-Tests laufen lassen
dbt test --select silver_orders --store-failures

# Kafka-Consumer-Lag checken
kafka-consumer-groups.sh --bootstrap-server kafka:9092   --group etl-silver-transform --describe | awk '{print $5, $6}'