Logs Estruturados: JSON, correlation IDs e levels com propósito

# ❌ Texto livre — o que a maioria tem
2024-01-15 14:23:11 ERROR Failed to process order 12345 for user 99 after 3 retries

# ✅ JSON estruturado — o que você quer ter
{
  "timestamp": "2024-01-15T14:23:11.847Z",
  "level": "error",
  "message": "Failed to process order after retries",
  "service": "order-service",
  "version": "1.4.2",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7",
  "order_id": "12345",
  "user_id": "99",
  "retry_count": 3,
  "error": "ConnectionTimeout",
  "error_detail": "payment-service:8080 unreachable after 5000ms",
  "duration_ms": 15234
}

# Loki LogQL — todos os erros de payment no último 1h
{service="order-service"} | json | level="error" | error_detail =~ "payment.*"

# Busca pelo trace_id específico (correlaciona com spans no Tempo)
{service=~".+"} | json | trace_id="4bf92f3577b34da6a3ce929d0e0e4736"

# Latência P99 derivada de logs (sem Prometheus)
{service="order-service"} | json | unwrap duration_ms | p99()

# Python: loguru + OpenTelemetry context bridge
import logging
import json
from opentelemetry import trace
from opentelemetry.trace import format_trace_id, format_span_id

class OtelJsonFormatter(logging.Formatter):
    """Injeta trace_id e span_id do contexto OTel ativo."""

    def format(self, record: logging.LogRecord) -> str:
        ctx = trace.get_current_span().get_span_context()

        log_entry = {
            "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
            "level": record.levelname.lower(),
            "message": record.getMessage(),
            "logger": record.name,
            "module": record.module,
            "line": record.lineno,
        }

        # Injetar trace context se span ativo
        if ctx.is_valid:
            log_entry["trace_id"] = format_trace_id(ctx.trace_id)
            log_entry["span_id"] = format_span_id(ctx.span_id)
            log_entry["trace_flags"] = ctx.trace_flags

        # Campos extras passados via extra={}
        for key, value in record.__dict__.items():
            if key not in logging.LogRecord.__dict__ and not key.startswith("_"):
                log_entry[key] = value

        # Stacktrace se exception
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_entry, default=str)


# Setup — uma vez no startup
handler = logging.StreamHandler()
handler.setFormatter(OtelJsonFormatter())
logging.getLogger().addHandler(handler)
logging.getLogger().setLevel(logging.INFO)

logger = logging.getLogger(__name__)

# Uso em qualquer lugar — trace_id é injetado automaticamente
def process_order(order_id: str, user_id: str):
    logger.info(
        "Processing order",
        extra={"order_id": order_id, "user_id": user_id}
    )
    # Output: {"timestamp":"...","level":"info","message":"Processing order",
    #          "trace_id":"4bf92f...","span_id":"00f067...","order_id":"123","user_id":"99"}

// Node.js: pino + OpenTelemetry (pino é 8× mais rápido que winston)
import pino from 'pino';
import { trace, context } from '@opentelemetry/api';

// Mixin lido a cada registro — injeta trace context
const otelMixin = () => {
  const span = trace.getActiveSpan();
  if (!span) return {};

  const ctx = span.spanContext();
  return {
    trace_id: ctx.traceId,
    span_id: ctx.spanId,
    trace_flags: ctx.traceFlags,
  };
};

const logger = pino({
  level: process.env.LOG_LEVEL ?? 'info',
  mixin: otelMixin,
  timestamp: pino.stdTimeFunctions.isoTime,
  // pino serializa em JSON por default — zero config adicional
});

export { logger };

// Uso
logger.info({ order_id: '123', user_id: '99' }, 'Processing order');
// Output: {"time":"2024-01-15T14:23:11.847Z","level":"info","msg":"Processing order",
//          "trace_id":"4bf92f...","span_id":"00f067...","order_id":"123","user_id":"99"}

# Schema base obrigatório (todos os serviços)
REQUIRED_FIELDS = {
    "timestamp",    # ISO 8601 com milissegundos
    "level",        # error | warn | info | debug
    "message",      # descrição human-readable do evento
    "service",      # nome do serviço (ex: "order-service")
    "version",      # versão do binário (ex: "1.4.2")
    "trace_id",     # se OTel ativo (W3C format)
    "span_id",      # se OTel ativo
}

# Campos opcionais recomendados por contexto
CONTEXT_FIELDS = {
    # HTTP handler
    "http_method", "http_path", "http_status", "duration_ms",

    # Database
    "db_operation", "db_table", "db_duration_ms",

    # Queue/messaging
    "queue_name", "message_id", "consumer_group",

    # Negócio
    "user_id",      # ⚠️ Cuidado LGPD/GDPR — anonimizar se necessário
    "order_id", "payment_id", "tenant_id",

    # Error
    "error",        # tipo/classe do erro
    "error_detail", # mensagem de detalhe
    "retry_count",  # número de tentativas anteriores
}

# structlog: o mais ergonômico em Python
import structlog
from opentelemetry import trace

# Configurar uma vez
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,     # contexto async-safe
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.BoundLogger,
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)

logger = structlog.get_logger()

# Contexto de request — FastAPI middleware
@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    span = trace.get_current_span()
    ctx = span.get_span_context()

    # Bind ao contexto da coroutine — disponível em TODOS os logs da request
    structlog.contextvars.bind_contextvars(
        trace_id=format(ctx.trace_id, "032x"),
        span_id=format(ctx.span_id, "016x"),
        method=request.method,
        path=request.url.path,
        request_id=request.headers.get("x-request-id"),
    )

    response = await call_next(request)

    logger.info(
        "request.completed",
        status=response.status_code,
    )
    # Output: {"event":"request.completed","level":"info","timestamp":"...",
    #          "trace_id":"...","method":"POST","path":"/checkout","status":200}

    structlog.contextvars.clear_contextvars()
    return response

# ❌ Eager evaluation — executa mesmo se DEBUG desligado
logger.debug(f"Cache keys: {json.dumps(cache.keys())}")  # dump() sempre executa

# ✅ Lazy — executa só se DEBUG ativo
logger.debug("Cache keys: %s", lambda: json.dumps(cache.keys()))

# ❌ Write síncrono em loop
for item in items:
    logger.info("Processing", extra={"item_id": item.id})  # I/O a cada iteração

# ✅ Agrupar ou usar rate limiting de logs
if len(items) > 100:
    logger.info("Processing batch", extra={"count": len(items)})
else:
    for item in items:
        logger.info("Processing", extra={"item_id": item.id})

# Configurar handler assíncrono (Python)
import logging.handlers
import queue

log_queue = queue.Queue(-1)  # unlimited
queue_handler = logging.handlers.QueueHandler(log_queue)
queue_listener = logging.handlers.QueueListener(
    log_queue,
    actual_handler,  # O handler real (StreamHandler, RotatingFileHandler)
    respect_handler_level=True,
)
queue_listener.start()  # Thread separada para I/O

# Rate-limited logging: logar no máximo N vezes/segundo
import time
from collections import defaultdict

class RateLimitedLogger:
    def __init__(self, logger, max_per_second: int = 100):
        self._logger = logger
        self._max = max_per_second
        self._counts: dict[str, int] = defaultdict(int)
        self._window_start = time.monotonic()

    def info(self, event: str, **kwargs):
        now = time.monotonic()
        if now - self._window_start > 1.0:
            self._counts.clear()
            self._window_start = now

        if self._counts[event] < self._max:
            self._counts[event] += 1
            self._logger.info(event, **kwargs)
        # Dropped — nunca bloqueia o caller