Stack Trace Analyzer em Python: O Parser Que Reconstrói o Caminho Completo do Erro em Sistemas Multi-Thread (Sem Framework)
Você já passou pela experiência de um bug que só aparece em produção, às 3 da manhã, e quando você vai olhar o log, encontra uma stack trace que parece ter sido escrita por alguém com Parkinson? Linhas se entrelaçam, threads aparecem do nada, e o traceback final não aponta para lugar nenhum que faça sentido?
Bem-vindo ao clube dos que sofrem com stack traces encadeadas em sistemas multi-thread. Eu passei 3 semanas caçando um deadlock intermitente num sistema de processamento de filas. O bug aparecia 1 vez a cada 500 execuções. Quando finalmente encontrei a causa, percebi que o Python já tinha todas as ferramentas para diagnosticar isso — eu só não sabia usá-las juntas.
Neste artigo, vou te mostrar como construir um analisador de stack traces encadeadas em Python puro que reconstrói o caminho completo de um erro, mesmo quando múltiplas threads estão envolvidas. Sem Sentry, sem Bugsnag, sem depender de serviço externo. Apenas traceback, faulthandler, threading e um pouco de SQLite para persistência.
O Problema Real: Stack Traces Que Mentem
Quando você tem uma aplicação com múltiplas threads e uma exception é lançada, o Python faz o básico: captura o traceback da thread que falhou e imprime. O problema é que, em sistemas com threads interdependentes, a thread que falha raramente é a thread que causou o problema.
Imagine este cenário: uma thread produtora enfileira dados, uma thread consumidora processa, e uma thread de heartbeat monitora a saúde do sistema. O heartbeat detecta timeout, tenta fazer cleanup, e explode com um RuntimeError: dictionary changed size during iteration. O traceback aponta para a thread de heartbeat. Mas o bug real? Está na thread produtora, que modificou o dicionário sem lock.
O traceback padrão te deixa na mão. Você precisa de algo que capture o estado de todas as threads no momento do crash.
A Anatomia de um Stack Trace Completo
Antes de construir a solução, vamos entender o que precisamos capturar:
- Thread ID e nome — identificação única de cada thread ativa
- Frame stack — a pilha de chamadas de cada thread no momento do crash
- Locals de cada frame — variáveis locais que podem revelar o estado do bug
- Relações de dependência — qual thread espera por qual, quem criou quem
- Timestamp preciso — para correlacionar com outros logs
O Python tem módulos para cada um desses pontos. O truque é costurá-los juntos.
Módulo 1: faulthandler — O Detector de Crashes Fatais
O faulthandler é subestimado pela maioria dos devs Python. Ele captura crashes que o sys.excepthook não alcança — segmentation faults, stack overflows, sinais POSIX como SIGSEGV e SIGABRT. E mais importante: ele pode dumpar o traceback de todas as threads.
import faulthandler
import sys
import os
def setup_faulthandler(log_path="crash_dump.log"):
"""Ativa o faulthandler para capturar crashes fatais."""
log_file = open(log_path, "w")
faulthandler.enable(file=log_file, all_threads=True)
# Captura SIGUSR1 para dump manual (kill -USR1 pid)
faulthandler.register(signal.SIGUSR1, file=log_file, all_threads=True)
return log_file
# Ativa no startup da aplicação
log_file = setup_faulthandler()
O detalhe crucial: all_threads=True é o que diferencia um dump útil de um inútil. Sem isso, você só vê a thread que recebeu o sinal.
Módulo 2: sys._current_frames() — O Raio-X das Threads
Essa é a arma secreta que quase ninguém conhece. sys._current_frames() retorna um dicionário mapeando thread ID → frame atual de todas as threads ativas. É exatamente o que precisamos para reconstruir o estado global no momento do crash.
import sys
import threading
import traceback
def capture_all_thread_frames():
"""Captura o frame stack de todas as threads ativas."""
frames = sys._current_frames()
thread_map = {t.ident: t for t in threading.enumerate()}
snapshot = {}
for thread_id, frame in frames.items():
thread = thread_map.get(thread_id)
thread_name = thread.name if thread else f"Unknown-{thread_id}"
# Extrai a stack completa do frame
stack = traceback.extract_stack(frame)
# Captura locals do frame mais recente
current_locals = {}
if frame and frame.f_locals:
current_locals = {
k: safe_repr(v)
for k, v in frame.f_locals.items()
if not k.startswith('__')
}
snapshot[thread_name] = {
"thread_id": thread_id,
"alive": thread.is_alive() if thread else False,
"daemon": thread.daemon if thread else None,
"stack": [(f.filename, f.lineno, f.name, f.line) for f in stack],
"locals": current_locals
}
return snapshot
def safe_repr(obj, max_len=200):
"""repr seguro que não explode com objetos complexos."""
try:
r = repr(obj)
return r[:max_len] + "..." if len(r) > max_len else r
except Exception:
return f"<unrepr {type(obj).__name__}>"
Repare no safe_repr(). Parece paranoia, mas quando você está capturando locals no momento de um crash, qualquer coisa pode dar errado. Objetos com __repr__ quebrado, referências circulares, generators exauridos — todos explodem se você não proteger.
Módulo 3: O Correlacionador — Ligando Threads ao Erro
Aqui é onde a mágica acontece. Ter os frames de todas as threads é ótimo, mas você precisa correlacionar qual thread causou o problema e quais foram afetadas.
import json
import sqlite3
from datetime import datetime
from pathlib import Path
class StackTraceAnalyzer:
def __init__(self, db_path="trace_history.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS crash_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
error_type TEXT,
error_message TEXT,
originating_thread TEXT,
full_snapshot TEXT,
correlation_hash TEXT
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS crash_patterns (
pattern_hash TEXT PRIMARY KEY,
first_seen TEXT,
last_seen TEXT,
occurrence_count INTEGER DEFAULT 1,
sample_snapshot TEXT
)
""")
def capture_crash(self, exception=None):
"""Captura um evento de crash completo."""
snapshot = capture_all_thread_frames()
current_thread = threading.current_thread().name
error_type = type(exception).__name__ if exception else "Unknown"
error_msg = str(exception) if exception else "No exception"
# Gera hash para correlação de padrões
pattern_key = self._compute_pattern_hash(snapshot, error_type)
event = {
"timestamp": datetime.utcnow().isoformat(),
"error_type": error_type,
"error_message": error_msg,
"originating_thread": current_thread,
"snapshot": snapshot,
"pattern_hash": pattern_key
}
self._persist(event)
return event
def _compute_pattern_hash(self, snapshot, error_type):
"""Hash baseado nos file:line da stack, não em valores voláteis."""
import hashlib
stack_signature = []
for thread_name, data in sorted(snapshot.items()):
for filename, lineno, funcname, _ in data["stack"][-3:]:
stack_signature.append(f"{filename}:{lineno}")
raw = error_type + "|" + "|".join(stack_signature)
return hashlib.sha256(raw.encode()).hexdigest()[:16]
def _persist(self, event):
with sqlite3.connect(self.db_path) as conn:
conn.execute(
"INSERT INTO crash_events (timestamp, error_type, error_message, "
"originating_thread, full_snapshot, correlation_hash) "
"VALUES (?, ?, ?, ?, ?, ?)",
(event["timestamp"], event["error_type"], event["error_message"],
event["originating_thread"], json.dumps(event["snapshot"], default=str),
event["pattern_hash"])
)
# Atualiza padrões
conn.execute("""
INSERT INTO crash_patterns (pattern_hash, first_seen, last_seen,
occurrence_count, sample_snapshot)
VALUES (?, ?, ?, 1, ?)
ON CONFLICT(pattern_hash) DO UPDATE SET
last_seen = excluded.last_seen,
occurrence_count = occurrence_count + 1
""", (event["pattern_hash"], event["timestamp"], event["timestamp"],
json.dumps(event["snapshot"], default=str)))
def find_recurring_patterns(self, min_occurrences=2):
"""Encontra padrões de crash recorrentes."""
with sqlite3.connect(self.db_path) as conn:
rows = conn.execute(
"SELECT pattern_hash, occurrence_count, first_seen, last_seen "
"FROM crash_patterns WHERE occurrence_count >= ? "
"ORDER BY occurrence_count DESC",
(min_occurrences,)
).fetchall()
return rows
Integrando com excepthook e threading.excepthook
O Python 3.8+ trouxe o threading.excepthook, que captura exceptions não tratadas em threads individuais. Combinado com sys.excepthook, você cobre todos os cenários:
analyzer = StackTraceAnalyzer()
def global_exception_handler(exc_type, exc_value, exc_tb):
"""Handler para exceções não tratadas na thread principal."""
event = analyzer.capture_crash(exc_value)
# Loga o snapshot completo
print(f"\n{'='*60}")
print(f"CRASH CAPTURADO — Pattern: {event['pattern_hash']}")
print(f"Thread de origem: {event['originating_thread']}")
print(f"Threads ativas: {len(event['snapshot'])}")
print(f"{'='*60}\n")
# Chama o handler padrão também
sys.__excepthook__(exc_type, exc_value, exc_tb)
def thread_exception_handler(args):
"""Handler para exceções não tratadas em threads."""
event = analyzer.capture_crash(args.exc_value)
print(f"THREAD CRASH: {args.thread.name} — Pattern: {event['pattern_hash']}")
# Instala os handlers
sys.excepthook = global_exception_handler
threading.excepthook = thread_exception_handler
🔧 Perrengue Real: Eu implementei o
threading.excepthooke achei que estava seguro. Duas semanas depois, descobri que exceptions capturadas portry/exceptgenéricos dentro de threads não passam pelo excepthook — óbvio, né? Mas eu não tinha pensado nisso. A solução foi adicionar um decorator@catch_and_analyzeem todas as funções de entrada de threads críticas. Não é elegante, mas captura tudo.
Detectando Race Conditions com Frame Diffing
Aqui vai a parte mais interessante: usar snapshots sequenciais para detectar race conditions em tempo real, antes mesmo do crash.
import time
from collections import defaultdict
class RaceConditionDetector:
def __init__(self, interval_ms=100):
self.interval = interval_ms / 1000
self.snapshots = []
self.max_snapshots = 50
self.shared_state_access = defaultdict(list)
def take_snapshot(self):
"""Captura um snapshot e detecta acesso concorrente a shared state."""
snapshot = capture_all_thread_frames()
self.snapshots.append({
"time": time.monotonic(),
"data": snapshot
})
if len(self.snapshots) > self.max_snapshots:
self.snapshots.pop(0)
# Detecta threads acessando as mesmas funções/arquivos
self._detect_concurrent_access(snapshot)
def _detect_concurrent_access(self, snapshot):
"""Identifica funções sendo executadas simultaneamente por múltiplas threads."""
active_functions = defaultdict(list)
for thread_name, data in snapshot.items():
if data["stack"]:
# Pega os últimos 3 frames
for filename, lineno, funcname, _ in data["stack"][-3:]:
key = f"{filename}:{funcname}"
active_functions[key].append(thread_name)
# Funções acessadas por 2+ threads simultaneamente
for key, threads in active_functions.items():
if len(threads) >= 2:
self.shared_state_access[key].append({
"time": time.monotonic(),
"threads": threads
})
def get_suspicious_patterns(self):
"""Retorna funções com acesso concorrente frequente."""
suspicious = []
for key, accesses in self.shared_state_access.items():
if len(accesses) >= 3:
suspicious.append({
"function": key,
"concurrent_accesses": len(accesses),
"threads_involved": list(set(
t for a in accesses for t in a["threads"]
))
})
return suspicious
O Relatório de Diagnóstico — Transformando Dados em Ação
Ter os dados é metade do trabalho. A outra metade é transformá-los em algo legível que te ajude a agir:
def generate_diagnostic_report(event, analyzer):
"""Gera um relatório legível do crash."""
lines = []
lines.append(f"# Crash Report — {event['timestamp']}")
lines.append(f"\n**Erro:** {event['error_type']}: {event['error_message']}")
lines.append(f"**Thread de origem:** {event['originating_thread']}")
lines.append(f"**Pattern hash:** {event['pattern_hash']}")
# Verifica se é um padrão recorrente
patterns = analyzer.find_recurring_patterns(min_occurrences=2)
for p in patterns:
if p[0] == event['pattern_hash']:
lines.append(f"\n⚠️ **PADRÃO RECORRENTE:** {p[1]} ocorrências desde {p[2]}")
lines.append(f"\n## Snapshot de Threads ({len(event['snapshot'])} ativas)")
for thread_name, data in sorted(event['snapshot'].items()):
status = "🟢" if data["alive"] else "🔴"
lines.append(f"\n### {status} {thread_name} (ID: {data['thread_id']})")
if data["stack"]:
lines.append("```")
for filename, lineno, funcname, line in data["stack"][-5:]:
short_file = filename.split("/")[-1]
lines.append(f" {short_file}:{lineno} in {funcname}()")
if line:
lines.append(f" {line}")
lines.append("```")
if data["locals"]:
lines.append("\n**Locals:**")
for k, v in list(data["locals"].items())[:10]:
lines.append(f"- `{k}` = {v}")
return "\n".join(lines)
Casos de Uso Práticos
1. Debug de Deadlocks
Quando duas threads esperam uma pela outra, o sistema trava sem exception. Mas com snapshots periódicos, você vê ambas as threads paradas no mesmo ponto:
def detect_deadlocks(analyzer, threshold_seconds=10):
"""Detecta threads paradas no mesmo frame por muito tempo."""
# Compara snapshots sequenciais
if len(analyzer.snapshots) < 2:
return []
latest = analyzer.snapshots[-1]["data"]
oldest = analyzer.snapshots[-min(10, len(analyzer.snapshots))]["data"]
time_diff = analyzer.snapshots[-1]["time"] - analyzer.snapshots[-min(10, len(analyzer.snapshots))]["time"]
if time_diff < threshold_seconds:
return []
stuck_threads = []
for thread_name in latest:
if thread_name in oldest:
if latest[thread_name]["stack"][-1] == oldest[thread_name]["stack"][-1]:
stuck_threads.append(thread_name)
return stuck_threads
2. Detecção de Lock Contention
Threads que passam muito tempo em threading.Lock.acquire() indicam contention. O snapshot mostra isso claramente quando você vê múltiplas threads paradas em acquire().
3. Correlação com Métricas de Performance
Salvando o pattern_hash junto com métricas de CPU/memória, você consegue responder perguntas como: "Os crashes com hash a3f7b2 acontecem quando a CPU está acima de 80%?"
Deploy: Colocando em Produção Sem Medo
Aqui vai a configuração mínima para ativar tudo no startup:
import signal
def bootstrap_trace_analyzer():
"""Ativa o analisador de stack traces no startup."""
analyzer = StackTraceAnalyzer(db_path="/var/log/app/traces.db")
detector = RaceConditionDetector(interval_ms=200)
# Handlers de exceção
sys.excepthook = lambda *args: (
analyzer.capture_crash(args[1]),
sys.__excepthook__(*args)
)
threading.excepthook = lambda args: (
analyzer.capture_crash(args.exc_value)
)
# faulthandler para crashes fatais
faulthandler.enable(all_threads=True)
# Snapshot periódico em background
def periodic_snapshot():
while True:
detector.take_snapshot()
time.sleep(detector.interval)
monitor = threading.Thread(
target=periodic_snapshot,
daemon=True,
name="trace-monitor"
)
monitor.start()
# SIGUSR1 para dump manual
def dump_handler(signum, frame):
event = analyzer.capture_crash()
report = generate_diagnostic_report(event, analyzer)
print(report)
signal.signal(signal.SIGUSR1, dump_handler)
return analyzer, detector
# No main da aplicação:
analyzer, detector = bootstrap_trace_analyzer()
Limitações e Armadilhas
Vou ser honesto sobre o que esse sistema não faz:
- Não substitui um profiler — o overhead de capturar locals em cada snapshot é real. Use em produção com intervalo generoso (200ms+).
- Locals podem conter dados sensíveis — o
safe_reprcaptura tudo, inclusive senhas que estão em variáveis locais. Filtre campos sensíveis. - Não detecta race conditions em C extensions — se o bug está no nível C do Python (ou numa lib como numpy), o frame Python não vai mostrar.
- SQLite pode ser gargalo — em crashes frequentes, o writes concorrentes no SQLite podem travar. Considere um WAL mode ou queue.
Conclusão: Pare de Adivinhar, Comece a Capturar
Race conditions e bugs intermitentes são o pesadelo de todo desenvolvedor porque parecem impossíveis de reproduzir. Mas a verdade é que, na maioria das vezes, o Python já estava te dando todas as ferramentas para diagnosticá-los — você só precisava costurá-las juntas.
O stack trace analyzer não é um framework. Não tem pip install. Não tem dashboard bonitinho. É um conjunto de funções Python que, quando combinadas, te dão uma visibilidade que nenhum serviço pago oferece de forma tão granular.
Da próxima vez que um bug intermitente aparecer em produção às 3 da manhã, você não vai precisar adivinhar. Vai dar kill -USR1 <pid> e ter a resposta em segundos.
Agora me diz: qual é o bug intermitente que está tirando seu sono agora? Aquele que você nunca consegue reproduzir no ambiente de desenvolvimento? Comenta aí que eu posso montar um analisador específico para o seu caso no próximo post.
Links úteis:
