6. Fallbacks and Health Checks

Description

Retry, circuit breaker, and rate limiting reduce errors — but they don't eliminate them. When all your retries are exhausted, when the circuit is open and the service is still down, you need something that serves the user anyway. Fallbacks are that final layer: degrading your service in a controlled way is better than an error 500. And health checks are the early warning signal that tells you something is about to fail before the users arrive. In this capsule you'll implement a complete fallback chain, a cache-based backup system, and real health checks that Kubernetes can consume.

Types of fallback and when to use each one

Fallback hierarchy (from highest to lowest quality):

1. Secondary Provider (same type of service, different provider)
   ├── Example: OpenAI fails → Anthropic
   ├── Quality: ~equivalent
   ├── Cost: may be different
   └── When: you have a contract with multiple providers

2. Model Downgrade (same provider, smaller/cheaper model)
   ├── Example: gpt-4o fails → gpt-4o-mini
   ├── Quality: lower (but functional)
   ├── Cost: lower
   └── When: the primary is the largest model, mini as fallback

3. Cached Response (response from a similar previous request)
   ├── Example: "What is the sentiment of 'I like it'?" → cache hit
   ├── Quality: exact if the context is the same
   ├── Cost: 0
   └── When: requests repeat frequently

4. Simplified Processing (simpler algorithm without LLM)
   ├── Example: sentiment with keyword matching instead of LLM
   ├── Quality: lower, but deterministic
   ├── Cost: 0 (doesn't call any API)
   └── When: you have a rule-based version of the algorithm

5. Static Default (predefined generic response)
   ├── Example: {"sentiment": "unknown", "score": 0.0, "confidence": 0.0}
   ├── Quality: minimal (only says "I don't know")
   ├── Cost: 0
   └── When: last resort when everything fails

6. Graceful Error (honest error with a clear message)
   ├── Example: {"error": "Service unavailable", "retry_after": 60}
   ├── Quality: there's no useful response, but it's honest
   └── When: when no fallback can give a reasonable response

Complete FallbackProvider

# 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):
    """All providers in the fallback chain failed."""
    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: tries multiple providers in order.
    
    The first provider that succeeds wins.
    If all fail, it raises FallbackExhaustedError.
    
    Lets you distinguish "degraded" responses (using a fallback)
    from normal responses.
    """
    
    def __init__(
        self,
        providers: list[LLMProvider],
        names: list[str] = None,        # Names for logging (optional)
        static_fallback: str = None,    # Static response if all fail
        on_fallback: Callable = None    # Callback when a fallback is triggered
    ):
        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
        
        # Metrics
        self._fallback_counts = {name: 0 for name in self._names}
        self._total_calls = 0
        self._degraded_calls = 0  # Calls that used a fallback
    
    def complete(self, messages: list[dict], **kwargs) -> str:
        """
        Try each provider in order.
        Return the result of the first provider that works.
        """
        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:
                    # It's not the primary — we're using a 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__
                )
        
        # All providers failed
        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 if most recent calls used a 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
        }

Communicating degradation to the user

# 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)
    
    # Detect whether a fallback was used
    degraded = False
    degraded_reason = None
    
    if isinstance(provider, FallbackProvider) and provider.is_degraded:
        degraded = True
        degraded_reason = "Service with reduced capacity — using backup model"
    
    response = {
        **result,
        "degraded": degraded,
        "degraded_reason": degraded_reason if degraded else None
    }
    
    # Log for degradation metrics
    if degraded:
        log.info("response_served_degraded", degraded_reason=degraded_reason)
    
    return response

# Why communicate the degradation?
# 1. The user can decide whether to trust the response
# 2. The product team can decide when to disable degraded features
# 3. Integration tests can verify that the fallback works
# 4. It's not ethical to return a lower-quality response as if it were normal

Response caching as a 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:
    """
    Wraps an LLMProvider with response caching.
    
    In normal mode: returns from cache on a hit, calls the provider on a miss.
    In fallback mode: if the provider fails, returns the last cached value.
    """
    
    def __init__(
        self,
        inner: LLMProvider,
        ttl_seconds: int = 300,      # 5 minutes by default
        fallback_ttl_seconds: int = 3600,  # In fallback, accepts cache up to 1h old
        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
        
        # Metrics
        self._hits = 0
        self._misses = 0
        self._fallback_hits = 0
    
    def _cache_key(self, messages: list[dict]) -> str:
        """Generate a cache key based on the messages."""
        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]:
        """Return the cached value if it exists and hasn't expired."""
        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:
        """Save to the cache, evicting if necessary."""
        if len(self._cache) >= self._max_size:
            # Evict the oldest (simple LRU)
            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)
        
        # Try fresh cache
        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 failed — try older cache as a 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  # If there's no cache, propagate the error

Real health checks (not cosmetic)

# 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: is the process alive and able to respond to HTTP?
    
    Kubernetes uses this to decide whether to restart the pod.
    It must be VERY simple — if this fails, the process is corrupt.
    Do NOT check external dependencies here.
    """
    return {"status": "alive", "timestamp": time.time()}

@router.get("/ready")
async def readiness():
    """
    Readiness probe: can the app receive traffic now?
    
    Kubernetes uses this to decide whether to send traffic to the pod.
    It does check dependencies — if OpenAI doesn't respond, don't send traffic.
    """
    checks = {}
    all_ready = True
    
    # Check configuration
    try:
        settings = get_settings()
        checks["config"] = {"status": "ok"}
    except Exception as e:
        checks["config"] = {"status": "error", "detail": str(e)[:100]}
        all_ready = False
    
    # Check 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  # Don't send traffic if the circuit is open
    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: are the external dependencies responding?
    
    Used for monitoring and alerts.
    NOT used by Kubernetes for routing (that's /ready).
    It can be slower and make real calls to dependencies.
    """
    settings = get_settings()
    checks = {}
    
    # Check OpenAI
    start = time.time()
    try:
        client = settings.create_openai_client()
        # Lightweight call: list models (much cheaper than a 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]
        }
    
    # Check 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
    )

Integration in main.py

# src/app/main.py (health checks integration fragment)
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 before any authentication middleware
    # so Kubernetes can access without auth
    app.include_router(health_router)
    
    # ... rest of the configuration
    
    return app

Exercises

Exercise 1: Deciding the fallback type

For an app that generates summaries of news articles:

1. OpenAI gpt-4o fails → which fallback to use?
2. All the LLM APIs are down → what to return?
3. The same article was processed 30 minutes ago → use cache?

Exercise 2: Health check for Kubernetes

What's the difference between GET /health/live and GET /health/ready? Why does Kubernetes need both?

Exercise 3: Designing the fallback chain

Your app has a /summarize endpoint that summarizes news articles. Design the complete fallback chain, indicating what you return at each level:

# Level 1: Primary provider (gpt-4o)
# → Complete and detailed summary

# Level 2: Secondary provider (gpt-4o-mini)
# → Shorter but functional summary
# → Communicate: "Using backup model"

# Level 3: Cache of previous responses
# → If this article was already summarized, return the cache
# → Communicate: "Showing previous summary"

# Level 4: Simplified processing
# → Extract the first 3 sentences of the article as a "summary"
# → No LLM, just text processing
# → Communicate: "Basic summary generated automatically"

# Level 5: Static default
# → {"summary": null, "status": "unavailable",
#    "message": "The summarization service is not available"}

Exercise 4: Implementing a degraded flag in the response

Modify the endpoint to communicate to the frontend when it's serving a degraded response:

@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

"My fallback never triggers"

Verify that the FallbackProvider receives the correct exceptions. If your RetryProvider uses reraise=True, it propagates the original exception. If it uses reraise=False, it propagates tenacity.RetryError. The FallbackProvider needs to catch the correct type. Add a temporary log in the catch to see what exception type arrives.

"The readiness health check always returns 503"

Check which checks are failing. The response includes a checks dict with the status of each dependency. If circuit_breakers shows "state": "open", it's because the circuit is open — normal during an outage. If config fails, you have a configuration problem. Call GET /health/deps to see the full detail of each dependency.

"Should I cache all LLM responses?"

Not necessarily. Cache when the input is repetitive and the response is deterministic (or acceptably similar). For sentiment analysis with the same texts, the cache is very effective. For chatbots with unique conversations, the cache rarely matches. Use a short TTL (5-15 minutes) to start and adjust based on your hit rate.

"Kubernetes restarts my pod constantly"

You're probably checking OpenAI in your liveness probe. If OpenAI has problems, your liveness fails → K8s restarts the pod → the new pod has the same problem → infinite loop. Rule of thumb: liveness only checks that the Python process is running. External dependencies go in readiness.

Summary

• Fallback hierarchy: secondary provider → model downgrade → cached response → simplified processing → static default → graceful error
• Always communicate degradation: the user and the monitoring system must know when a fallback is used
• Liveness vs Readiness: liveness is "am I alive?", readiness is "can I receive traffic?"
• Real health checks: verify that OpenAI responds, not just that the HTTP server works
• Cache as a fallback: useful for repetitive requests, with an extended TTL in fallback mode

Additional resources

1. Kubernetes Probes — Liveness/readiness documentation
2. Graceful Degradation (MDN) — The concept
3. Caching Strategies — Cache patterns
4. Microsoft — Health Endpoint Monitoring Pattern — The health checks pattern