#!/usr/bin/env python3 """ Nami Health Monitor — checks all services and alerts on state changes. Runs every 5 minutes via cron. Only sends Telegram when status changes. Usage: python3 /opt/nami-army/health_monitor.py State file: /opt/nami-army/health_state.json """ import json import os import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path STATE_FILE = Path("/opt/nami-army/health_state.json") # ─── Service Definitions ────────────────────────────────────────────── SERVICES = { "hermes-gateway": { "name": "Hermes Gateway", "emoji": "🧠", "type": "port", "port": 8644, "critical": True }, "nami-bridge": { "name": "Nami Bridge", "emoji": "🌉", "type": "systemd", "critical": True }, "nami-status-api": { "name": "Status API", "emoji": "📡", "type": "systemd", "port": 8699, "critical": False }, "laopatana-stat-lab": { "name": "LaoPatana", "emoji": "📊", "type": "systemd", "port": 3000, "critical": False }, "hanoi-stats-analyzer": { "name": "Hanoi Lottery", "emoji": "🎲", "type": "systemd", "port": 3002, "critical": False }, "clipboardbypao": { "name": "ClipboardByPao", "emoji": "📋", "type": "systemd", "port": 3001, "critical": False }, "gold-signal-os": { "name": "Gold Signal", "emoji": "🥇", "type": "systemd", "port": 8002, "critical": False }, "miroshark-backend": { "name": "MiroShark Backend", "emoji": "🦈", "type": "systemd", "port": 5001, "critical": False }, "miroshark-frontend": { "name": "MiroShark Frontend", "emoji": "🦈", "type": "systemd", "port": 3003, "critical": False }, "nami-api-gateway": { "name": "API Gateway", "emoji": "🌸", "type": "systemd", "port": 8700, "critical": False }, "miroshark-oracle": { "name": "MiroShark Oracle", "emoji": "🔮", "type": "systemd", "port": 8003, "critical": False }, "nginx": { "name": "Nginx", "emoji": "🌐", "type": "systemd", "critical": True }, "postgresql": { "name": "PostgreSQL", "emoji": "🗄️", "type": "systemd", "critical": True } } def check_systemd(service_name: str) -> dict: """Check if a systemd service is active.""" try: result = subprocess.run( ["systemctl", "is-active", service_name], capture_output=True, text=True, timeout=5 ) stdout = result.stdout.strip() return {"status": "up" if stdout == "active" else "down", "detail": stdout} except Exception as e: return {"status": "down", "detail": str(e)} def check_port(port: int) -> dict: """Check if a port is listening.""" try: result = subprocess.run( ["ss", "-tlnp"], capture_output=True, text=True, timeout=5 ) listening = f":{port} " in result.stdout return {"status": "up" if listening else "down", "detail": f"Port {port}"} except Exception as e: return {"status": "down", "detail": str(e)} def check_all() -> dict: """Check all services and return status dict.""" results = {} now = datetime.now(timezone.utc).isoformat() for svc_id, svc_def in SERVICES.items(): # Check systemd if applicable if svc_def["type"] in ("systemd",): result = check_systemd(svc_id) elif svc_def["type"] == "port": result = check_port(svc_def["port"]) else: result = {"status": "unknown", "detail": "unknown type"} results[svc_id] = { "name": svc_def["name"], "emoji": svc_def["emoji"], "status": result["status"], "detail": result.get("detail", ""), "critical": svc_def.get("critical", False), "last_checked": now } return results def load_previous_state() -> dict: """Load last known state from file.""" if STATE_FILE.exists(): try: return json.loads(STATE_FILE.read_text()) except: pass return {} def save_state(state: dict): """Save current state to file.""" STATE_FILE.parent.mkdir(parents=True, exist_ok=True) STATE_FILE.write_text(json.dumps(state, indent=2)) def detect_changes(previous: dict, current: dict) -> list: """Detect services that changed state.""" changes = [] for svc_id, svc_data in current.items(): prev_status = previous.get(svc_id, {}).get("status") curr_status = svc_data["status"] if prev_status and prev_status != curr_status: changes.append({ "service": svc_id, "name": svc_data["name"], "emoji": svc_data["emoji"], "from": prev_status, "to": curr_status, "critical": svc_data["critical"], "detail": svc_data.get("detail", "") }) return changes def format_telegram_message(changes: list, all_results: dict) -> str: """Format alert message for Telegram.""" lines = ["🏥 **Nami Health Monitor**\n"] # Down alerts first (most important) downs = [c for c in changes if c["to"] == "down"] recoveries = [c for c in changes if c["to"] == "up"] if downs: lines.append("🔴 **DOWN:**") for c in downs: lines.append(f" {c['emoji']} {c['name']} — was {c['from']} → now DOWN") lines.append("") if recoveries: lines.append("🟢 **RECOVERED:**") for c in recoveries: lines.append(f" {c['emoji']} {c['name']} — back UP") # Current summary up_count = sum(1 for s in all_results.values() if s["status"] == "up") down_count = sum(1 for s in all_results.values() if s["status"] == "down") critical_down = sum(1 for s in all_results.values() if s["status"] == "down" and s.get("critical")) total = len(all_results) lines.append("") lines.append(f"📊 **Status:** {up_count}/{total} up" + (f" · **{critical_down} critical down**" if critical_down else "")) if down_count > 0: lines.append("") lines.append("❌ **Down services:**") for svc_id, svc_data in all_results.items(): if svc_data["status"] == "down": lines.append(f" {svc_data['emoji']} {svc_data['name']} — {svc_data.get('detail', '?')}") lines.append(f"\n_{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} · VPS @ 178.104.181.132_") return "\n".join(lines) def main(): # Check all services results = check_all() # Load previous state previous = load_previous_state() # Detect changes changes = detect_changes(previous, results) # Print summary to stdout print("🏥 Nami Health Monitor") print("=" * 50) up = sum(1 for s in results.values() if s["status"] == "up") down = sum(1 for s in results.values() if s["status"] == "down") total = len(results) print(f"Status: {up}/{total} up, {down} down\n") for svc_id, svc_data in results.items(): icon = "✅" if svc_data["status"] == "up" else "❌" crit = " 🔴CRITICAL" if svc_data.get("critical") and svc_data["status"] == "down" else "" print(f" {icon} {svc_data['emoji']} {svc_data['name']:20s} {svc_data['status']:4s}{crit}") # Print changes if changes: print("\n📢 Changes detected:") for c in changes: arrow = "🔴" if c["to"] == "down" else "🟢" print(f" {arrow} {c['emoji']} {c['name']}: {c['from']} → {c['to']}") # Output Telegram message (for cron capture) print("\n--- TELEGRAM ALERT ---") print(format_telegram_message(changes, results)) else: print("\n✅ No changes — all stable") # Save state for next run save_state(results) if __name__ == "__main__": main()