Observabilidad Amigable para Operaciones: Guía de Implementación en FastAPI para Logs, Métricas y Trazas — Request ID, Logs JSON, Prometheus, OpenTelemetry y Diseño de Dashboards

Resumen (Pirámide Invertida)

• El objetivo es establecer “visibilidad” para ejecutar FastAPI en producción de modo que puedas responder a incidentes y mejorar el rendimiento rápidamente.
• Comienza añadiendo un Request ID, logs estructurados (JSON) y checks de liveness/readiness.
• Luego, expón métricas de la app (latencia, rendimiento, tasa de error — las métricas RED) en formato Prometheus y prepara condiciones de alerta.
• Finalmente, introduce trazado distribuido con OpenTelemetry para visualizar desglose de latencias entre APIs externas y bases de datos e identificar cuellos de botella.

Quién se Beneficia (Personas Concretas)

• Aprendiz A (estudiante, dev en solitario):
  La API funciona pero a veces se vuelve lenta por causas desconocidas. Quiere construir visibilidad paso a paso, empezando con logs y métricas.
• Equipo Pequeño B (taller de 3 personas):
  El análisis de causa raíz tarda demasiado durante incidentes. Quiere Request IDs para rastreo y un flujo de autonotificación cuando sube la tasa de errores.
• Equipo SaaS C (startup):
  La latencia se propaga entre múltiples servicios. Quiere trazado distribuido con OpenTelemetry para ver rápidamente dónde se invierte el tiempo.

1. El Mínimo Conjunto de Observabilidad: Empieza Solo con Esto

La observabilidad se apoya en tres pilares: logs, métricas y trazas. Conjunto mínimo inicial:

1. Request ID (correlation ID)
2. Logs estructurados (JSON)
3. Health checks (liveness/readiness)

1.1 Middleware para Adjuntar un Request ID

# app/obsv/request_id.py
import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp, Receive, Scope, Send

REQUEST_ID_HEADER = "X-Request-ID"

class RequestIdMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: ASGIApp) -> None:
        super().__init__(app)

    async def dispatch(self, request, call_next):
        rid = request.headers.get(REQUEST_ID_HEADER) or str(uuid.uuid4())
        response = await call_next(request)
        response.headers[REQUEST_ID_HEADER] = rid
        # Guardarlo en request.state facilita referenciarlo en logs, etc.
        request.state.request_id = rid
        return response

1.2 Logging en JSON (solo logging estándar)

# app/obsv/logging.py
import json, logging, sys
from typing import Any

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            "time": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        if hasattr(record, "request_id"):
            payload["request_id"] = record.request_id
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False)

def setup_logging(level: str = "INFO"):
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.handlers[:] = [handler]
    root.setLevel(level.upper())

1.3 Integración en la App (con Health Checks)

# app/main.py
from fastapi import FastAPI, Request
from app.obsv.request_id import RequestIdMiddleware, REQUEST_ID_HEADER
from app.obsv.logging import setup_logging
import logging

app = FastAPI(title="Observability-Ready API")
app.add_middleware(RequestIdMiddleware)
setup_logging("INFO")
log = logging.getLogger("app")

@app.get("/health/liveness")
def liveness():
    return {"status": "alive"}

@app.get("/health/readiness")
def readiness():
    # En producción, verifica conectividad a DB, etc.
    return {"status": "ready"}

@app.get("/")
def root(request: Request):
    # Incluir Request ID en logs
    extra = {"request_id": getattr(request.state, "request_id", None)}
    log.info("hello request", extra=extra)
    return {"ok": True, "request_id": request.headers.get(REQUEST_ID_HEADER)}

Puntos de decisión

• Adjunta el Request ID tanto en el header como en la respuesta para facilitar referencias cruzadas con logs de upstream (Nginx/ALB).
• Los logs JSON son fáciles de agregar y buscar—una vez migras, no querrás volver al texto plano.

2. Métricas: Publica RED con Prometheus

Las métricas son series numéricas para observar tendencias en el tiempo. Cubre primero RED:

• Rate: número de requests
• Errors: número de errores (4xx/5xx)
• Duration: distribución de latencia (histograma)

2.1 Añade un Endpoint de Prometheus

# app/obsv/metrics.py
import time
from typing import Callable
from starlette.middleware.base import BaseHTTPMiddleware
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response

REQ_COUNTER = Counter("http_requests_total", "request count", ["method", "path", "code"])
REQ_ERRORS = Counter("http_requests_error_total", "error count", ["method", "path", "code"])
REQ_LATENCY = Histogram(
    "http_request_duration_seconds",
    "request latency",
    ["method", "path", "code"],
    buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5]
)

class PrometheusMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next: Callable):
        start = time.perf_counter()
        response = await call_next(request)
        dur = time.perf_counter() - start
        labels = {
            "method": request.method,
            "path": request.url.path,  # En prod, normaliza params a plantillas
            "code": str(response.status_code),
        }
        REQ_COUNTER.labels(**labels).inc()
        REQ_LATENCY.labels(**labels).observe(dur)
        if response.status_code >= 400:
            REQ_ERRORS.labels(**labels).inc()
        return response

def metrics_endpoint():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

2.2 Integra en la App

# app/main.py (adiciones)
from app.obsv.metrics import PrometheusMiddleware, metrics_endpoint

app.add_middleware(PrometheusMiddleware)

@app.get("/metrics")
def metrics():
    return metrics_endpoint()

2.3 Dashboard Mínimo (PromQL)

• Latencia P90

  histogram_quantile(
    0.90,
    sum(rate(http_request_duration_seconds_bucket[5m])) by (le)
  )
  

• Tasa de Error

  sum(rate(http_requests_error_total[5m])) /
  sum(rate(http_requests_total[5m]))
  

• QPS

  sum(rate(http_requests_total[1m]))
  

Puntos de decisión

• Normaliza parámetros de ruta (p. ej., /users/123) para evitar explosión de cardinalidad. Usa nombres de ruta de Starlette o un mapeo propio.
• Ajusta buckets para que P50/P90/P99 sean estables y útiles.

3. Haz los Logs Más Útiles: Resumen por Request

Pon latencia, estado, path, Request ID y (si aplica) user ID en una sola línea para buscar fácilmente.

# app/obsv/access_log.py
import time, logging
from starlette.middleware.base import BaseHTTPMiddleware
from typing import Callable

log = logging.getLogger("access")

class AccessLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next: Callable):
        start = time.perf_counter()
        response = await call_next(request)
        dur_ms = int((time.perf_counter() - start) * 1000)
        rid = getattr(request.state, "request_id", None)
        extra = {"request_id": rid}
        log.info(
            f'{request.client.host} {request.method} {request.url.path} '
            f'{response.status_code} {dur_ms}ms',
            extra=extra
        )
        return response

Integración en la app:

# app/main.py (adiciones)
from app.obsv.access_log import AccessLogMiddleware
app.add_middleware(AccessLogMiddleware)

Puntos de decisión

• Los access logs deben ser una línea por request.
• En caso de duda, estilo NCSA + campos extra (request_id/bytes, etc.) es suficiente.

4. Trazado Distribuido con OpenTelemetry

El trazado distribuido visualiza cada segmento de procesamiento (span) de un request en el tiempo. Es potente para hallar cuellos de botella.

4.1 Dependencias (Ejemplo)

opentelemetry-distro
opentelemetry-exporter-otlp
opentelemetry-instrumentation-fastapi
opentelemetry-instrumentation-requests

4.2 Ejecución de Una Línea (Mínimo)

opentelemetry-instrument \
  --traces_exporter otlp \
  --service_name fastapi-app \
  uvicorn app.main:app --host 0.0.0.0 --port 8000

Configura endpoint y muestreo vía variables de entorno (ejemplos):

OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317
OTEL_TRACES_SAMPLER=parentbased_traceidratio
OTEL_TRACES_SAMPLER_ARG=0.1  # muestreo 10%

4.3 Crear un Span Manualmente en Código

# app/obsv/trace.py
from opentelemetry import trace
tracer = trace.get_tracer("app")

def heavy_calc(x: int) -> int:
    with tracer.start_as_current_span("heavy_calc"):
        # Trabajo real (ejemplo)
        s = 0
        for i in range(x):
            s += i * i
        return s

4.4 Visualiza también HTTP/DB Externos

• Con opentelemetry-instrumentation-requests, las llamadas HTTP via requests generan spans automáticamente.
• Añade instrumentación de DB (p. ej., SQLAlchemy) para capturar tiempos de query.

Puntos de decisión

• Usa muestreo en producción (tasa fija o tail sampling).
• No es obligatorio que Request ID y Trace ID coincidan, pero ofrece enlaces cruzados para navegación log↔trace.

5. Dashboards y Alertas: Qué Vigilar

Vistas de alta frecuencia:

• RED: QPS, tasa de errores, latencia P90
• Recursos: CPU, memoria, GC count, FD count
• DB: conexiones, conteo de consultas lentas, P95 de query
• APIs externas: tasa de fallo, conteo de reintentos
• Colas (p. ej., Celery): longitud, demora, tasa de fallos

Ejemplos de Alertas

• Tasa de error > 2% sostenida por 5 minutos
• P90 > 300 ms sostenida por 10 minutos
• Fallos de readiness (detener despliegues progresivos)

Puntos de decisión

• Añade duraciones para que picos breves no te despierten constantemente.
• Empieza con un conjunto pequeño y curado de alertas para reducir falsos positivos.

6. Plantilla End-to-End (Mínima)

# app/main.py (resumen)
from fastapi import FastAPI, Request
from app.obsv.request_id import RequestIdMiddleware
from app.obsv.logging import setup_logging
from app.obsv.metrics import PrometheusMiddleware, metrics_endpoint
from app.obsv.access_log import AccessLogMiddleware
import logging

setup_logging("INFO")
app = FastAPI(title="Obs API")
app.add_middleware(RequestIdMiddleware)
app.add_middleware(PrometheusMiddleware)
app.add_middleware(AccessLogMiddleware)
log = logging.getLogger("app")

@app.get("/health/liveness")
def liveness():
    return {"status": "alive"}

@app.get("/health/readiness")
def readiness():
    return {"status": "ready"}

@app.get("/metrics")
def metrics():
    return metrics_endpoint()

@app.get("/calc")
def calc(n: int, request: Request):
    s = sum(i*i for i in range(n))
    log.info("calc done", extra={"request_id": getattr(request.state, "request_id", None)})
    return {"n": n, "sum": s}

7. Consejos Operacionales: Pequeños Detalles que Importan en Producción

• No loguees PII. Limítate a mensajes e IDs.
• Niveles de log: empieza en INFO, usa ERROR para excepciones y WARN para fallos de validación esperados.
• Usa tasas de muestreo y verbosidad diferentes para staging vs producción.
• Registra eventos de negocio críticos (p. ej., creación de órdenes) en logs y métricas.
• Durante despliegues, confía en readiness para el monitoreo externo y no romper rolling updates.

8. Seguridad y Privacidad

• Nunca incluyas tokens o contraseñas en trazas o logs. Enmascara headers.
• Loguear cuerpos de petición/respuesta automáticamente es arriesgado (fuga de PII). Prefiere resúmenes.
• No incluyas datos identificables en métricas (diseña labels con cuidado).
• Protege UIs de monitoreo (Grafana/Jaeger, etc.) con auth y restricciones de red.

9. Errores Comunes y Remedios

10. Planes de Adopción de Ejemplo por Audiencia

• Dev en solitario:
  1. Request ID + logs JSON → 2) /metrics → 3) Dashboard con solo P90 y tasa de error
• Equipo pequeño:
  1. Stack de monitoreo (Prometheus/Grafana) → 2) Dos alertas (tasa de error/P90) → 3) Trazado al 10%
• SaaS en crecimiento:
  1. Política de correlation ID entre servicios → 2) Spans a medida para dominios clave → 3) Visualiza métricas KPI de negocio

11. Hoja de Ruta (Despliegue por Fases)

1. Conjunto mínimo: Request ID, logs JSON, health checks.
2. Métricas: publica RED vía Prometheus con dashboard simple.
3. Alertas: empieza con dos — tasa de error y P90.
4. Trazado: introduce OpenTelemetry al 10%; visualiza APIs externas y DB clave.
5. Afinado: normalización de paths y limpieza de labels, refinamiento de dashboards, logging de auditoría para eventos de negocio clave.

12. Referencias

• FastAPI
  • FastAPI Official Docs
  • Middleware
• Starlette
  • Starlette Middleware
• Prometheus / Grafana
  • prometheus_client (Python)
  • Prometheus Docs
  • Grafana Docs
• OpenTelemetry
  • OpenTelemetry Python
  • Instrumentation for FastAPI
  • OTLP Exporter
• Logging
  • Python logging

Conclusión

• La ruta más rápida hacia una observabilidad exitosa es empezar con el conjunto mínimo: Request ID, logs JSON y health checks.
• Luego, publica las métricas RED vía Prometheus y prepara dashboards/alertas para P90 y tasa de error.
• Finalmente, adopta OpenTelemetry para trazado distribuido y entender el desglose de latencias en APIs externas y BD, atacando primero los problemas más impactantes.
• Vigila la cardinalidad, el manejo de PII y el mantenimiento de alertas, y haz de la “visibilidad” un lenguaje compartido en el equipo. Las operaciones se estabilizan y los ciclos de mejora se aceleran.