""" API Health Check Monitor v2.1 ============================== Monitor portable de salud de APIs con dashboard en tiempo real. Configurable completamente via apis_config.json sin tocar código. Desplegable en Docker, Kubernetes o cualquier servidor. SOBRE LOS REQUESTS QUE GENERA ESTE MONITOR ------------------------------------------- Cada health check genera requests HTTP INTENCIONALES hacia los endpoints configurados. Este es el comportamiento correcto de todo monitor de salud (idéntico a los Kubernetes liveness probes, AWS health checks, Pingdom, etc.). Todos los requests se identifican en logs del servidor destino con: User-Agent: API-Health-Monitor/2.1 X-Health-Probe: true X-Monitor-Source: api-health-monitor Para filtrar estos requests en sus logs: Nginx: access_log off if ($http_x_health_probe = "true"); Apache: SetEnvIf X-Health-Probe true HEALTHCHECK App: if request.headers.get('X-Health-Probe') == 'true': skip_logging() Para reducir frecuencia: aumentar check_interval_seconds en apis_config.json. ENDPOINTS HTTP DISPONIBLES (puerto 8000) ----------------------------------------- GET /dashboard.html → Dashboard interactivo GET /results.json → Último ciclo de health checks (JSON) GET /apis_config.json → Configuración actual GET /api/health → Health check del propio monitor (para K8s probes) GET /api/history → Historial de checks desde SQLite ?env=PROD&limit=500&since=2024-01-01T00:00:00 GET /api/stats → Estadísticas agregadas (uptime, latencia, etc.) ?env=PROD&hours=24 POST /api/settings → Actualiza global_config en apis_config.json Body: {"check_interval_seconds": 10} """ import json import base64 import time import statistics import sys import os import sqlite3 import urllib.request import urllib.error import urllib.parse import ssl import threading from datetime import datetime, timedelta from http.server import SimpleHTTPRequestHandler, HTTPServer, ThreadingHTTPServer # ══════════════════════════════════════════════════════════════════════════════ # CONFIGURACIÓN CENTRAL # ══════════════════════════════════════════════════════════════════════════════ CONFIG_PATH = os.environ.get('CONFIG_FILE_PATH', 'apis_config.json') DB_PATH = os.environ.get('DB_PATH', 'monitor_history.db') LATENCY_HISTORY: dict = {} _db_conn_lock = threading.Lock() # ══════════════════════════════════════════════════════════════════════════════ # 1. CARGA DE CONFIGURACIÓN # ══════════════════════════════════════════════════════════════════════════════ def load_config() -> tuple: """ Carga/recarga apis_config.json en cada ciclo. Permite cambios en caliente desde el dashboard sin reiniciar. """ try: with open(CONFIG_PATH, 'r', encoding='utf-8') as f: data = json.load(f) global_cfg = data.get("global_config", {}) envs = data.get("environments", {}) apis = data.get("apis", []) external_sites = data.get("external_sites", []) return global_cfg, envs, apis, external_sites except FileNotFoundError: print(f"❌ CRITICAL: Config file not found: '{CONFIG_PATH}'") print(" Ensure apis_config.json is in the same directory as this script.") sys.exit(1) except json.JSONDecodeError as e: print(f"❌ CRITICAL: Invalid JSON in '{CONFIG_PATH}': {e}") sys.exit(1) except Exception as e: print(f"❌ CRITICAL: Cannot load config: {e}") sys.exit(1) def save_global_config_patch(patch: dict) -> bool: """Actualiza parcialmente global_config — con validación estricta de tipos y rangos.""" VALID_FIELDS = { "check_interval_seconds": (int, 5, 300), "request_timeout_seconds": (int, 3, 30), "anomaly_zscore_threshold": (float, 1.0, 5.0), "anomaly_history_size": (int, 5, 100), "history_retention_days": (int, 1, 90), "ssl_verify": bool, } filtered = {} for k, v in patch.items(): if k not in VALID_FIELDS: continue rule = VALID_FIELDS[k] if rule is bool: if not isinstance(v, bool): return False filtered[k] = v else: expected_type, min_val, max_val = rule # Aceptar int o float para campos numéricos if not isinstance(v, (int, float)): return False val = expected_type(v) if not (min_val <= val <= max_val): return False filtered[k] = val if not filtered: return False try: with open(CONFIG_PATH, 'r', encoding='utf-8') as f: data = json.load(f) data.setdefault("global_config", {}).update(filtered) with open(CONFIG_PATH, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) return True except Exception as e: print(f"❌ Error saving settings: {e}") return False # ══════════════════════════════════════════════════════════════════════════════ # 2. PERSISTENCIA SQLite (OPCIONAL) # ══════════════════════════════════════════════════════════════════════════════ def init_db() -> sqlite3.Connection: """ Inicializa la base de datos SQLite para historial de checks. SQLite es parte de la stdlib de Python — sin dependencias externas. Archivo: monitor_history.db (mismo directorio que el script). """ conn = sqlite3.connect(DB_PATH, check_same_thread=False) conn.execute("PRAGMA journal_mode=WAL") # Escrituras concurrentes seguras conn.execute("PRAGMA synchronous=NORMAL") # Balance rendimiento/durabilidad conn.execute(""" CREATE TABLE IF NOT EXISTS health_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, ambiente TEXT NOT NULL, api_name TEXT NOT NULL, namespace TEXT, url_check TEXT, status_code INTEGER, status_text TEXT, status_color TEXT, response_time_ms INTEGER, tags_summary TEXT, is_anomaly INTEGER DEFAULT 0, z_score REAL DEFAULT 0.0 ) """) # Índices para queries rápidas en el dashboard de Reports conn.execute("CREATE INDEX IF NOT EXISTS idx_ts ON health_events(timestamp)") conn.execute("CREATE INDEX IF NOT EXISTS idx_env ON health_events(ambiente)") conn.execute("CREATE INDEX IF NOT EXISTS idx_api ON health_events(api_name)") conn.execute("CREATE INDEX IF NOT EXISTS idx_env_ts ON health_events(ambiente, timestamp)") conn.commit() print(f"✅ SQLite iniciado: {DB_PATH}") return conn def save_to_db(conn: sqlite3.Connection, results: list, retention_days: int = 7): """Persiste el ciclo actual en SQLite y limpia registros viejos.""" if not conn or not results: return try: with _db_conn_lock: conn.executemany(""" INSERT INTO health_events (timestamp, ambiente, api_name, namespace, url_check, status_code, status_text, status_color, response_time_ms, tags_summary, is_anomaly, z_score) VALUES (?,?,?,?,?,?,?,?,?,?,?,?) """, [ ( r["timestamp"], r["ambiente"], r["api_name"], r.get("namespace"), r.get("url_check"), r.get("status_code"), r.get("status_text"), r.get("status_color"), _parse_ms(r.get("response_time")), r.get("tags_summary"), 1 if r.get("is_anomaly") else 0, r.get("z_score_value", 0.0), ) for r in results ]) # Limpieza automática según retención configurada cutoff = (datetime.now() - timedelta(days=retention_days)).strftime("%Y-%m-%d %H:%M:%S") conn.execute("DELETE FROM health_events WHERE timestamp < ?", (cutoff,)) conn.commit() except Exception as e: print(f"⚠️ SQLite write error: {e}") def query_history(conn: sqlite3.Connection, env: str = None, api: str = None, limit: int = 500, since: str = None) -> list: """Consulta el historial de eventos desde SQLite.""" parts, params = [], [] if env: parts.append("ambiente = ?"); params.append(env) if api: parts.append("api_name = ?"); params.append(api) if since: parts.append("timestamp >= ?"); params.append(since) where = ("WHERE " + " AND ".join(parts)) if parts else "" params.append(limit) rows = conn.execute(f""" SELECT timestamp, ambiente, api_name, namespace, status_code, status_text, status_color, response_time_ms, tags_summary, is_anomaly, z_score FROM health_events {where} ORDER BY timestamp DESC LIMIT ? """, params).fetchall() cols = ["timestamp", "ambiente", "api_name", "namespace", "status_code", "status_text", "status_color", "response_time_ms", "tags_summary", "is_anomaly", "z_score"] return [dict(zip(cols, r)) for r in rows] def query_stats(conn: sqlite3.Connection, env: str = None, hours: int = 24) -> dict: """ Estadísticas agregadas para el panel de Reports del dashboard. Retorna uptime, latencia promedio, APIs más lentas, más inestables, etc. """ since = (datetime.now() - timedelta(hours=hours)).strftime("%Y-%m-%d %H:%M:%S") env_filter = "AND ambiente = ?" if env else "" params_base = [since] + ([env] if env else []) total = conn.execute(f"SELECT COUNT(*) FROM health_events WHERE timestamp >= ? {env_filter}", params_base).fetchone()[0] errors = conn.execute(f"SELECT COUNT(*) FROM health_events WHERE timestamp >= ? {env_filter} AND (status_code = 0 OR status_code >= 400)", params_base).fetchone()[0] anomalies = conn.execute(f"SELECT COUNT(*) FROM health_events WHERE timestamp >= ? {env_filter} AND is_anomaly = 1", params_base).fetchone()[0] lat_row = conn.execute(f""" SELECT AVG(response_time_ms), MIN(response_time_ms), MAX(response_time_ms) FROM health_events WHERE timestamp >= ? {env_filter} AND response_time_ms IS NOT NULL """, params_base).fetchone() slowest = conn.execute(f""" SELECT api_name, AVG(response_time_ms) as avg_lat FROM health_events WHERE timestamp >= ? {env_filter} AND response_time_ms IS NOT NULL GROUP BY api_name ORDER BY avg_lat DESC LIMIT 5 """, params_base).fetchall() # Tendencia por hora (últimas 24h) hourly = conn.execute(f""" SELECT strftime('%Y-%m-%dT%H:00:00', timestamp) as hour, COUNT(*) as total, SUM(CASE WHEN status_code = 0 OR status_code >= 400 THEN 1 ELSE 0 END) as errors, AVG(response_time_ms) as avg_lat FROM health_events WHERE timestamp >= ? {env_filter} GROUP BY hour ORDER BY hour """, params_base).fetchall() return { "period_hours": hours, "total_checks": total, "total_errors": errors, "total_anomalies": anomalies, "uptime_pct": round((total - errors) / total * 100, 2) if total else 100, "avg_latency_ms": round(lat_row[0] or 0), "min_latency_ms": lat_row[1] or 0, "max_latency_ms": lat_row[2] or 0, "slowest_apis": [{"api": r[0], "avg_ms": round(r[1])} for r in slowest], "hourly_trend": [ {"hour": r[0], "total": r[1], "errors": r[2], "avg_lat": round(r[3] or 0)} for r in hourly ], } def _parse_ms(response_time: str) -> int | None: """Convierte '125 ms' → 125. Retorna None si no es parseable.""" if response_time and response_time != "N/A": try: return int(str(response_time).replace(" ms", "").strip()) except (ValueError, AttributeError): pass return None # ══════════════════════════════════════════════════════════════════════════════ # 3. DETECCIÓN DE ANOMALÍAS (Z-Score estadístico) # ══════════════════════════════════════════════════════════════════════════════ def detect_anomaly(api_id: str, current_ms, threshold: float = 3.0, history_size: int = 20, min_samples: int = 5) -> tuple: """ Detecta anomalías de latencia usando Z-Score. Z = |latencia_actual - media_histórica| / desviación_estándar Un Z-Score > threshold indica que la latencia es estadísticamente anómala. Retorna (is_anomaly: bool, z_score: float). Por qué Z-Score (y no solo un umbral fijo): • Un umbral fijo de "500ms = malo" es arbitrario — depende de cada API. • Z-Score adapta el umbral al comportamiento histórico de cada API. • Una API que siempre responde en 2000ms no dispara alerta por latencia. • Una API que pasa de 50ms a 150ms SÍ es una anomalía estadística. """ if current_ms is None: return False, 0.0 if api_id not in LATENCY_HISTORY: LATENCY_HISTORY[api_id] = [] history = LATENCY_HISTORY[api_id] if len(history) < min_samples: history.append(current_ms) return False, 0.0 mean = statistics.mean(history) stdev = statistics.stdev(history) or 1.0 z = abs(current_ms - mean) / stdev history.append(current_ms) if len(history) > history_size: history.pop(0) return z > threshold, round(z, 2) # ══════════════════════════════════════════════════════════════════════════════ # 4. LÓGICA DE HTTP REQUEST (con headers de identificación) # ══════════════════════════════════════════════════════════════════════════════ def apply_ssl_config(global_cfg: dict): """ Configura SSL según la configuración. ssl_verify=false → bypass de certificados (útil en entornos internos con certs auto-firmados). Por defecto: ssl_verify=true (comportamiento seguro). """ if not global_cfg.get("ssl_verify", True): try: ssl._create_default_https_context = ssl._create_unverified_context return True except Exception: pass else: # Restaurar comportamiento por defecto (seguro) try: ssl._create_default_https_context = ssl.create_default_context except Exception: pass return False def check_single_url(url: str, timeout: int, user_agent: str) -> tuple: """ Hace un GET a una URL y retorna (status_code, elapsed_ms, color, status_text). Incluye headers de identificación para que los logs del servidor destino puedan reconocer e ignorar estos requests si es necesario. """ req = urllib.request.Request(url) req.add_header("User-Agent", user_agent) req.add_header("Accept", "application/json, text/plain, */*") req.add_header("X-Health-Probe", "true") req.add_header("X-Monitor-Source", "api-health-monitor") start = time.time() try: with urllib.request.urlopen(req, timeout=timeout) as response: elapsed_ms = int((time.time() - start) * 1000) status_code = response.status detail = "OK" try: body = response.read(4096).decode("utf-8", errors="ignore") if body: data = json.loads(body) raw = data.get("message", data.get("status", data.get("description", "OK"))) detail = str(raw) if not isinstance(raw, dict) else "UP" except Exception: pass return status_code, elapsed_ms, "green", f"OK ({status_code}) — {detail}" except urllib.error.HTTPError as e: elapsed_ms = int((time.time() - start) * 1000) server_header = e.headers.get("Server", "").lower() envoy_flags = e.headers.get("x-envoy-response-flags", "") if e.code == 503: return 503, elapsed_ms, "orange", "SERVER FAIL (503) — Upstream/Pod unavailable" elif e.code == 404: is_gateway = ("istio-envoy" in server_header or "nginx" in server_header or "traefik" in server_header or envoy_flags == "NR") label = "INFRA 404 — Gateway: no route found" if is_gateway else "APP 404 — Endpoint not found" return 404, elapsed_ms, "blue" if is_gateway else "red", label elif e.code in (401, 403): return e.code, elapsed_ms, "orange", f"AUTH ERROR ({e.code}) — Unauthorized or Forbidden" else: return e.code, elapsed_ms, "orange", f"HTTP ERROR ({e.code})" except urllib.error.URLError as e: return 0, None, "red", f"CONNECTION FAILED — {str(e.reason)[:80]}" except Exception as e: return 0, None, "red", f"UNKNOWN ERROR — {type(e).__name__}: {str(e)[:60]}" # ══════════════════════════════════════════════════════════════════════════════ # 5. CICLO DE HEALTH CHECKS # ══════════════════════════════════════════════════════════════════════════════ def run_health_checks(global_cfg: dict, envs: dict, apis: list, external_sites: list, db_conn=None) -> list: """ Ejecuta todos los health checks configurados en un ciclo. Combina: environments × apis (base_url + path) + external_sites (URL directa). Escribe results.json y persiste en SQLite si está habilitado. """ results = [] current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") timeout = int(global_cfg.get("request_timeout_seconds", 5)) user_agent = global_cfg.get("user_agent", "API-Health-Monitor/2.1") default_strat = global_cfg.get("default_health_strategy", ["/actuator/health", "/healthcheck", "/health", "/status", ""]) z_threshold = float(global_cfg.get("anomaly_zscore_threshold", 3.0)) hist_size = int(global_cfg.get("anomaly_history_size", 20)) min_samples = int(global_cfg.get("anomaly_min_samples", 5)) # Aplicar configuración SSL en cada ciclo (puede cambiar en caliente) ssl_bypassed = apply_ssl_config(global_cfg) if ssl_bypassed: print(" ⚠️ SSL verification disabled (ssl_verify: false in config)") # Filtro opcional por ambiente único (via variable de entorno) target_env = os.environ.get("ENVIRONMENT_KEY_ENV", "").strip() monitored_envs = {target_env: envs[target_env]} if target_env and target_env in envs else envs print(f"[{current_time}] ▶ Checking {len(monitored_envs)} env(s) × {len(apis)} API(s) + {len(external_sites)} site(s)") # ── 5a. APIs por environment ────────────────────────────────────────────── for env_key, env_data in monitored_envs.items(): url_base = env_data.get("url_base") or env_data.get("url_ingress_base", "") tags_raw = env_data.get("tags", {}) tags_str = ", ".join(f"{k}:{v}" for k, v in tags_raw.items()) if tags_raw else "—" for api in apis: health_paths = api.get("health_strategy", default_strat) api_path = api.get("path", "") api_id = f"{env_key}__{api['name']}" last_result = None found = False for h_path in health_paths: url = url_base + api_path + h_path code, ms, color, text = check_single_url(url, timeout, user_agent) if code == 200 or code == 503 or code == 0: found = True last_result = (url, code, ms, color, text) break elif code == 404: last_result = (url, code, ms, color, text) # Seguir probando paths else: last_result = (url, code, ms, color, text) found = True break if not found and last_result: url, code, ms, color, text = last_result if code == 404: text = "CLIENT ERROR (404) — All health paths returned 404. Check routing." color = "red" elif not last_result: url, code, ms, color, text = url_base + api_path, 0, None, "red", "NO PATHS CONFIGURED" is_anomaly, z = detect_anomaly(api_id, ms, z_threshold, hist_size, min_samples) results.append({ "timestamp": current_time, "ambiente": env_key, "api_name": api["name"], "namespace": env_data.get("namespace", env_key.lower()), "url_check": url, "status_code": code, "status_text": text, "status_color": color, "response_time": f"{ms} ms" if ms is not None else "N/A", "tags_summary": tags_str, "is_anomaly": is_anomaly, "z_score_value": z, }) # ── 5b. External sites (URL directa) ───────────────────────────────────── for site in external_sites: url = site.get("url", "") if not url: continue code, ms, color, text = check_single_url(url, timeout, user_agent) api_id = f"SITES__{site['name']}" is_anomaly, z = detect_anomaly(api_id, ms, z_threshold, hist_size, min_samples) results.append({ "timestamp": current_time, "ambiente": "SITES", "api_name": site["name"], "namespace": "external", "url_check": url, "status_code": code, "status_text": text, "status_color": color, "response_time": f"{ms} ms" if ms is not None else "N/A", "tags_summary": site.get("description", ""), "is_anomaly": is_anomaly, "z_score_value": z, }) # ── 5c. Persistencia ────────────────────────────────────────────────────── try: with open("results.json", "w", encoding="utf-8") as f: json.dump(results, f, indent=2, ensure_ascii=False) except Exception as e: print(f" ❌ Error saving results.json: {e}") if db_conn: retention = int(global_cfg.get("history_retention_days", 7)) save_to_db(db_conn, results, retention) ok_count = sum(1 for r in results if 200 <= (r["status_code"] or 0) < 300) err_count = len(results) - ok_count print(f" ✅ {ok_count} OK | ❌ {err_count} errors | 📦 {len(results)} total") return results # ══════════════════════════════════════════════════════════════════════════════ # 6. SERVIDOR WEB CON ENDPOINTS REST # ══════════════════════════════════════════════════════════════════════════════ # Referencia al conn SQLite compartida (se asigna en main) _shared_db_conn = None ADMIN_USER = os.environ.get("MONITOR_ADMIN_USER", "admin") ADMIN_PASS = os.environ.get("MONITOR_ADMIN_PASS", "") def check_auth(headers) -> bool: """Valida autenticación Basic para endpoints de administración.""" auth = headers.get("Authorization", "") if not auth.startswith("Basic "): return False try: decoded = base64.b64decode(auth[6:]).decode("utf-8") user, password = decoded.split(":", 1) return user == ADMIN_USER and password == ADMIN_PASS and password != "" except Exception: return False class MonitorRequestHandler(SimpleHTTPRequestHandler): """ Servidor HTTP que combina: - Archivos estáticos (dashboard.html, results.json, etc.) - Endpoints REST (/api/*) para integración con el dashboard y sistemas externos Diseñado para ser ligero y sin dependencias externas (solo stdlib). """ def do_GET(self): parsed = urllib.parse.urlparse(self.path) path = parsed.path qs = urllib.parse.parse_qs(parsed.query) # ── API: Health del propio monitor ───────────────────────────────── if path == "/api/health": self._json(200, { "status": "UP", "timestamp": datetime.now().isoformat(), "version": "2.1", "uptime": "Monitor is running", }) # ── API: Historial desde SQLite ───────────────────────────────────── elif path == "/api/history": if not _shared_db_conn: self._json(503, {"error": "SQLite not enabled. Set use_sqlite: true in global_config."}) return try: env = qs.get("env", [None])[0] api = qs.get("api", [None])[0] limit = int(qs.get("limit", ["500"])[0]) since = qs.get("since", [None])[0] limit = min(limit, 5000) # Seguridad: max 5000 registros data = query_history(_shared_db_conn, env, api, limit, since) self._json(200, data) except Exception as e: self._json(500, {"error": str(e)}) # ── API: Estadísticas agregadas ───────────────────────────────────── elif path == "/api/stats": if not _shared_db_conn: self._json(503, {"error": "SQLite not enabled. Set use_sqlite: true in global_config."}) return try: env = qs.get("env", [None])[0] hours = int(qs.get("hours", ["24"])[0]) hours = min(hours, 720) # Max 30 días data = query_stats(_shared_db_conn, env, hours) self._json(200, data) except Exception as e: self._json(500, {"error": str(e)}) # ── Archivos estáticos (dashboard.html, results.json, etc.) ───────── else: super().do_GET() def do_POST(self): path = urllib.parse.urlparse(self.path).path if path == "/api/settings": if not check_auth(self.headers): self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="Monitor Admin"') self.send_header("Content-Length", "0") self.end_headers() return try: length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) patch = json.loads(body) if save_global_config_patch(patch): self._json(200, {"ok": True, "updated": patch}) else: self._json(400, {"ok": False, "error": "No valid keys to update"}) except json.JSONDecodeError: self._json(400, {"ok": False, "error": "Invalid JSON body"}) except Exception as e: self._json(500, {"ok": False, "error": str(e)}) else: self._json(404, {"error": "Not found"}) def do_OPTIONS(self): """Pre-flight CORS — necesario si el dashboard se accede desde otro origen.""" self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def _json(self, code: int, data): """Helper: serializa y envía una respuesta JSON.""" payload = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(payload) def log_message(self, fmt, *args): """Suprime logs de polling del dashboard (results.json, apis_config.json).""" msg = args[0] if args else "" skip_patterns = ["results.json", "apis_config.json", "sw.js", "manifest.json", "api/health"] if not any(p in str(msg) for p in skip_patterns): super().log_message(fmt, *args) def run_dashboard_server(): """Inicia el servidor HTTP en un thread background.""" try: server = ThreadingHTTPServer(("0.0.0.0", 8000), MonitorRequestHandler) print(f"🌐 Dashboard: http://localhost:8000/dashboard.html") print(f" API: http://localhost:8000/api/health") server.serve_forever() except OSError as e: print(f"❌ Cannot start HTTP server on port 8000: {e}") print(" Verify port 8000 is not in use.") os._exit(1) # ══════════════════════════════════════════════════════════════════════════════ # 7. PUNTO DE ENTRADA PRINCIPAL # ══════════════════════════════════════════════════════════════════════════════ if __name__ == "__main__": # Cambiar al directorio del script (para servir archivos estáticos correctamente) script_dir = os.path.dirname(os.path.abspath(sys.argv[0])) if script_dir: os.chdir(script_dir) print("=" * 62) print(" 🔍 API Health Check Monitor v2.1") print("=" * 62) # Cargar config inicial global_cfg, envs, apis, external_sites = load_config() app_name = global_cfg.get("app_name", "API Health Monitor") print(f" App: {app_name}") print(f" Config: {CONFIG_PATH}") print(f" Entornos: {list(envs.keys())}") print(f" APIs: {len(apis)}") print(f" Sites: {len(external_sites)}") print(f" Intervalo: {global_cfg.get('check_interval_seconds', 5)}s") print(f" Timeout: {global_cfg.get('request_timeout_seconds', 5)}s") print(f" SSL verify: {global_cfg.get('ssl_verify', True)}") # SQLite (opcional) db_conn = None if global_cfg.get("use_sqlite", False): db_conn = init_db() _shared_db_conn = db_conn retention = global_cfg.get("history_retention_days", 7) print(f" SQLite: Enabled ({retention}d retención) → {DB_PATH}") else: print(f" SQLite: Disabled (set use_sqlite: true to enable history)") print() print(" ℹ️ Requests se identifican con:") print(" User-Agent: API-Health-Monitor/2.1") print(" X-Health-Probe: true") print() print(" Ctrl+C para detener") print("=" * 62) # Servidor HTTP en background server_thread = threading.Thread(target=run_dashboard_server, daemon=True) server_thread.start() # Loop principal de monitoreo while True: try: global_cfg, envs, apis, external_sites = load_config() interval = int(global_cfg.get("check_interval_seconds", 5)) run_health_checks(global_cfg, envs, apis, external_sites, db_conn) print(f" ⏳ Next check in {interval}s — Edit apis_config.json or use ⚙️ Settings to change\n") time.sleep(interval) except KeyboardInterrupt: print("\n\n🛑 Monitor stopped. Goodbye!") if db_conn: db_conn.close() break except Exception as e: print(f"⚠️ Cycle error: {e} — Retrying in 10s...") time.sleep(10)