distributed tracing python puro rastreamento logs microservicos

Distributed Tracing em Python Puro: O Sistema Que Rastreia Requisicoes Entre Multiplos Servicos e Conecta Logs Automaticamente (Sem Jaeger, Sem OpenTelemetry)

PorOlivetto

Voce ja passou pela seguinte situacao: um usuario reporta um erro, voce abre os logs do servico A e encontra a mensagem “Erro ao processar requisicao”. Mas qual requisicao? De onde veio? Para onde foi? Voce pula para o servico B, depois para o C, e tem tres terminais abertos tentando correlacionar timestamps manualmente, torcendo para que os relogios estejam sincronizados.

Se isso te parece familiar, bem-vindo ao clube. Esse e o pesadelo classico de quem trabalha com sistemas distribuidos sem distributed tracing.

A boa noticia? Voce nao precisa de Jaeger, Zipkin, OpenTelemetry SDK ou qualquer dependencia pesada para resolver isso. Em Python puro, com cerca de 150 linhas de codigo, e possivel implementar um sistema de rastreamento distribuido que gera trace IDs, propaga contexto entre servicos e agrupa logs automaticamente.

Neste artigo, vou te mostrar como construi um tracer minimalista que salvou minha pele em producao mais vezes do que posso contar.

O Problema Real: Logs Sem Contexto Sao Inuteis

Vamos ser honestos: logs sao essenciais. Mas logs sem correlacao sao como paginas de um livro embaralhadas. Voce tem todas as informacoes, mas nenhuma narrativa.

Considere este cenario real que enfrentei semana passada:

  • Usuario clica em “Finalizar Compra”
  • API Gateway recebe a requisicao e valida o token
  • Servico de Pedidos cria o pedido no banco
  • Servico de Pagamento processa a cobranca
  • Servico de Estoque reserva os produtos
  • Servico de Notificacao envia o email de confirmacao

Se o email nao chega, onde voce comeca a investigar? Sem um identificador unico que percorra toda essa cadeia, voce esta basicamente adivinhando.

tela debug codigo distributed tracing python puro observabilidade
Tela de debug: sem trace ID, cada servico e uma ilha isolada de informacao.

A Solucao: Trace ID Propagation

O conceito e simples: cada requisicao recebe um ID unico (trace_id) na origem. Esse ID e propagado atraves de todos os servicos envolvidos, e cada log gerado inclui esse identificador.

Na pratica, funciona assim:

  1. Geracao: O servico de entrada cria um UUID unico
  2. Propagacao: O ID e passado via header HTTP (X-Trace-ID)
  3. Contexto: Cada servico extrai o ID do header e o inclui em todos os logs
  4. Agregacao: Ferramentas de log agregam mensagens pelo trace_id

A beleza dessa abordagem e que ela nao exige nenhuma biblioteca externa. Voce so precisa de tres ingredientes:

  • contextvars (nativo do Python 3.7+)
  • Headers HTTP (qualquer framework web suporta)
  • Um logger minimamente customizado

Implementacao em Python Puro

Vamos ao codigo. Primeiro, precisamos de um mecanismo para armazenar o trace_id no contexto da requisicao atual. Python tem o modulo contextvars (disponivel desde 3.7) que e perfeito para isso:

import contextvars
import uuid
import logging
import time
from functools import wraps

# Variavel de contexto para armazenar o trace_id
trace_id_var = contextvars.ContextVar('trace_id', default=None)

class TracedLogger:
    def __init__(self, name):
        self.logger = logging.getLogger(name)

    def _format_message(self, msg):
        trace_id = trace_id_var.get()
        if trace_id:
            return f"[{trace_id}] {msg}"
        return msg

    def info(self, msg, *args, **kwargs):
        self.logger.info(self._format_message(msg), *args, **kwargs)

    def error(self, msg, *args, **kwargs):
        self.logger.error(self._format_message(msg), *args, **kwargs)

    def warning(self, msg, *args, **kwargs):
        self.logger.warning(self._format_message(msg), *args, **kwargs)

    def debug(self, msg, *args, **kwargs):
        self.logger.debug(self._format_message(msg), *args, **kwargs)

Simples, nao? Agora todo log gerado por esse logger incluira automaticamente o trace_id se ele estiver no contexto. Zero dependencias externas.

Middleware para Propagacao Automatica

Para frameworks web como Flask ou FastAPI, podemos criar middlewares que extraem o trace_id dos headers HTTP e o colocam no contexto:

# Middleware para Flask
from flask import request

def trace_middleware(app):
    @app.before_request
    def before_request():
        # Extrai trace_id do header ou gera um novo
        trace_id = request.headers.get('X-Trace-ID')
        if not trace_id:
            trace_id = str(uuid.uuid4())[:8]
        trace_id_var.set(trace_id)

    @app.after_request
    def after_request(response):
        # Adiciona trace_id na resposta para debug
        trace_id = trace_id_var.get()
        if trace_id:
            response.headers['X-Trace-ID'] = trace_id
        return response

E para FastAPI, o padrao e similar:

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class TraceMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        trace_id = request.headers.get('X-Trace-ID')
        if not trace_id:
            trace_id = str(uuid.uuid4())[:8]
        trace_id_var.set(trace_id)

        response = await call_next(request)
        response.headers['X-Trace-ID'] = trace_id
        return response

Para chamadas HTTP entre servicos, precisamos propagar o trace_id nos headers de saida:

import urllib.request
import urllib.error

class TracedHTTPClient:
    @staticmethod
    def request(url, method='GET', data=None, headers=None):
        headers = headers or {}

        # Adiciona trace_id aos headers se existir no contexto
        trace_id = trace_id_var.get()
        if trace_id:
            headers['X-Trace-ID'] = trace_id

        req = urllib.request.Request(
            url, data=data, headers=headers, method=method
        )

        try:
            with urllib.request.urlopen(req) as response:
                return response.read(), response.status
        except urllib.error.HTTPError as e:
            return e.read(), e.code

Span: Medindo Duracao de Operacoes

Alem do trace_id, e util medir quanto tempo cada operacao leva. Isso nos permite identificar gargalos. Vamos criar um decorador de span:

def traced_span(operation_name):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            logger = TracedLogger(func.__module__)
            start_time = time.time()

            logger.info(f"Iniciando operacao: {operation_name}")

            try:
                result = func(*args, **kwargs)
                duration = time.time() - start_time
                logger.info(
                    f"Operacao {operation_name} concluida em {duration:.3f}s"
                )
                return result
            except Exception as e:
                duration = time.time() - start_time
                logger.error(
                    f"Operacao {operation_name} falhou apos {duration:.3f}s: {e}"
                )
                raise

        return wrapper
    return decorator

Uso pratico:

@traced_span("processar_pagamento")
def processar_pagamento(pedido_id, valor):
    # Simula processamento
    time.sleep(0.5)
    return {"status": "aprovado", "pedido_id": pedido_id}
codigo monitoramento performance distributed tracing span python
Spans revelam exatamente onde o tempo esta sendo gasto em cada servico.

Exemplo Completo: Sistema de Pedidos com 3 Servicos

Vamos juntar tudo em um exemplo pratico. Imagine tres servicos: API Gateway, Servico de Pedidos e Servico de Pagamento.

API Gateway (porta 8000)

from flask import Flask, request, jsonify
import json

app = Flask(__name__)
trace_middleware(app)
logger = TracedLogger('gateway')

@app.route('/pedidos', methods=['POST'])
def criar_pedido():
    logger.info("Recebendo requisicao para criar pedido")

    dados = request.get_json()

    client = TracedHTTPClient()
    response, status = client.request(
        'http://servico-pedidos:8001/pedidos',
        method='POST',
        data=json.dumps(dados).encode(),
        headers={'Content-Type': 'application/json'}
    )

    if status == 201:
        logger.info("Pedido criado com sucesso")
        return jsonify(json.loads(response)), 201
    else:
        logger.error(f"Falha ao criar pedido: status {status}")
        return jsonify({"erro": "Falha ao processar pedido"}), 500

if __name__ == '__main__':
    app.run(port=8000)

Servico de Pedidos (porta 8001)

app = Flask(__name__)
trace_middleware(app)
logger = TracedLogger('pedidos')

@app.route('/pedidos', methods=['POST'])
def processar_pedido():
    logger.info("Processando novo pedido")

    dados = request.get_json()
    pedido_id = str(uuid.uuid4())[:8]

    client = TracedHTTPClient()
    response, status = client.request(
        'http://servico-pagamento:8002/pagamentos',
        method='POST',
        data=json.dumps({
            "pedido_id": pedido_id,
            "valor": dados['valor']
        }).encode(),
        headers={'Content-Type': 'application/json'}
    )

    if status == 200:
        logger.info(f"Pedido {pedido_id} aprovado")
        return jsonify({
            "pedido_id": pedido_id,
            "status": "aprovado"
        }), 201
    else:
        logger.error(f"Pagamento recusado para pedido {pedido_id}")
        return jsonify({"erro": "Pagamento recusado"}), 400

Servico de Pagamento (porta 8002)

app = Flask(__name__)
trace_middleware(app)
logger = TracedLogger('pagamento')

@traced_span("validar_cartao")
def validar_cartao(valor):
    time.sleep(0.3)
    return True

@traced_span("processar_cobranca")
def processar_cobranca(pedido_id, valor):
    time.sleep(0.2)
    return {"transacao_id": str(uuid.uuid4())[:8]}

@app.route('/pagamentos', methods=['POST'])
def processar_pagamento():
    logger.info("Processando pagamento")

    dados = request.get_json()
    pedido_id = dados['pedido_id']
    valor = dados['valor']

    logger.info(f"Validando cartao para pedido {pedido_id}")

    if not validar_cartao(valor):
        logger.warning(f"Cartao recusado para pedido {pedido_id}")
        return jsonify({"erro": "Cartao recusado"}), 400

    logger.info(f"Processando cobranca para pedido {pedido_id}")
    resultado = processar_cobranca(pedido_id, valor)

    logger.info(f"Pagamento aprovado: transacao {resultado['transacao_id']}")
    return jsonify({"status": "aprovado", **resultado}), 200

Analisando Logs com Trace ID

Agora, quando voce faz uma requisicao ao gateway, todos os logs dos tres servicos terao o mesmo trace_id. Veja como isso aparece na pratica:

# Requisicao de exemplo
curl -X POST http://localhost:8000/pedidos \
  -H "Content-Type: application/json" \
  -d '{"produto_id": 123, "valor": 99.90}'

Logs gerados (todos com o mesmo trace_id a1b2c3d4):

# API Gateway
[21:00:01] [a1b2c3d4] INFO gateway: Recebendo requisicao para criar pedido
[21:00:01] [a1b2c3d4] INFO gateway: Pedido criado com sucesso

# Servico de Pedidos
[21:00:01] [a1b2c3d4] INFO pedidos: Processando novo pedido
[21:00:01] [a1b2c3d4] INFO pedidos: Pedido x7y8z9w0 aprovado

# Servico de Pagamento
[21:00:01] [a1b2c3d4] INFO pagamento: Processando pagamento
[21:00:01] [a1b2c3d4] INFO pagamento: Iniciando operacao: validar_cartao
[21:00:01] [a1b2c3d4] INFO pagamento: Operacao validar_cartao concluida em 0.301s
[21:00:01] [a1b2c3d4] INFO pagamento: Iniciando operacao: processar_cobranca
[21:00:01] [a1b2c3d4] INFO pagamento: Operacao processar_cobranca concluida em 0.201s
[21:00:01] [a1b2c3d4] INFO pagamento: Pagamento aprovado: transacao m4n5o6p7

Para investigar qualquer problema, basta filtrar pelo trace_id:

grep "a1b2c3d4" /var/log/servicos/*.log

E voce tem a historia completa da requisicao, de ponta a ponta. Sem adivinhacao, sem correlacao manual de timestamps.

Agregacao Automatica com Python

Para facilitar ainda mais a analise, podemos criar um script que agrupa logs por trace_id automaticamente:

import re
from collections import defaultdict

def agrupar_logs_por_trace(log_files):
    traces = defaultdict(list)
    pattern = re.compile(r'\[([a-f0-9]{8})\]')

    for log_file in log_files:
        with open(log_file, 'r') as f:
            for line in f:
                match = pattern.search(line)
                if match:
                    trace_id = match.group(1)
                    traces[trace_id].append(line.strip())

    for trace_id in traces:
        traces[trace_id].sort()

    return dict(traces)

def mostrar_trace(trace_id, traces):
    if trace_id not in traces:
        print(f"Trace {trace_id} nao encontrado")
        return

    print(f"\n{'='*60}")
    print(f"TRACE: {trace_id}")
    print(f"{'='*60}")

    for log_line in traces[trace_id]:
        print(log_line)

    print(f"{'='*60}\n")

# Uso
log_files = [
    '/var/log/gateway.log',
    '/var/log/pedidos.log',
    '/var/log/pagamento.log'
]

traces = agrupar_logs_por_trace(log_files)
mostrar_trace('a1b2c3d4', traces)

Visualizacao ASCII de Spans

Para ter uma visao mais clara da cadeia de chamadas e suas duracoes, podemos gerar uma visualizacao ASCII direto no terminal:

def visualizar_trace_ascii(trace_id, traces):
    logs = traces.get(trace_id, [])

    print(f"\nTrace: {trace_id}\n")

    spans = []
    for log in logs:
        if 'Iniciando operacao:' in log:
            op_name = log.split('Iniciando operacao:')[1].strip()
            spans.append({'name': op_name})
        elif 'concluida em' in log:
            duration = log.split('concluida em')[1].split('s')[0].strip()
            if spans:
                spans[-1]['duration'] = float(duration)

    for span in spans:
        duration = span.get('duration', 0)
        bar_length = int(duration * 50)
        bar = '#' * bar_length + '-' * (50 - bar_length)
        print(f"{span['name']:<25} |{bar}| {duration:.3f}s")

# Saida:
# Trace: a1b2c3d4
#
# validar_cartao          |###############-----------------------------------| 0.301s
# processar_cobranca      |##########----------------------------------------| 0.201s

Isso e incrivelmente util quando voce esta investigando um incidente as 3 da manha e precisa de uma visao rapida do que aconteceu.

Integracao com Logs Estruturados (JSON)

Para ambientes de producao, logs estruturados (JSON) sao muito mais faceis de processar. Vamos adaptar nosso logger:

import json
from datetime import datetime

class StructuredTracedLogger:
    def __init__(self, service_name):
        self.service_name = service_name

    def _log(self, level, msg, **extra_fields):
        trace_id = trace_id_var.get()

        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'level': level,
            'service': self.service_name,
            'trace_id': trace_id,
            'message': msg,
            **extra_fields
        }

        print(json.dumps(log_entry))

    def info(self, msg, **extra):
        self._log('INFO', msg, **extra)

    def error(self, msg, **extra):
        self._log('ERROR', msg, **extra)

    def warning(self, msg, **extra):
        self._log('WARNING', msg, **extra)

# Uso
logger = StructuredTracedLogger('pagamento')
logger.info("Processando pagamento", pedido_id="x7y8z9w0", valor=99.90)

# Saida:
# {"timestamp": "2026-07-28T21:00:01", "level": "INFO",
#  "service": "pagamento", "trace_id": "a1b2c3d4",
#  "message": "Processando pagamento",
#  "pedido_id": "x7y8z9w0", "valor": 99.90}

Com logs estruturados, ferramentas como ELK Stack, Loki ou ate mesmo jq tornam-se extremamente poderosas:

# Filtrar todos os logs de um trace especifico
cat /var/log/servicos/*.log | jq 'select(.trace_id == "a1b2c3d4")'

# Contar erros por servico
cat /var/log/servicos/*.log | \
  jq 'select(.level == "ERROR")' | \
  jq -s 'group_by(.service) | map({service: .[0].service, count: length})'

# Top 10 traces mais lentos
cat /var/log/servicos/*.log | \
  jq -s 'group_by(.trace_id) | map({trace: .[0].trace_id, logs: length}) | sort_by(.logs) | reverse | .[:10]'

Sampling Inteligente: Nem Todo Trace Precisa Ser Armazenado

Em producao com alto volume, armazenar 100% dos traces e caro e desnecessario. Um sampler inteligente mantem os traces importantes e descarta os rotineiros:

import random

class SmartSampler:
    def __init__(self, sample_rate=0.1, always_keep_errors=True):
        self.sample_rate = sample_rate
        self.always_keep_errors = always_keep_errors
        self._has_error = False

    def should_sample(self):
        return random.random() < self.sample_rate

    def mark_error(self):
        self._has_error = True

    def should_keep_trace(self):
        # Sempre mantem traces com erros
        if self.always_keep_errors and self._has_error:
            return True
        # Amostragem aleatoria para o resto
        return self.should_sample()

# Uso no middleware
sampler = SmartSampler(sample_rate=0.1)

@app.errorhandler(500)
def handle_error(e):
    sampler.mark_error()
    return jsonify({"error": str(e)}), 500

A regra e simples: traces com erros sao sempre armazenados. Os outros seguem uma taxa de amostragem configuravel. Voce mantem 100% da visibilidade sobre problemas e reduz drasticamente o custo de armazenamento.

Limitacoes e Quando Usar Ferramentas Especializadas

Essa implementacao minimalista e perfeita para:

  • Projetos pequenos a medios (ate 20 servicos)
  • Prototipagem rapida e validacao de arquitetura
  • Ambientes com restricoes de dependencias
  • Aprendizado dos conceitos fundamentais de distributed tracing

Para sistemas maiores, considere migrar para:

  • OpenTelemetry: Padrao da industria, suporte a multiplas linguagens, exportadores para Jaeger, Zipkin, Grafana Tempo
  • Jaeger: Visualizacao avancada de traces com timeline interativa
  • Zipkin: Alternativa leve ao Jaeger, facil de operar
  • Grafana Tempo: Integracao nativa com Grafana e Loki

A boa noticia? Os conceitos que voce aprendeu aqui (trace_id, span, propagacao de contexto, sampling) sao exatamente os mesmos usados por essas ferramentas. A migracao e muito mais suave quando voce entende os fundamentos.

Conclusao: Simplicidade Que Funciona

Distributed tracing nao precisa ser complicado. Com contextvars, um logger adaptado e headers HTTP, voce tem 80% do valor de solucoes enterprise com 1% da complexidade.

O que construímos aqui:

  • TracedLogger - Logger que inclui trace_id automaticamente
  • TraceMiddleware - Extrai/gera trace_id em cada requisicao
  • TracedHTTPClient - Propaga trace_id em chamadas entre servicos
  • traced_span - Decorador que mede duracao de operacoes
  • SmartSampler - Sampling inteligente que prioriza erros
  • Agregador de logs - Agrupa e visualiza traces completos

Tudo isso sem instalar uma unica dependencia externa. O melhor debug e aquele que voce entende de ponta a ponta.

Se voce curtiu esse post, da uma olhada nos artigos sobre Log Sampler Adaptativo e Log Fingerprint com SimHash que publicamos recentemente. Eles complementam perfeitamente o tracing que construímos hoje.

E voce? Qual foi o bug mais dificil de rastrear em um sistema distribuido? Conta ai nos comentarios.

Qual automacao de observabilidade voce quer ver no proximo post? Deixa sua sugestao e eu transformo em tutorial pratico.

Posts Similares