Logging Estruturado em Profundidade

O módulo 05 introduziu logging estruturado como prática de cross-cutting concern. Este conceito vai mais fundo: o que diferencia logging de produção de logging de desenvolvimento, quais campos são obrigatórios para investigação real, como levels se tornam problema de performance em escala, como sampling de logs funciona sem perder os eventos que importam, e como os principais sistemas de log aggregation funcionam por dentro — incluindo seus custos reais.

Logging estruturado é a base de toda investigação em produção. Sem logs bons, traces e métricas perdem metade do valor — você sabe que um span foi lento, mas não sabe o contexto de negócio que causou isso. Com logs ruins (texto não estruturado, campos ausentes, sem correlação), a investigação vira grep manual em gigabytes de texto.

Campos obrigatórios em produção

Todo log em produção deve ter, no mínimo:

{
  "timestamp":   "2026-05-09T14:23:01.123456Z",  // ISO 8601, microsegundos, UTC
  "level":       "INFO",                           // DEBUG/INFO/WARN/ERROR/FATAL
  "service":     "order-service",                  // nome do serviço
  "version":     "2.1.0",                          // versão do binário/imagem
  "environment": "production",                     // prod/staging/dev
  "host":        "pod-order-7d9f8b-xk2p9",        // hostname/pod
  "trace_id":    "4bf92f3577b34da6a3ce929d0e0e4736", // W3C TraceContext
  "span_id":     "00f067aa0ba902b7",
  "event":       "order_placed",                   // nome do evento — não mensagem livre
  "message":     "Pedido ord-123 criado..."        // descrição legível
}

Campos adicionais dependem do domínio, mas alguns aparecem em quase todo serviço web:

{
  // Contexto HTTP
  "http_method":    "POST",
  "http_path":      "/api/orders",
  "http_status":    201,
  "http_latency_ms": 87,
  "request_id":     "req-abc123",   // ID único da requisição HTTP
  "user_agent":     "Mozilla/5.0...",

  // Contexto de negócio (varia por domínio)
  "customer_id":  "cust-456",
  "order_id":     "ord-123",
  "item_count":   3,
  "total_value":  299.90,

  // Contexto de erro (quando aplicável)
  "error_type":    "InsufficientInventoryError",
  "error_message": "SKU-789: requested 5, available 2",
  "stack_trace":   "..."   // apenas em ERROR/FATAL
}

Levels — quando cada um se aplica e por que DEBUG é perigoso em produção

DEBUG: detalhe de execução para desenvolvimento. Variáveis intermediárias, entrada de funções, estado interno. Em produção, DEBUG é desabilitado por padrão — em sistemas de alto throughput, a serialização e I/O de logs DEBUG podem consumir 10-30% da CPU. Se você precisar de DEBUG em produção para investigar, habilite por serviço por um período curto e reverta.

INFO: eventos de negócio significativos com resultado bem-sucedido. Pedido criado, pagamento processado, usuário autenticado. Cada INFO deve ser um evento que faria sentido no log de auditoria do negócio. Se você tem mais de 100 logs INFO por request em serviços normais, está logando demais — reveja o que é realmente significativo.

WARN: condição inesperada que o sistema tratou, mas que merece atenção. Retry bem-sucedido (o sistema se recuperou, mas a condição foi anormal), fallback ativado, validação de dados suspeita mas não crítica, operação próxima de limite (fila 80% cheia). WARNs não requerem ação imediata mas devem ser monitorados — aumento súbito de WARNs frequentemente precede um ERROR.

ERROR: falha que impactou o usuário ou degradou o serviço. A operação falhou, o usuário recebeu erro, o dado não foi processado. Todo ERROR deve ter contexto suficiente para investigação: o que estava sendo feito, com quais parâmetros, qual foi a exceção, qual é o impacto. Nunca logue ERROR e swallow a exception — se é ERROR, o chamador precisa saber.

FATAL/CRITICAL: falha irrecuperável que está causando o shutdown do processo. Banco de dados não conectou na inicialização, certificado inválido, dependência crítica indisponível. Após um FATAL, o processo tipicamente encerra ou entra em estado de falha total.

// Antipadrão comum: ERROR para tudo que deu errado
log.Error("Produto não encontrado para ID {ProductId}", productId);
// Problema: 404 é um resultado esperado, não um erro do sistema

// Correto: level proporcional ao impacto real
if (product == null)
    log.Information("Produto {ProductId} não encontrado — retornando 404", productId);
    // INFO: resultado de negócio esperado

if (dbConnection.State != ConnectionState.Open)
    log.Error("Falha de conexão ao banco durante checkout para cliente {CustomerId}", customerId);
    // ERROR: falha real que impacta o usuário

Correlation ID e Request ID

Em sistemas distribuídos, uma operação de negócio atravessa múltiplos serviços e gera logs em cada um. Sem identificadores correlacionados, reconstituir a jornada de um request é impossível. Dois identificadores são essenciais:

Trace ID: o ID do trace distribuído (W3C TraceContext). Gerado pelo primeiro serviço que recebe o request e propagado via header traceparent. Presente em todos os logs de todos os serviços que participam do request. Permite correlacionar logs de todos os serviços que trataram uma requisição.

Correlation ID (ou Request ID): ID específico do request HTTP dentro de um serviço. Pode ser o mesmo que o trace ID, ou pode ser um ID gerado pelo API Gateway ou load balancer antes mesmo do trace começar. O cliente da API pode enviar um X-Request-Id que você preserva e retorna — isso permite ao cliente correlacionar sua chamada com os logs do servidor.

// Middleware para propagar Correlation ID em ASP.NET Core
app.Use(async (ctx, next) => {
    // Pega do header se cliente enviou, ou gera novo
    var correlationId = ctx.Request.Headers["X-Request-Id"].FirstOrDefault()
                        ?? ctx.TraceIdentifier;  // ASP.NET gera um por request

    // Adiciona a todos os logs deste request via LogContext
    using (LogContext.PushProperty("RequestId", correlationId))
    using (LogContext.PushProperty("ClientIp", ctx.Connection.RemoteIpAddress)) {
        ctx.Response.Headers["X-Request-Id"] = correlationId;  // ecoa de volta
        await next();
    }
});

Sampling de logs — alta frequência sem custo explosivo

Em sistemas de alto throughput (10k+ req/s), logar 100% dos eventos em INFO pode gerar gigabytes por hora — com custo de storage e egress que rapidamente se torna proibitivo. Sampling permite reduzir o volume sem perder os eventos que importam.

Estratégias de sampling:

Tail-based sampling por trace: coleta todos os spans do trace em um buffer temporário e decide se mantém ou descarta baseado no resultado final. Traces com erro são sempre mantidos (100%); traces de sucesso são mantidos em 1-10%. Implementado no OTel Collector com o processador tail_sampling.

Head-based sampling: decide no início do trace se vai coletar (geralmente aleatório). Simples, mas pode descartar traces que acabam falhando — exatamente os mais importantes.

Sampling por evento de negócio: eventos críticos de negócio (pagamento, cancelamento, autenticação) nunca são amostrados — 100% de coleta. Eventos de baixo valor (health check, listagens de catálogo) são amostrados em 1-5%. Implementado na aplicação.

// OTel Collector — tail-based sampling policy
processors:
  tail_sampling:
    decision_wait: 10s      # aguarda 10s para ver o resultado do trace
    num_traces: 100000      # traces em buffer simultâneo
    policies:
      - name: errors-always
        type: status_code
        status_code: { status_codes: [ERROR] }   # 100% dos traces com erro

      - name: slow-requests
        type: latency
        latency: { threshold_ms: 1000 }          # 100% dos traces lentos

      - name: sample-happy-path
        type: probabilistic
        probabilistic: { sampling_percentage: 5 } # 5% dos demais

Log aggregation — os sistemas e seus trade-offs

Elasticsearch + Kibana (ELK Stack)

O sistema mais completo — indexação full-text de todos os campos, KQL como linguagem de query poderosa, Kibana Discover para exploração interativa. A indexação permite busca por qualquer campo sem configuração prévia.

O preço: Elasticsearch é caro em recursos. A indexação de todos os campos consome CPU e memória significativas. Em produção com volume alto, o custo de operação de um cluster Elasticsearch (ou Elastic Cloud) é substancial. A recomendação para índices de log: use ILM (Index Lifecycle Management) para mover índices antigos para nós "warm" (disco barato) e deletar após o período de retenção.

# ILM policy para logs de 30 dias
PUT _ilm/policy/logs-policy
{
  "policy": {
    "phases": {
      "hot":    { "min_age": "0ms",  "actions": { "rollover": { "max_size": "50gb", "max_age": "1d" } } },
      "warm":   { "min_age": "7d",   "actions": { "shrink": { "number_of_shards": 1 } } },
      "cold":   { "min_age": "15d",  "actions": { "freeze": {} } },
      "delete": { "min_age": "30d",  "actions": { "delete": {} } }
    }
  }
}

Loki + Grafana

Loki (Grafana Labs) adota uma filosofia diferente: não indexa o conteúdo dos logs — apenas os labels (metadados). O conteúdo é comprimido e armazenado como chunks, similar ao Prometheus para métricas. Resultado: armazenamento muito mais barato, mas queries de conteúdo são mais lentas (leitura sequencial dos chunks).

Loki é ideal quando você já tem Grafana e quer um sistema de logs integrado. Os labels do Loki devem ser poucos e de baixa cardinalidade (service, environment, pod) — os mesmos que você usaria em métricas Prometheus. Alta cardinalidade em labels do Loki causa os mesmos problemas que em Prometheus.

# LogQL — query language do Loki
# Seleciona logs do order-service em produção
{service="order-service", environment="production"}

# Filtra por conteúdo + extrai campo
{service="order-service"} |= "order_placed"
  | json
  | customer_id = "cust-456"

# Métrica de taxa de erros (metric query)
sum(rate({service="order-service"} |= "ERROR" [5m])) by (pod)

# Latência P99 extraída dos logs (se você loga elapsed_ms)
quantile_over_time(0.99,
  {service="order-service"} | json | unwrap elapsed_ms [5m]
) by (endpoint)

CloudWatch Logs Insights

Para times full-AWS, CloudWatch Logs Insights oferece queries SQL-like sobre logs sem infraestrutura adicional. O custo é por dado escaneado — queries amplas em períodos longos ficam caras. A estratégia: use log groups por serviço, defina retenção por grupo (15-90 dias), e use filtros de subscription para enviar logs críticos para S3 de longo prazo.

# CloudWatch Logs Insights — query de erros com contexto
fields @timestamp, level, event, customer_id, order_id, error_message
| filter level = "ERROR"
| filter service = "order-service"
| sort @timestamp desc
| limit 100

# Contagem de erros por tipo nos últimos 24h
filter level = "ERROR"
| stats count(*) as error_count by error_type
| sort error_count desc

Custo real de logging

Logging tem custo real que times frequentemente subestimam: storage (logs crescem linearmente com tráfego), compute (indexação e compressão), egress (enviar logs para sistemas centralizados em cloud custa por GB), e operação (manutenção do cluster de logs).

Ordem de magnitude de custo em 2026: Elasticsearch self-hosted ~$500-2000/mês para 50GB/dia com retenção de 30 dias. Elastic Cloud ~$3000-8000/mês para o mesmo volume. Loki self-hosted ~$200-500/mês. Datadog Logs ~$2.55/GB ingerido — em 50GB/dia são ~$3800/mês. CloudWatch ~$0.50/GB ingerido + $0.005/1000 queries.

Estratégia de custo: log menos e melhor. Eventos INFO devem ser acionáveis — se ninguém consulta aquele log, não escreva. Use sampling agressivo para paths de alto volume. Filtre logs de health check e readiness probe antes mesmo de enviar para o aggregator. Defina retenção curta (7-15 dias) com archival para S3 se precisar de histórico longo.

Comparativo entre linguagens — logging de produção

// C# — configuração completa de logging de produção

// Program.cs
builder.Host.UseSerilog((ctx, services, cfg) => cfg
    .ReadFrom.Configuration(ctx.Configuration)  // appsettings.json para levels
    .ReadFrom.Services(services)
    .Enrich.FromLogContext()
    .Enrich.WithMachineName()
    .Enrich.WithEnvironmentName()
    .Enrich.WithProperty("service", "order-service")
    .Enrich.WithProperty("version", Assembly.GetEntryAssembly()!
        .GetCustomAttribute<AssemblyInformationalVersionAttribute>()!
        .InformationalVersion)
    .WriteTo.Async(a => a
        .Elasticsearch(new ElasticsearchSinkOptions(new Uri("http://elasticsearch:9200")) {
            IndexFormat     = "logs-order-service-{0:yyyy.MM.dd}",
            AutoRegisterTemplate = true,
            TemplateName    = "logs",
            BatchPostingLimit = 100,
            Period          = TimeSpan.FromSeconds(2),
            QueueSizeLimit  = 100_000,          // buffer local para picos
            EmitEventFailure = EmitEventFailureHandling.WriteToSelfLog,
        })
    )
    // Fallback para console se Elasticsearch indisponível
    .WriteTo.Console(
        outputTemplate: "[{Timestamp:HH:mm:ss} {Level:u3}] {Message:lj}{NewLine}{Exception}",
        restrictedToMinimumLevel: LogEventLevel.Warning  // console apenas WARN+
    )
);

// appsettings.Production.json — levels por namespace
{
  "Serilog": {
    "MinimumLevel": {
      "Default": "Information",
      "Override": {
        "Microsoft": "Warning",
        "Microsoft.EntityFrameworkCore.Database.Command": "Warning",
        "System.Net.Http.HttpClient": "Warning",
        "Grpc": "Warning"
      }
    }
  }
}

// Middleware para correlation ID
app.Use(async (ctx, next) => {
    var reqId = ctx.Request.Headers["X-Request-Id"].FirstOrDefault()
                ?? Activity.Current?.Id
                ?? ctx.TraceIdentifier;
    using (LogContext.PushProperty("RequestId", reqId))
    using (LogContext.PushProperty("RemoteIp",  ctx.Connection.RemoteIpAddress?.ToString())) {
        ctx.Response.Headers["X-Request-Id"] = reqId;
        await next();
    }
});

# Python — structlog configurado para produção com Loki

import logging
import structlog
from opentelemetry._logs import set_logger_provider
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter

# OTel Log Provider — exporta para Loki via OTLP
logger_provider = LoggerProvider()
logger_provider.add_log_record_processor(
    BatchLogRecordProcessor(
        OTLPLogExporter(endpoint="http://otel-collector:4317"),
        max_export_batch_size=512,
        export_timeout_millis=5000,
    )
)
set_logger_provider(logger_provider)

# Processors do structlog para produção
def drop_health_checks(logger, method, event_dict):
    if event_dict.get("http_path") in ("/health", "/ready", "/metrics"):
        raise structlog.DropEvent()  # não loga health checks
    return event_dict

def add_level_number(logger, method, event_dict):
    level_map = {"debug": 10, "info": 20, "warning": 30, "error": 40, "critical": 50}
    event_dict["level_num"] = level_map.get(event_dict.get("level", ""), 0)
    return event_dict

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        drop_health_checks,
        add_level_number,
        inject_trace_context,   # processor do conceito 01
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ],
    logger_factory=structlog.PrintLoggerFactory(),
    cache_logger_on_first_use=True,  # performance: evita recriar loggers
)

# FastAPI middleware para correlation ID e contexto de request
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
import uuid

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = request.headers.get("x-request-id") or str(uuid.uuid4())

        structlog.contextvars.bind_contextvars(
            request_id=request_id,
            http_method=request.method,
            http_path=request.url.path,
            client_ip=request.client.host if request.client else None,
        )

        response = await call_next(request)

        structlog.contextvars.bind_contextvars(http_status=response.status_code)
        structlog.contextvars.clear_contextvars()

        response.headers["x-request-id"] = request_id
        return response

// Go — logging de produção com slog + filtros + sampling

package logging

import (
    "context"
    "log/slog"
    "math/rand"
    "net/http"
    "os"
    "strings"
)

// Handler com filtros de produção
type productionHandler struct {
    slog.Handler
    sampleRate float64 // 0.0-1.0 para logs INFO de baixo valor
}

func (h productionHandler) Enabled(ctx context.Context, level slog.Level) bool {
    return h.Handler.Enabled(ctx, level)
}

func (h productionHandler) Handle(ctx context.Context, r slog.Record) error {
    // Filtra health checks
    if path, ok := ctx.Value(ctxKeyPath{}).(string); ok {
        if path == "/health" || path == "/ready" || path == "/metrics" {
            return nil
        }
    }

    // Sampling: INFO de operações rotineiras amostrado
    if r.Level == slog.LevelInfo {
        if tag, ok := ctx.Value(ctxKeyTag{}).(string); ok && tag == "routine" {
            if rand.Float64() > h.sampleRate {
                return nil
            }
        }
    }

    return h.Handler.Handle(ctx, r)
}

func New(sampleRate float64) *slog.Logger {
    base := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
        Level:     slog.LevelInfo,
        AddSource: false, // desabilitado em produção — alto overhead
        ReplaceAttr: func(groups []string, a slog.Attr) slog.Attr {
            // Renomear "time" → "timestamp" para compatibilidade com Elastic/Loki
            if a.Key == "time" {
                return slog.Attr{Key: "timestamp", Value: a.Value}
            }
            // Renomear "msg" → "message"
            if a.Key == "msg" {
                return slog.Attr{Key: "message", Value: a.Value}
            }
            return a
        },
    })

    return slog.New(productionHandler{
        Handler:    traceHandler{base},  // do conceito 01
        sampleRate: sampleRate,
    }).With(
        slog.String("service", "order-service"),
        slog.String("version", Version),
    )
}

// Middleware HTTP para correlation ID e contexto
func LoggingMiddleware(log *slog.Logger, next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        reqID := r.Header.Get("X-Request-Id")
        if reqID == "" {
            reqID = newRequestID()
        }

        ctx := context.WithValue(r.Context(), ctxKeyPath{}, r.URL.Path)

        rw := &responseWriter{ResponseWriter: w}
        next.ServeHTTP(rw, r.WithContext(ctx))

        w.Header().Set("X-Request-Id", reqID)

        // Não loga health checks
        if strings.HasPrefix(r.URL.Path, "/health") {
            return
        }

        log.InfoContext(ctx, "http_request",
            slog.String("request_id",  reqID),
            slog.String("method",      r.Method),
            slog.String("path",        r.URL.Path),
            slog.Int("status",         rw.status),
            slog.Int64("latency_ms",   rw.latencyMs),
            slog.String("remote_addr", r.RemoteAddr),
        )
    })
}

Como praticar

1. Audite os logs do seu projeto atual: abra os logs de produção e avalie — você consegue responder "o que aconteceu com a requisição do usuário X às 14h23 de ontem?" sem grep em texto livre? Liste os campos ausentes e os logs com nível incorreto. Proponha um schema mínimo de campos obrigatórios para o projeto.
   Critério: lista documentada de campos ausentes; pelo menos 3 exemplos de log com nível errado e justificativa; schema proposto com no mínimo 8 campos obrigatórios incluindo trace_id, service, version, environment.
2. Configure sampling de logs: em um serviço de alto volume, adicione um processor no OTel Collector que descarta 99% dos logs INFO de endpoints /health, /metrics e /catalog, e mantém 100% de todos os ERRORs, WARNs e logs com trace_id de traces amostrados. Meça o impacto no volume gerado.
   Critério: volume de logs reduzido em pelo menos 80% sem perder nenhum ERROR/WARN; 100% dos logs de traces amostrados são retidos; teste verificando que um ERROR gerado durante sampling aparece no destino.
3. Implemente correlation IDs de ponta a ponta: configure um API Gateway para gerar X-Request-Id e propagá-lo. Configure todos os serviços downstream para ler o header, incluir nos logs como campo request_id, e repassar em chamadas HTTP saintes. Trace um request manual e confirme o mesmo ID em todos os logs.
   Critério: dado um request_id, você encontra os logs de todos os 3+ serviços envolvidos na requisição em uma única query no sistema de log aggregation; sem necessidade de saber qual serviço processou a requisição a priori.
4. Configure Loki + Grafana localmente (Docker Compose). Envie logs de um serviço Go/Python/C# via OTLP ou Promtail. Escreva queries LogQL para: (1) taxa de erros por serviço nos últimos 30 minutos, (2) distribuição de latência extraída de logs estruturados via | unwrap latency_ms, (3) todos os logs de um trace_id específico.
   Critério: as 3 queries retornam resultados corretos com dados reais; query de trace_id retorna logs de múltiplos serviços em ordem cronológica; dashboard básico criado com os 3 painéis.
5. Calcule o custo real de logging do seu sistema ou de um sistema hipotético de 50 req/s com 5 serviços. Estime o volume diário em GB, aplique os preços publicados de Datadog Logs, Elastic Cloud, e Loki self-hosted (compute + S3). Identifique os 3 maiores contribuidores de volume e avalie se são necessários.
   Critério: análise com números concretos (GB/dia calculado, custo mensal de cada opção); recomendação de qual sistema usar com justificativa; proposta de sampling que reduz custo em pelo menos 50% sem perda de informação crítica.