Rastreamento de correlation ID em sistema distribuído Python - monitoramento de logs de erros

Correlation ID em Python Puro: O Rastreador Que Conecta Logs de 5 Microsserviços Sem Precisar de Elasticsearch

PorOlivetto

O Log Que Não Conecta Nada é Só Ruído

Você já abriu o arquivo de log de um serviço, encontrou um erro, correu para o outro serviço relacionado, e não soube se os dois eventos pertenciam à mesma requisição? Se a resposta é sim, você não está sozinho. A maioria dos times trata logs como um depósito de lixo digital: guarda tudo, não encontra nada.

Correlation ID é o conceito mais subestimado em observabilidade de sistemas distribuídos. E hoje vou te mostrar como implementar um rastreador completo em Python puro — sem OpenTelemetry, sem Jaeger, sem aquela stack inteira que exige um Kubernetes dedicado só pra rodar.

Quando um usuário faz uma requisição e ela passa por API Gateway → Auth Service → Order Service → Payment Service → Notification Service, você precisa de um fio condutor que conecte todos esses pedaços. Sem correlation ID, você tem cinco logs separados. Com correlation ID, você tem uma história completa.

Como um Correlation ID Funciona na Prática

O conceito é ridículo de simples: gerar um identificador único quando a requisição chega, propagá-lo por todos os serviços envolvidos, e incluí-lo em cada linha de log que qualquer serviço produzir. Parece trivial, né? Então por que 80% dos sistemas não fazem isso?

Porque implementar de forma consistente é mais difícil do que parece. Você precisa de:

  • Geração centralizada na borda do sistema
  • Propagação via headers HTTP entre serviços
  • Context threading dentro de cada serviço (especialmente com threads/async)
  • Formatação automática dos logs — ninguém quer escrever correlation_id=xyz manualmente em cada logger.info()

Vou resolver cada um desses problemas.

Camada 1: O Gerador de Correlation ID

Primeiro, vamos criar o módulo central que gera e armazena o correlation ID em contexto local de thread. Python já tem contextvars pra isso — é literalmente feito para armazenar estado por requisição.

# correlation_id.py
import uuid
import contextvars

# Context variable que sobrevive entre chamadas no mesmo request
_current_correlation_id: contextvars.ContextVar[str] = contextvars.ContextVar(
    "correlation_id", default=None
)

def generate_correlation_id() -> str:
    """Gera um novo correlation ID no formato UUID4 compacto."""
    return str(uuid.uuid4())

def set_correlation_id(correlation_id: str | None = None) -> str:
    """Define o correlation ID no contexto atual.
    Se None, gera um novo automaticamente."""
    cid = correlation_id or generate_correlation_id()
    _current_correlation_id.set(cid)
    return cid

def get_correlation_id() -> str | None:
    """Recupera o correlation ID do contexto atual."""
    return _current_correlation_id.get()

def clear_correlation_id() -> None:
    """Limpa o correlation ID do contexto (útil pós-request)."""
    _current_correlation_id.set(None)

Repare que usei contextvars.ContextVar e não threading.local(). Por quê? Porque contextvars funciona tanto com threads quanto com async/await. Se algum dia você migrar pra FastAPI com endpoints async, o correlation ID continua funcionando sem mudar uma linha.

Camada 2: O Formatter de Log Automático

Aqui é onde a mágica acontece. Em vez de todo desenvolvedor lembrar de adicionar o correlation ID em cada log, vamos criar um logging.Formatter que injeta automaticamente:

# log_formatter.py
import logging
from correlation_id import get_correlation_id

class CorrelationIdFormatter(logging.Formatter):
    """Formatter que injeta correlation_id em toda linha de log."""

    def format(self, record: logging.LogRecord) -> str:
        # Adiciona correlation_id ao log record
        record.correlation_id = get_correlation_id() or "no-request"
        # Delega para o formatador padrão com nosso template
        return super().format(record)


def setup_correlation_logging():
    """Configura o logging raiz com o formatter de correlation ID."""
    formatter = CorrelationIdFormatter(
        fmt="%(asctime)s [%(correlation_id)s] %(levelname)s %(name)s: %(message)s",
        datefmt="%Y-%m-%dT%H:%M:%S"
    )

    handler = logging.StreamHandler()
    handler.setFormatter(formatter)

    root = logging.getLogger()
    root.addHandler(handler)
    root.setLevel(logging.DEBUG)

    return root

Agora, qualquer logger.info("pedido criado") vai sair automaticamente como:

2026-07-07T21:05:33 [a3f7b2c1-...] INFO order_service: pedido criado
2026-07-07T21:05:34 [a3f7b2c1-...] DEBUG payment_gateway: tentando cobrança
2026-07-07T21:05:36 [a3f7b2c1-...] ERROR payment_gateway: timeout na cobrança - gateway indisponível

Três serviços diferentes, o mesmo correlation ID. Trace completo com grep.

Camada 3: Middleware HTTP para Extração e Propagação

Os serviços precisam extrair o correlation ID dos headers de entrada e repassá-lo nos headers de saída. Vamos criar um middleware genérico que funciona com qualquer framework WSGI/ASGI:

# middleware.py
import logging
from functools import wraps
from correlation_id import set_correlation_id, get_correlation_id, generate_correlation_id

CORRELATION_HEADER = "X-Correlation-ID"
logger = logging.getLogger(__name__)

def extract_correlation_id(request) -> str:
    """Extrai correlation ID do header ou gera um novo."""
    # Flask: request.headers.get()
    # FastAPI: request.headers.get()
    # Django: request.META.get('HTTP_X_CORRELATION_ID')
    cid = request.headers.get(CORRELATION_HEADER)
    return set_correlation_id(cid)

def inject_correlation_id(headers: dict) -> dict:
    """Injeta correlation ID em headers de requisições outgoing."""
    cid = get_correlation_id()
    if cid:
        headers[CORRELATION_HEADER] = cid
    return headers

def correlation_middleware(func):
    """Decorator para envolver handlers HTTP com correlation ID."""
    @wraps(func)
    def wrapper(request, *args, **kwargs):
        cid = extract_correlation_id(request)
        logger.info(f"[{cid}] Request recebido: {request.method} {getattr(request, 'path', '/')}")
        try:
            response = func(request, *args, **kwargs)
            return response
        except Exception as exc:
            logger.exception(f"[{cid}] Erro não tratado: {exc}")
            raise
    return wrapper

Esse middleware faz três coisas essenciais:

  1. Extrai o correlation ID do header X-Correlation-ID se ele existir (caso a requisição venha de outro serviço)
  2. Gera um novo ID se for a requisição de entrada na borda do sistema
  3. Loga automaticamente o ID no início e no fim de cada request

Camada 4: Propagação entre Microsserviços

A parte mais negligenciada é a comunicação entre serviços. Se seu Order Service chama o Payment Service via requests ou httpx, o correlation ID morre na próxima chamada HTTP — a menos que você o injete explicitamente.

# http_client.py
import requests
import httpx
from correlation_id import get_correlation_id
from middleware import CORRELATION_HEADER, inject_correlation_id

def get_with_correlation(url: str, **kwargs) -> requests.Response:
    """Wrapper para requests.get que injeta correlation ID."""
    headers = kwargs.pop("headers", {})
    inject_correlation_id(headers)
    return requests.get(url, headers=headers, **kwargs)

def post_with_correlation(url: str, json: dict, **kwargs) -> requests.Response:
    """Wrapper para requests.post que injeta correlation ID."""
    headers = kwargs.pop("headers", {})
    inject_correlation_id(headers)
    return requests.post(url, json=json, headers=headers, **kwargs)


class CorrelationHTTPXClient:
    """Cliente httpx com correlation ID automático em todas as requisições."""

    def __init__(self):
        self.client = httpx.Client()

    def request(self, method: str, url: str, **kwargs):
        headers = kwargs.pop("headers", {})
        inject_correlation_id(headers)
        return self.client.request(method, url, headers=headers, **kwargs)

    def close(self):
        self.client.close()

Com isso, toda chamada entre serviços carrega o correlation ID automaticamente. Você não precisa lembrar de adicionar o header em cada requests.get() do código.

Camada 5: Agregador de Logs por Correlation ID

Ok, agora seus logs têm correlation ID. Mas como encontrar todos os logs de uma requisição específica quando algo dá errado? Você não precisa de Elasticsearch pra isso — especialmente se o volume de logs for gerenciável.

# log_aggregator.py
import re
from pathlib import Path
from collections import defaultdict
from typing import Generator

LOG_PATTERN = re.compile(
    r"(?P<timestamp>\S+)\s+\[(?P<correlation_id>[^\]]+)\]\s+"
    r"(?P<level>\S+)\s+(?P<service>\S+):\s+(?P<message>.*)"
)

def grep_correlation_id(log_dir: str, correlation_id: str) -> list[dict]:
    """Varre todos os arquivos de log em um diretório e retorna
    todas as entradas que correspondem ao correlation ID."""
    entries = []
    log_path = Path(log_dir)

    for log_file in log_path.glob("*.log"):
        with open(log_file) as f:
            for line in f:
                match = LOG_PATTERN.search(line)
                if match and match.group("correlation_id") == correlation_id:
                    entries.append(match.groupdict())

    # Ordena por timestamp para reconstruir a narrativa
    entries.sort(key=lambda e: e["timestamp"])
    return entries


def reconstruct_request_flow(entries: list[dict]) -> str:
    """Reconstrói a narrativa completa de uma requisição a partir dos logs."""
    flow = []
    for entry in entries:
        status_icon = {
            "INFO": "✅",
            "DEBUG": "🔍",
            "WARNING": "⚠️",
            "ERROR": "❌",
            "CRITICAL": "💀",
        }.get(entry["level"], "📝")

        flow.append(
            f"{status_icon} {entry['timestamp']} [{entry['service']}] "
            f"{entry['message']}"
        )
    return "\n".join(flow)

Com isso, quando algo der errado, você roda:

$ python -c "
from log_aggregator import grep_correlation_id, reconstruct_request_flow
entries = grep_correlation_id('/var/log/app', 'a3f7b2c1-...')
print(reconstruct_request_flow(entries))
"

E recebe algo assim:

✅ 2026-07-07T21:05:33 [api_gateway] Request recebido: POST /api/orders
✅ 2026-07-07T21:05:33 [auth_service] Token validado para usuário 42
✅ 2026-07-07T21:05:34 [order_service] Pedido #1234 criado
🔍 2026-07-07T21:05:34 [payment_gateway] Iniciando cobrança R$ 149.90
❌ 2026-07-07T21:05:36 [payment_gateway] Timeout na cobrança - gateway indisponível
⚠️ 2026-07-07T21:05:36 [order_service] Rollback do pedido #1234

Cinco serviços, um correlation ID, narrativa completa. Sem precisar abrir cinco terminais diferentes.

Cenário Real: O Bug Que Só Apareceu em Produção

Esse é o tipo de bug que só aparece em produção, sob carga real, com timing específico. Sem correlation ID, você acha que é problema de rede. Com correlation ID, você vê exatamente onde a cadeia quebrou.

Async/await: O Desafio Extra

Se você usa asyncio com FastAPI ou aiohttp, o contextvars continua funcionando — mas há uma pegadinha. Se você spawnar tasks manualmente com asyncio.create_task(), o contexto não é propagado automaticamente para a nova task em versões antigas do Python.

# async_correlation.py
import asyncio
from correlation_id import get_correlation_id, set_correlation_id
import contextvars

async def process_payment_async(order_id: int):
    cid = get_correlation_id()
    logger.info(f"[{cid}] Processando pagamento assíncrono para pedido {order_id}")
    # Simula chamada ao gateway
    await asyncio.sleep(0.5)
    logger.info(f"[{cid}] Pagamento processado para pedido {order_id}")

async def handle_order(request):
    cid = set_correlation_id(request.headers.get("X-Correlation-ID"))
    logger.info(f"[{cid}] Processando pedido")

    # CORRETO: copia contexto explicitamente para a task
    ctx = contextvars.copy_context()
    task = asyncio.create_task(process_payment_async(42))

    await task
    return {"status": "ok", "correlation_id": cid}

A partir do Python 3.11, asyncio.create_task() propaga o contexto automaticamente. Mas se seu código precisa rodar em 3.9 ou 3.10 (e sim, muita gente ainda roda), você precisa do copy_context() explicitamente.

Estrutura Final do Projeto

Quando tudo está junto, a estrutura fica assim:

meu-projeto/
├── correlation_id.py      # Geração e storage do ID
├── log_formatter.py       # Formatter com injection automática
├── middleware.py           # Extração/propagação HTTP
├── http_client.py         # Wrappers de requests/httpx
├── log_aggregator.py      # Busca e reconstrução de traces
└── app.py                 # Sua aplicação

Total: ~200 linhas de código. Sem dependências. Sem infraestrutura adicional. Sem necessidade de time de plataforma pra manter uma stack de observabilidade.

Quando Escalar para Algo Mais Robusto

Essa implementação resolve 90% dos problemas de rastreamento para times pequenos e médios. Mas existem limites:

  • Volume acima de 10k requests/minuto: grep em arquivos de log fica lento. Considere Loki (mais leve que Elasticsearch) ou OpenSearch.
  • Múltiplos times: Se outro time precisa acessar seus logs, um centralizador (Grafana, Datadog) faz sentido.
  • Spans e métricas: Correlation ID rastreia fluxo, mas não duração de cada etapa. Para métricas de latência, adicione spans (aí sim, OpenTelemetry vale a pena).

Mas pra resolver o problema agora, sem burocracia, sem infra extra, sem reuniões de alinhamento com o time de SRE — essa implementação entrega valor hoje.

O Que Fazer Depois de Implementar

Depois que o correlation ID está rodando, o próximo passo natural é adicionar structured logging (JSON ao invés de texto puro). Isso torna o grep_correlation_id desnecessário — você pode usar jq diretamente nos logs:

$ cat /var/log/app/app.log | jq 'select(.correlation_id == "a3f7b2c1-...")'

Mas isso é assunto pra outro post. O importante é: comece com o básico que funciona. Correlation ID é o alicerce. Structured logging, métricas, traces — tudo vem depois.

Se você implementou correlation ID no seu sistema e quer compartilhar como foi, ou se tem alguma dúvida sobre alguma parte da implementação, deixa nos comentários. E me diz: qual automação ou ferramenta de engenharia você quer ver implementada do zero em Python puro no próximo post?

Porque a gente não para aqui. O log é só o começo.

Posts Similares