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"""

Monitoramento Heineken House

🍺

Pausa na breja e nos drinks.
O ETL caiu.

O pipeline de dados do WiFi da Heineken House está parado há mais de {STALE_THRESHOLD_H}h. Os dados do dashboard podem estar desatualizados. Dê uma olhada quando puder.

Última fonte {source}	Última execução {last_run_str}	Tempo parado {tempo_str}

Para verificar no servidor

$ docker exec wifi_etl_worker tail -50 /var/log/wifi-etl.log
$ docker logs wifi_etl_worker

Este alerta não será repetido nas próximas {COOLDOWN_H}h. Se o problema persistir, um novo aviso será enviado automaticamente.

""" 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()