392 lines
13 KiB
Python
392 lines
13 KiB
Python
"""
|
||
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"},
|
||
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()
|