API-Design Teil 16: Observability

Eine API ohne Observability ist wie Autofahren mit verbundenen Augen. Du merkst erst, dass etwas kaputt ist, wenn Kunden sich beschweren. Mit guter Observability siehst du Probleme, bevor sie eskalieren – und kannst sie schnell debuggen.

Dieser Artikel zeigt die drei Säulen der Observability: Logs, Metrics und Traces. Plus: Audit-Logs für Security und Compliance.

Zielbild

Nach diesem Artikel kannst du:

• Structured Logging mit einheitlichem Schema implementieren
• Request-IDs und Trace-IDs für Korrelation einsetzen
• Die wichtigsten API-Metriken definieren und messen
• Distributed Tracing für Microservices einrichten
• Audit-Logs für sicherheitsrelevante Aktionen führen
• PII in Logs vermeiden oder maskieren

Kernfragen

Bevor du weiterliest, versuche diese Fragen für dein Projekt zu beantworten:

1. Welche Metriken braucht ihr von Tag 1? Latenz? Error Rate? Throughput?
2. Wie korreliert ihr Requests? Request-ID? Trace-ID?
3. Welche Aktionen müssen auditiert werden? Login? Datenänderungen?
4. Wo landen eure Logs? Elasticsearch? CloudWatch? Loki?
5. Wie verhindert ihr PII in Logs? Maskierung? Filtering?

Die drei Säulen

Observability basiert auf drei komplementären Datentypen.

Wann was?

Frage	Tool
Was ist um 14:23 passiert?	Logs
Wie viele Errors hatten wir heute?	Metrics
Warum war dieser Request langsam?	Traces
Wer hat diese Daten geändert?	Audit-Logs

Structured Logging

Unstrukturierte Logs sind schwer zu parsen und zu durchsuchen.

Unstrukturiert (schlecht)

2026-01-17 14:23:45 ERROR Failed to create order for user 123: insufficient balance

Strukturiert (gut)

{
  "timestamp": "2026-01-17T14:23:45.123Z",
  "level": "error",
  "message": "Failed to create order",
  "request_id": "req_abc123",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "user_id": "user_123",
  "error_code": "INSUFFICIENT_BALANCE",
  "order_amount": 150.00,
  "available_balance": 100.00,
  "service": "order-service",
  "environment": "production"
}

Log-Schema

Definiere ein einheitliches Schema für alle Services:

Feld	Typ	Pflicht	Beschreibung
timestamp	ISO 8601	Ja	Zeitpunkt (UTC)
level	String	Ja	debug/info/warn/error
message	String	Ja	Menschenlesbare Beschreibung
request_id	String	Ja*	Eindeutige Request-ID
trace_id	String	Nein	OpenTelemetry Trace-ID
span_id	String	Nein	OpenTelemetry Span-ID
user_id	String	Nein	Anonymisiert wenn PII
tenant_id	String	Nein	Multi-Tenant
service	String	Ja	Service-Name
environment	String	Ja	dev/staging/production
error_code	String	Nein	Bei Fehlern
duration_ms	Number	Nein	Bei Timing-relevanten Logs

request_id ist für Request-Logs Pflicht, bei Hintergrundjobs optional.

Log Levels

Level	Verwendung	Beispiel
debug	Entwicklung, verbose	SQL-Queries, Cache-Hits
info	Normale Operationen	Request completed, Job started
warn	Potenzielle Probleme	Retry erfolgreich, Deprecated API
error	Fehler, aber Service läuft	Validation failed, External API error
fatal	Service kann nicht weiterlaufen	DB-Connection lost, Config invalid

Implementierung (Audit-Logging)

import structlog
import uuid

# Structlog konfigurieren
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.JSONRenderer()
    ]
)

logger = structlog.get_logger()

# Middleware: Request-Context setzen
async def logging_middleware(request, call_next):
    request_id = request.headers.get("X-Request-Id") or str(uuid.uuid4())
    traceparent = request.headers.get("traceparent")
    trace_id = traceparent.split("-")[1] if traceparent else None

    # Context für alle Logs in diesem Request
    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=request_id,
        trace_id=trace_id,
        path=request.url.path,
        method=request.method
    )

    logger.info("Request started")

    response = await call_next(request)

    logger.info("Request completed", status=response.status_code)
    return response

Request-ID und Trace-ID

Korrelation ist der Schlüssel zum Debugging in verteilten Systemen.

Request-ID

Eindeutige ID für jeden API-Request.

# Request
GET /v1/orders HTTP/1.1
X-Request-Id: req_abc123

# Response
HTTP/1.1 200 OK
X-Request-Id: req_abc123

Header: X-Request-Id ist de-facto Standard, aber nicht RFC-spezifiziert.

Generierung:

import uuid

def generate_request_id() -> str:
    return f"req_{uuid.uuid4().hex[:12]}"

Regel: Wenn Client eine Request-ID sendet, diese validieren (Format/Length) und dann verwenden. Sonst generieren.

Trace-ID (OpenTelemetry)

Verbindet alle Spans eines verteilten Requests.

Standard: W3C Trace Context (traceparent, optional tracestate).

traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01
tracestate: vendorkey=value

Korrelation in Logs

Alle Logs eines Requests haben dieselbe Request-ID (und idealerweise dieselbe Trace-ID):

{ "request_id": "req_abc123", "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", "message": "Validating order", "service": "order-service" }
{ "request_id": "req_abc123", "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", "message": "Charging payment", "service": "payment-service" }
{ "request_id": "req_abc123", "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", "message": "Order created", "service": "order-service" }

Suche: request_id:"req_abc123" → Alle relevanten Logs.

Header-Propagation

Trace-Context muss zwischen Services weitergegeben werden:

from opentelemetry.propagate import inject

# Ausgehender Request
async def call_payment_service(order_id: str, request_id: str):
    headers = {"X-Request-Id": request_id}
    inject(headers)  # setzt traceparent/tracestate
    return await http_client.post(
        f"{PAYMENT_URL}/charge",
        headers=headers,
        json={"order_id": order_id}
    )

Metrics

Metriken beantworten: Wie verhält sich das System insgesamt?

Die vier goldenen Signale

Google's SRE Book definiert vier kritische Metriken:

Signal	Frage	Metrik
Latency	Wie schnell?	Response Time (p50, p95, p99)
Traffic	Wie viel?	Requests per Second
Errors	Wie oft kaputt?	Error Rate (%)
Saturation	Wie voll?	CPU, Memory, Connections

API-spezifische Metriken

Metrik	Typ	Labels
http_requests_total	Counter	method, route, status_class
http_request_duration_seconds	Histogram	method, route, status_class
http_request_size_bytes	Histogram	method, route
http_response_size_bytes	Histogram	method, route
http_active_requests	Gauge	method, route

Wichtig: Verwende route (z. B. /v1/orders/{id}) statt rohem Path, um High-Cardinality-Probleme zu vermeiden.

Prometheus-Instrumentierung

import time
from prometheus_client import Counter, Histogram, Gauge

# Metriken definieren
REQUEST_COUNT = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "route", "status_class"]
)

REQUEST_LATENCY = Histogram(
    "http_request_duration_seconds",
    "HTTP request latency",
    ["method", "route", "status_class"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)

ACTIVE_REQUESTS = Gauge(
    "http_active_requests",
    "Active HTTP requests",
    ["method", "route"]
)

# Middleware
async def metrics_middleware(request, call_next):
    route_obj = request.scope.get("route")
    route = route_obj.path if route_obj else "unknown"
    method = request.method

    ACTIVE_REQUESTS.labels(method=method, route=route).inc()
    start_time = time.time()
    response = None

    try:
        response = await call_next(request)
        return response
    finally:
        duration = time.time() - start_time
        status = response.status_code if response else 500
        status_class = f"{status // 100}xx"

        REQUEST_COUNT.labels(method=method, route=route, status_class=status_class).inc()
        REQUEST_LATENCY.labels(method=method, route=route, status_class=status_class).observe(duration)
        ACTIVE_REQUESTS.labels(method=method, route=route).dec()

Percentile vs. Average

Durchschnitt lügt:

Beispiel	Latenz	Hinweis
9 Requests	100ms
1 Request	10.000ms
Durchschnitt	1.090ms	Alles OK
p99	10.000ms	Ein User wartet 10 Sekunden!

Wichtige Percentile:

Percentile	Bedeutung
p50 (Median)	Typische Erfahrung
p95	95% der Requests sind schneller
p99	Worst-Case für 1% der Requests
p99.9	Extreme Outliers

SLIs und SLOs

SLI (Service Level Indicator): Was wir messen.

SLI: Anteil der Requests mit Latenz < 500ms

SLO (Service Level Objective): Unser Ziel.

SLO: 99% der Requests haben Latenz < 500ms

Beispiel-SLOs:

SLI	SLO
Availability	99.9%
Latency (p95)	< 200ms
Latency (p99)	< 500ms
Error Rate	< 0.1%

Distributed Tracing

Traces zeigen den Weg eines Requests durch alle Services.

Konzepte

Begriff	Bedeutung
Trace	Ein kompletter Request-Flow
Span	Eine einzelne Operation innerhalb des Trace
Parent Span	Der aufrufende Span
Context	Trace-ID + Span-ID + Flags

Beispiel-Trace

OpenTelemetry Setup

from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

# Provider konfigurieren
resource = Resource.create({"service.name": "order-service"})
provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# Auto-Instrumentierung
FastAPIInstrumentor.instrument_app(app)
HTTPXClientInstrumentor().instrument()

# Manueller Span
tracer = trace.get_tracer(__name__)

async def process_order(order_id: str):
    with tracer.start_as_current_span("process_order") as span:
        span.set_attribute("order.id", order_id)

        # Nested Span
        with tracer.start_as_current_span("validate_inventory"):
            await check_inventory(order_id)

        with tracer.start_as_current_span("charge_payment"):
            await charge_payment(order_id)

Span-Attribute

span.set_attribute("http.method", "POST")
span.set_attribute("http.url", "/v1/orders")
span.set_attribute("http.status_code", 201)
span.set_attribute("user.id", user_id)
span.set_attribute("order.id", order_id)
span.set_attribute("order.amount", 150.00)

Hinweis: Vermeide PII in Span-Attributen (z. B. E-Mail, vollständige Adresse). Nutze IDs oder gehashte Werte.

Trace-Sampling

Bei hohem Traffic: Nicht jeden Request tracen.

Strategie	Beschreibung
Always On	Alles tracen (Dev/Staging)
Probabilistic	X% der Requests
Rate Limiting	Max N Traces/Sekunde
Tail-Based	Nur langsame/fehlerhafte Requests

from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased

# 10% der Requests tracen (respektiert Upstream-Sampling)
sampler = ParentBased(TraceIdRatioBased(0.1))
provider = TracerProvider(sampler=sampler)

Audit-Logs

Audit-Logs dokumentieren sicherheitsrelevante Aktionen für Compliance und Forensik.

Was auditieren?

Kategorie	Beispiele
Authentifizierung	Login, Logout, Failed Login, Token Refresh
Autorisierung	Permission Denied, Role Change
Datenänderungen	Create, Update, Delete von sensitiven Daten
Admin-Aktionen	User erstellen, Rollen ändern, Config ändern
Datenzugriff	PII-Zugriff, Export, Bulk-Download
System	Service Start/Stop, Config Reload

Audit-Log-Schema

{
  "timestamp": "2026-01-17T14:23:45.123Z",
  "event_type": "user.login.success",
  "actor": {
    "id": "user_123",
    "type": "user",
    "ip": "192.168.1.100",
    "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
  },
  "target": {
    "type": "session",
    "id": "sess_abc123"
  },
  "context": {
    "request_id": "req_xyz789",
    "tenant_id": "tenant_acme",
    "mfa_used": true
  },
  "outcome": "success",
  "metadata": {
    "auth_method": "password",
    "new_session": true
  }
}

Event-Types

# Authentication
user.login.success
user.login.failure
user.logout
user.mfa.enabled
user.password.changed
user.password.reset

# Authorization
permission.denied
role.assigned
role.revoked
scope.elevated

# Data
resource.created
resource.updated
resource.deleted
resource.exported

# Admin
admin.user.created
admin.user.deleted
admin.config.changed
admin.impersonation.started

Implementierung

from datetime import datetime
import json

class AuditLogger:
    def __init__(self, sink):
        self.sink = sink

    async def log(
        self,
        event_type: str,
        actor: dict,
        target: dict = None,
        outcome: str = "success",
        metadata: dict = None,
        context: dict = None
    ):
        event = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "event_type": event_type,
            "actor": actor,
            "target": target,
            "outcome": outcome,
            "context": context or {},
            "metadata": metadata or {}
        }

        # Audit-Logs sind append-only und unveränderlich
        await self.sink.write(event)

# Verwendung
audit = AuditLogger(sink=audit_storage)

await audit.log(
    event_type="resource.deleted",
    actor={"id": current_user.id, "type": "user", "ip": request.client.host},
    target={"type": "order", "id": order_id},
    outcome="success",
    context={"request_id": request_id, "tenant_id": tenant_id}
)

Audit-Log-Speicherung

Anforderung	Lösung
Unveränderlich	Append-only Storage, Write-Once
Langzeit-Aufbewahrung	Compliance-Anforderungen beachten
Durchsuchbar	Indexiert, aber nicht für Real-Time
Sicher	Verschlüsselt, separater Zugriff

Audit-Logs sollten minimal gehalten werden (kein Klartext-PII, keine Secrets).

PII in Logs

Personenbezogene Daten in Logs sind ein DSGVO-Risiko.

Was ist PII?

Kategorie	Beispiele
Direkt identifizierend	Name, E-Mail, Telefon, Adresse
Indirekt identifizierend	IP-Adresse, User-Agent, Device-ID
Sensitiv	Passwort, Kreditkarte, Gesundheitsdaten
Quasi-Identifier	Geburtsdatum + PLZ + Geschlecht

Strategien

1. Nicht loggen:

# FALSCH
logger.info("User logged in", email=user.email)

# RICHTIG
logger.info("User logged in", user_id=user.id)

2. Maskieren:

def mask_email(email: str) -> str:
    local, domain = email.split("@")
    return f"{local[:2]}***@{domain}"

# j***@example.com

IP-Adressen kannst du z. B. auf /24 kürzen (IPv4) oder /64 (IPv6), wenn eine grobe Lokalisierung reicht.

3. Hashen:

import hmac
import hashlib

def hash_pii(value: str, salt: bytes) -> str:
    return hmac.new(salt, value.encode(), hashlib.sha256).hexdigest()[:16]

4. Tokenisieren:

PII → Token, Mapping in separater, geschützter DB.

Log-Sanitization

SENSITIVE_FIELDS = {"password", "token", "api_key", "credit_card", "ssn", "authorization", "cookie"}

def sanitize_log_data(data: dict) -> dict:
    sanitized = {}
    for key, value in data.items():
        if key.lower() in SENSITIVE_FIELDS:
            sanitized[key] = "[REDACTED]"
        elif key.lower() == "email":
            sanitized[key] = mask_email(value)
        else:
            sanitized[key] = value
    return sanitized

Regeln & Anti-Patterns

Do

• Structured Logging (JSON) mit einheitlichem Schema
• Request-ID in jedem Log und jeder Response
• Trace-ID für Service-übergreifende Korrelation
• Die vier goldenen Signale messen (Latency, Traffic, Errors, Saturation)
• Percentile statt Durchschnitt (p95, p99)
• Audit-Logs für sicherheitsrelevante Aktionen
• PII maskieren oder vermeiden

Don't

• Unstrukturierte Text-Logs
• Logs ohne Request-ID
• Nur Durchschnittswerte für Latenz
• PII in Logs (E-Mail, Name, IP ungehasht)
• Passwörter, Tokens, Keys in Logs
• Audit-Logs veränderbar speichern
• Zu viele Labels (High Cardinality Problem)

Artefakt: Observability-Checkliste

# Observability-Checkliste

## Logging

### Schema

- [ ] Einheitliches JSON-Schema für alle Services
- [ ] Pflichtfelder: timestamp, level, message, request_id, service, environment
- [ ] Log-Levels definiert und konsistent verwendet

### Korrelation

- [ ] Request-ID in allen Logs
- [ ] Request-ID in Response-Header
- [ ] Trace-ID bei Distributed Tracing
- [ ] Context-Propagation zwischen Services

### PII

- [ ] Liste sensitiver Felder definiert
- [ ] Sanitization/Maskierung implementiert
- [ ] Keine Passwörter/Tokens in Logs
- [ ] Log-Retention-Policy (DSGVO)

## Metrics

### Basis-Metriken

- [ ] `http_requests_total` (Counter)
- [ ] `http_request_duration_seconds` (Histogram)
- [ ] `http_active_requests` (Gauge)
- [ ] Error Rate berechenbar

### SLIs/SLOs

| SLI           | SLO     | Aktuell |
|---------------|---------|---------|
| Availability  | 99.9%   | ___%    |
| Latency (p95) | < 200ms | ___ms   |
| Latency (p99) | < 500ms | ___ms   |
| Error Rate    | < 0.1%  | ___%    |

### Alerts

- [ ] Alert bei Error Rate > 1%
- [ ] Alert bei p99 Latency > 1s
- [ ] Alert bei Availability < 99.5%
- [ ] On-Call-Rotation definiert

## Tracing

### Setup

- [ ] OpenTelemetry SDK integriert
- [ ] Auto-Instrumentierung für HTTP
- [ ] Auto-Instrumentierung für DB
- [ ] Exporter konfiguriert (Jaeger/Zipkin/OTLP)

### Sampling

- [ ] Sampling-Strategie definiert
- [ ] Production: Probabilistic oder Tail-Based
- [ ] Dev/Staging: Always On

### Span-Attribute

- [ ] HTTP-Attribute (method, url, status)
- [ ] User-Attribute (user_id, tenant_id)
- [ ] Business-Attribute (order_id, amount)

## Audit-Logs

### Events

- [ ] Login/Logout
- [ ] Failed Authentication
- [ ] Permission Denied
- [ ] Sensitive Data Access
- [ ] Data Modification (Create/Update/Delete)
- [ ] Admin Actions
- [ ] Config Changes

### Speicherung

- [ ] Append-only/Immutable
- [ ] Verschlüsselt
- [ ] Retention: 7 Jahre
- [ ] Separater Zugriff

### Schema

```json
{
  "timestamp": "ISO 8601",
  "event_type": "category.action.outcome",
  "actor": {
    "id",
    "type",
    "ip"
  },
  "target": {
    "type",
    "id"
  },
  "outcome": "success|failure",
  "context": {
    "request_id",
    "tenant_id"
  },
  "metadata": {}
}
```

## Tools

| Bereich    | Tool                                |
|------------|-------------------------------------|
| Logging    | Elasticsearch / Loki / CloudWatch   |
| Metrics    | Prometheus / Datadog / CloudWatch   |
| Tracing    | Jaeger / Zipkin / Tempo             |
| Dashboards | Grafana / Datadog                   |
| Alerting   | Alertmanager / PagerDuty / Opsgenie |

## Dashboards

- [ ] API Overview (Traffic, Errors, Latency)
- [ ] Per-Endpoint Breakdown
- [ ] Error Details
- [ ] Dependency Health
- [ ] SLO Dashboard

Checkliste

Bevor du zum nächsten Artikel gehst, prüfe:

• [ ] Structured Logging (JSON) ist implementiert
• [ ] Log-Schema ist definiert und dokumentiert
• [ ] Request-ID ist in allen Logs und Responses
• [ ] Trace-ID für Distributed Tracing
• [ ] Basis-Metriken (Latency, Traffic, Errors) werden erfasst
• [ ] Percentile (p95, p99) statt Durchschnitt
• [ ] SLIs und SLOs sind definiert
• [ ] Alerts sind konfiguriert
• [ ] Audit-Logs für sicherheitsrelevante Aktionen
• [ ] PII wird maskiert oder nicht geloggt
• [ ] Observability-Checkliste ist dokumentiert

Wie es weitergeht

Im nächsten Teil geht es um Versionierung und Deprecation: Wie du API-Versionen managst, Breaking Changes ankündigst und alte Versionen graceful abschaltest.

Alle Teile der Serie: Serie: API-Design