Logging y Monitoreo

Observabilidad — Logging estructurado y tracking de errores

Propósito

Esta guía establece cómo implementar logging y monitoreo en aplicaciones del DTI. El objetivo es tener visibilidad del comportamiento de las aplicaciones en todos los ambientes.

Logging estructurado

Formato JSON

Todo log debe ser JSON estructurado para facilitar análisis:

{
  "timestamp": "2026-05-15T10:30:00Z",
  "level": "INFO",
  "message": "Usuario inició sesión",
  "request_id": "550e8400-e29b-41d4-a716-446655440000",
  "user_id": 123,
  "ip": "192.168.1.100",
  "module": "auth",
  "action": "login"
}

Implementación Python (FastAPI)

# app/core/logging.py
import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
        }
        if hasattr(record, "request_id"):
            log_data["request_id"] = record.request_id
        if hasattr(record, "user_id"):
            log_data["user_id"] = record.user_id
        return json.dumps(log_data)

# Configuración
def setup_logging():
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())
    logger = logging.getLogger("app")
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)

Implementación TypeScript

// src/lib/logger.ts
interface LogEntry {
  timestamp: string;
  level: 'DEBUG' | 'INFO' | 'WARNING' | 'ERROR' | 'CRITICAL';
  message: string;
  module?: string;
  requestId?: string;
  userId?: string;
  [key: string]: unknown;
}

export const logger = {
  info(message: string, context?: Partial<LogEntry>) {
    console.log(JSON.stringify({
      timestamp: new Date().toISOString(),
      level: 'INFO',
      message,
      ...context
    }));
  },

  error(message: string, error?: Error, context?: Partial<LogEntry>) {
    console.error(JSON.stringify({
      timestamp: new Date().toISOString(),
      level: 'ERROR',
      message,
      error: error?.message,
      stack: error?.stack,
      ...context
    }));
  }
};

Niveles de log

Logging en endpoints

# FastAPI - logging de requests
from fastapi import Request
import uuid

@app.middleware("http")
async def log_requests(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id

    logger.info(
        "Incoming request",
        extra={
            "request_id": request_id,
            "method": request.method,
            "path": request.url.path,
            "ip": request.client.host
        }
    )

    response = await call_next(request)

    logger.info(
        "Request completed",
        extra={
            "request_id": request_id,
            "status_code": response.status_code
        }
    )
    return response

Error tracking

Configuración de Sentry

# FastAPI - Sentry
import sentry_sdk

sentry_sdk.init(
    dsn=os.getenv("SENTRY_DSN"),
    environment=os.getenv("APP_ENV"),
    traces_sample_rate=0.1
)

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    sentry_sdk.capture_exception(exc, {
        "request_id": request.state.request_id,
        "path": request.url.path
    })
    return JSONResponse({"error": "Internal server error"}, status_code=500)

// Next.js - Sentry
import * as Sentry from '@sentry/nextjs';

Sentry.init({
  dsn: process.env.SENTRY_DSN,
  environment: process.env.NODE_ENV,
});

export default Sentry.captureException;

Categorización de errores

# app/core/exceptions.py
from enum import Enum

class ErrorCategory(Enum):
    CLIENT_ERROR = "4xx"  # Errores del cliente
    SERVER_ERROR = "5xx"  # Errores del servidor
    AUTH_ERROR = "auth"    # Errores de autenticación
    VALIDATION_ERROR = "validation"  # Errores de validación

def categorize_error(status_code: int) -> ErrorCategory:
    if 400 <= status_code < 500:
        return ErrorCategory.CLIENT_ERROR
    return ErrorCategory.SERVER_ERROR

Logging en producción

Log rotation

# /etc/logrotate.d/app
/var/log/app/*.log {
    daily
    rotate 7
    compress
    delaycompress
    notifempty
    create 0640 www-data www-data
    sharedscripts
    postrotate
        docker kill -s USR1 $(docker ps -q --filter name=app) 2>/dev/null || true
    endscript
}

Centralización de logs

Para análisis avanzado, centralizar logs en:

• ELK Stack (Elasticsearch, Logstash, Kibana)
• CloudWatch (AWS)
• Datadog

# Ejemplo: envío de logs a Logstash
logging:
  handlers:
    logstash:
      class: logging.handlers.SocketHandler
      host: logstash.unach.cl
      port: 5044
  loggers:
    app:
      handlers: [console, logstash]
      level: INFO

Contexto de request

# FastAPI - propagar request_id
from contextvars import ContextVar
request_id_ctx: ContextVar[str] = ContextVar("request_id", default="")

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    request_id_ctx.set(request_id)
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

# Uso en cualquier lugar
def get_request_id():
    return request_id_ctx.get()

Checklist

• ¿Los logs están en formato JSON estructurado?
• ¿Se incluye request_id en cada log?
• ¿No se loggean passwords, tokens o datos sensibles?
• ¿Los errores 5xx se envían a error tracker?
• ¿Hay log rotation configurado en producción?
• ¿El nivel DEBUG no está activo en producción?

Referencias

• Seguridad
• Autenticación y Roles
• Environment Variables