Observability & Site Reliability Engineering¶

import logging
import structlog

def configure_logging(level: str = "INFO") -> None:
    """Configure structlog for production: JSON output, timestamps, context vars."""
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,      # merge per-request context
            structlog.stdlib.add_logger_name,
            structlog.stdlib.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),  # ISO-8601 timestamp
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer(),          # output as JSON
        ],
        wrapper_class=structlog.make_filtering_bound_logger(
            getattr(logging, level.upper(), logging.INFO)
        ),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
    )

configure_logging(level="INFO")
logger = structlog.get_logger()

# FastAPI middleware: inject correlation ID into all log calls for this request
from fastapi import Request
import uuid

async def correlation_id_middleware(request: Request, call_next):
    trace_id = request.headers.get("traceparent", str(uuid.uuid4()))
    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        trace_id=trace_id,
        method=request.method,
        path=request.url.path,
    )
    response = await call_next(request)
    structlog.contextvars.clear_contextvars()
    return response

# Usage in business logic — context fields from middleware are automatically included
def process_payment(order_id: str, amount: float, user_id: int) -> dict:
    log = logger.bind(order_id=order_id, user_id=user_id, amount=amount)
    log.info("payment_started")
    try:
        result = gateway.charge(order_id=order_id, amount=amount)
        log.info("payment_succeeded", transaction_id=result["id"])
        return result
    except gateway.DeclinedError as exc:
        log.warning("payment_declined", reason=str(exc))
        raise
    except Exception as exc:
        log.error("payment_error", exc_info=True)
        raise

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

# Define metrics (module-level singletons)
HTTP_REQUESTS_TOTAL = Counter(
    "http_requests_total",
    "Total HTTP requests",
    labelnames=["method", "path", "status_code"],
)

HTTP_REQUEST_DURATION_SECONDS = Histogram(
    "http_request_duration_seconds",
    "HTTP request latency in seconds",
    labelnames=["method", "path"],
    buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0],
)

ACTIVE_REQUESTS = Gauge(
    "http_active_requests",
    "Number of HTTP requests currently being processed",
)

PAYMENT_FAILURES_TOTAL = Counter(
    "payment_failures_total",
    "Total payment failures",
    labelnames=["reason"],
)

class PrometheusMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        # Normalize path to avoid high-cardinality labels (e.g., /users/42 -> /users/{id})
        path = request.url.path
        method = request.method

        ACTIVE_REQUESTS.inc()
        start = time.perf_counter()
        try:
            response = await call_next(request)
            return response
        finally:
            duration = time.perf_counter() - start
            status = getattr(response, "status_code", 500)

            HTTP_REQUESTS_TOTAL.labels(method=method, path=path, status_code=status).inc()
            HTTP_REQUEST_DURATION_SECONDS.labels(method=method, path=path).observe(duration)
            ACTIVE_REQUESTS.dec()

# FastAPI setup
from fastapi import FastAPI
from prometheus_client import make_asgi_app

app = FastAPI()
app.add_middleware(PrometheusMiddleware)

# Mount /metrics endpoint
metrics_app = make_asgi_app()
app.mount("/metrics", metrics_app)

# Usage in business logic
def process_payment(amount: float, token: str) -> dict:
    try:
        result = gateway.charge(amount, token)
        return result
    except gateway.DeclinedError:
        PAYMENT_FAILURES_TOTAL.labels(reason="card_declined").inc()
        raise
    except gateway.FraudError:
        PAYMENT_FAILURES_TOTAL.labels(reason="fraud_detected").inc()
        raise

from opentelemetry import trace
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.sdk.resources import Resource
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor

def configure_tracing(service_name: str, otlp_endpoint: str = "http://localhost:4317") -> trace.Tracer:
    """Configure OpenTelemetry with OTLP export (works with Jaeger, Tempo, Datadog, Honeycomb)."""
    resource = Resource.create({"service.name": service_name, "deployment.environment": "production"})
    provider = TracerProvider(resource=resource)
    exporter = OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True)
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return trace.get_tracer(service_name)

tracer = configure_tracing("checkout-service")

# Auto-instrument FastAPI (injects spans for every HTTP request automatically)
FastAPIInstrumentor.instrument_app(app)

# Auto-instrument outgoing HTTP calls (propagates traceparent header)
HTTPXClientInstrumentor().instrument()

# Auto-instrument SQLAlchemy (captures every query as a span)
SQLAlchemyInstrumentor().instrument(engine=engine)

# Manual spans for business-logic operations
def process_payment(order_id: str, amount: float) -> dict:
    with tracer.start_as_current_span("payment.charge") as span:
        span.set_attribute("order.id", order_id)
        span.set_attribute("payment.amount", amount)
        span.set_attribute("payment.currency", "USD")
        span.add_event("Calling payment gateway")

        try:
            result = gateway.charge(order_id=order_id, amount=amount)
            span.set_attribute("payment.transaction_id", result["id"])
            span.set_status(trace.StatusCode.OK)
            return result
        except Exception as exc:
            span.record_exception(exc)
            span.set_status(trace.StatusCode.ERROR, str(exc))
            raise

# Inject trace context into structlog for correlated logs + traces
from opentelemetry import trace as otel_trace

def get_current_trace_context() -> dict:
    ctx = otel_trace.get_current_span().get_span_context()
    if ctx.is_valid:
        return {
            "trace_id": format(ctx.trace_id, "032x"),
            "span_id": format(ctx.span_id, "016x"),
        }
    return {}

# Call this in your logging middleware to add trace context to every log line
structlog.contextvars.bind_contextvars(**get_current_trace_context())

from dataclasses import dataclass
from datetime import timedelta

@dataclass
class SLO:
    name: str
    target: float           # e.g. 0.995 for 99.5%
    window_days: int = 28

    @property
    def error_budget(self) -> float:
        """Fraction of requests that can fail."""
        return 1.0 - self.target

    def budget_remaining(self, current_error_rate: float) -> float:
        """Fraction of error budget remaining (0.0 = exhausted, 1.0 = fully remaining)."""
        if current_error_rate >= self.error_budget:
            return 0.0
        return 1.0 - (current_error_rate / self.error_budget)

    def hours_until_exhausted(self, current_error_rate: float) -> float | None:
        """
        If the current error rate continues unchanged, how many hours until
        the error budget is exhausted?
        Returns None if budget is not being consumed.
        """
        if current_error_rate <= 0:
            return None
        window_hours = self.window_days * 24
        # Budget consumed per hour at this error rate
        consumption_rate = current_error_rate / self.error_budget
        if consumption_rate <= 0:
            return None
        return window_hours * (1 - 0) / consumption_rate

    def burn_rate(self, current_error_rate: float) -> float:
        """
        Burn rate: 1.0 = consuming budget at exactly the sustainable rate.
        > 1.0 = consuming faster than sustainable (will exhaust before window ends).
        """
        return current_error_rate / self.error_budget

# Example: calculate budget status
checkout_slo = SLO(name="checkout-availability", target=0.995, window_days=28)

current_error_rate = 0.012   # 1.2% of requests are failing

print(f"SLO target:            {checkout_slo.target:.1%}")
print(f"Error budget:          {checkout_slo.error_budget:.2%}")
print(f"Current error rate:    {current_error_rate:.2%}")
print(f"Burn rate:             {checkout_slo.burn_rate(current_error_rate):.1f}x sustainable")
print(f"Budget remaining:      {checkout_slo.budget_remaining(current_error_rate):.1%}")
remaining_hours = checkout_slo.hours_until_exhausted(current_error_rate)
if remaining_hours:
    print(f"Budget exhausted in:   {remaining_hours:.1f} hours")

# Output:
# SLO target:            99.5%
# Error budget:          0.50%
# Current error rate:    1.20%
# Burn rate:             2.4x sustainable
# Budget remaining:      0.0%  (already exceeded)
# Budget exhausted in:   280.0 hours (at the SLO target consumption rate)