Módulo 5: Structured Logging para AI Systems

5. JSON Logs con structlog

Siguiente →

Descripción

JSON logs no son solo un formato — son la interfaz entre tu app y cualquier sistema de observabilidad (Datadog, Loki, CloudWatch, ELK). Un log JSON bien formado es queryable por máquinas, indexable por sistemas de log aggregation, y dashboardable sin configuración adicional. Esta cápsula implementa la configuración completa de structlog para producción, los procesadores esenciales para apps AI, y los patrones de querying que convierten los logs en un sistema de diagnóstico real.

JSON Lines: el formato estándar

¿Por qué "JSON Lines" y no JSON puro?

JSON puro:
[
  {"event": "request_1", "tokens": 450},
  {"event": "request_2", "tokens": 320}
]
Problema: Para añadir una entrada, debes leer TODO el archivo, parsear, añadir, reescribir.
Con millones de logs: imposible.

JSON Lines (JSONL):
{"event": "request_1", "tokens": 450}
{"event": "request_2", "tokens": 320}
Ventajas:
→ Append-only: cada log es una línea, se añade al final
→ Parseo incremental: puedes leer una línea sin cargar todo el archivo
→ Fácil de grep, sed, awk
→ jq funciona perfecto
→ Todos los sistemas de log aggregation lo soportan nativamente
→ Stream-friendly: stdout pipe a otro proceso

Configuración completa de structlog

# src/logging_config.py
import logging
import os
import sys
import structlog
from typing import Any

def get_log_level() -> int:
    """Lee el log level desde env var o usa INFO por defecto."""
    level_str = os.getenv("LOG_LEVEL", "INFO").upper()
    return getattr(logging, level_str, logging.INFO)

def configure_structlog(
    env: str = None,
    log_level: int = None,
    json_output: bool = None
) -> None:
    """
    Configura structlog para el entorno especificado.
    
    Args:
        env: "production", "development", o "testing"
        log_level: Override del nivel de log
        json_output: Override del formato (True=JSON, False=Console)
    """
    env = env or os.getenv("ENVIRONMENT", "development")
    log_level = log_level or get_log_level()
    
    # Determinar si usar JSON o Console renderer
    use_json = json_output if json_output is not None else (env == "production")
    
    # ─── Procesadores para interceptar logging estándar de Python ───────────
    # structlog puede capturar logs de librerías que usan logging.getLogger()
    logging.basicConfig(
        format="%(message)s",
        stream=sys.stdout,
        level=log_level
    )
    
    # ─── Procesadores compartidos (dev y prod) ───────────────────────────────
    shared_processors = [
        # Leer request_id (y otros campos) del contexto de contextvars
        structlog.contextvars.merge_contextvars,
        
        # Añadir level (info, warning, error, etc.)
        structlog.processors.add_log_level,
        
        # Timestamp en formato ISO 8601
        structlog.processors.TimeStamper(fmt="iso"),
        
        # Nombre del logger (útil para identificar el módulo)
        structlog.stdlib.add_logger_name,
        
        # Renderizar stack traces en excepciones
        structlog.processors.StackInfoRenderer(),
    ]
    
    if use_json:
        # ─── Producción: JSON Lines ──────────────────────────────────────────
        processors = shared_processors + [
            # Formatear la excepción como objeto JSON (no como texto)
            structlog.processors.format_exc_info,
            # Serializar como JSON (una línea por evento)
            structlog.processors.JSONRenderer()
        ]
    else:
        # ─── Desarrollo: Console legible ─────────────────────────────────────
        processors = shared_processors + [
            # Colores y formato humano para la terminal
            structlog.dev.ConsoleRenderer(
                colors=True,
                exception_formatter=structlog.dev.plain_traceback
            )
        ]
    
    structlog.configure(
        processors=processors,
        # Nivel de filtrado: logs por debajo de log_level se ignoran
        wrapper_class=structlog.make_filtering_bound_logger(log_level),
        # context_class: dict es suficiente para la mayoría de casos
        context_class=dict,
        # Logger factory: usa print (stdout) por defecto
        logger_factory=structlog.PrintLoggerFactory(),
        # Cache de la configuración para performance
        cache_logger_on_first_use=True,
    )

# Llamar al inicio de la app:
# configure_structlog()

Procesadores custom para apps AI

# src/logging_config.py (continuación)
import hashlib
from typing import MutableMapping

def add_app_metadata(
    logger: Any,
    method_name: str,
    event_dict: MutableMapping[str, Any]
) -> MutableMapping[str, Any]:
    """
    Añade metadatos de la app a todos los logs.
    Útil para identificar la versión del app en sistemas de aggregation.
    """
    event_dict["app"] = os.getenv("APP_NAME", "production-best-practices")
    event_dict["version"] = os.getenv("APP_VERSION", "0.1.0")
    event_dict["env"] = os.getenv("ENVIRONMENT", "development")
    return event_dict

def sanitize_sensitive_fields(
    logger: Any,
    method_name: str,
    event_dict: MutableMapping[str, Any]
) -> MutableMapping[str, Any]:
    """
    Elimina o enmascara campos que no deben aparecer en logs.
    Safety net para evitar que PII o secrets lleguen a los logs.
    """
    FIELDS_TO_REMOVE = {"password", "api_key", "secret", "token", "authorization"}
    FIELDS_TO_HASH = {"user_id", "email", "phone"}
    
    for field in list(event_dict.keys()):
        field_lower = field.lower()
        
        if any(sensitive in field_lower for sensitive in FIELDS_TO_REMOVE):
            event_dict[field] = "[REDACTED]"
        
        elif any(pii in field_lower for pii in FIELDS_TO_HASH):
            # Hash para analytics sin exponer el valor real
            value = str(event_dict[field])
            event_dict[field] = hashlib.sha256(value.encode()).hexdigest()[:12]
    
    return event_dict

def truncate_long_strings(
    logger: Any,
    method_name: str,
    event_dict: MutableMapping[str, Any],
    max_length: int = 500
) -> MutableMapping[str, Any]:
    """
    Trunca strings largos para evitar logs enormes.
    En INFO, ningún string debería ser mayor de 500 chars.
    """
    for key, value in event_dict.items():
        if isinstance(value, str) and len(value) > max_length:
            event_dict[key] = value[:max_length] + f"...[truncated {len(value)} chars]"
    return event_dict

# Configuración con procesadores custom:
def configure_structlog_with_custom_processors():
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            add_app_metadata,           # Custom: metadata de la app
            sanitize_sensitive_fields,  # Custom: remover secrets
            truncate_long_strings,      # Custom: truncar campos largos
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.stdlib.add_logger_name,
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer()
        ]
    )

Loguear excepciones correctamente

import structlog
log = structlog.get_logger()

# ─── Forma incorrecta ───────────────────────────────────────────────────────
try:
    response = client.chat.completions.create(...)
except Exception as e:
    log.error("failed", error=str(e))  # Solo el mensaje, sin stack trace

# ─── Forma correcta ────────────────────────────────────────────────────────

# Opción 1: exc_info=True (incluye stack trace completo en el log)
try:
    response = client.chat.completions.create(...)
except Exception as e:
    log.error(
        "llm_request_failed",
        error_type=type(e).__name__,
        exc_info=True  # ← structlog incluye el traceback en el log
    )

# Opción 2: log.exception (shorthand de log.error con exc_info=True)
try:
    response = client.chat.completions.create(...)
except Exception as e:
    log.exception("llm_request_failed", error_type=type(e).__name__)

# Con JSONRenderer, el output será:
# {
#   "event": "llm_request_failed",
#   "error_type": "RateLimitError",
#   "exception": "Traceback (most recent call last):\n  ...",
#   "level": "error",
#   "timestamp": "2024-01-15T10:30:00Z"
# }

Querying avanzado con jq

# Instalar jq:
# macOS: brew install jq
# Ubuntu: sudo apt install jq

# ─── Queries básicos ────────────────────────────────────────────────────────

# Ver todos los eventos únicos
jq -r '.event' logs.json | sort | uniq -c | sort -rn

# Filtrar por nivel
jq 'select(.level == "error")' logs.json

# Filtrar por evento
jq 'select(.event == "llm_request_completed")' logs.json

# ─── Cost queries ───────────────────────────────────────────────────────────

# Costo total
jq -s '[.[].total_cost_usd // 0] | add' logs.json

# Requests más caros (top 10)
jq -s 'map(select(.total_cost_usd != null)) | sort_by(-.total_cost_usd) | .[0:10] | .[] | {request_id, total_cost_usd, total_tokens, model}' logs.json

# Costo promedio por modelo
jq -s '
  map(select(.event == "llm_request_completed"))
  | group_by(.model)
  | map({
      model: .[0].model,
      avg_cost: ([.[].total_cost_usd] | add / length),
      count: length
    })
' logs.json

# ─── Performance queries ────────────────────────────────────────────────────

# Requests lentos (>5 segundos)
jq 'select(.duration_ms > 5000)' logs.json | jq -r '.request_id'

# Latencia promedio
jq -s '[map(select(.duration_ms != null)) | .[].duration_ms] | add / length' logs.json

# ─── Guardrail queries ──────────────────────────────────────────────────────

# Cuántos guardrails se activaron
jq 'select(.event == "guardrail_activated")' logs.json | wc -l

# Por tipo de guardrail
jq 'select(.event == "guardrail_activated") | .guardrail_type' logs.json | sort | uniq -c

# ─── Error queries ──────────────────────────────────────────────────────────

# Todos los errores con su request_id
jq 'select(.level == "error") | {timestamp, request_id, event, error_type}' logs.json

# ─── Time-based queries ─────────────────────────────────────────────────────

# Logs de la última hora (timestamp es ISO 8601)
jq 'select(.timestamp > "2024-01-15T09:00:00Z")' logs.json

# ─── Tracing: historia de un request ───────────────────────────────────────
jq 'select(.request_id == "a1b2c3d4")' logs.json | jq -s 'sort_by(.timestamp) | .[]'

Script Python para análisis de logs

# scripts/log_analyzer.py
# Alternativa a jq cuando necesitas lógica más compleja

import json
import sys
from pathlib import Path
from collections import defaultdict

def read_jsonl(path: str) -> list:
    """Lee un archivo JSON Lines."""
    logs = []
    with open(path) as f:
        for line in f:
            line = line.strip()
            if line:
                try:
                    logs.append(json.loads(line))
                except json.JSONDecodeError:
                    pass
    return logs

def print_report(logs: list):
    """Genera un reporte legible de los logs."""
    
    # Separar por tipo de log
    llm_logs = [l for l in logs if l.get("event") == "llm_request_completed"]
    error_logs = [l for l in logs if l.get("level") == "error"]
    guardrail_logs = [l for l in logs if l.get("event") == "guardrail_activated"]
    
    print("=" * 60)
    print("AI SYSTEM LOG ANALYSIS REPORT")
    print("=" * 60)
    
    # Summary
    print(f"\n📊 SUMMARY")
    print(f"  Total LLM requests:  {len(llm_logs)}")
    print(f"  Total errors:        {len(error_logs)}")
    print(f"  Guardrail activations: {len(guardrail_logs)}")
    
    # Cost summary
    if llm_logs:
        total_cost = sum(l.get("total_cost_usd", 0) for l in llm_logs)
        avg_cost = total_cost / len(llm_logs)
        max_cost_log = max(llm_logs, key=lambda l: l.get("total_cost_usd", 0))
        
        print(f"\n💰 COST SUMMARY")
        print(f"  Total cost:          ${total_cost:.6f}")
        print(f"  Avg cost/request:    ${avg_cost:.8f}")
        print(f"  Most expensive:      ${max_cost_log.get('total_cost_usd', 0):.6f}")
        print(f"    → request_id: {max_cost_log.get('request_id', 'unknown')}")
        print(f"    → tokens: {max_cost_log.get('total_tokens', 0)}")
        print(f"    → model: {max_cost_log.get('model', 'unknown')}")
    
    # Performance summary
    if llm_logs:
        durations = [l.get("duration_ms", 0) for l in llm_logs]
        avg_duration = sum(durations) / len(durations)
        slow_requests = [l for l in llm_logs if l.get("duration_ms", 0) > 5000]
        
        print(f"\n⚡ PERFORMANCE")
        print(f"  Avg duration:        {avg_duration:.0f}ms")
        print(f"  Slow (>5s) requests: {len(slow_requests)}")
    
    # Errors
    if error_logs:
        print(f"\n❌ ERRORS ({len(error_logs)} total)")
        error_types = defaultdict(int)
        for l in error_logs:
            error_types[l.get("error_type", "unknown")] += 1
        for error_type, count in sorted(error_types.items(), key=lambda x: -x[1]):
            print(f"  {error_type}: {count}")
    
    # Guardrails
    if guardrail_logs:
        print(f"\n🛡️  GUARDRAILS ({len(guardrail_logs)} activations)")
        guard_types = defaultdict(int)
        for l in guardrail_logs:
            guard_types[l.get("guardrail_type", "unknown")] += 1
        for guard_type, count in sorted(guard_types.items(), key=lambda x: -x[1]):
            print(f"  {guard_type}: {count}")
    
    print("\n" + "=" * 60)

if __name__ == "__main__":
    log_file = sys.argv[1] if len(sys.argv) > 1 else "logs/app.json"
    logs = read_jsonl(log_file)
    print_report(logs)

Configuración de log rotation

# Para evitar que los archivos de log crezcan indefinidamente:

import logging
from logging.handlers import TimedRotatingFileHandler
import structlog

def configure_file_logging(log_dir: str = "logs"):
    """
    Configura logging a archivo con rotación diaria.
    Retiene logs de los últimos 30 días.
    """
    import os
    os.makedirs(log_dir, exist_ok=True)
    
    # Handler para archivo con rotación diaria
    file_handler = TimedRotatingFileHandler(
        filename=f"{log_dir}/app.json",
        when="midnight",        # Rotar a medianoche
        interval=1,             # Cada 1 día
        backupCount=30,         # Retener 30 días
        encoding="utf-8"
    )
    file_handler.setLevel(logging.INFO)
    
    # También loguear a stdout (para sistemas como Docker/Kubernetes)
    # que recogen logs del stdout del contenedor
    stdout_handler = logging.StreamHandler(sys.stdout)
    stdout_handler.setLevel(logging.INFO)
    
    logging.basicConfig(
        handlers=[file_handler, stdout_handler],
        format="%(message)s",
        level=logging.INFO
    )
    
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer()
        ],
        logger_factory=structlog.stdlib.LoggerFactory(),
    )

Testing del sistema de logging

# tests/unit/test_json_logs.py
import json
import io
import pytest
import structlog

@pytest.fixture
def json_log_capture():
    """Captura logs como JSON para tests de integridad."""
    output = io.StringIO()
    
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer()
        ],
        logger_factory=structlog.PrintLoggerFactory(file=output),
        cache_logger_on_first_use=False,  # Importante: no cachear en tests
    )
    yield output
    structlog.reset_defaults()

def parse_logs(output: io.StringIO) -> list:
    """Parsea logs JSON Lines a lista de dicts."""
    return [
        json.loads(line)
        for line in output.getvalue().strip().split("\n")
        if line.strip()
    ]

def test_each_log_is_valid_json(json_log_capture):
    """Cada log debe ser JSON válido."""
    log = structlog.get_logger()
    log.info("event_1", field_a="value_a")
    log.warning("event_2", field_b=42)
    log.error("event_3", field_c=True)
    
    # No debe lanzar JSONDecodeError
    logs = parse_logs(json_log_capture)
    assert len(logs) == 3

def test_log_has_required_fields(json_log_capture):
    """Cada log debe tener event, level y timestamp."""
    log = structlog.get_logger()
    log.info("test_event", some_field="some_value")
    
    logs = parse_logs(json_log_capture)
    assert len(logs) == 1
    
    entry = logs[0]
    assert "event" in entry
    assert "level" in entry
    assert "timestamp" in entry

def test_context_vars_appear_in_all_logs(json_log_capture):
    """Los campos del contexto deben aparecer en todos los logs."""
    structlog.contextvars.bind_contextvars(request_id="test-req-99")
    
    log = structlog.get_logger()
    log.info("event_a")
    log.info("event_b")
    log.warning("event_c")
    
    structlog.contextvars.clear_contextvars()
    
    logs = parse_logs(json_log_capture)
    for entry in logs:
        assert entry.get("request_id") == "test-req-99", \
            f"request_id missing from: {entry}"

def test_exception_logged_as_json(json_log_capture):
    """Las excepciones deben aparecer como campo JSON, no como texto suelto."""
    log = structlog.get_logger()
    
    try:
        raise ValueError("test error")
    except ValueError:
        log.exception("something_failed")
    
    logs = parse_logs(json_log_capture)
    assert len(logs) == 1
    entry = logs[0]
    
    # La excepción debe estar como campo en el JSON
    assert "exception" in entry or "exc_info" in entry
    # No debe aparecer en el campo event
    assert "Traceback" not in entry.get("event", "")

Ejercicios

Ejercicio 1: Configurar structlog para producción

Escribe la configuración completa de structlog para una app en producción que:

  1. Usa JSON Lines
  2. Incluye request_id del contexto
  3. Añade level y timestamp
  4. Redacta campos con "password" o "secret" en el nombre
Ver solución 
import structlog, os, hashlib

def redact_secrets(logger, method_name, event_dict):
    for key in list(event_dict.keys()):
        if any(s in key.lower() for s in ["password", "secret", "api_key"]):
            event_dict[key] = "[REDACTED]"
    return event_dict

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        redact_secrets,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer()
    ]
)

Ejercicio 2: jq one-liners

Escribe los comandos jq para:

  1. Contar cuántos requests completaron con éxito hoy
  2. Listar los request_ids de todos los errores
  3. Encontrar el modelo más utilizado
Ver solución 
# 1. Requests completados hoy (asumiendo fecha 2024-01-15)
jq 'select(.event == "request_completed" and (.timestamp | startswith("2024-01-15")))' logs.json | wc -l

# 2. request_ids de errores
jq -r 'select(.level == "error") | .request_id' logs.json

# 3. Modelo más utilizado
jq -r 'select(.event == "llm_request_completed") | .model' logs.json | sort | uniq -c | sort -rn | head -1

Resumen

  • JSON Lines: una línea por evento, append-only, stream-friendly — el formato estándar
  • structlog + JSONRenderer para producción, ConsoleRenderer para desarrollo
  • Procesadores en orden: contextvars → add_log_level → timestamp → (custom) → JSONRenderer
  • Procesadores custom: sanitize secrets, truncate long strings, add app metadata
  • jq es suficiente para queries ad-hoc; un script Python para análisis recurrentes
  • Log rotation para evitar archivos ilimitados — 30 días de retención como punto de partida

Recursos adicionales

  1. structlog Documentation — Documentación completa con ejemplos
  2. jq Manual — Referencia completa de jq
  3. JSON Lines format — Especificación del formato
  4. Loki — Sistema de log aggregation que indexa JSON logs nativamente
  5. Structured Logging in Python (Real Python) — Tutorial completo
← Anterior4. Token y Cost TrackingSiguiente →6. Debugging de Sistemas No-Determinísticos