6. Fallbacks y Health Checks

Descripción

Retry, circuit breaker y rate limiting reducen los errores — pero no los eliminan. Cuando todos tus retries se agotan, cuando el circuit está abierto y el servicio sigue caído, necesitas algo que sirva al usuario de todas formas. Los fallbacks son esa capa final: degradar tu servicio de forma controlada es mejor que un error 500. Y los health checks son la señal de alerta temprana que te dice que algo va a fallar antes de que lleguen los usuarios. En esta cápsula vas a implementar una cadena de fallbacks completa, un sistema de cache como respaldo, y health checks reales que Kubernetes puede consumir.

Tipos de fallback y cuándo usar cada uno

Jerarquía de fallbacks (de mayor a menor calidad):

1. Secondary Provider (mismo tipo de servicio, diferente proveedor)
   ├── Ejemplo: OpenAI falla → Anthropic
   ├── Calidad: ~equivalente
   ├── Costo: puede ser diferente
   └── Cuándo: tienes contrato con múltiples proveedores

2. Model Downgrade (mismo proveedor, modelo más pequeño/barato)
   ├── Ejemplo: gpt-4o falla → gpt-4o-mini
   ├── Calidad: menor (pero funcional)
   ├── Costo: menor
   └── Cuándo: el primary es el modelo más grande, mini como fallback

3. Cached Response (respuesta de un request anterior similar)
   ├── Ejemplo: "¿Cuál es el sentimiento de 'Me gusta' ?" → cache hit
   ├── Calidad: exacta si el contexto es el mismo
   ├── Costo: 0
   └── Cuándo: los requests se repiten frecuentemente

4. Simplified Processing (algoritmo más simple sin LLM)
   ├── Ejemplo: sentiment con keyword matching en vez de LLM
   ├── Calidad: menor, pero determinístico
   ├── Costo: 0 (no llama a ningún API)
   └── Cuándo: tienes una versión rule-based del algoritmo

5. Static Default (respuesta genérica predefinida)
   ├── Ejemplo: {"sentiment": "unknown", "score": 0.0, "confidence": 0.0}
   ├── Calidad: mínima (solo dice "no sé")
   ├── Costo: 0
   └── Cuándo: último recurso cuando todo falla

6. Graceful Error (error honesto con mensaje claro)
   ├── Ejemplo: {"error": "Servicio no disponible", "retry_after": 60}
   ├── Calidad: no hay respuesta útil, pero es honesto
   └── Cuándo: cuando ningún fallback puede dar una respuesta razonable

FallbackProvider completo

# src/infrastructure/fallback_provider.py
from typing import Optional, Callable
import time
import structlog
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
from src.infrastructure.error_classifier import ErrorCategory

log = structlog.get_logger()

class FallbackExhaustedError(LLMProviderError):
    """Todos los providers en la fallback chain fallaron."""
    def __init__(self, failures: list[dict]):
        super().__init__(
            message="All providers in fallback chain failed",
            category=ErrorCategory.OUTAGE,
            should_retry=False
        )
        self.failures = failures

class FallbackProvider:
    """
    Fallback chain: intenta múltiples providers en orden.
    
    El primer provider que tenga éxito gana.
    Si todos fallan, lanza FallbackExhaustedError.
    
    Permite distinguir respuestas "degradadas" (usando un fallback)
    de respuestas normales.
    """
    
    def __init__(
        self,
        providers: list[LLMProvider],
        names: list[str] = None,        # Nombres para logging (opcional)
        static_fallback: str = None,    # Respuesta estática si todos fallan
        on_fallback: Callable = None    # Callback cuando se activa un fallback
    ):
        if len(providers) < 2:
            raise ValueError("FallbackProvider requires at least 2 providers")
        
        self._providers = providers
        self._names = names or [type(p).__name__ for p in providers]
        self._static_fallback = static_fallback
        self._on_fallback = on_fallback
        
        # Métricas
        self._fallback_counts = {name: 0 for name in self._names}
        self._total_calls = 0
        self._degraded_calls = 0  # Calls que usaron un fallback
    
    def complete(self, messages: list[dict], **kwargs) -> str:
        """
        Intenta cada provider en orden.
        Devuelve el resultado del primer provider que funcione.
        """
        self._total_calls += 1
        failures = []
        primary_failed = False
        
        for i, (provider, name) in enumerate(zip(self._providers, self._names)):
            try:
                result = provider.complete(messages, **kwargs)
                
                if primary_failed:
                    # No es el primary — estamos usando un fallback
                    self._degraded_calls += 1
                    self._fallback_counts[name] += 1
                    
                    log.warning(
                        "fallback_provider_used",
                        primary_failed=self._names[0],
                        fallback_used=name,
                        fallback_index=i,
                        failures_before=[f["provider"] for f in failures]
                    )
                    
                    if self._on_fallback:
                        self._on_fallback(name, failures)
                
                return result
            
            except Exception as e:
                primary_failed = True
                failures.append({
                    "provider": name,
                    "error_type": type(e).__name__,
                    "error_message": str(e)[:200]
                })
                
                log.warning(
                    "provider_failed_trying_next",
                    provider=name,
                    next_provider=self._names[i + 1] if i + 1 < len(self._names) else "static_fallback",
                    error_type=type(e).__name__
                )
        
        # Todos los providers fallaron
        if self._static_fallback is not None:
            log.error(
                "all_providers_failed_using_static",
                failures=[f["provider"] for f in failures]
            )
            self._degraded_calls += 1
            return self._static_fallback
        
        raise FallbackExhaustedError(failures=failures)
    
    @property
    def is_degraded(self) -> bool:
        """True si la mayoría de calls recientes usaron fallback."""
        if self._total_calls == 0:
            return False
        return (self._degraded_calls / self._total_calls) > 0.5
    
    def get_metrics(self) -> dict:
        return {
            "total_calls": self._total_calls,
            "degraded_calls": self._degraded_calls,
            "degraded_percent": round(
                (self._degraded_calls / self._total_calls * 100) if self._total_calls > 0 else 0, 1
            ),
            "fallback_counts": self._fallback_counts
        }

Comunicar degradación al usuario

# src/app/routers/sentiment.py
from fastapi import Depends
from src.infrastructure.fallback_provider import FallbackProvider
from src.domain.sentiment_service import analyze_sentiment
import structlog

log = structlog.get_logger()

@router.post("/analyze")
async def analyze_sentiment_endpoint(
    body: AnalyzeRequest,
    provider: LLMProvider = Depends(get_llm_provider)
):
    result = analyze_sentiment(body.text, provider)
    
    # Detectar si se usó un fallback
    degraded = False
    degraded_reason = None
    
    if isinstance(provider, FallbackProvider) and provider.is_degraded:
        degraded = True
        degraded_reason = "Servicio con capacidad reducida — usando modelo de respaldo"
    
    response = {
        **result,
        "degraded": degraded,
        "degraded_reason": degraded_reason if degraded else None
    }
    
    # Logear para métricas de degradación
    if degraded:
        log.info("response_served_degraded", degraded_reason=degraded_reason)
    
    return response

# ¿Por qué comunicar la degradación?
# 1. El usuario puede decidir si confiar en la respuesta
# 2. El equipo de producto puede decidir cuándo desactivar features degradados
# 3. Los tests de integración pueden verificar que el fallback funciona
# 4. No es ético devolver una respuesta de menor calidad como si fuera normal

Response caching como fallback

# src/infrastructure/cache_provider.py
from typing import Optional
import time
import hashlib
import json
import structlog
from src.infrastructure.llm_provider import LLMProvider

log = structlog.get_logger()

class CachedProvider:
    """
    Wrappea un LLMProvider con caching de respuestas.
    
    En modo normal: devuelve del caché si hay hit, llama al provider si hay miss.
    En modo fallback: si el provider falla, devuelve el último valor en caché.
    """
    
    def __init__(
        self,
        inner: LLMProvider,
        ttl_seconds: int = 300,      # 5 minutos por defecto
        fallback_ttl_seconds: int = 3600,  # En fallback, acepta caché de hasta 1h
        max_cache_size: int = 1000
    ):
        self._inner = inner
        self._ttl = ttl_seconds
        self._fallback_ttl = fallback_ttl_seconds
        self._cache: dict[str, dict] = {}
        self._max_size = max_cache_size
        
        # Métricas
        self._hits = 0
        self._misses = 0
        self._fallback_hits = 0
    
    def _cache_key(self, messages: list[dict]) -> str:
        """Genera una clave de caché basada en los mensajes."""
        content = json.dumps(messages, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()[:16]
    
    def _get_from_cache(self, key: str, max_age: int) -> Optional[str]:
        """Devuelve el valor del caché si existe y no ha expirado."""
        if key in self._cache:
            entry = self._cache[key]
            age = time.time() - entry["timestamp"]
            if age < max_age:
                return entry["value"]
        return None
    
    def _set_cache(self, key: str, value: str) -> None:
        """Guarda en el caché, evictando si es necesario."""
        if len(self._cache) >= self._max_size:
            # Evict el más antiguo (LRU simple)
            oldest_key = min(self._cache, key=lambda k: self._cache[k]["timestamp"])
            del self._cache[oldest_key]
        
        self._cache[key] = {
            "value": value,
            "timestamp": time.time()
        }
    
    def complete(self, messages: list[dict], **kwargs) -> str:
        key = self._cache_key(messages)
        
        # Intentar caché fresco
        cached = self._get_from_cache(key, self._ttl)
        if cached is not None:
            self._hits += 1
            log.debug("cache_hit", key=key)
            return cached
        
        self._misses += 1
        
        try:
            result = self._inner.complete(messages, **kwargs)
            self._set_cache(key, result)
            return result
        except Exception as e:
            # Provider falló — intentar caché más antiguo como fallback
            stale_cached = self._get_from_cache(key, self._fallback_ttl)
            if stale_cached is not None:
                self._fallback_hits += 1
                log.warning(
                    "cache_stale_fallback_used",
                    key=key,
                    provider_error=type(e).__name__
                )
                return stale_cached
            raise  # Si no hay caché, propagar el error

Health checks reales (no cosméticos)

# src/health/checks.py
from fastapi import APIRouter
from fastapi.responses import JSONResponse
import time
import structlog
from src.config import get_settings

log = structlog.get_logger()
router = APIRouter(prefix="/health", tags=["health"])

@router.get("/live")
async def liveness():
    """
    Liveness probe: ¿está el proceso vivo y puede responder HTTP?
    
    Kubernetes usa esto para decidir si reiniciar el pod.
    Debe ser MUY simple — si esto falla, el proceso está corrupto.
    NO verificar dependencias externas aquí.
    """
    return {"status": "alive", "timestamp": time.time()}

@router.get("/ready")
async def readiness():
    """
    Readiness probe: ¿puede la app recibir tráfico ahora?
    
    Kubernetes usa esto para decidir si enviar tráfico al pod.
    Sí verifica dependencias — si OpenAI no responde, no enviar tráfico.
    """
    checks = {}
    all_ready = True
    
    # Verificar configuración
    try:
        settings = get_settings()
        checks["config"] = {"status": "ok"}
    except Exception as e:
        checks["config"] = {"status": "error", "detail": str(e)[:100]}
        all_ready = False
    
    # Verificar circuit breakers
    try:
        from src.app.dependencies import get_circuit_breaker_metrics
        cb_metrics = get_circuit_breaker_metrics()
        any_open = any(m["state"] == "open" for m in cb_metrics.values())
        checks["circuit_breakers"] = {
            "status": "degraded" if any_open else "ok",
            "details": cb_metrics
        }
        if any_open:
            all_ready = False  # No enviar tráfico si el circuit está abierto
    except Exception as e:
        checks["circuit_breakers"] = {"status": "unknown", "detail": str(e)[:100]}
    
    status_code = 200 if all_ready else 503
    
    return JSONResponse(
        content={
            "status": "ready" if all_ready else "not_ready",
            "checks": checks,
            "timestamp": time.time()
        },
        status_code=status_code
    )

@router.get("/deps")
async def dependency_check():
    """
    Dependency health check: ¿están las dependencias externas respondiendo?
    
    Usado para monitoring y alertas.
    NOT usado por Kubernetes para routing (eso es /ready).
    Puede ser más lento y hacer llamadas reales a dependencias.
    """
    settings = get_settings()
    checks = {}
    
    # Verificar OpenAI
    start = time.time()
    try:
        client = settings.create_openai_client()
        # Llamada ligera: listar modelos (mucho más barato que un completion)
        models = client.models.list()
        latency_ms = (time.time() - start) * 1000
        checks["openai"] = {
            "status": "ok",
            "latency_ms": round(latency_ms, 1),
            "models_available": len(list(models.data)) > 0
        }
    except Exception as e:
        latency_ms = (time.time() - start) * 1000
        checks["openai"] = {
            "status": "error",
            "latency_ms": round(latency_ms, 1),
            "error": type(e).__name__,
            "detail": str(e)[:200]
        }
    
    # Verificar rate limit status
    try:
        from src.app.dependencies import get_rate_limiter_metrics
        rate_metrics = get_rate_limiter_metrics()
        checks["rate_limits"] = {
            "status": "ok",
            "details": rate_metrics
        }
    except Exception as e:
        checks["rate_limits"] = {"status": "unknown"}
    
    all_ok = all(c.get("status") == "ok" for c in checks.values())
    
    log.info(
        "dependency_check_completed",
        all_ok=all_ok,
        checks={k: v.get("status") for k, v in checks.items()}
    )
    
    return JSONResponse(
        content={"checks": checks, "timestamp": time.time()},
        status_code=200 if all_ok else 503
    )

Integración en main.py

# src/app/main.py (fragmento de integración de health checks)
from fastapi import FastAPI
from src.health.checks import router as health_router

def create_app() -> FastAPI:
    app = FastAPI(title="Sentiment Analysis API")
    
    # Health checks antes de cualquier middleware de autenticación
    # para que Kubernetes pueda acceder sin auth
    app.include_router(health_router)
    
    # ... resto de la configuración
    
    return app

Ejercicios

Ejercicio 1: Decidir el tipo de fallback

Para una app que genera resúmenes de artículos de noticias:

1. OpenAI gpt-4o falla → ¿qué fallback usar?
2. Todas las APIs de LLM están caídas → ¿qué devolver?
3. El mismo artículo fue procesado hace 30 minutos → ¿usar cache?

Ejercicio 2: Health check para Kubernetes

¿Qué diferencia hay entre GET /health/live y GET /health/ready? ¿Por qué Kubernetes necesita ambos?

Ejercicio 3: Diseñar la cadena de fallbacks

Tu app tiene un endpoint /summarize que resume artículos de noticias. Diseña la cadena de fallbacks completa, indicando qué devuelves en cada nivel:

# Nivel 1: Primary provider (gpt-4o)
# → Resumen completo y detallado

# Nivel 2: Secondary provider (gpt-4o-mini)
# → Resumen más corto pero funcional
# → Comunicar: "Usando modelo de respaldo"

# Nivel 3: Cache de respuestas previas
# → Si este artículo ya fue resumido, devolver el cache
# → Comunicar: "Mostrando resumen previo"

# Nivel 4: Procesamiento simplificado
# → Extraer las primeras 3 oraciones del artículo como "resumen"
# → Sin LLM, solo text processing
# → Comunicar: "Resumen básico generado automáticamente"

# Nivel 5: Static default
# → {"summary": null, "status": "unavailable",
#    "message": "El servicio de resúmenes no está disponible"}

Ejercicio 4: Implementar degraded flag en la respuesta

Modifica el endpoint para comunicar al frontend cuándo está sirviendo una respuesta degradada:

@router.post("/summarize")
async def summarize(body: SummarizeRequest, provider = Depends(get_llm_provider)):
    result = summarize_article(body.text, provider)
    
    response = {
        **result,
        "degraded": False,
        "degraded_reason": None,
        "provider_used": "primary"
    }
    
    if isinstance(provider, FallbackProvider):
        if provider.is_degraded:
            response["degraded"] = True
            response["degraded_reason"] = provider.degraded_reason
            response["provider_used"] = provider.last_used_provider
    
    return response

Troubleshooting

"Mi fallback nunca se activa"

Verifica que el FallbackProvider recibe las excepciones correctas. Si tu RetryProvider hace reraise=True, propaga la excepción original. Si hace reraise=False, propaga tenacity.RetryError. El FallbackProvider necesita catchear el tipo correcto. Agrega un log temporal en el catch para ver qué tipo de excepción llega.

"El health check de readiness siempre devuelve 503"

Revisa qué checks están fallando. La respuesta incluye un checks dict con el estado de cada dependencia. Si circuit_breakers muestra "state": "open", es porque el circuit está abierto — normal durante un outage. Si config falla, tienes un problema de configuración. Llama a GET /health/deps para ver el detalle completo de cada dependencia.

"¿Debo cachear todas las respuestas del LLM?"

No necesariamente. Cachea cuando el input es repetitivo y la respuesta es determinista (o aceptablemente similar). Para análisis de sentimiento con los mismos textos, el cache es muy efectivo. Para chatbots con conversaciones únicas, el cache rara vez matchea. Usa un TTL corto (5-15 minutos) para empezar y ajusta según tu hit rate.

"Kubernetes reinicia mi pod constantemente"

Probablemente estás chequeando OpenAI en tu liveness probe. Si OpenAI tiene problemas, tu liveness falla → K8s reinicia el pod → el pod nuevo tiene el mismo problema → loop infinito. Regla de oro: liveness solo verifica que el proceso Python está corriendo. Las dependencias externas van en readiness.

Resumen

• Fallback hierarchy: secondary provider → model downgrade → cached response → simplified processing → static default → graceful error
• Siempre comunicar degradación: el usuario y el sistema de monitoring deben saber cuando se usa un fallback
• Liveness vs Readiness: liveness es "¿estoy vivo?", readiness es "¿puedo recibir tráfico?"
• Health checks reales: verificar que OpenAI responde, no solo que el servidor HTTP funciona
• Cache como fallback: útil para requests repetitivos, con TTL extendido en modo fallback

Recursos adicionales

1. Kubernetes Probes — Documentación de liveness/readiness
2. Graceful Degradation (MDN) — El concepto
3. Caching Strategies — Patrones de cache
4. Microsoft — Health Endpoint Monitoring Pattern — El patrón de health checks