""" observability.py — NutriFlow API Instrumentación completa: logs JSON estructurados, métricas Prometheus y correlation IDs distribuidos para FastAPI.

Skill: observabilidad-python-logs-metricas-trazas (CULTIVA IA) Dependencias: structlog, prometheus_client, fastapi, httpx """

─── 1. LOGGING ESTRUCTURADO ─────────────────────────────────────────────────

import logging import structlog from contextvars import ContextVar import uuid import time from contextlib import contextmanager from functools import wraps

Variable de contexto para propagar el correlation ID

correlation_id: ContextVar[str] = ContextVar("correlation_id", default="")

def configure_logging(log_level: str = "INFO") -> None: """Configura structlog con salida JSON y campos consistentes.""" structlog.configure( processors=[ structlog.contextvars.merge_contextvars, # Inyecta correlation_id automáticamente structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.JSONRenderer(), ], wrapper_class=structlog.make_filtering_bound_logger( getattr(logging, log_level.upper()) ), context_class=dict, logger_factory=structlog.PrintLoggerFactory(), cache_logger_on_first_use=True, )

def set_correlation_id(cid: str | None = None) -> str: """Establece correlation ID en el contexto actual.""" cid = cid or str(uuid.uuid4()) correlation_id.set(cid) structlog.contextvars.bind_contextvars(correlation_id=cid) return cid

─── 2. CUATRO SEÑALES DORADAS — PROMETHEUS ───────────────────────────────────

from prometheus_client import Counter, Histogram, Gauge

LATENCIA: cuánto tardan los requests

REQUEST_LATENCY = Histogram( "nutriflow_http_request_duration_seconds", "Latencia de requests HTTP en segundos", ["method", "endpoint", "status", "plan_type"], buckets=[0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], )

TRÁFICO: tasa de requests

REQUEST_COUNT = Counter( "nutriflow_http_requests_total", "Total de requests HTTP", ["method", "endpoint", "status", "clinic_tier"], )

ERRORES: tasa de errores

ERROR_COUNT = Counter( "nutriflow_http_errors_total", "Total de errores HTTP", ["method", "endpoint", "error_type"], )

SATURACIÓN: conexiones de BD en uso

DB_POOL_USAGE = Gauge( "nutriflow_db_connection_pool_used", "Conexiones de BD activas", )

Métrica específica: duración de generación de planes (sub-operaciones)

LLM_CALL_DURATION = Histogram( "nutriflow_llm_call_duration_seconds", "Duración de llamadas al servicio LLM", ["plan_type"], buckets=[0.5, 1, 2, 4, 8, 15, 30], )

─── 3. MIDDLEWARE FASTAPI — CORRELATION ID ──────────────────────────────────

from fastapi import FastAPI, Request, Response import httpx

app = FastAPI(title="NutriFlow API", version="2.1.0")

@app.middleware("http") async def correlation_middleware(request: Request, call_next): """ Middleware que: 1. Lee X-Correlation-ID entrante (o genera uno nuevo) 2. Lo inyecta en el contexto de logs 3. Lo devuelve en la cabecera de respuesta """ cid = request.headers.get("X-Correlation-ID") or str(uuid.uuid4()) set_correlation_id(cid)

logger = structlog.get_logger()
logger.info(
    "request_started",
    method=request.method,
    path=request.url.path,
    client_ip=request.client.host if request.client else "unknown",
)

response = await call_next(request)
response.headers["X-Correlation-ID"] = cid
return response

─── 4. PROPAGACIÓN A SERVICIOS DOWNSTREAM ───────────────────────────────────

async def call_llm_service(patient_id: str, plan_type: str) -> dict: """ Llama al servicio LLM interno propagando el correlation ID. Cardinalidad: NO pasamos patient_id como label de métrica. """ logger = structlog.get_logger() cid = correlation_id.get()

async with httpx.AsyncClient(timeout=30.0) as client:
    logger.info(
        "llm_service_call_started",
        service="nutrition-llm-service",
        plan_type=plan_type,
    )
    try:
        response = await client.post(
            "http://nutrition-llm-service/generate",
            json={"patient_id": patient_id, "plan_type": plan_type},
            headers={"X-Correlation-ID": cid},   # <── propagación
        )
        response.raise_for_status()
        logger.info(
            "llm_service_call_completed",
            service="nutrition-llm-service",
            status_code=response.status_code,
        )
        return response.json()
    except httpx.TimeoutException as e:
        logger.error(
            "llm_service_timeout",
            service="nutrition-llm-service",
            error=str(e),
        )
        ERROR_COUNT.labels(
            method="POST",
            endpoint="/api/v1/plans/generate",
            error_type="TimeoutException",
        ).inc()
        raise

─── 5. CONTEXT MANAGER timed_operation ──────────────────────────────────────

@contextmanager def timed_operation(name: str, **extra_fields): """ Context manager para medir y loguear sub-operaciones. Emite DEBUG al inicio, INFO al completar y ERROR si falla. """ logger = structlog.get_logger() start = time.perf_counter() logger.debug("operation_started", operation=name, **extra_fields)

try:
    yield
except Exception as e:
    elapsed_ms = (time.perf_counter() - start) * 1000
    logger.error(
        "operation_failed",
        operation=name,
        duration_ms=round(elapsed_ms, 2),
        error_type=type(e).__name__,
        error=str(e),
        **extra_fields,
    )
    raise
else:
    elapsed_ms = (time.perf_counter() - start) * 1000
    logger.info(
        "operation_completed",
        operation=name,
        duration_ms=round(elapsed_ms, 2),
        **extra_fields,
    )

─── 6. ENDPOINTS INSTRUMENTADOS ─────────────────────────────────────────────

from fastapi import HTTPException from pydantic import BaseModel

class PlanRequest(BaseModel): patient_id: str plan_type: str = "standard" # standard | premium | clinic_pro clinic_tier: str = "basic" # free | basic | enterprise

@app.post("/api/v1/plans/generate") async def generate_plan(body: PlanRequest): """ Genera un plan nutricional personalizado. - Alta latencia esperada (3-8s por LLM) - Instrumentado con timed_operation para sub-etapas """ logger = structlog.get_logger() start = time.perf_counter()

try:
    patient_data = None
    plan_data = None

    # Sub-operación 1: fetch del paciente
    with timed_operation("fetch_patient", patient_id=body.patient_id):
        # En producción: await patient_repo.get(body.patient_id)
        patient_data = {"id": body.patient_id, "age": 34, "restrictions": ["gluten"]}

    # Sub-operación 2: llamada al LLM
    llm_start = time.perf_counter()
    with timed_operation("call_llm", plan_type=body.plan_type):
        plan_data = await call_llm_service(body.patient_id, body.plan_type)
    llm_duration = time.perf_counter() - llm_start
    LLM_CALL_DURATION.labels(plan_type=body.plan_type).observe(llm_duration)

    # Sub-operación 3: escritura en caché
    with timed_operation("cache_write", plan_type=body.plan_type):
        # En producción: await redis.setex(f"plan:{body.patient_id}", 3600, plan_data)
        pass

    duration = time.perf_counter() - start
    REQUEST_COUNT.labels(
        method="POST",
        endpoint="/api/v1/plans/generate",
        status="200",
        clinic_tier=body.clinic_tier,      # Cardinalidad acotada ✓
    ).inc()
    REQUEST_LATENCY.labels(
        method="POST",
        endpoint="/api/v1/plans/generate",
        status="200",
        plan_type=body.plan_type,           # Cardinalidad acotada ✓
    ).observe(duration)

    logger.info(
        "plan_generated",
        plan_type=body.plan_type,
        clinic_tier=body.clinic_tier,
        duration_ms=round(duration * 1000, 2),
    )
    return {"status": "ok", "plan": plan_data}

except Exception as e:
    duration = time.perf_counter() - start
    REQUEST_COUNT.labels(
        method="POST",
        endpoint="/api/v1/plans/generate",
        status="500",
        clinic_tier=body.clinic_tier,
    ).inc()
    REQUEST_LATENCY.labels(
        method="POST",
        endpoint="/api/v1/plans/generate",
        status="500",
        plan_type=body.plan_type,
    ).observe(duration)
    raise HTTPException(status_code=500, detail="Error generando plan")

@app.get("/api/v1/patients/{patient_id}") async def get_patient(patient_id: str, clinic_tier: str = "basic"): """ Obtiene datos de un paciente. SLA objetivo: <100ms p99. Nivel de log: DEBUG para lookups individuales (INFO solo en ERROR). """ logger = structlog.get_logger() start = time.perf_counter()

with timed_operation("fetch_patient_db", patient_id=patient_id):
    # En producción: await db.fetch_one(...)
    patient = {"id": patient_id, "name": "Ana García", "clinic_tier": clinic_tier}

duration = time.perf_counter() - start
REQUEST_COUNT.labels(
    method="GET",
    endpoint="/api/v1/patients/{patient_id}",
    status="200",
    clinic_tier=clinic_tier,
).inc()
REQUEST_LATENCY.labels(
    method="GET",
    endpoint="/api/v1/patients/{patient_id}",
    status="200",
    plan_type="none",
).observe(duration)

# DEBUG: información interna, no llena logs de producción
logger.debug("patient_cache_lookup", patient_id=patient_id, cache_hit=False)

return patient

@app.post("/api/v1/intake") async def register_intake(data: dict, clinic_tier: str = "basic"): """ Registro de ingesta diaria — alto tráfico (~500 req/min). Nivel WARNING si la ingesta supera el límite calórico recomendado. """ logger = structlog.get_logger() start = time.perf_counter()

calories = data.get("calories", 0)
limit = data.get("daily_limit", 2000)

if calories > limit * 0.9:
    logger.warning(
        "caloric_limit_approaching",
        calories=calories,
        limit=limit,
        pct_used=round(calories / limit * 100, 1),
    )

REQUEST_COUNT.labels(
    method="POST",
    endpoint="/api/v1/intake",
    status="201",
    clinic_tier=clinic_tier,
).inc()
REQUEST_LATENCY.labels(
    method="POST",
    endpoint="/api/v1/intake",
    status="201",
    plan_type="none",
).observe(time.perf_counter() - start)

return {"status": "registered"}

─── 7. STARTUP ───────────────────────────────────────────────────────────────

@app.on_event("startup") async def startup(): configure_logging("INFO") logger = structlog.get_logger() logger.info("nutriflow_api_started", version="2.1.0", env="production")

─── EJEMPLO DE LOGS JSON EMITIDOS ───────────────────────────────────────────

""" Salida estructurada real al procesar POST /api/v1/plans/generate:

{"level":"info","timestamp":"2026-06-15T09:12:03.421Z","correlation_id":"a3f7-...-9e2b","event":"request_started","method":"POST","path":"/api/v1/plans/generate","client_ip":"10.0.0.5"} {"level":"debug","timestamp":"2026-06-15T09:12:03.422Z","correlation_id":"a3f7-...-9e2b","event":"operation_started","operation":"fetch_patient","patient_id":"pt_8842"} {"level":"info","timestamp":"2026-06-15T09:12:03.489Z","correlation_id":"a3f7-...-9e2b","event":"operation_completed","operation":"fetch_patient","duration_ms":67.1,"patient_id":"pt_8842"} {"level":"debug","timestamp":"2026-06-15T09:12:03.490Z","correlation_id":"a3f7-...-9e2b","event":"operation_started","operation":"call_llm","plan_type":"premium"} {"level":"info","timestamp":"2026-06-15T09:12:06.821Z","correlation_id":"a3f7-...-9e2b","event":"operation_completed","operation":"call_llm","duration_ms":3331.4,"plan_type":"premium"} {"level":"info","timestamp":"2026-06-15T09:12:06.822Z","correlation_id":"a3f7-...-9e2b","event":"plan_generated","plan_type":"premium","clinic_tier":"enterprise","duration_ms":3401.2}

Métricas Prometheus expuestas en /metrics: nutriflow_http_request_duration_seconds_bucket{method="POST",endpoint="/api/v1/plans/generate",status="200",plan_type="premium",le="5"} 14 nutriflow_http_requests_total{method="POST",endpoint="/api/v1/plans/generate",status="200",clinic_tier="enterprise"} 147 nutriflow_llm_call_duration_seconds_bucket{plan_type="premium",le="4"} 89 nutriflow_db_connection_pool_used 12 """