From 9d1a49a1dd655363fd1d9ac82039387bb41407a3 Mon Sep 17 00:00:00 2001 From: Rafael Lopes Date: Mon, 29 Jun 2026 18:21:33 -0300 Subject: [PATCH] FEAT: Adicionado watchdog para verificar cron da Heineken House --- .env.example | 7 + app/core/config.py | 9 +- app/watchdog.py | 316 +++++++++++++++++++++++++++++++++++++++++++++ infra/crontab | 3 + 4 files changed, 334 insertions(+), 1 deletion(-) create mode 100644 app/watchdog.py diff --git a/.env.example b/.env.example index 3d2f3d6..552ddbc 100644 --- a/.env.example +++ b/.env.example @@ -28,6 +28,13 @@ WIFIRE_GROUP_ID= # Alternativa ao ESTABLISHMENT_ID — preencha um dos # Logging LOG_LEVEL=INFO + +# Watchdog (alertas por email se o ETL parar) +MAIL_HOST=10.0.120.60 +MAIL_PORT=25 +MAIL_FROM=Goleiro +WATCHDOG_RECIPIENT_EMAILS=caio.marques@sothis.com.br +WATCHDOG_CC_EMAILS=rafael.lopes@sothis.com.br ``` --- diff --git a/app/core/config.py b/app/core/config.py index 2fc2637..4562334 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -24,4 +24,11 @@ WIFIRE_API_KEY = os.getenv("WIFIRE_API_KEY", "") WIFIRE_ESTABLISHMENT_ID = os.getenv("WIFIRE_ESTABLISHMENT_ID", "") WIFIRE_GROUP_ID = os.getenv("WIFIRE_GROUP_ID", "") -LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") \ No newline at end of file +LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") + +# Watchdog / Email +MAIL_HOST = os.getenv("MAIL_HOST", "") +MAIL_PORT = int(os.getenv("MAIL_PORT", "25")) +MAIL_FROM = os.getenv("MAIL_FROM", "") +WATCHDOG_RECIPIENT_EMAILS = os.getenv("WATCHDOG_RECIPIENT_EMAILS", "") +WATCHDOG_CC_EMAILS = os.getenv("WATCHDOG_CC_EMAILS", "") \ No newline at end of file diff --git a/app/watchdog.py b/app/watchdog.py new file mode 100644 index 0000000..0b95a2f --- /dev/null +++ b/app/watchdog.py @@ -0,0 +1,316 @@ +import os +import smtplib +import logging +from datetime import datetime, timezone, timedelta +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart + +import psycopg2 + +from app.core.config import ( + DB_HOST, DB_PORT, DB_NAME, DB_USER, DB_PASSWORD, + MAIL_HOST, MAIL_PORT, MAIL_FROM, + WATCHDOG_RECIPIENT_EMAILS, WATCHDOG_CC_EMAILS, +) + +STALE_THRESHOLD_H = 1 +COOLDOWN_H = 2 +COOLDOWN_FILE = '/tmp/etl_alert_sent' + +logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s') +log = logging.getLogger(__name__) + + +def get_last_run(): + conn = psycopg2.connect( + host=DB_HOST, port=DB_PORT, dbname=DB_NAME, + user=DB_USER, password=DB_PASSWORD, connect_timeout=5, + ) + try: + cur = conn.cursor() + cur.execute("SELECT source, last_run_at FROM watermarks ORDER BY last_run_at DESC LIMIT 1") + return cur.fetchone() + finally: + conn.close() + + +def in_cooldown(): + if not os.path.exists(COOLDOWN_FILE): + return False + mtime = datetime.fromtimestamp(os.path.getmtime(COOLDOWN_FILE), tz=timezone.utc) + return (datetime.now(timezone.utc) - mtime) < timedelta(hours=COOLDOWN_H) + + +def mark_cooldown(): + with open(COOLDOWN_FILE, 'w') as f: + f.write(datetime.now(timezone.utc).isoformat()) + + +def clear_cooldown(): + if os.path.exists(COOLDOWN_FILE): + os.remove(COOLDOWN_FILE) + + +def build_html(source, last_run_at, horas, mins): + last_run_brt = last_run_at - timedelta(hours=3) + last_run_str = last_run_brt.strftime('%d/%m %H:%M BRT') + tempo_str = f"{horas}h{mins:02d}min" + + return f""" + + + + + + + + + +""" + + +def build_plain(source, last_run_at, horas, mins): + last_run_brt = last_run_at - timedelta(hours=3) + return ( + f"HEINEKEN HOUSE — ETL PARADO\n" + f"{'=' * 40}\n\n" + f"Pausa na breja e nos drinks. O ETL caiu.\n\n" + f"Última fonte : {source}\n" + f"Última execução : {last_run_brt.strftime('%d/%m/%Y %H:%M BRT')}\n" + f"Tempo parado : {horas}h{mins:02d}min\n\n" + f"Para verificar:\n" + f" docker exec wifi_etl_worker tail -50 /var/log/wifi-etl.log\n" + f" docker logs wifi_etl_worker\n\n" + f"Este alerta não será repetido nas próximas {COOLDOWN_H}h.\n" + ) + + +def send_alert(subject, source, last_run_at, horas, mins): + recipients = [e.strip() for e in WATCHDOG_RECIPIENT_EMAILS.split(',') if e.strip()] + cc = [e.strip() for e in WATCHDOG_CC_EMAILS.split(',') if e.strip()] + + msg = MIMEMultipart('alternative') + msg['From'] = MAIL_FROM + msg['To'] = ', '.join(recipients) + msg['Subject'] = subject + if cc: + msg['Cc'] = ', '.join(cc) + + msg.attach(MIMEText(build_plain(source, last_run_at, horas, mins), 'plain', 'utf-8')) + msg.attach(MIMEText(build_html(source, last_run_at, horas, mins), 'html', 'utf-8')) + + with smtplib.SMTP(MAIL_HOST, MAIL_PORT, timeout=10) as smtp: + smtp.sendmail(MAIL_FROM, recipients + cc, msg.as_string()) + + log.info(f"Alerta enviado → {recipients + cc}") + + +def send_db_error_alert(error): + recipients = [e.strip() for e in WATCHDOG_RECIPIENT_EMAILS.split(',') if e.strip()] + cc = [e.strip() for e in WATCHDOG_CC_EMAILS.split(',') if e.strip()] + + subject = "[ETL ALERTA] Falha ao conectar no banco de dados" + body = ( + f"HEINEKEN HOUSE — ETL ALERTA\n\n" + f"O watchdog não conseguiu conectar no banco de dados.\n\n" + f"Erro: {error}\n\n" + f"Para verificar:\n" + f" docker logs wifi_etl_worker\n" + f" docker exec wifi_etl_worker tail -50 /var/log/wifi-etl.log\n" + ) + + msg = MIMEText(body, 'plain', 'utf-8') + msg['From'] = MAIL_FROM + msg['To'] = ', '.join(recipients) + msg['Subject'] = subject + if cc: + msg['Cc'] = ', '.join(cc) + + with smtplib.SMTP(MAIL_HOST, MAIL_PORT, timeout=10) as smtp: + smtp.sendmail(MAIL_FROM, recipients + cc, msg.as_string()) + + +def main(): + log.info("=== WATCHDOG START ===") + + try: + row = get_last_run() + except Exception as e: + log.error(f"Erro ao conectar no banco: {e}") + if not in_cooldown(): + send_db_error_alert(e) + mark_cooldown() + return + + if not row: + log.warning("Nenhum watermark encontrado — ETL nunca rodou?") + return + + source, last_run_at = row + age = datetime.now(timezone.utc) - last_run_at + horas = int(age.total_seconds() // 3600) + mins = int((age.total_seconds() % 3600) // 60) + + log.info(f"Último ETL: {source} em {last_run_at} ({horas}h{mins:02d}min atrás)") + + if age > timedelta(hours=STALE_THRESHOLD_H): + if in_cooldown(): + log.info("ETL parado mas em cooldown — alerta já enviado recentemente") + return + + subject = f"[ETL ALERTA] Heineken House — pipeline parado há {horas}h{mins:02d}min" + send_alert(subject, source, last_run_at, horas, mins) + mark_cooldown() + else: + log.info("ETL saudável") + clear_cooldown() + + log.info("=== WATCHDOG DONE ===") + + +if __name__ == '__main__': + main() diff --git a/infra/crontab b/infra/crontab index 4657ec8..028983c 100644 --- a/infra/crontab +++ b/infra/crontab @@ -5,3 +5,6 @@ # A cada 5 minutos: executa o ETL */5 * * * * root cd /app && /usr/local/bin/python -m app.main >> /var/log/wifi-etl.log 2>&1 +# A cada 15 minutos: watchdog verifica se o ETL está rodando +*/15 * * * * root cd /app && /usr/local/bin/python -m app.watchdog >> /var/log/wifi-etl-watchdog.log 2>&1 +