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. É aquele cenário clássico onde você tem três abas de terminal abertas, grepando timestamps e rezando pra coincidir alguma coisa.

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. Sem precisar convencer seu gerente a aprovar mais uma ferramenta na infraestrutura.

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.

A diferença entre “temos um erro” e “temos um erro na chamada do Payment Service porque o gateway de pagamento voltou 503 às 21:05:36” é a diferença entre passar a noite toda no PagerDuty e dormir tranquilo.

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 — alguém precisa criar o ID no ponto de entrada
  • Propagação via headers HTTP entre serviços — o ID precisa viajar de serviço em serviço
  • Context threading dentro de cada serviço — especialmente com threads e async/await, o ID precisa sobreviver ao câmbio de contexto
  • Formatação automática dos logs — ninguém quer escrever correlation_id=xyz manualmente em cada logger.info() do projeto

Cada um desses pontos é uma armadilha em potencial. Vou resolver todos eles.

Camada 1: O Gerador de Correlation ID

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

# 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."""
    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. O threading.local() morre quando você muda de thread ou entra num event loop async. Com contextvars, o correlation ID sobrevive a qualquer tipo de concorrência que o Python jogar em cima.

Se algum dia você migrar pra FastAPI com endpoints async, o correlation ID continua funcionando sem mudar uma linha. Essa é a beleza de usar a ferramenta certa desde o começo.

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 — o que obviamente não vai acontecer — vamos criar um logging.Formatter que injeta automaticamente o ID em toda linha de log, sem esforço.

# 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:
        record.correlation_id = get_correlation_id() or "no-request"
        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

Três serviços diferentes, o mesmo correlation ID. Trace completo com um grep. Você não precisa mais cruzar timestamps manualmente ou abrir cinco terminais ao mesmo tempo.

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 ou ASGI — Flask, FastAPI, Django, o que for.

# middleware.py
import logging
from functools import wraps
from correlation_id import set_correlation_id, get_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."""
    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 — o caso onde a requisição já veio de outro serviço interno
  2. Gera um novo ID se for a requisição de entrada na borda do sistema — o caso do usuário acessando pela primeira vez
  3. Loga automaticamente o ID no início e no fim de cada request — pra você ter bookends claros de onde a requisição começou e terminou

O decorator @wraps(func) é importante aqui: ele preserva o nome e o docstring da função original, o que é crucial pra debugging e pra ferramentas de profiling que olham o __name__ das funções.

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. E se você tem 47 pontos no código onde faz chamadas HTTP, ninguém vai lembrar de adicionar o header em todos eles.

A solução é criar wrappers automáticos:

# 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. Basta trocar as chamadas diretas de requests.get() por get_with_correlation(). Se você tem um service layer bem definido, essa troca é trivial — poucos pontos de alteração.

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

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):
    """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())
    entries.sort(key=lambda e: e["timestamp"])
    return entries

def reconstruct_request_flow(entries):
    """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']}] {entry['message']}"
        )
    return "\n".join(flow)

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

$ python -c "from log_aggregator import *; 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
⚠️ 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. Sem precisar ser detective de timestamp.

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. Em staging, tudo funcionava porque o ambiente era mais lento e os timeouts eram maiores. Sem correlation ID, você acha que é problema de rede e fica rodando ping e traceroute como se fosse 2010. 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 que quebra muita gente. Se você spawnar tasks manualmente com asyncio.create_task(), o contexto não é propagado automaticamente para a nova task em versões mais 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}")
    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 essas versões em produção — você precisa do copy_context() explicitamente. Sem isso, o correlation ID morre na task async e seus logs ficam com no-request no meio do fluxo.

Estrutura Final do Projeto

Quando tudo está junto, a estrutura do projeto fica assim:

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

Total: ~200 linhas de código. Sem dependências. Sem infraestrutura adicional. Sem necessidade de time de plataforma pra manter uma stack de observabilidade. Sem precisar convencer ninguém a aprovar orçamento.

Quando Escalar para Algo Mais Robusto

Essa implementação resolve 90% dos problemas de rastreamento para times pequenos e médios. Mas existem limites que você precisa reconhecer com honestidade:

  • Volume acima de 10k requests/minuto: grep em arquivos de log fica lento. Nesse ponto, considere Loki (mais leve que Elasticsearch) ou OpenSearch. O correlation ID continua sendo o mesmo — só muda onde você armazena.
  • Múltiplos times: Se outro time precisa acessar seus logs, um centralizador (Grafana, Datadog, New Relic) faz sentido. Mas o correlation ID via header continua sendo o mecanismo de propagação.
  • Spans e métricas: Correlation ID rastreia fluxo, mas não duração de cada etapa. Para métricas de latência por serviço, adicione spans — aí sim, OpenTelemetry vale a pena.
  • Logs em múltiplos nós: Se seus serviços rodam em containers efêmeros, você precisa de um log shipper (Fluentd, Vector) pra centralizar os arquivos antes de fazer o grep.

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. Não amanhã. Não depois do próximo sprint. 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-...")'

O segundo passo é adicionar health checks com correlation ID — quando um serviço faz health check no outro, inclua o correlation ID pra poder rastrear até as verificações de saúde. Isso parece overkill até o dia em que o health check é o culpado de um loop de retry infinito.

O terceiro passo é correlation ID em filas de mensagem — se você usa RabbitMQ, Kafka ou SQS, injete o correlation ID nas mensagens também. Assim o rastreamento não quebra quando a comunicação deixa de ser síncrona.

Mas isso é assunto pra outros posts. O importante é: comece com o básico que funciona. Correlation ID é o alicerce. Structured logging, métricas, traces distribuídos — 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