Patrón Structured Logging

Overview

El structured logging emite log entries como JSON objects con campos consistentes en vez de free-form text. Esto hace los logs machine-parseable, searchable en log aggregation tools (ELK, Datadog, Splunk), y filterable por cualquier field. Un structured log entry incluye un timestamp, level, message, y contextual fields como correlation ID, user ID, service name, y duration. El patrón reemplaza print(f"User {user_id} did {action}") con logger.info("user_action", user_id=user_id, action=action) — produciendo {"timestamp": "...", "level": "INFO", "message": "user_action", "user_id": 42, "action": "login"}.

When to Use

• Cualquier aplicación de producción que produce logs consumidos por aggregation tools
• Microservices donde los logs de múltiples services necesitan ser correlated
• Aplicaciones donde necesitás buscar y filtrar logs por structured fields
• Compliance requirements que demandan specific log fields
• Debuggear production issues donde grep en text logs es insuficiente

When NOT to Use

• CLI tools o scripts donde human-readable output es preferred
• Development debugging donde print() es más rápido
• Aplicaciones con muy bajo log volume donde text logs alcanzan
• Embedded systems con strict memory constraints

Solution

Python con structlog

# Python — structlog para structured logging
import structlog
import logging

# Configure structlog
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()

# Usage
logger.info("user_login", user_id=42, ip_address="192.168.1.1", method="oauth")
logger.error("payment_failed", order_id=123, reason="insufficient_funds", amount=99.99)
logger.warning("rate_limit_approaching", endpoint="/api/orders", current_rate=95, limit=100)

# Output:
# {"event": "user_login", "user_id": 42, "ip_address": "192.168.1.1",
#  "method": "oauth", "level": "info", "timestamp": "2026-01-15T10:00:00Z"}

Python con standard logging + JSON formatter

# Python — standard logging con JSON formatter
import logging
import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname.lower(),
            "message": record.getMessage(),
            "logger": record.name,
        }

        # Agregar extra fields
        for key, value in record.__dict__.items():
            if key not in ("name", "msg", "args", "levelname", "levelno",
                          "pathname", "filename", "module", "exc_info",
                          "exc_text", "stack_info", "lineno", "funcName",
                          "created", "msecs", "relativeCreated", "thread",
                          "threadName", "processName", "process", "message"):
                log_entry[key] = value

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

        return json.dumps(log_entry, default=str)

# Setup
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger = logging.getLogger("myapp")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Usage
logger.info("Order created", extra={"order_id": 123, "customer_id": 456, "total": 99.99})
logger.error("Database connection failed", extra={"host": "db.example.com", "port": 5432})

JavaScript con Winston

// JavaScript — Winston structured logging
const winston = require('winston');

const logger = winston.createLogger({
  level: process.env.LOG_LEVEL || 'info',
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.errors({ stack: true }),
    winston.format.json()
  ),
  defaultMeta: {
    service: 'order-service',
    version: '1.0.0',
    environment: process.env.NODE_ENV || 'development',
  },
  transports: [
    new winston.transports.Console(),
    new winston.transports.File({
      filename: 'logs/error.log',
      level: 'error',
    }),
    new winston.transports.File({
      filename: 'logs/combined.log',
    }),
  ],
});

// Usage
logger.info('order_created', {
  orderId: 123,
  customerId: 456,
  total: 99.99,
  items: 3,
});

logger.error('payment_failed', {
  orderId: 123,
  reason: 'insufficient_funds',
  amount: 99.99,
  paymentMethod: 'credit_card',
});

logger.warn('rate_limit_warning', {
  endpoint: '/api/orders',
  currentRate: 95,
  limit: 100,
  windowMs: 60000,
});

JavaScript con pino

// JavaScript — pino para high-performance structured logging
const pino = require('pino');

const logger = pino({
  level: process.env.LOG_LEVEL || 'info',
  timestamp: pino.stdTimeFunctions.isoTime,
  formatters: {
    level: (label) => ({ level: label }),
  },
  base: {
    service: 'order-service',
    version: '1.0.0',
  },
});

// Usage
logger.info({ orderId: 123, customerId: 456, total: 99.99 }, 'order_created');
logger.error({ orderId: 123, reason: 'insufficient_funds' }, 'payment_failed');

// Child loggers con persistent context
const requestLogger = logger.child({ requestId: 'abc-123', userId: 42 });
requestLogger.info('processing request');
requestLogger.info('request completed', { durationMs: 150 });

Java con SLF4J + Logstash encoder

// Java — Logback con Logstash JSON encoder
// pom.xml: net.logstash.logback:logstash-logback-encoder

// logback.xml
/*
<configuration>
  <appender name="JSON" class="ch.qos.logback.core.ConsoleAppender">
    <encoder class="net.logstash.logback.encoder.LogstashEncoder">
      <customFields>{"service":"order-service","version":"1.0.0"}</customFields>
    </encoder>
  </appender>
  <root level="INFO">
    <appender-ref ref="JSON" />
  </root>
</configuration>
*/

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

public class OrderService {
    private static final Logger logger = LoggerFactory.getLogger(OrderService.class);

    public void createOrder(OrderRequest request) {
        MDC.put("orderId", String.valueOf(request.getOrderId()));
        MDC.put("customerId", String.valueOf(request.getCustomerId()));

        logger.info("order_created total={} items={}",
            request.getTotal(), request.getItems().size());

        try {
            processPayment(request);
            logger.info("payment_successful amount={}", request.getTotal());
        } catch (PaymentException e) {
            logger.error("payment_failed reason={} amount={}",
                e.getReason(), request.getTotal(), e);
        } finally {
            MDC.clear();
        }
    }
}

.NET con Serilog

// C# — Serilog structured logging
using Serilog;
using Serilog.Formatting.Json;

Log.Logger = new LoggerConfiguration()
    .Enrich.WithProperty("Service", "order-service")
    .Enrich.WithProperty("Version", "1.0.0")
    .Enrich.FromLogContext()
    .WriteTo.Console(new JsonFormatter())
    .Writeto.File(new JsonFormatter(), "logs/app.log")
    .CreateLogger();

// Usage con structured properties
Log.Information("Order {OrderId} created for customer {CustomerId} with total {Total}",
    order.Id, order.CustomerId, order.Total);

Log.Error("Payment failed for order {OrderId}: {Reason}",
    order.Id, ex.Message);

// Usando LogContext para scoped properties
using (LogContext.PushProperty("CorrelationId", correlationId))
{
    Log.Information("Processing order {OrderId}", order.Id);
    // Todos los logs dentro de este scope incluyen CorrelationId
}

Log levels y cuándo usarlos

# Python — log level guidelines
import structlog
logger = structlog.get_logger()

# DEBUG — detailed diagnostic info, disabled en production
logger.debug("cache_hit", key="user:42", ttl=300)

# INFO — general operational events (request processed, order created)
logger.info("order_created", order_id=123, total=99.99)

# WARNING — algo unexpected pero not fatal (rate limit approaching, retry needed)
logger.warning("retry_attempt", attempt=2, max_attempts=3, delay_ms=500)

# ERROR — un failure que debería ser investigado (payment failed, db error)
logger.error("payment_failed", order_id=123, reason="insufficient_funds")

# CRITICAL — system-wide failure requiring immediate action
logger.critical("database_unreachable", host="db.example.com", port=5432)

Contextual logging con bound fields

# Python — bind context al logger
import structlog
logger = structlog.get_logger()

# Bind persistent context
request_logger = logger.bind(
    correlation_id="abc-123",
    user_id=42,
    endpoint="/api/orders",
)

# Todos los logs subsiguientes incluyen bound fields
request_logger.info("request_started")
request_logger.info("db_query_executed", query="SELECT * FROM orders", duration_ms=15)
request_logger.info("request_completed", status_code=200, duration_ms=150)

# Output:
# {"correlation_id": "abc-123", "user_id": 42, "endpoint": "/api/orders",
#  "event": "request_started", "level": "info", "timestamp": "..."}

ELK integration

# Logstash pipeline configuration para structured logs
input:
  file:
    path: "/var/log/myapp/*.log"
    codec: json
    start_position: "beginning"

filter:
  # Ensure timestamp es parsed correctly
  date:
    match: ["timestamp", "ISO8601"]
    target: "@timestamp"

  # Add service field si missing
  if ![service]:
    mutate:
      add_field:
        service: "unknown"

output:
  elasticsearch:
    hosts: ["http://elasticsearch:9200"]
    index: "myapp-logs-%{+YYYY.MM.dd}"

Datadog integration

# datadog.yaml — collect structured logs
logs:
  - type: file
    path: "/var/log/myapp/*.log"
    service: "order-service"
    source: "python"
    sourcecategory: "sourcecode"
    log_processing_rules:
      - type: multi_line
        name: "json_logs"
        pattern: '^\{'

Variants

Async logging para high throughput

# Python — async logging con queue handler
import logging
import logging.handlers
import queue

log_queue = queue.Queue(-1)  # Unlimited size

# Queue handler — non-blocking
queue_handler = logging.handlers.QueueHandler(log_queue)

# Queue listener — procesa logs en un separate thread
file_handler = logging.FileHandler("app.log")
file_handler.setFormatter(JsonFormatter())

listener = logging.handlers.QueueListener(log_queue, file_handler)
listener.start()

logger = logging.getLogger("myapp")
logger.addHandler(queue_handler)
logger.setLevel(logging.INFO)

# Logs se enqueue sin blocking el main thread
logger.info("high_throughput_log", extra={"count": 1000000})

Sampling para high-volume logs

// JavaScript — pino con sampling para high-volume logs
const pino = require('pino');

const logger = pino({
  level: 'debug',
  // Logear solo 10% de debug logs en production
  levelVal: process.env.NODE_ENV === 'production' ? 20 : 10,
});

// O usar una custom sampling strategy
const sampledLogger = pino({
  timestamp: pino.stdTimeFunctions.isoTime,
  base: { service: 'api' },
}, pino.destination({ sync: false }));

// Samplear 1 in every 100 requests at debug level
let requestCount = 0;
function logRequest(req) {
  requestCount++;
  if (requestCount % 100 === 0 || req.path !== '/health') {
    sampledLogger.info({ path: req.path, method: req.method }, 'request_received');
  }
}

Redacting sensitive fields

# Python — redact sensitive fields en logs
import structlog
import re

def redact_sensitive(processor, logger, event_dict):
    sensitive_keys = {"password", "token", "api_key", "credit_card", "ssn"}
    for key in list(event_dict.keys()):
        if key.lower() in sensitive_keys:
            event_dict[key] = "[REDACTED]"
        elif isinstance(event_dict[key], str):
            # Redact credit card numbers en cualquier string field
            event_dict[key] = re.sub(
                r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
                '[REDACTED]',
                event_dict[key],
            )
    return event_dict

structlog.configure(
    processors=[
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        redact_sensitive,
        structlog.processors.JSONRenderer(),
    ],
)

Best Practices

• For a deeper guide, see Structured Logging: JSON Logs, Correlation IDs, Aggregation.

• Usá JSON format en production — machine-parseable, searchable en aggregation tools

• Incluí context, no solo messages — logger.info("order_created", order_id=123) no logger.info("Order 123 created")

• Usá consistent field names — user_id en todos lados, no userId en un service y user_id en otro

• Incluí un timestamp en ISO 8601 — 2026-01-15T10:00:00.123Z

• Usá appropriate log levels — DEBUG para dev, INFO para operations, ERROR para failures

• Bind request context — correlation ID, user ID, endpoint deberían estar en cada log entry para un request

• Redact sensitive data — passwords, tokens, credit card numbers nunca deberían aparecer en logs

• Usá async logging para high throughput — no dejes que logging blockee request processing

• Logeá en boundaries — service entry/exit, external calls, state transitions

• Incluí duration para operations — duration_ms field para cualquier timed operation

Common Mistakes

• String interpolation en vez de structured fields: logger.info(f"User {user_id} logged in") en vez de logger.info("user_login", user_id=user_id). La structured version es searchable por user_id.
• Inconsistent field names: user_id en un service, userId en otro. Estandarizá across todos los services.
• Loggear sensitive data: passwords, tokens, PII en logs. Redact u omití estos fields.
• Demasiados log levels: usar los 6 levels inconsistentemente. Stickte a DEBUG, INFO, WARN, ERROR.
• No incluir timestamps: depender del log shipper para agregar timestamps. Incluilos en el log entry mismo.
• Loggear en hot loops: loggear dentro de tight loops genera enormous volume. Sampleá o agregá.