SLI (Service Level Indicator)
  ↓ gemessen über
SLO (Service Level Objective)
  ↓ Commitment-Basis für
SLA (Service Level Agreement)
  ↓ Bruch löst aus
Konsequenzen (Credits, Eskalationen, etc.)

Error Budget = 1 - SLO-Ziel

Für Freshness-SLO von 99,5% (über 30 Tage = 43.200 Minuten):
Error Budget = 0,5% × 43.200 = 216 Minuten erlaubter Lag-Verletzung pro Monat

-- models/monitoring/sli_freshness.sql
SELECT
    table_name,
    MAX(event_timestamp) AS latest_record,
    EXTRACT(EPOCH FROM (NOW() - MAX(event_timestamp))) / 60 AS lag_minutes,
    CASE
        WHEN EXTRACT(EPOCH FROM (NOW() - MAX(event_timestamp))) / 60 <= 15 THEN 'OK'
        WHEN EXTRACT(EPOCH FROM (NOW() - MAX(event_timestamp))) / 60 <= 30 THEN 'WARNING'
        ELSE 'BREACH'
    END AS slo_status
FROM (
    SELECT 'fact_geopolitical_events' AS table_name, event_timestamp
    FROM {{ ref('fact_geopolitical_events') }}
    UNION ALL
    SELECT 'agg_risk_scores_hourly', hour_timestamp
    FROM {{ ref('agg_risk_scores_hourly') }}
) t
GROUP BY table_name;

-- dbt-Test: Row-Counts zwischen Source und Target vergleichen
-- tests/assert_completeness.sql
{% set expected_minimum_rows = 100 %}

SELECT COUNT(*) AS missing_count
FROM (
    SELECT event_id FROM {{ source('raw', 'events') }}
    EXCEPT
    SELECT event_id FROM {{ ref('fact_geopolitical_events') }}
) missing
HAVING COUNT(*) > {{ expected_minimum_rows }};

# models/schema.yml
models:
  - name: fact_geopolitical_events
    description: "Kern-Fact-Tabelle für geopolitische Events"
    meta:
      slo_freshness_minutes: 15
      slo_completeness_pct: 99.5
      slo_owner: "platform-team@harbinger.com"

    columns:
      - name: event_id
        tests:
          - unique
          - not_null

      - name: country_code
        tests:
          - not_null
          - accepted_values:
              values: "{{ var('valid_country_codes') }}"

      - name: severity_score
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0.0
              max_value: 10.0

      - name: event_timestamp
        tests:
          - not_null
          - dbt_utils.recency:
              datepart: minute
              interval: 30  # Daten müssen innerhalb 30 Minuten vorliegen

graph TD
    subgraph Datenplattform
        A[Airflow Pipelines] --> B[fact / dim / agg tables]
        C[dbt Jobs] --> B
    end

    subgraph SLO-Monitoring
        B --> D[SLI Probe Queries
alle 5 Min]
        D --> E[slo_measurements Tabelle
TimescaleDB]
        E --> F[SLO Calculator
Rolling 30-Tage-Fenster]
        F --> G[slo_status Tabelle
PostgreSQL]
    end

    subgraph Alerting
        G --> H{SLO Status Check}
        H -->|WARNING: Budget zu 50% verbraucht| I[Slack Alert]
        H -->|CRITICAL: Budget aufgebraucht| J[PagerDuty]
        H -->|BREACH| K[Incident Auto-Created
+ Freeze-Flag gesetzt]
    end

    subgraph Dashboards
        G --> L[Grafana SLO Dashboard]
        E --> L
    end

# slo_calculator.py — läuft als Kubernetes-CronJob alle 5 Minuten

import psycopg2
from datetime import datetime, timedelta, timezone

SLO_CONFIG = {
    "fact_geopolitical_events": {
        "freshness_slo_minutes": 15,
        "completeness_slo_pct": 99.5,
        "window_days": 30,
    },
    "agg_risk_scores_hourly": {
        "freshness_slo_minutes": 90,
        "completeness_slo_pct": 99.0,
        "window_days": 30,
    },
}

def calculate_error_budget(table: str, slo_pct: float, conn) -> dict:
    # Aktuellen Error-Budget-Verbrauch für eine Tabelle berechnen
    window_start = datetime.now(timezone.utc) - timedelta(days=30)
    
    with conn.cursor() as cur:
        sql = (
            "SELECT COUNT(*) AS total_measurements, "
            "SUM(CASE WHEN slo_met THEN 1 ELSE 0 END) AS passing_measurements, "
            "ROUND(100.0 * SUM(CASE WHEN slo_met THEN 1 ELSE 0 END) / COUNT(*), 4) AS actual_slo_pct "
            "FROM slo_measurements "
            "WHERE table_name = %s AND measured_at >= %s"
        )
        cur.execute(sql, (table, window_start))
        
        row = cur.fetchone()
        total, passing, actual_pct = row
        
        allowed_failures = total * (1 - slo_pct / 100)
        actual_failures = total - passing
        budget_remaining_pct = max(0, (allowed_failures - actual_failures) / allowed_failures * 100)
        
        return {
            "table": table,
            "slo_target_pct": slo_pct,
            "actual_pct": float(actual_pct),
            "budget_remaining_pct": budget_remaining_pct,
            "status": "OK" if budget_remaining_pct > 50
                      else "WARNING" if budget_remaining_pct > 0
                      else "EXHAUSTED",
        }

# Prometheus-Alert-Rules für Data-SLOs
groups:
  - name: data_platform_slo
    rules:
      - alert: DataFreshnessBreachImmediate
        expr: |
          (
            data_slo_error_budget_burn_rate{table="fact_geopolitical_events"} > 14.4
          )
        for: 2m
        labels:
          severity: critical
          team: platform
        annotations:
          summary: "Data-SLO-Error-Budget brennt schnell: {{ $labels.table }}"
          description: |
            Tabelle {{ $labels.table }} verbrennt Error-Budget mit Rate {{ $value }}×.
            Budget wird in {{ printf "%.1f" (1 / $value * 720) }} Minuten aufgebraucht sein.
            Runbook: https://wiki.harbinger.com/runbooks/data-slo-breach

## Incident: [Titel]
**Datum**: YYYY-MM-DD
**Dauer**: X Stunden Y Minuten
**SLO-Breach**: Freshness-SLO für `fact_geopolitical_events` (Lag > 15 min für 3,2 h)
**Impact**: Risk-Score-Dashboard zeigte veraltete Daten; 3 Analyst-Teams betroffen

### Timeline
- 14:32 — Airflow-DAG `event_enrichment` startet; Upstream-API rate-limited
- 14:45 — Lag überschreitet 15-Minuten-SLO-Schwelle; Alert feuert nicht (Bug in Alert-Config)
- 17:12 — On-Call sieht veraltetes Dashboard bei Routine-Check
- 17:18 — Incident erklärt; Pipeline-Rerun gestartet
- 17:41 — Datenfrische wiederhergestellt; SLO erfüllt

### Root-Cause
Upstream-News-API begann ab 14:30 wegen erhöhter Ingestion-Volumes rate-limiting.
Retry-Logic nutzte Exponential-Backoff, max Retries (3) wurden lautlos verbraucht.

### Beitragende Faktoren
- Alert-Fehlkonfig: `for: 30m` statt `for: 2m` beim Freshness-Alert
- Kein Dead-Letter-Monitoring für gescheiterte API-Calls

### Action-Items
| Action | Owner | Bis |
| --- | --- | --- |
| Alert-`for`-Dauer auf 2m fixen | @platform-team | 2026-05-20 |
| DLQ-Monitoring für API-Ingestion hinzufügen | @platform-team | 2026-05-25 |
| Adaptive Rate-Limit-Handling implementieren | @data-eng | 2026-06-01 |