Painel de monitoramento de servidor com métricas de sistemas de automação e dead man switch

Dead Man’s Switch Local: Detecte Quando Suas Automações Ficam Silenciosas Antes do Desastre

PorOlivetto

O Silêncio é o Pior Erro: Dead Man’s Switch Para Detectar Quando Suas Automações Param de Falar

Tem 3 da manhã. Seu cron job de backup rodou — ou melhor, deveria ter rodado. O script não travou com um traceback bonito que você pode depurar. Ele simplesmente não executou. Ninguém foi notificado. Ninguém percebeu. E três semanas depois, quando o disco finalmente corrompeu, o último backup válido era de um mês atrás.

Esse é o tipo de falha que nenhum try/except pega. Porque o problema não é o código quebrar. O problema é o código calar.

Depois de perder dados de um projeto por causa de um agendador que silenciosamente parou de disparar (o systemd timer simplesmente… sumiu do journal, sem log, sem erro), eu construí um dead man’s switch local. Um sistema que não espera o erro acontecer — ele detecta quando o silêncio se torna anormal e escala alertas antes que o desastre seja irreversível.

Neste artigo, vou te mostrar exatamente como implementar esse padrão em Python, com SQLite para persistência de estado, escalation em camadas e notificações que realmente chegam. Se você tem automações rodando em casa ou no servidor e nunca recebeu um “tá tudo ok” delas… você já está em risco.

Servidor de monitoramento com logs de heartbeat e sistema de dead man switch

O Que é um Dead Man’s Switch (e Por Que Você Precisa de Um)

O termo vem do mundo ferroviário e industrial: um dispositivo que exige interação periódica do operador. Se o operador para de interagir — porque desmaiou, saiu ou qualquer outra coisa — o sistema automaticamente dispara uma ação de segurança.

Traduzindo para automação: cada script, cron job ou serviço periódico precisa enviar um heartbeat (sinal de vida) dentro de uma janela de tempo. Se o heartbeat não chega, algo está errado. O switch detecta o silêncio e escala.

A diferença entre um simples monitoramento de “script rodou ou não” e um dead man’s switch é a presunção de vida. Você não está verificando se o script falhou — está verificando se ele existiu quando deveria. São coisas fundamentalmente diferentes.

Os Três Modos de Silêncio

Na minha experiência, automações silenciam de três formas:

  • Silêncio total: o agendador (cron, systemd timer, scheduler) nunca disparou. O script nem chegou a ser chamado.
  • Silêncio parcial: o script rodou, mas travou antes do heartbeat. Pode ser um deadlock, um timeout de rede, ou aquele bug que só aparece às 3h de domingo.
  • Silêncio malicioso: o script rodou, completou, mas o heartbeat falhou porque o endpoint de notificação estava fora do ar. Aqui o problema é o canal, não a automação.

Um bom dead man’s switch precisa distinguir esses cenários — ou pelo menos alertar sobre o silêncio antes que você precise distinguir.

Arquitetura do Sistema

O sistema que eu construí tem quatro componentes:

  1. Heartbeat Sender: decorator ou chamada explícita que registra “estou vivo” no SQLite
  2. Dead Man’s Switch Checker: processo que varifica se todos os heartbeats esperados chegaram dentro da janela
  3. Alert Escalator: pipeline de notificação em camadas (log → arquivo → email → Telegram → chamada)
  4. Dashboard de Status: CLI simples para ver quem está vivo, quem está morto e há quanto tempo

Implementação: O Heartbeat Sender

Primeiro, o banco de dados. SQLite é perfeito aqui porque é single-file, zero configuração e aguenta milhões de writes sem piscar:

import sqlite3
import time
from datetime import datetime, timedelta
from pathlib import Path
from functools import wraps

DB_PATH = Path("/var/lib/automente/heartbeats.db")

def init_db():
    """Cria tabelas se não existirem."""
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(DB_PATH))
    conn.execute("""
        CREATE TABLE IF NOT EXISTS heartbeats (
            task_name TEXT PRIMARY KEY,
            last_heartbeat REAL NOT NULL,
            consecutive_misses INTEGER DEFAULT 0,
            status TEXT DEFAULT 'alive',
            expected_interval_sec INTEGER NOT NULL,
            last_success_note TEXT DEFAULT ''
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS heartbeat_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            task_name TEXT NOT NULL,
            timestamp REAL NOT NULL,
            note TEXT DEFAULT ''
        )
    """)
    conn.commit()
    return conn

def register_task(conn, task_name: str, expected_interval_sec: int):
    """Registra ou atualiza uma tarefa monitorada."""
    conn.execute("""
        INSERT INTO heartbeats (task_name, last_heartbeat, expected_interval_sec)
        VALUES (?, ?, ?)
        ON CONFLICT(task_name) DO UPDATE SET
            expected_interval_sec = excluded.expected_interval_sec
    """, (task_name, time.time(), expected_interval_sec))
    conn.commit()

Agora o decorator que torna o heartbeat trivial de adicionar em qualquer automação:

def heartbeat(task_name: str, interval_sec: int = 3600):
    """
    Decorator que registra heartbeat antes e depois da execução.
    Se a função travar, o post-mortem nunca chega = dead man's switch ativa.
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            conn = init_db()
            register_task(conn, task_name, interval_sec)

            # Heartbeat de INÍCIO
            conn.execute("""
                UPDATE heartbeats SET
                    last_heartbeat = ?,
                    status = 'running'
                WHERE task_name = ?
            """, (time.time(), task_name))
            conn.commit()

            try:
                result = func(*args, **kwargs)

                # Heartbeat de SUCESSO
                conn.execute("""
                    UPDATE heartbeats SET
                        last_heartbeat = ?,
                        status = 'alive',
                        consecutive_misses = 0,
                        last_success_note = 'OK'
                    WHERE task_name = ?
                """, (time.time(), task_name))
                conn.commit()
                return result

            except Exception as e:
                # Heartbeat de FALHA (ainda conta como sinal de vida,
                # mas marca o status)
                conn.execute("""
                    UPDATE heartbeats SET
                        last_heartbeat = ?,
                        status = 'error',
                        last_success_note = ?
                    WHERE task_name = ?
                """, (time.time(), str(e)[:200], task_name))
                conn.commit()
                raise
            finally:
                conn.close()
        return wrapper
    return decorator

Dashboard de alertas de automação com indicadores de sistemas críticos e notificações

O Checker: Quem Detecta o Silêncio

O checker é um processo separado que roda a cada N minutos e varifica se todos os heartbeats estão dentro da janela esperada. A lógica é simples mas eficiente:

def check_dead_men(conn, escalation_levels=None):
    """
    Varifica todos os tasks e identifica os que estão silenciosos.
    Retorna lista de violações com nível de severidade.
    """
    if escalation_levels is None:
        escalation_levels = [
            # (multiplicador do intervalo, nível, ação)
            (1.5, 'warning', 'log'),
            (3.0, 'critical', 'telegram'),
            (6.0, 'emergency', 'phone'),
        ]

    now = time.time()
    violations = []

    cursor = conn.execute("""
        SELECT task_name, last_heartbeat, expected_interval_sec,
               consecutive_misses, status
        FROM heartbeats
    """)

    for task_name, last_hb, interval, misses, status in cursor:
        elapsed = now - last_hb
        overdue_ratio = elapsed / interval if interval > 0 else float('inf')

        for multiplier, level, action in escalation_levels:
            if overdue_ratio >= multiplier:
                # Incrementa counter de misses consecutivos
                conn.execute("""
                    UPDATE heartbeats SET
                        consecutive_misses = consecutive_misses + 1,
                        status = ?
                    WHERE task_name = ?
                """, (level, task_name))

                violations.append({
                    'task': task_name,
                    'level': level,
                    'action': action,
                    'elapsed_min': round(elapsed / 60, 1),
                    'expected_min': round(interval / 60, 1),
                    'last_status': status,
                })
                break  # Pega o nível mais alto aplicável

    conn.commit()
    return violations

🔧 O Perrengue Real: Na primeira versão, eu usei schedule + threading.Timer para o checker. Funcionou por duas semanas. Depois o processo Python simplesmente vazou memória até o OOM killer matar tudo. O próprio monitor morreu de morte silenciosa. Lição: o checker precisa ser o processo mais estúpido e leve possível. Hoje rodo como um script standalone via cron a cada 5 minutos — sem dependências, sem threads, sem stateful runtime. Se o cron quebrar, pelo menos o systemd journal registra.

Escalation: Quando o Alerta Precisa Chegar

Notificação em camada é o conceito mais subestimado em monitoramento pessoal. A ideia é: o primeiro alerta é discreto, o segundo é irritante, o terceiro é “liga pro on-call às 3h da manhã”.

import smtplib
import requests
from email.message import EmailMessage

def escalate(violations):
    """Dispara notificações baseadas no nível de severidade."""
    for v in violations:
        msg = (
            f"🚨 DEAD MAN'S SWITCH — {v['level'].upper()}\n"
            f"Task: {v['task']}\n"
            f"Esperado a cada: {v['expected_min']}min\n"
            f"Último heartbeat: {v['elapsed_min']}min atrás\n"
            f"Status anterior: {v['last_status']}\n"
        )

        if v['action'] == 'log':
            with open('/var/log/automente/dms_alerts.log', 'a') as f:
                f.write(f"{datetime.now().isoformat()} {msg}\n")

        elif v['action'] == 'telegram':
            # Envia pro seu bot do Telegram
            telegram_send(v['task'], msg)

        elif v['action'] == 'phone':
            # Aqui entra o escalation nuclear:
            # CallMeBot, Twilio, ou até um webhook pro seu telefone
            emergency_call(v['task'], msg)

O segredo do telegram_send é usar o mesmo padrão de heartbeat: se o Telegram estiver fora do ar, o próprio envio de notificação falha — e isso é um heartbeat do canal de alerta. Dead man’s switch dentro do dead man’s switch.

Dashboard CLI: Quem Está Vivo, Quem Está Morto

Nada substitui um comando rápido no terminal para ver o estado geral:

def dashboard():
    """Mostra status de todas as tarefas monitoradas."""
    conn = init_db()
    now = time.time()

    print(f"{'TASK':<30} {'STATUS':<12} {'ÚLTIMO HB':<18} {'ATRASO':<12}")
    print("-" * 72)

    cursor = conn.execute("""
        SELECT task_name, last_heartbeat, expected_interval_sec,
               status, consecutive_misses
        FROM heartbeats ORDER BY last_heartbeat ASC
    """)

    for name, last_hb, interval, status, misses in cursor:
        elapsed = now - last_hb
        overdue = max(0, elapsed - interval)
        overdue_str = f"+{overdue/60:.0f}min" if overdue > 0 else "OK"

        status_icon = {
            'alive': '🟢',
            'running': '🟡',
            'error': '🔴',
            'warning': '🟠',
            'critical': '⚠️',
            'emergency': '🚨',
        }.get(status, '⚪')

        hb_ago = datetime.fromtimestamp(last_hb).strftime('%H:%M:%S')
        print(f"{status_icon} {name:<27} {status:<12} {hb_ago:<18} {overdue_str}")

    conn.close()

Resultado no terminal:

🟢 backup_diario                 alive        02:15:00         OK
🟡 sync_drive                    running      01:45:00         +32min
🔴 email_processor               error        23:00:00         +420min
⚠️  webhook_listener              warning      22:30:00         +90min

Integrando Com Suas Automações Existentes

O beauty desse sistema é que a migração é incremental. Você não precisa reescrever nada:

# Antes (sua automação atual):
def process_emails():
    # ... 200 linhas de código ...
    pass

# Depois (com dead man's switch):
@heartbeat("email_processor", interval_sec=7200)  # espera heartbeat a cada 2h
def process_emails():
    # ... 200 linhas de código ...
    pass

Um decorator. Uma linha. E agora sua automação tem um sistema de detecção de silêncio enterprise-grade — sem pagar R$500/mês num SaaS que você nem precisa.

Automacões Que Merecem Dead Man's Switch Imediatamente

  • Backups: o clássico. Se o backup parou de rodar, você só descobre quando precisa.
  • Sync de arquivos: Dropbox/S3/local. Se o sync para, os dados divergem silenciosamente.
  • Webhook receivers: se seu endpoint parou de receber, o sistema upstream pode estar enviando pro void.
  • Cleaners/recyclers: scripts que limpam temporários, rotacionam logs. Se param, o disco enche e tudo quebra.
  • Health checks externos: se seu script de monitoramento de um serviço terceiro parou, você perdeu visibilidade.

Configurando o Checker Como Serviço

A forma mais confiável que encontrei de rodar o checker é via cron — porque se o cron quebra, o syslog registra, e isso é detectável por um nível acima:

# /etc/cron.d/automente-dms
# Roda o checker a cada 5 minutos
*/5 * * * * root /usr/bin/python3 /opt/automente/dms_checker.py >> /var/log/automente/dms_checker.log 2>&1

# O próprio checker.py registra heartbeat de SI MESMO
# Se o cron parar, o heartbeat do checker para também
# E um nível superior (systemd timer alternativo) detecta

Sim, é redundância de redundância. E sim, isso é proposital. Porque o único bug pior que uma automação que falha é uma automação que falha e ninguém sabe.

Conclusão: O Silêncio Não é Ouro — É Débito Técnico

Eu passei muito tempo tratando automações como "set and forget". Funciona até não funcionar. E quando não funciona, geralmente é no pior momento possível, com o maior impacto possível, e zero chance de recovery sem perda de dados.

Um dead man's switch não é sobre paranoia. É sobre responsabilidade técnica. Se você coloca um processo para rodar sozinho, você é responsável por saber quando ele para. E a única forma escalável de fazer isso é automatizar a detecção do silêncio.

O código completo está no padrão que mostrei aqui: decorator simples, SQLite como state store, checker standalone via cron, escalation em camadas. Cerca de 150 linhas de Python. Zero dependências externas (exceto requests pro Telegram). Custo: R$0. Valor: impagável quando seu backup de 3 meses te salva no pior dia.

Quer Ver Qual Automação Próxima?

Esse dead man's switch resolve um problema específico. Mas e o seu? Qual automação sua te acordaria de madrugada se parasse sem avisar? Me conta nos comentários ou manda um direct — o próximo artigo pode ser sobre exatamente aquilo que está te deixando ansioso.

E se você ainda não tem dead man's switch em nenhuma das suas automações... talvez devesse parar de ler e começar a implementar. Antes do próximo silêncio.

Anti-Padrões Que Eu Já Cometi (Para Você Não Repetir)

Se tem uma coisa que anos de automação me ensinaram é que os piores bugs são os que você cria tentando resolver outros bugs. Aqui vão os erros reais que cometi construindo esse sistema:

1. Usar o Mesmo Canal Para Alerta e Heartbeat

Na versão 1, o checker mandava alertas pelo mesmo webhook que recebia os heartbeats. Quando o webhook caiu, o checker não conseguiu alertar. O sistema de monitoramento ficou cego exatamente quando mais precisava enxergar. A correção foi simples: heartbeat usa canal A, alert usa canal B. E o heartbeat do checker usa canal C. Três canais, três pontos de falha independentes.

2. Armazenar Estado em Memória

Sim, eu sei. SQLite é óbvio. Mas na primeira tentativa, eu usei um dicionário Python em memória. O processo rodou por 11 dias. Reiniciei o servidor. Perdi todo o histórico de misses consecutivos. As automações que estavam em estado "critical" voltaram para "alive" como se nada tivesse acontecido. O checker perdeu a noção de gravidade. SQLite resolveu isso permanentemente — estado persistente sobrevive a reinícios, crashes e atualizações de kernel.

3. Intervalo Fixo Para Tudo

Definir "todo heartbeat deve chegar a cada hora" parece razoável até você ter um backup que roda semanalmente e um webhook que recebe pings a cada 30 segundos. O sistema precisa de intervalos configuráveis por task. O campo expected_interval_sec na tabela de heartbeats resolve isso. Cada tarefa define seu próprio ritmo de sobrevivência.

4. Não Testar o Caminho de Falha

Construir o sistema é fácil. Testar se ele funciona quando as coisas quebram é que é o trabalho real. Meu ritual mensal: mato deliberadamente 3 automações aleatórias e verifico se o dead man's switch detecta, escala e notifica dentro do SLA esperado. Se não detecta em 15 minutos, o sistema tem bug. E bug de monitoramento é o tipo mais traiçoeiro de bug — porque o bug é o sistema que deveria te contar sobre bugs.

Posts Similares