""" Health Monitor ============== Runs every 5 minutes via cron. Checks processes, resources, SSL, cache logs, and runs nightly database backups. Alerts via Discord webhook on any failure. Auto-fixes safe/known issues (Tier 1). Alerts and waits for human decision on everything else (Tier 2/3). Deploy to: /srv/health_monitor.py Cron entry (/etc/cron.d/health_monitor): */5 * * * * root python3 /srv/health_monitor.py >> /var/log/health_monitor.log 2>&1 Secrets loaded from /srv/api/.env: DISCORD_WEBHOOK=https://discord.com/api/webhooks/... DB_PASS=... """ import json import os import shutil import subprocess import time from datetime import datetime, timezone from pathlib import Path from dotenv import load_dotenv load_dotenv("/srv/api/.env") # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- DISCORD_WEBHOOK = os.getenv("DISCORD_WEBHOOK", "") DB_PASS = os.getenv("DB_PASS", "Shlevison2k17") DB_USER = "verimund_user" DB_NAME = "verimund" BACKUP_DIR = Path("/srv/backups/db") BACKUP_KEEP_DAYS = 7 LOG_DIR = Path("/var/log") STATE_FILE = Path("/srv/health_monitor_state.json") DISK_WARN_PCT = 80 RAM_WARN_PCT = 90 CPU_WARN_PCT = 90 SSL_WARN_DAYS = 14 DOMAINS = [ "verimundsolutions.com", "api.verimundsolutions.com", "git.verimundsolutions.com", "admin.verimundsolutions.com", ] CACHE_LOGS = { "stable": LOG_DIR / "cache_builder_stable.log", "beta": LOG_DIR / "cache_builder_beta.log", } # --------------------------------------------------------------------------- # State (tracks last daily/weekly run so we don't repeat within the window) # --------------------------------------------------------------------------- def _load_state() -> dict: if STATE_FILE.exists(): try: return json.loads(STATE_FILE.read_text()) except Exception: pass return {} def _save_state(state: dict) -> None: STATE_FILE.write_text(json.dumps(state, indent=2)) def _should_run(state: dict, key: str, interval_seconds: int) -> bool: last = state.get(key, 0) return (time.time() - last) >= interval_seconds def _mark_run(state: dict, key: str) -> None: state[key] = time.time() # --------------------------------------------------------------------------- # Discord alerts # --------------------------------------------------------------------------- def alert(message: str, tier: int = 2) -> None: """Send a Discord webhook message. Never raises — alerting must not crash the monitor.""" prefix = {1: "✅ [AUTO-FIXED]", 2: "⚠ [TIER 2]", 3: "🚨 [TIER 3 — ACTION REQUIRED]"}.get(tier, "ℹ") full = f"{prefix} {message}" print(full, flush=True) if not DISCORD_WEBHOOK: return try: import urllib.request payload = json.dumps({"content": full}).encode() req = urllib.request.Request( DISCORD_WEBHOOK, data=payload, headers={"Content-Type": "application/json", "User-Agent": "DiscordBot (https://verimundsolutions.com, 1.0)"}, method="POST", ) urllib.request.urlopen(req, timeout=10) except Exception as e: print(f" [alert] Discord send failed: {e}", flush=True) def log(message: str) -> None: print(f"[{datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} UTC] {message}", flush=True) # --------------------------------------------------------------------------- # Process checks # --------------------------------------------------------------------------- def _pm2_list() -> list[str]: """Return list of PM2 process names that are 'online'.""" try: out = subprocess.check_output( ["pm2", "jlist"], stderr=subprocess.DEVNULL, text=True ) procs = json.loads(out) return [p["name"] for p in procs if p.get("pm2_env", {}).get("status") == "online"] except Exception: return [] def check_processes() -> None: log("Checking processes...") # Nginx r = subprocess.run(["systemctl", "is-active", "nginx"], capture_output=True, text=True) if r.stdout.strip() != "active": subprocess.run(["systemctl", "start", "nginx"], capture_output=True) alert("Nginx was down — restarted.", tier=1) # PostgreSQL r = subprocess.run(["systemctl", "is-active", "postgresql"], capture_output=True, text=True) if r.stdout.strip() != "active": subprocess.run(["systemctl", "start", "postgresql"], capture_output=True) alert("PostgreSQL was down — restarted.", tier=1) # PM2 itself r = subprocess.run(["systemctl", "is-active", "pm2-root"], capture_output=True, text=True) if r.stdout.strip() != "active": subprocess.run(["systemctl", "start", "pm2-root"], capture_output=True) alert("PM2 service was down — restarted.", tier=1) # PM2-managed processes online = _pm2_list() for name in ("api",): if name not in online: subprocess.run(["pm2", "restart", name], capture_output=True) alert(f"PM2 process '{name}' was down — restarted.", tier=1) # API health check (HTTP) try: import urllib.request urllib.request.urlopen("http://127.0.0.1:8000/docs", timeout=5) except Exception: subprocess.run(["pm2", "restart", "api"], capture_output=True) alert("API server not responding on port 8000 — restarted via PM2.", tier=1) # --------------------------------------------------------------------------- # Resource checks # --------------------------------------------------------------------------- def check_resources() -> None: log("Checking resources...") # Disk usage = shutil.disk_usage("/") pct = usage.used / usage.total * 100 if pct >= DISK_WARN_PCT: _clean_old_logs() # Re-check after cleanup usage = shutil.disk_usage("/") pct = usage.used / usage.total * 100 if pct >= DISK_WARN_PCT: alert( f"Disk usage is {pct:.1f}% after auto-cleanup. Manual intervention may be needed.", tier=3 if pct >= 90 else 2, ) else: alert(f"Disk was {pct:.1f}% — old logs cleaned, now {pct:.1f}%.", tier=1) # RAM with open("/proc/meminfo") as f: lines = {line.split(":")[0]: int(line.split()[1]) for line in f if ":" in line} total = lines.get("MemTotal", 1) avail = lines.get("MemAvailable", 1) ram_pct = (total - avail) / total * 100 if ram_pct >= RAM_WARN_PCT: alert(f"RAM usage is {ram_pct:.1f}% ({(total - avail) // 1024} MB used of {total // 1024} MB).", tier=2) # CPU (1-minute load average vs core count) load1 = os.getloadavg()[0] cores = os.cpu_count() or 1 cpu_pct = load1 / cores * 100 if cpu_pct >= CPU_WARN_PCT: alert(f"CPU load is {cpu_pct:.1f}% (load avg {load1:.2f} on {cores} cores).", tier=2) def _clean_old_logs() -> None: """Delete log files older than 30 days to free disk space.""" now = time.time() cutoff = now - 30 * 86400 cleaned = 0 for path in LOG_DIR.glob("*.log*"): if path.stat().st_mtime < cutoff: try: path.unlink() cleaned += 1 except Exception: pass if cleaned: log(f" Cleaned {cleaned} old log file(s).") # --------------------------------------------------------------------------- # SSL certificate checks (daily) # --------------------------------------------------------------------------- def check_ssl(state: dict) -> None: if not _should_run(state, "ssl_check", 86400): return log("Checking SSL certificates...") _mark_run(state, "ssl_check") for domain in DOMAINS: try: r = subprocess.run( ["certbot", "certificates", "--domain", domain], capture_output=True, text=True, ) for line in r.stdout.splitlines(): if "VALID:" in line: days = int(line.split("VALID:")[1].split("day")[0].strip()) if days <= SSL_WARN_DAYS: log(f" SSL for {domain} expires in {days} days — renewing...") result = subprocess.run( ["certbot", "renew", "--cert-name", domain, "--non-interactive"], capture_output=True, text=True, ) if result.returncode == 0: alert(f"SSL cert for {domain} renewed ({days} days remaining).", tier=1) else: alert( f"SSL cert for {domain} expires in {days} days but renewal FAILED.\n" f"```{result.stderr[-500:]}```", tier=3, ) except Exception as e: log(f" SSL check failed for {domain}: {e}") # --------------------------------------------------------------------------- # Cache log checks (daily) # --------------------------------------------------------------------------- def check_cache_logs(state: dict) -> None: if not _should_run(state, "cache_log_check", 86400): return log("Checking cache builder logs...") _mark_run(state, "cache_log_check") for channel, log_path in CACHE_LOGS.items(): if not log_path.exists(): continue try: content = log_path.read_text(errors="replace") lines = content.splitlines() recent = "\n".join(lines[-50:]) if "Traceback" in recent or "Error" in recent: alert( f"Cache builder ({channel}) log contains errors:\n```{recent[-800:]}```", tier=2, ) elif "0 rows" in recent or "0 tickers" in recent.lower(): alert( f"Cache builder ({channel}) may have produced 0 rows:\n```{recent[-800:]}```", tier=2, ) except Exception as e: log(f" Could not read cache log for {channel}: {e}") # --------------------------------------------------------------------------- # Database backup (daily) # --------------------------------------------------------------------------- def run_backup(state: dict) -> None: if not _should_run(state, "db_backup", 86400): return log("Running database backup...") _mark_run(state, "db_backup") BACKUP_DIR.mkdir(parents=True, exist_ok=True) date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") out_path = BACKUP_DIR / f"verimund_{date_str}.sql.gz" env = os.environ.copy() env["PGPASSWORD"] = DB_PASS dump = subprocess.run( ["pg_dump", "-U", DB_USER, "-h", "127.0.0.1", DB_NAME], capture_output=True, env=env, ) if dump.returncode != 0: alert( f"pg_dump failed:\n```{dump.stderr.decode()[-500:]}```", tier=3, ) return import gzip with gzip.open(out_path, "wb") as f: f.write(dump.stdout) size_mb = out_path.stat().st_size / 1024 / 1024 log(f" Backup written: {out_path} ({size_mb:.1f} MB)") # Sync to R2 if rclone is configured r = subprocess.run( ["rclone", "sync", str(BACKUP_DIR), "r2:verimund-backups/db/"], capture_output=True, text=True, ) if r.returncode != 0: alert(f"rclone sync to R2 failed:\n```{r.stderr[-400:]}```", tier=2) else: log(" R2 sync complete.") # Remove backups older than BACKUP_KEEP_DAYS cutoff = time.time() - BACKUP_KEEP_DAYS * 86400 for f in BACKUP_DIR.glob("verimund_*.sql.gz"): if f.stat().st_mtime < cutoff: f.unlink() log(f" Deleted old backup: {f.name}") # --------------------------------------------------------------------------- # Log compression (weekly) # --------------------------------------------------------------------------- def compress_old_logs(state: dict) -> None: if not _should_run(state, "log_compress", 7 * 86400): return log("Compressing old logs...") _mark_run(state, "log_compress") import gzip cutoff = time.time() - 7 * 86400 compressed = 0 for path in LOG_DIR.glob("*.log"): if path.stat().st_mtime < cutoff: gz_path = path.with_suffix(".log.gz") try: with open(path, "rb") as f_in, gzip.open(gz_path, "wb") as f_out: f_out.write(f_in.read()) path.unlink() compressed += 1 except Exception as e: log(f" Could not compress {path.name}: {e}") if compressed: log(f" Compressed {compressed} log file(s).") # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> None: log("=== Health monitor start ===") state = _load_state() check_processes() check_resources() check_ssl(state) check_cache_logs(state) run_backup(state) compress_old_logs(state) _save_state(state) log("=== Health monitor done ===") if __name__ == "__main__": main()