Error Fingerprinting com SimHash em Python Puro: O Agrupador Que Identifica Bugs Duplicados Entre Milhões de Stack Traces (Sem Elasticsearch)
3 da manhã. O PagerDuty toca. Você abre o log e vê 47.000 erros — todos parecidos, mas nenhum idêntico. O estagiário já tentou resolver ontem, marcou como “intermitente”, e agora o problema voltou com juros. Se você já passou por isso, sabe: o problema não é o bug. É que ninguém percebeu que eram 47.000 instâncias do mesmo bug.
Hoje você vai construir um sistema de error fingerprinting com SimHash em Python puro que calcula a assinatura de cada stack trace, agrupa erros semanticamente similares e te mostra, em tempo real, quais bugs são novos e quais são velhos conhecidos disfarçados. Sem Elasticsearch. Sem Datadog. Sem dependências externas. Apenas Python, matemática e um pouco de teimosia.
O Problema Que Ninguém Resolve Com grep
Vamos ser honestos: quando o log explode, a primeira reação é grep ERROR application.log | wc -l. Funciona pra contar. Não funciona pra entender.
O motivo é simples: dois stack traces do mesmo bug quase nunca são idênticos. Um tem timestamp diferente. Outro tem um ID de request no meio. Um terceiro pegou um caminho ligeiramente diferente no código e tem duas frames extras. O grep vê três erros diferentes. Seu cérebro, se tivesse tempo, veria um só.
Aqui entra o SimHash. É um algoritmo de fingerprinting que transforma textos (no nosso caso, stack traces) em hashes de 64 bits onde textos similares produzem hashes similares. Não é magia — é álgebra linear aplicada a strings. A distância de Hamming entre dois hashes SimHash te diz, numericamente, quão parecidos dois erros são.
SimHash em 30 Linhas: O Coração do Sistema
Antes de construir o agrupador, precisamos do SimHash funcionando. A implementação é surpreendentemente compacta:
import hashlib
import re
from collections import Counter
def tokenize_stack_trace(trace: str) -> list[str]:
"""Extrai tokens relevantes do stack trace, ignorando ruído."""
# Remove timestamps, UUIDs, IDs numéricos e paths absolutos
cleaned = re.sub(r'\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}[.\d]*[Z]?', '<TIMESTAMP>', trace)
cleaned = re.sub(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', '<UUID>', cleaned)
cleaned = re.sub(r'/[\w./\-]+\.py', '<FILEPATH>', cleaned)
cleaned = re.sub(r'\b\d{4,}\b', '<NUM>', cleaned)
# Tokeniza em palavras e símbolos significativos
return re.findall(r'[a-zA-Z_]\w*|[{}()\[\].,;:@#$%&*+\-=/]', cleaned)
def simhash(tokens: list[str], hashbits: int = 64) -> int:
"""Calcula o SimHash de uma lista de tokens."""
vector = [0] * hashbits
token_counts = Counter(tokens)
for token, count in token_counts.items():
# Hash do token usando MD5 (rápido e suficiente pra fingerprinting)
token_hash = int(hashlib.md5(token.encode()).hexdigest(), 16)
# Usa apenas os primeiros `hashbits` bits do hash
token_hash = token_hash & ((1 << hashbits) - 1)
weight = count # Frequência como peso
for i in range(hashbits):
if token_hash & (1 << i):
vector[i] += weight
else:
vector[i] -= weight
# Converte o vetor em fingerprint binário
fingerprint = 0
for i in range(hashbits):
if vector[i] > 0:
fingerprint |= (1 << i)
return fingerprint
def hamming_distance(hash1: int, hash2: int, hashbits: int = 64) -> int:
"""Calcula a distância de Hamming entre dois hashes."""
xor = hash1 ^ hash2
return bin(xor).count('1')
def are_similar(hash1: int, hash2: int, threshold: int = 10) -> bool:
"""Dois hashes são similares se a distância de Hamming é menor que o threshold."""
return hamming_distance(hash1, hash2) <= thresholdEsse código faz três coisas fundamentais:
- tokenize_stack_trace normaliza o ruído — timestamps viram
<TIMESTAMP>, UUIDs viram<UUID>, paths viram<FILEPATH>. Isso garante que dois erros com timestamps diferentes mas mesma causa raiz produzam tokens quase idênticos. - simhash constrói um vetor de 64 dimensões onde cada dimensão acumula o peso dos tokens que têm bit 1 naquela posição, e subtrai o peso dos que têm bit 0. O fingerprint final é o sinal de cada dimensão.
- hamming_distance conta quantos bits diferem entre dois fingerprints. Stack traces do mesmo bug tipicamente têm distância 2-8. Bugs diferentes têm distância 20+.
Por Que SimHash e Não MinHash ou TLS?
Boa pergunta. Existem pelo menos três algoritmos populares de locality-sensitive hashing:
- MinHash: excelente para similaridade de conjuntos (Jaccard), mas requer múltiplas funções hash e é mais lento pra comparações em tempo real.
- SimHash: um único hash por documento, comparação O(1) via XOR + popcount. Ideal para streaming.
- TLS (Threshold LSH): mais preciso, mas muito mais complexo de implementar sem bibliotecas.
Para fingerprinting de erros, o SimHash vence porque:
- O hash tem tamanho fixo (64 bits = 8 bytes). Cabe em qualquer lugar.
- A comparação é uma operação XOR seguida de contagem de bits. Python faz isso em nanosegundos.
- O threshold de similaridade (distância de Hamming ≤ 10) é intuitivo e ajustável.
- Não precisa de treinamento, modelo, ou configuração complexa.
O Agrupador: Clusterizando Erros em Tempo Real
Agora a parte interessante. Vamos construir um processador que lê stack traces de um arquivo de log, calcula o fingerprint de cada um, e agrupa os similares automaticamente:
import json
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class ErrorCluster:
"""Representa um grupo de erros similares."""
fingerprint: int
first_seen: datetime
last_seen: datetime
count: int = 1
sample_trace: str = ""
sample_file: str = ""
sample_line: int = 0
error_type: str = ""
@dataclass
class ErrorFingerprinter:
"""Motor de fingerprinting e agrupamento de erros."""
clusters: dict[int, ErrorCluster] = field(default_factory=dict)
threshold: int = 10 # Distância de Hamming máxima para considerar similar
total_processed: int = 0
def process_trace(self, trace: str, source_file: str = "", timestamp: datetime = None) -> dict:
"""Processa um stack trace e retorna info do cluster."""
if timestamp is None:
timestamp = datetime.now()
tokens = tokenize_stack_trace(trace)
fingerprint = simhash(tokens)
self.total_processed += 1
# Busca cluster existente mais próximo
best_match = None
best_distance = float('inf')
for existing_fp, cluster in self.clusters.items():
dist = hamming_distance(fingerprint, existing_fp)
if dist < best_distance:
best_distance = dist
best_match = existing_fp
# Se encontrou cluster similar, atualiza
if best_match is not None and best_distance <= self.threshold:
cluster = self.clusters[best_match]
cluster.count += 1
cluster.last_seen = timestamp
return {
"status": "matched",
"cluster_fingerprint": best_match,
"distance": best_distance,
"cluster_count": cluster.count,
"first_seen": cluster.first_seen.isoformat(),
"error_type": cluster.error_type
}
# Novo cluster
error_type = self._extract_error_type(trace)
new_cluster = ErrorCluster(
fingerprint=fingerprint,
first_seen=timestamp,
last_seen=timestamp,
sample_trace=trace[:2000], # Armazena amostra limitada
sample_file=source_file,
error_type=error_type
)
self.clusters[fingerprint] = new_cluster
return {
"status": "new_cluster",
"cluster_fingerprint": fingerprint,
"error_type": error_type,
"total_clusters": len(self.clusters)
}
def _extract_error_type(self, trace: str) -> str:
"""Extrai o tipo de erro da primeira linha do stack trace."""
first_line = trace.strip().split('\n')[0]
# Tenta capturar o nome da exceção
match = re.search(r'([A-Z]\w*(?:Error|Exception|Fault|Failure))', first_line)
if match:
return match.group(1)
# Fallback: primeiros 80 chars
return first_line[:80].strip()
def get_report(self) -> dict:
"""Gera relatório completo de todos os clusters."""
sorted_clusters = sorted(
self.clusters.values(),
key=lambda c: c.count,
reverse=True
)
return {
"total_errors_processed": self.total_processed,
"unique_bugs_found": len(self.clusters),
"deduplication_ratio": round(
(1 - len(self.clusters) / max(self.total_processed, 1)) * 100, 2
),
"clusters": [
{
"fingerprint": hex(c.fingerprint),
"error_type": c.error_type,
"count": c.count,
"first_seen": c.first_seen.isoformat(),
"last_seen": c.last_seen.isoformat(),
"sample_preview": c.sample_trace[:200] + "..."
}
for c in sorted_clusters
]
}O fluxo é direto: para cada stack trace, calcula o SimHash, compara com todos os clusters existentes, e ou incrementa o cluster mais próximo (se a distância for ≤ 10) ou cria um novo. Simples. Efficaz. Sem Kafka, sem Redis, sem microsserviço dedicado.
Leitor de Log: Integração Com o Mundo Real
Agora precisamos alimentar o fingerprinter com dados reais. Stack traces em logs geralmente aparecem em blocos multi-linha. Aqui vai um parser que lida com os formatos mais comuns (Python traceback, Java stack trace, Node.js error):
import re
from pathlib import Path
def extract_stack_traces(log_path: str) -> list[dict]:
"""Extrai stack traces individuais de um arquivo de log."""
content = Path(log_path).read_text(encoding='utf-8', errors='replace')
traces = []
# Padrão Python traceback
python_pattern = re.compile(
r'(Traceback \(most recent call last\):.*?)(?=\n\d{4}-|\nTraceback|\Z)',
re.DOTALL
)
for match in python_pattern.finditer(content):
traces.append({
"trace": match.group(1).strip(),
"language": "python",
"source": log_path
})
# Padrão Java/Node (Exception at ...)
java_pattern = re.compile(
r'((?:[A-Z]\w*(?:Error|Exception)).*?)(?=\n\d{4}-|\n[A-Z]\w*Error|\Z)',
re.DOTALL
)
for match in java_pattern.finditer(content):
trace_text = match.group(1).strip()
if len(trace_text.split('\n')) >= 3: # Pelo menos 3 linhas = stack trace real
lang = "java" if ".java:" in trace_text else "nodejs"
traces.append({
"trace": trace_text,
"language": lang,
"source": log_path
})
return traces
def process_log_file(log_path: str, threshold: int = 10) -> dict:
"""Pipeline completo: extrai traces, faz fingerprinting, gera relatório."""
fingerprinter = ErrorFingerprinter(threshold=threshold)
traces = extract_stack_traces(log_path)
print(f"📂 Lendo: {log_path}")
print(f"🔍 Encontrados {len(traces)} stack traces")
for i, trace_info in enumerate(traces, 1):
result = fingerprinter.process_trace(
trace=trace_info["trace"],
source_file=trace_info["source"]
)
if result["status"] == "new_cluster":
print(f" 🆕 [{i}/{len(traces)}] Novo bug: {result['error_type']}")
elif result["status"] == "matched" and result["cluster_count"] % 100 == 0:
print(f" 🔁 [{i}/{len(traces)}] Cluster '{result['error_type']}' agora tem {result['cluster_count']} ocorrências")
report = fingerprinter.get_report()
print(f"\n{'='*60}")
print(f"📊 RELATÓRIO FINAL")
print(f"{'='*60}")
print(f"Total de erros processados: {report['total_errors_processed']}")
print(f"Bugs únicos identificados: {report['unique_bugs_found']}")
print(f"Taxa de deduplicação: {report['deduplication_ratio']}%")
print(f"{'='*60}")
for i, cluster in enumerate(report['clusters'][:10], 1):
print(f"\n🐛 Bug #{i}: {cluster['error_type']}")
print(f" Ocorrências: {cluster['count']}")
print(f" Primeiro: {cluster['first_seen']}")
print(f" Último: {cluster['last_seen']}")
print(f" Fingerprint: {cluster['fingerprint']}")
return reportCom isso, você tem um pipeline completo. Joga um arquivo de log, ele extrai os stack traces, calcula fingerprints, agrupa, e te entrega um relatório ordenado por frequência. Os bugs que mais acontecem aparecem primeiro. Coincidência? Nunca.
Modo Streaming: Processando Logs em Tempo Real
Arquivos são ótimos para análise forense. Mas em produção, você quer streaming. Aqui vai um watcher que monitora um log em tempo real e processa cada novo stack trace assim que aparece:
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class LogWatcher(FileSystemEventHandler):
"""Monitora mudanças em arquivos de log e processa novos stack traces."""
def __init__(self, log_path: str, threshold: int = 10):
self.log_path = log_path
self.fingerprinter = ErrorFingerprinter(threshold=threshold)
self.last_position = 0
self.buffer = ""
def on_modified(self, event):
if event.src_path != self.log_path:
return
with open(self.log_path, 'r', encoding='utf-8', errors='replace') as f:
f.seek(self.last_position)
new_content = f.read()
self.last_position = f.tell()
if not new_content:
return
self.buffer += new_content
# Tenta extrair stack traces completos do buffer
while True:
# Procura por início de traceback
match = re.search(r'(Traceback \(most recent call last\):.*?)(?=\n\d{4}-|\nTraceback|\Z)',
self.buffer, re.DOTALL)
if not match:
break
trace = match.group(1).strip()
self.buffer = self.buffer[match.end():]
result = self.fingerprinter.process_trace(trace, source_file=self.log_path)
if result["status"] == "new_cluster":
print(f"🚨 NOVO BUG DETECTADO: {result['error_type']}")
print(f" Total de clusters: {result['total_clusters']}")
else:
if result["cluster_count"] == 2:
print(f"⚠️ BUG RECORRENTE: {result['error_type']} (2ª ocorrência!)")
elif result["cluster_count"] % 50 == 0:
print(f"🔥 BUG EM SURTO: {result['error_type']} ({result['cluster_count']} ocorrências)")
def start(self):
"""Inicia o monitoramento."""
print(f"👁️ Monitorando: {self.log_path}")
print(f" Threshold de similaridade: {self.fingerprinter.threshold}")
print(f" Pressione Ctrl+C para parar.\n")
# Processa conteúdo existente
with open(self.log_path, 'r') as f:
self.last_position = f.seek(0, 2) # Vai pro final
observer = Observer()
observer.schedule(self, path=str(Path(self.log_path).parent), recursive=False)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
print(f"\n📊 Relatório final:")
report = self.fingerprinter.get_report()
print(f" Erros processados: {report['total_errors_processed']}")
print(f" Bugs únicos: {report['unique_bugs_found']}")
observer.join()Sim, precisa do watchdog (pip install watchdog). É a única dependência externa, e é opcional — se você preferir, pode usar polling com os.stat() e time.sleep(). O ponto é: com menos de 100 linhas de código, você tem um sistema de detecção de surtos de erros rodando na sua máquina.
Otimização: Bucketing Com Prefixo de Hash
Quando você passa de 10.000 clusters, comparar cada novo fingerprint contra todos os existentes fica lento — O(n) por inserção. A solução clássica é bucketing por prefixo:
class OptimizedFingerprinter(ErrorFingerprinter):
"""Versão otimizada com bucketing por prefixo de hash."""
def __init__(self, threshold: int = 10, prefix_bits: int = 8):
super().__init__(threshold=threshold)
self.prefix_bits = prefix_bits
self.buckets: dict[int, list[int]] = {} # prefix -> list of fingerprints
def _get_prefix(self, fingerprint: int) -> int:
"""Extrai os primeiros N bits como prefixo."""
return fingerprint >> (64 - self.prefix_bits)
def process_trace(self, trace: str, source_file: str = "", timestamp=None) -> dict:
if timestamp is None:
timestamp = datetime.now()
tokens = tokenize_stack_trace(trace)
fingerprint = simhash(tokens)
self.total_processed += 1
prefix = self._get_prefix(fingerprint)
# Busca apenas no bucket do prefixo e buckets adjacentes
candidates = set()
for p in range(max(0, prefix - 1), min(256, prefix + 2)):
if p in self.buckets:
candidates.update(self.buckets[p])
best_match = None
best_distance = float('inf')
for existing_fp in candidates:
dist = hamming_distance(fingerprint, existing_fp)
if dist < best_distance:
best_distance = dist
best_match = existing_fp
if best_match is not None and best_distance <= self.threshold:
cluster = self.clusters[best_match]
cluster.count += 1
cluster.last_seen = timestamp
return {
"status": "matched",
"cluster_fingerprint": best_match,
"distance": best_distance,
"cluster_count": cluster.count,
"first_seen": cluster.first_seen.isoformat(),
"error_type": cluster.error_type
}
# Novo cluster
error_type = self._extract_error_type(trace)
new_cluster = ErrorCluster(
fingerprint=fingerprint,
first_seen=timestamp,
last_seen=timestamp,
sample_trace=trace[:2000],
sample_file=source_file,
error_type=error_type
)
self.clusters[fingerprint] = new_cluster
if prefix not in self.buckets:
self.buckets[prefix] = []
self.buckets[prefix].append(fingerprint)
return {
"status": "new_cluster",
"cluster_fingerprint": fingerprint,
"error_type": error_type,
"total_clusters": len(self.clusters)
}A ideia é simples: se dois fingerprints são similares (distância de Hamming ≤ 10), eles muito provavelmente compartilham os primeiros 8 bits. Então, ao invés de comparar contra todos os clusters, comparamos apenas contra os que estão no mesmo “balde” (ou baldes adjacentes). Reduz de O(n) para O(1) amortizado.
Persistência: Salvando e Carregando Clusters
Se o script reiniciar, você perde todos os clusters. Vamos resolver isso com SQLite — sem ORM, sem complicação:
import sqlite3
class PersistentFingerprinter(OptimizedFingerprinter):
"""Fingerprinter com persistência em SQLite."""
def __init__(self, db_path: str = "error_fingerprints.db", threshold: int = 10):
super().__init__(threshold=threshold)
self.db_path = db_path
self._init_db()
self._load_clusters()
def _init_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS error_clusters (
fingerprint INTEGER PRIMARY KEY,
error_type TEXT NOT NULL,
first_seen TEXT NOT NULL,
last_seen TEXT NOT NULL,
count INTEGER DEFAULT 1,
sample_trace TEXT,
sample_file TEXT
)
""")
def _load_clusters(self):
with sqlite3.connect(self.db_path) as conn:
for row in conn.execute("SELECT * FROM error_clusters"):
fp, error_type, first_seen, last_seen, count, sample, src = row
cluster = ErrorCluster(
fingerprint=fp,
first_seen=datetime.fromisoformat(first_seen),
last_seen=datetime.fromisoformat(last_seen),
count=count,
sample_trace=sample or "",
sample_file=src or "",
error_type=error_type
)
self.clusters[fp] = cluster
prefix = self._get_prefix(fp)
if prefix not in self.buckets:
self.buckets[prefix] = []
self.buckets[prefix].append(fp)
self.total_processed += count
print(f"📦 Carregados {len(self.clusters)} clusters existentes")
def _save_cluster(self, cluster: ErrorCluster):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT OR REPLACE INTO error_clusters
(fingerprint, error_type, first_seen, last_seen, count, sample_trace, sample_file)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
cluster.fingerprint, cluster.error_type,
cluster.first_seen.isoformat(), cluster.last_seen.isoformat(),
cluster.count, cluster.sample_trace, cluster.sample_file
))Agora seus clusters sobrevivem a restarts. Pode rodar o fingerprinter como um daemon, como um cron job, ou como parte do seu pipeline de CI/CD. Os dados ficam.
Dashboard Textual: Relatórios Que Qualquer Um Entende
Dados sem visualização são dados esquecidos. Aqui vai um gerador de relatório que qualquer pessoa do time consegue ler — sem precisar abrir Grafana:
def generate_text_report(fingerprinter: ErrorFingerprinter) -> str:
"""Gera relatório formatado em texto puro."""
report = fingerprinter.get_report()
lines = []
lines.append("╔" + "═"*58 + "╗")
lines.append("║" + " RELATÓRIO DE ERROR FINGERPRINTING".center(58) + "║")
lines.append("╠" + "═"*58 + "╣")
lines.append(f"║ Erros totais processados: {report['total_errors_processed']:<30}║")
lines.append(f"║ Bugs únicos identificados: {report['unique_bugs_found']:<29}║")
lines.append(f"║ Taxa de deduplicação: {report['deduplication_ratio']}%{'':<23}║")
lines.append("╠" + "═"*58 + "╣")
lines.append("║" + " TOP BUGS (por frequência)".center(58) + "║")
lines.append("╠" + "═"*58 + "╣")
for i, cluster in enumerate(report['clusters'][:15], 1):
count_bar = "█" * min(cluster['count'] // 10, 20) + "░" * max(0, 20 - cluster['count'] // 10)
lines.append(f"║ #{i:<3} {cluster['error_type'][:30]:<30} {count_bar} ║")
lines.append(f"║ {cluster['count']} ocorrências | {cluster['first_seen'][:10]} → {cluster['last_seen'][:10]}{'':<10}║")
lines.append("╚" + "═"*58 + "╝")
return "\n".join(lines)
def export_to_json(fingerprinter: ErrorFingerprinter, output_path: str):
"""Exporta relatório completo em JSON."""
report = fingerprinter.get_report()
with open(output_path, 'w') as f:
json.dump(report, f, indent=2, ensure_ascii=False)
print(f"📄 Relatório exportado: {output_path}")O dashboard textual é intencional. Ele funciona em SSH, em logs de CI, em emails, em mensagens do Slack. Não precisa de servidor web, não precisa de frontend. É informação pura, acessível de qualquer terminal.
Integração Com Alertas: Notificações Automáticas
O sistema é bom, mas se ele só fala quando você olha, você vai esquecer de olhar. Vamos adicionar alertas automáticos:
import smtplib
from email.mime.text import MIMEText
class AlertingFingerprinter(PersistentFingerprinter):
"""Fingerprinter com alertas automáticos."""
def __init__(self, *args, alert_config: dict = None, **kwargs):
super().__init__(*args, **kwargs)
self.alert_config = alert_config or {}
self.alert_threshold = self.alert_config.get('new_bug_threshold', 1)
self.surge_threshold = self.alert_config.get('surge_threshold', 50)
self.surge_window_minutes = self.alert_config.get('surge_window', 5)
def process_trace(self, trace: str, source_file: str = "", timestamp=None) -> dict:
result = super().process_trace(trace, source_file, timestamp)
if result["status"] == "new_cluster":
self._send_alert(
"🚨 NOVO BUG DETECTADO",
f"Tipo: {result['error_type']}\n"
f"Arquivo: {source_file}\n"
f"Fingerprint: {hex(result['cluster_fingerprint'])}\n"
f"Total de clusters ativos: {result['total_clusters']}"
)
elif result["status"] == "matched":
cluster = self.clusters[result["cluster_fingerprint"]]
# Alerta de surto: muitas ocorrências em pouco tempo
elapsed = (cluster.last_seen - cluster.first_seen).total_seconds() / 60
if cluster.count >= self.surge_threshold and elapsed <= self.surge_window_minutes:
self._send_alert(
"🔥 SURTO DE BUG DETECTADO",
f"Tipo: {result['error_type']}\n"
f"Ocorrências: {cluster.count} em {elapsed:.1f} minutos\n"
f"Taxa: {cluster.count / max(elapsed, 0.1):.1f} erros/min"
)
return result
def _send_alert(self, subject: str, body: str):
"""Envia alerta por email (configurável para webhook/Slack/Telegram)."""
config = self.alert_config
if not config.get('smtp_host'):
print(f"[ALERT] {subject}\n{body}\n")
return
msg = MIMEText(body)
msg['Subject'] = f"[AutoMente] {subject}"
msg['From'] = config['smtp_from']
msg['To'] = config['smtp_to']
try:
with smtplib.SMTP(config['smtp_host'], config.get('smtp_port', 587)) as server:
server.starttls()
server.login(config['smtp_user'], config['smtp_pass'])
server.send_message(msg)
except Exception as e:
print(f"⚠️ Falha ao enviar alerta: {e}")O princípio é: bugs novos geram alerta imediato. Bugs em surto (muitas ocorrências em janela curta) geram alerta de urgência. Você pode trocar o SMTP por webhook do Slack, bot do Telegram, ou chamada HTTP pra qualquer sistema de incidentes. O mecanismo de alerta é plugável.
Calibrando o Threshold: Quando 10 Não É 10
O threshold de distância de Hamming ≤ 10 funciona bem para a maioria dos casos. Mas “a maioria” não é “todos”. Aqui vão algumas diretrizes de calibração:
- Threshold 6-8: Stack traces pequenos (<10 frames). Menos tolerante — só agrupa erros muito parecidos. Bom para projetos com poucos arquivos.
- Threshold 10-12: Stack traces médios (10-30 frames). O sweet spot para a maioria das aplicações Python, Java e Node.
- Threshold 14-16: Stack traces grandes (30+ frames) com muita variação superficial. Bom para frameworks que geram traces verbosos (Django, Spring).
Uma técnica útil é calcular a matriz de distância entre todos os clusters e verificar se há clusters que deveriam estar fundidos:
def find_merge_candidates(fingerprinter: ErrorFingerprinter, merge_threshold: int = 15) -> list[dict]:
"""Encontra clusters que poderiam ser fundidos."""
fingerprints = list(fingerprinter.clusters.keys())
candidates = []
for i in range(len(fingerprints)):
for j in range(i + 1, len(fingerprints)):
dist = hamming_distance(fingerprints[i], fingerprints[j])
if fingerprinter.threshold < dist <= merge_threshold:
c1 = fingerprinter.clusters[fingerprints[i]]
c2 = fingerprinter.clusters[fingerprints[j]]
candidates.append({
"cluster_1": c1.error_type,
"cluster_2": c2.error_type,
"distance": dist,
"combined_count": c1.count + c2.count
})
return sorted(candidates, key=lambda x: x['combined_count'], reverse=True)Se essa função retorna muitos pares, seu threshold está baixo demais. Se não retorna nenhum e você sabe que tem bugs duplicados, está alto demais. É um processo iterativo, não uma ciência exata.
Comparação Honesta: SimHash vs Elasticsearch
Vamos falar de elefantes na sala. Se você já usa Elasticsearch com Kibana, por que trocar (ou complementar) com SimHash?
- Elasticsearch é melhor para busca livre, dashboards interativos, e queries complexas. Ele não faz fingerprinting nativo de stack traces — você precisa de plugins ou scripts customizados.
- SimHash local é melhor para detecção de duplicatas em tempo real, alertas automáticos, e ambientes sem infraestrutura pesada. Roda em qualquer máquina com Python.
A resposta honesta: não é pra trocar. É pra complementar. Use o SimHash como primeira linha de defesa — detecta surtos e bugs novos antes mesmo de chegarem ao Elasticsearch. Quando você abrir o Kibana, já sabe quais são os 3-5 bugs que importam, ao invés de se perder em 47.000 entradas.
Onde Ir Daqui
O que construímos aqui é funcional e completo. Mas tem espaço pra crescer:
- API REST: envolva o fingerprinter com FastAPI e exponha endpoints para consulta de clusters via HTTP.
- Pesos por token: dê mais peso aos tokens do topo do stack trace (geralmente mais relevantes que frames de framework).
- Integração com Git: quando um novo cluster aparece, automaticamente faça
git lognos arquivos mencionados pra ver quais commits recentes podem ter causado. - Machine learning: use os clusters como features para treinar um classificador que prevê a severidade de novos bugs.
E Agora?
Você tem em mãos um sistema de fingerprinting de erros que roda em qualquer máquina, não depende de serviço externo, e te diz em segundos o que sua equipe levaria dias pra descobrir manualmente. Copie o código. Rode nos seus logs. Veja quantos bugs “diferentes” são, na verdade, o mesmo problema com roupas diferentes.
E me conta nos comentários: qual automação de debug você quer ver desmontada aqui no AutoMente? Tem algum perrengue com logs que ninguém resolve? Manda que a gente transforma em post.
