Cabos de rede azuis conectados em switch de data center representando sistemas distribuídos

Engine de Correlacao de Erros Distribuidos em Python: Quando o Bug Esta na Orquestracao, Nao no Violino Solo

PorOlivetto

Correlacao de erros distribuidos e a habilidade que separa engenheiros que acham bugs de engenheiros que entendem sistemas. E eu vou te contar por que.

Eu ja passei 4 horas debugando um servico que nao estava quebrado. O culpado? Um servico completamente diferente, tres containers acima, que estava devolvendo timeouts silenciosos. O primeiro servico apenas reagia ao timeout. Mas seus logs gritavam ConnectionError! enquanto o verdadeiro assassino dormia tranquilo, sem um ERROR sequer no nome.

Se voce ja viveu isso, sabe o que e. O log do microservico A aponta para o B. O B aponta para o C. O C, aquele canalha, nao loga nada porque “e so um cache, nao precisa de log”. Spoiler: precisa. Sempre precisa.

O Problema: Logs que Mentem por Omissao

O maior erro da observabilidade moderna e achar que mais logs = mais visibilidade. Mentira. Mais logs = mais ruido. E ruido distribuido entre 12 servicos e pior que silencio total, porque te da a ilusao de que voce sabe o que esta acontecendo.

O que falta nao e volume. E correlacao. A capacidade de olhar para um erro no servico de pagamentos e saber, sem chutar, que ele foi causado por uma race condition no servico de estoque 47 milissegundos antes.

Hoje vamos construir uma engine de correlacao de erros distribuidos em Python. Nada de ELK Stack caro. Nada de Splunk. Nada de SaaS que cobra por GB de log. So Python, SQLite e um pouco de arrogancia intelectual.

A Arquitetura: Tres Camadas de Inteligencia

A engine tem tres componentes que trabalham em conjunto:

  1. Coletor com Contexto: cada log carrega um trace_id e metadados de servico.
  2. Correlacionador Temporal: encontra erros que aconteceram proximos no tempo, com peso para mesmos trace_id.
  3. Agrupador Semantico: clusteriza erros similares usando fingerprinting de mensagem e stack trace.
Codigo Python em tela mostrando sistema de correlacao de erros distribuidos
Arquitetura de tres camadas para correlacao inteligente de falhas.

1. O Coletor: Ensinando Seus Logs a se Apresentarem

Primeira regra: todo log de erro precisa de contexto. Nao basta “Connection refused”. Quem tentou conectar? Para onde? Em qual trace?

import json
import uuid
import logging
from datetime import datetime, timezone

class ContextualLogger:
    def __init__(self, service_name):
        self.service_name = service_name
        self.logger = logging.getLogger(service_name)

    def error(self, message, exception=None, target_service=None, **kwargs):
        trace_id = kwargs.get('trace_id', str(uuid.uuid4()))
        entry = {
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'service': self.service_name,
            'level': 'ERROR',
            'trace_id': trace_id,
            'message': str(message),
            'exception_type': type(exception).__name__ if exception else None,
            'target_service': target_service,
            'metadata': kwargs
        }
        self.logger.error(json.dumps(entry, ensure_ascii=False))
        return trace_id

Notem: o metodo retorna o trace_id. Isso e crucial. O servico chamador pega esse ID e propaga para o proximo. Se o pagamento chama o estoque, o trace_id viaja junto. Quando tudo explode, voce tem um fio de Ariadne.

2. O Correlacionador Temporal: Encontrando o Culpado pelo Horario

O principio e simples: erros que acontecem proximos no tempo, especialmente com o mesmo trace_id, provavelmente estao relacionados. Mas “proximo” e relativo. Em microservicos, 500ms e uma eternidade. Em pipelines de dados, 5 minutos e instantaneo.

import sqlite3
from datetime import datetime, timedelta

class ErrorCorrelationEngine:
    def __init__(self, db_path='errors.db'):
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        self.conn.execute("""CREATE TABLE IF NOT EXISTS errors (
    id INTEGER PRIMARY KEY,
    timestamp TEXT,
    trace_id TEXT,
    service TEXT,
    message TEXT,
    exception_type TEXT,
    fingerprint TEXT
)""")
        self.conn.execute("""CREATE INDEX IF NOT EXISTS idx_trace_time
ON errors(trace_id, timestamp)""")
        self.conn.commit()

    def ingest(self, error_entry):
        fp = f"{error_entry.get('exception_type','Unknown')}:"
        fp += error_entry['message'][:60]
        self.conn.execute("""INSERT INTO errors (timestamp, trace_id, service, message, exception_type, fingerprint)
VALUES (?, ?, ?, ?, ?, ?)""", (
            error_entry['timestamp'],
            error_entry.get('trace_id'),
            error_entry['service'],
            error_entry['message'],
            error_entry.get('exception_type'),
            fp
        ))
        self.conn.commit()

    def find_correlated(self, error_id, time_window_seconds=5):
        row = self.conn.execute('SELECT * FROM errors WHERE id = ?', (error_id,)).fetchone()
        if not row:
            return []
        base_time = datetime.fromisoformat(row[1])
        window = timedelta(seconds=time_window_seconds)
        cursor = self.conn.execute("""SELECT * FROM errors
WHERE timestamp BETWEEN ? AND ?
AND id != ?
ORDER BY timestamp""", (
            (base_time - window).isoformat(),
            (base_time + window).isoformat(),
            error_id
        ))
        results = []
        for r in cursor.fetchall():
            score = 0
            if r[2] and r[2] == row[2]:
                score += 100
            if r[6] == row[6]:
                score += 50
            if r[3] == row[3]:
                score -= 20
            results.append({
                'id': r[0], 'service': r[3], 'message': r[4],
                'score': score, 'same_trace': r[2] == row[2]
            })
        return sorted(results, key=lambda x: x['score'], reverse=True)

O coracao esta no scoring. Mesmo trace_id vale 100 pontos – e a prova de fogo. Mesmo fingerprint vale 50 – indica que o mesmo tipo de erro esta se propagando. Erro no mesmo servico perde pontos, porque provavelmente e retry, nao causa raiz.

3. O Agrupador Semantico: Nao Precisamos Ver o Mesmo Erro 500 Vezes

Se seu servico de estoque caiu, voce nao quer 500 entradas identicas. Quer saber: “o servico de estoque esta com ConnectionRefused, afetando 847 requisicoes, comecou as 14:32”.

from collections import defaultdict

class ErrorGrouper:
    def __init__(self, engine):
        self.engine = engine

    def group_recent(self, minutes=10):
        since = (datetime.now(timezone.utc) - timedelta(minutes=minutes)).isoformat()
        rows = self.engine.conn.execute("""SELECT fingerprint, service, message, COUNT(*) as cnt,
MIN(timestamp) as first, MAX(timestamp) as last
FROM errors
WHERE timestamp > ?
GROUP BY fingerprint, service
ORDER BY cnt DESC""", (since,)).fetchall()
        groups = []
        for fp, svc, msg, cnt, first, last in rows:
            groups.append({
                'fingerprint': fp,
                'service': svc,
                'sample_message': msg,
                'count': cnt,
                'first_seen': first,
                'last_seen': last,
                'affected_traces': self._count_traces(fp, since)
            })
        return groups

    def _count_traces(self, fingerprint, since):
        row = self.engine.conn.execute("""SELECT COUNT(DISTINCT trace_id) FROM errors
WHERE fingerprint = ? AND timestamp > ? AND trace_id IS NOT NULL""", (fingerprint, since)).fetchone()
        return row[0] if row else 0
Servidores e codigo em tela representando monitoramento de erros distribuidos
Agrupamento inteligente transforma milhares de linhas de log em sinais claros.

O Perrengue: Quando a Correlacao Encontra o Caos Real

Colocando Tudo Junto: O Pipeline Completo

Aqui esta o script que roda a cada minuto via cron, lendo logs de stdin:

#!/usr/bin/env python3
import json
import sys
from error_correlation_engine import ErrorCorrelationEngine, ErrorGrouper

def main():
    engine = ErrorCorrelationEngine('/var/log/correlator/errors.db')
    grouper = ErrorGrouper(engine)
    for line in sys.stdin:
        try:
            entry = json.loads(line)
            if entry.get('level') == 'ERROR':
                engine.ingest(entry)
        except json.JSONDecodeError:
            continue
    groups = grouper.group_recent(minutes=5)
    for g in groups:
        if g['count'] > 10:
            print(f"🚨 {g['service']}: {g['fingerprint']}")
            print(f"   Afetou {g['count']} requisicoes em {g['affected_traces']} traces")
            print(f"   Primeiro: {g['first_seen']} | Ultimo: {g['last_seen']}")

if __name__ == '__main__':
    main()

Links Internos para Seu Arsenal

Se voce curtiu isso, provavelmente tambem vai querer:

Conclusao: A Orquestracao Precisa de um Maestro

Debugar sistemas distribuidos sem correlacao e como tentar montar um quebra-cabeca com pecas de caixas diferentes. Cada servico mostra uma parte da verdade, mas nenhum mostra o quadro completo.

A engine que construimos aqui nao e perfeita. Nao tem machine learning. Nao tem dashboards bonitos. Mas funciona. E funciona sem depender de um SaaS que vai dobrar de preco no ano que vem.

O ponto nao e ter a ferramenta mais sofisticada. E ter alguma ferramenta que conecte os pontos antes de voce gastar 4 horas cacando um fantasma no servico errado.

Qual automacao voce quer ver aqui?

Fala pra gente: voce tem um cenario de debug distribuido que nao se encaixa nisso? Um tipo de correlacao que sente falta? Deixa nos comentarios. O Lab da Garra esta sempre aberto.

– Olivetto, cacando bugs em multiplas dimensoes temporais.

Posts Similares