Production Debugging

import logging
import structlog
 
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
)
log = structlog.get_logger()
 
def handle_request(request_id: str, trace_id: str, tenant: str) -> None:
    structlog.contextvars.bind_contextvars(
        request_id=request_id, trace_id=trace_id, tenant=tenant
    )
    log.info("order_created", order_id="ord_123")

# Live stack sample (no restart, needs ptrace or CAP_SYS_PTRACE)
py-spy top --pid $(pgrep -f "gunicorn.*api:app")
py-spy dump --pid <pid> --lines 50

"""prod_debug_helpers.py - minimal toolkit for correlating prod evidence."""
from __future__ import annotations
 
import json
import os
import sys
import time
from contextlib import contextmanager
from dataclasses import dataclass, asdict
 
import structlog
 
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger()
 
 
@dataclass(frozen=True)
class DeployContext:
    git_sha: str
    python_version: str
    service: str
 
 
def deploy_context() -> DeployContext:
    return DeployContext(
        git_sha=os.getenv("GIT_SHA", "unknown"),
        python_version=sys.version.split()[0],
        service=os.getenv("SERVICE_NAME", "api"),
    )
 
 
@contextmanager
def request_scope(request_id: str, trace_id: str, **extra: str):
    structlog.contextvars.bind_contextvars(
        request_id=request_id,
        trace_id=trace_id,
        **extra,
        **asdict(deploy_context()),
    )
    start = time.perf_counter()
    try:
        yield
    finally:
        log.info("request_finished", duration_ms=(time.perf_counter() - start) * 1000)
 
 
def simulate_slow_path() -> None:
    with request_scope("req-9f2", "trace-abc", tenant="acme"):
        log.info("cache_miss", key="user:42")
        time.sleep(0.05)  # stands in for DB round-trip
        log.warning("downstream_retry", attempt=2, service="billing")
 
 
if __name__ == "__main__":
    simulate_slow_path()
    print("Filter prod logs: trace_id=trace-abc AND level>=warning")

# Enable faulthandler on startup for native hangs
import faulthandler
faulthandler.enable()
 
# Temporary debug in one pod only (never commit)
import logging
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)