Automação que Falha em Silêncio: O Sistema de Monitoramento de Erros em Python Que Salva Seu Sono
Se você já acordou às 3h da manhã porque uma automação falhou silenciosamente e ninguém percebeu — esse post é pra você.
Não aquele tipo de falha dramática com traceback de 400 linhas. Pior: o tipo que parece que está funcionando, mas na verdade está processando dados errados, enviando notificações duplicadas ou simplesmente ignorando erros porque o except: engoliu tudo sem deixar rastro.
Depois de perder mais horas do que eu gostaria de admitir caçando bugs que “só acontecem em produção”, eu desenvolvi uma abordagem sistemática para monitorar erros em automações Python que me salvou de inúmeros perrengues. E o melhor: funciona com ferramentas que você já tem instaladas.
O Problema Silencioso das Automações que “Funcionam”
Tem uma armadilha clássica que pega todo mundo que trabalha com automação: o script rodou com código de saída 0, então está tudo certo, né?
Errado. Um exit code zero só significa que o Python não crashou. Não significa que seus dados estão corretos, que a API respondeu o que você esperava, ou que o arquivo de destino não está vazio.
Eu já vi automações que:
- Processaram 10.000 registros mas pularam 3.000 porque o CSV tinha encoding diferente
- Enviaram 47 emails duplicados porque o lock file não funcionou
- Gravaram dados num banco de testes em vez do de produção por 2 semanas
Todas com exit code zero. Todas “funcionando perfeitamente” até alguém reclamar.
Por que o logging tradicional não basta
O print() não é logging. E mesmo logging.info() solto pelo código sem estrutura é tão útil quanto um bilhete na geladeira que ninguém lê.
O que você precisa é de um sistema de detecção de anomalias nos próprios logs — algo que analise os padrões de erro e te avise antes do problema escalar.
Monitoramento de Erros com Python: A Abordagem em Camadas
A estratégia que eu uso divide o monitoramento em três camadas. Cada uma captura um tipo diferente de falha que a anterior deixa escapar.
Camada 1: Validação de Resultado (Pós-execução)
Todo script de automação deve validar o próprio resultado. Não confie no “rodou sem erro”. Verifique.
import json
import logging
from pathlib import Path
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class ValidationResult:
success: bool
checks_passed: int = 0
checks_failed: int = 0
details: list = field(default_factory=list)
def add_check(self, name: str, passed: bool, detail: str = ""):
if passed:
self.checks_passed += 1
logger.info(f"✓ Check '{name}' passou")
else:
self.checks_failed += 1
logger.error(f"✗ Check '{name}' falhou: {detail}")
self.details.append(f"{name}: {detail}")
self.success = self.checks_failed == 0
def report(self) -> str:
status = "SUCESSO" if self.success else "FALHA"
return (
f"[{status}] {self.checks_passed} OK, "
f"{self.checks_failed} falhas\n"
+ "\n".join(f" - {d}" for d in self.details)
)
def validate_sync(source_path: Path, dest_path: Path) -> ValidationResult:
result = ValidationResult(success=True)
# Check 1: arquivo destino existe?
result.add_check(
"destino_existe",
dest_path.exists(),
f"Arquivo {dest_path} não encontrado"
)
# Check 2: tamanho razoável?
if dest_path.exists():
size_ok = dest_path.stat().st_size > 100
result.add_check(
"tamanho_valido",
size_ok,
f"Arquivo com apenas {dest_path.stat().st_size} bytes"
)
# Check 3: integridade JSON
if dest_path.exists():
try:
json.loads(dest_path.read_text())
result.add_check("json_valido", True)
except json.JSONDecodeError as e:
result.add_check("json_valido", False, str(e))
return result
Essa estrutura parece exagero? Eu também achava — até o dia em que uma automação de migração corrompeu 200 arquivos e eu só descobri porque o check_failed acionou o alerta.
🔥 O perrengue que me ensinou: Em 2024, eu configurei um job noturno para consolidar relatórios de 12 fontes diferentes. O script rodava “perfeitamente” — exit 0, sem traceback. O problema? Uma das APIs tinha mudado o formato da resposta de
data.resultsparadata.items. O script simplesmente não pegava os dados daquela fonte. Durante 3 semanas, os relatórios ficaram incompletos e ninguém notou até a auditoria mensal. Se eu tivesse validação de resultado, o alerta teria disparado no primeiro run.
Camada 2: Health Check Periódico
Validação pós-execução pega erros no final. Mas e quando o script trava no meio? Para isso, você precisa de heartbeats — o script registra que está vivo em intervalos regulares.
import sqlite3
import time
from datetime import datetime, timedelta
class HealthMonitor:
def __init__(self, db_path: str = "/tmp/automation_health.db"):
self.db = sqlite3.connect(db_path)
self._setup()
def _setup(self):
self.db.execute("""
CREATE TABLE IF NOT EXISTS heartbeats (
job_name TEXT PRIMARY KEY,
last_beat TEXT,
status TEXT,
details TEXT
)
""")
self.db.commit()
def heartbeat(self, job_name: str, status: str = "running", details: str = ""):
now = datetime.now().isoformat()
self.db.execute(
"INSERT OR REPLACE INTO heartbeats VALUES (?, ?, ?, ?)",
(job_name, now, status, details)
)
self.db.commit()
def check_stale(self, job_name: str, threshold_minutes: int = 30) -> bool:
row = self.db.execute(
"SELECT last_beat FROM heartbeats WHERE job_name = ?",
(job_name,)
).fetchone()
if not row:
return True # nunca rodou = problema
last = datetime.fromisoformat(row[0])
return datetime.now() - last > timedelta(minutes=threshold_minutes)
def get_all_status(self) -> dict:
rows = self.db.execute("SELECT * FROM heartbeats").fetchall()
return {
r[0]: {"last_beat": r[1], "status": r[2], "details": r[3]}
for r in rows
}
Com isso, um processo separado pode verificar a cada 15 minutos se algum job está “preso” e disparar um alerta via Telegram, email ou webhook.
Camada 3: Análise de Padrão de Erro
Aqui é onde a mágica acontece. Em vez de só registrar erros, você analisa os padrões para detectar anomalias antes que virem incidentes.
from collections import Counter, defaultdict
from datetime import datetime, timedelta
import json
class ErrorPatternAnalyzer:
def __init__(self, log_file: str):
self.log_file = log_file
self.errors = self._parse_errors()
def _parse_errors(self) -> list:
errors = []
try:
with open(self.log_file) as f:
for line in f:
if '"level": "ERROR"' in line:
entry = json.loads(line)
errors.append(entry)
except FileNotFoundError:
pass
return errors
def error_rate_last_hours(self, hours: int = 24) -> float:
cutoff = datetime.now() - timedelta(hours=hours)
recent = [
e for e in self.errors
if datetime.fromisoformat(e.get("timestamp", "2000-01-01")) > cutoff
]
return len(recent) / hours
def top_error_types(self, n: int = 5) -> list:
counter = Counter(
e.get("error_type", "unknown") for e in self.errors
)
return counter.most_common(n)
def detect_anomaly(self, baseline_rate: float, threshold_multiplier: float = 3.0) -> dict:
current_rate = self.error_rate_last_hours(1)
is_anomaly = current_rate > baseline_rate * threshold_multiplier
return {
"anomaly_detected": is_anomaly,
"current_rate": round(current_rate, 2),
"baseline_rate": baseline_rate,
"threshold": round(baseline_rate * threshold_multiplier, 2),
"action": "INVESTIGAR AGORA" if is_anomaly else "OK"
}
Essa análise simples identifica quando a taxa de erros sai do padrão normal. Se suas automações normalmente geram 2 erros por hora e de repente aparecem 15 em uma hora, algo está errado — mesmo que nenhum erro individual pareça crítico.
Integrando Tudo: O Script Watchdog
Agora vamos juntar as três camadas num único script que você pode rodar como um serviço ou cron job:
#!/usr/bin/env python3
"""
watchdog.py — Monitor de saúde de automações
Usar com: python watchdog.py --check --alert-threshold 3
"""
import argparse
import sys
import requests
from datetime import datetime
def send_alert(message: str, webhook_url: str):
"""Envia alerta via webhook (Telegram, Slack, etc.)"""
payload = {"text": f"🚨 WATCHDOG: {message}"}
try:
requests.post(webhook_url, json=payload, timeout=10)
except Exception as e:
print(f"Falha ao enviar alerta: {e}", file=sys.stderr)
def run_checks(config_path: str, threshold: float):
monitor = HealthMonitor()
analyzer = ErrorPatternAnalyzer("/var/log/automation/errors.jsonl")
alerts = []
# Check 1: jobs travados
for job_name in ["sync_dados", "gerar_relatorios", "backup_db"]:
if monitor.check_stale(job_name, threshold_minutes=30):
alerts.append(f"⏰ Job '{job_name}' sem heartbeat há 30+ min")
# Check 2: taxa de erro anormal
result = analyzer.detect_anomaly(baseline_rate=2.0, threshold_multiplier=threshold)
if result["anomaly_detected"]:
alerts.append(
f"📈 Taxa de erro anormal: {result['current_rate']}/h "
f"(baseline: {result['baseline_rate']}/h)"
)
# Check 3: top erros recentes
top_errors = analyzer.top_error_types(3)
for error_type, count in top_errors:
if count > 10:
alerts.append(f"🔥 {error_type}: {count} ocorrências recentes")
if alerts:
message = "\n".join([
f"⚠️ ALERTAS WATCHDOG - {datetime.now():%d/%m %H:%M}",
"",
*alerts,
"",
"Verifique: https://automente.com.br/log-de-erros/"
])
print(message)
# Descomente para alertas reais:
# send_alert(message, "SEU_WEBHOOK_AQUI")
return 1
print(f"✅ Todos os checks OK - {datetime.now():%d/%m %H:%M}")
return 0
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Watchdog de automações")
parser.add_argument("--check", action="store_true", help="Rodar verificações")
parser.add_argument("--alert-threshold", type=float, default=3.0,
help="Multiplicador para detecção de anomalia")
args = parser.parse_args()
if args.check:
sys.exit(run_checks("config.yaml", args.alert_threshold))
else:
parser.print_help()
Alertas Inteligentes: Quando e Como Notificar
Um sistema de monitoramento que grita a cada 5 minutos é tão útil quanto um que não grita nunca. O segredo está na inteligência dos alertas:
Regra do 3-2-1 para alertas
- 3 níveis de severidade: INFO (log), WARNING (canal do time), CRITICAL (página no celular)
- 2 canais mínimos: Nunca dependa de um único canal. Se o Telegram cair, o email precisa funcionar.
- 1 ação clara: Cada alerta deve dizer O QUE fazer, não só QUE algo está errado.
from enum import Enum
class Severity(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
def format_alert(severity: Severity, job: str, message: str) -> dict:
templates = {
Severity.INFO: {
"prefix": "ℹ️",
"action": "Apenas para registro",
"channels": ["log"]
},
Severity.WARNING: {
"prefix": "⚠️",
"action": f"Verifique o job '{job}' em até 1h",
"channels": ["telegram", "log"]
},
Severity.CRITICAL: {
"prefix": "🚨",
"action": f"AÇÃO IMEDIATA: job '{job}' requer atenção agora",
"channels": ["telegram", "sms", "log"]
}
}
tpl = templates[severity]
return {
"title": f"{tpl['prefix']} {severity.value.upper()} - {job}",
"message": message,
"action_required": tpl["action"],
"channels": tpl["channels"],
"timestamp": datetime.now().isoformat()
}
Erros que Eu Cometi (Para Você Não Cometer)
Vou ser direto aqui. Essas são as armadilhas que me custaram horas de debugging:
- Logrotate sem compressão: logs de 2GB que travavam a análise porque eu esqueci de configurar o
logrotate. Resultado: disco cheio, automação parada, cliente irritado. - Timezone inconsistente: parte dos logs em UTC, parte em BRT. A análise de padrão ficava completamente distorcida.
- Alerta sem contexto: “Erro no job sync” sem dizer qual erro, quando começou ou qual o impacto. O alerta virou ruído e todo mundo parou de prestar atenção.
- Health check no lugar errado: eu colocava o heartbeat antes do processamento. Se o script travasse, ele registrava “vivo” mesmo estando morto. Sempre registre depois de cada etapa.
Próximos Passos: Do Reativo ao Preditivo
Com essas três camadas funcionando, você já está numa posição muito melhor que 90% das pessoas que rodam automações. Mas dá para ir além:
- Baseline automática: em vez de definir uma taxa de erro fixa, use uma média móvel dos últimos 7 dias. Isso adapta o limiar automaticamente.
- Correlação de erros: quando dois jobs falham juntos, não são duas falhas — é uma causa raiz só. Agrupe por timestamp para encontrar dependências ocultas.
- Auto-recuperação: para erros conhecidos (timeout de API, arquivo bloqueado), tente recovery automático antes de alertar.
O monitoramento de erros em automações não é sobre paranoia — é sobre confiança. Confiança de que quando algo der errado, você vai saber antes do cliente. Confiança de que seus scripts não estão “funcionando” enquanto destroem dados silenciosamente. Confiança de que você pode dormir tranquilo.
Quer Levar Isso a Sério?
Se você quer que eu escreva sobre como conectar esse watchdog a alertas via Telegram com botões de ação, ou como criar um dashboard web simples para visualizar o status de todas as suas automações em tempo real — me diz nos comentários.
Qual erro de automação te pegou de surpresa e você gostaria que tivesse detectado antes? Conta aqui — os melhores perrengues viram próximos posts.
Esse post faz parte da série Log de Erros — onde transformamos cada falha em lição.
