#!/usr/bin/env python3 from __future__ import annotations import concurrent.futures import json import os import threading import time import urllib.error import urllib.request from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse ROOT = Path(__file__).resolve().parent STATIC_DIR = ROOT / "static" CONFIG_PATH = Path(os.environ.get("TARGETS_FILE", ROOT / "config" / "targets.json")).expanduser() APP_BIND = os.environ.get("APP_BIND", "0.0.0.0").strip() or "0.0.0.0" APP_PORT = int(os.environ.get("APP_PORT", "6666")) PROBE_TIMEOUT_SECONDS = float(os.environ.get("PROBE_TIMEOUT_SECONDS", "3")) PROBE_CACHE_TTL_SECONDS = float(os.environ.get("PROBE_CACHE_TTL_SECONDS", "5")) PROBE_MAX_WORKERS = int(os.environ.get("PROBE_MAX_WORKERS", "12")) _CACHE_LOCK = threading.Lock() _CACHE_PAYLOAD: dict | None = None _CACHE_AT = 0.0 def utc_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() def load_targets() -> list[dict]: payload = json.loads(CONFIG_PATH.read_text(encoding="utf-8")) hosts = payload.get("hosts") or [] normalized: list[dict] = [] for host in hosts: services = host.get("services") or [] for service in services: normalized.append( { "host": str(host.get("name") or "").strip(), "group": str(host.get("group") or "").strip(), "ip": str(host.get("ip") or "").strip(), "service": str(service.get("name") or "").strip(), "kind": str(service.get("kind") or "java").strip(), "port": int(service.get("port")), "path": str(service.get("path") or "/").strip(), } ) return normalized def _healthy_from_body(service_kind: str, status_code: int, body_text: str) -> bool: lowered = body_text.lower() if service_kind == "golang": return status_code == 200 and ('"code":"ok"' in lowered or '"code": "ok"' in lowered or '"status":"ok"' in lowered) if status_code in {401, 403}: return True if status_code != 200: return False if '"status":"up"' in lowered or '"status": "up"' in lowered: return True if body_text.strip() == "UP": return True return "up" in lowered and "status" in lowered def probe_target(target: dict) -> dict: url = f"http://{target['ip']}:{target['port']}{target['path']}" started = time.perf_counter() request = urllib.request.Request(url, headers={"User-Agent": "hy-app-monitor/1.0"}) base = { **target, "url": url, "statusCode": 0, "latencyMs": 0, "ok": False, "detail": "", } try: with urllib.request.urlopen(request, timeout=PROBE_TIMEOUT_SECONDS) as response: body_text = response.read().decode("utf-8", errors="replace") latency_ms = int((time.perf_counter() - started) * 1000) ok = _healthy_from_body(target["kind"], response.status, body_text) return { **base, "statusCode": int(response.status), "latencyMs": latency_ms, "ok": ok, "detail": body_text[:220].replace("\n", " ").strip(), } except urllib.error.HTTPError as exc: body_text = exc.read().decode("utf-8", errors="replace") latency_ms = int((time.perf_counter() - started) * 1000) ok = _healthy_from_body(target["kind"], exc.code, body_text) return { **base, "statusCode": int(exc.code), "latencyMs": latency_ms, "ok": ok, "detail": body_text[:220].replace("\n", " ").strip() or str(exc), } except Exception as exc: # noqa: BLE001 latency_ms = int((time.perf_counter() - started) * 1000) return { **base, "latencyMs": latency_ms, "detail": str(exc), } def build_monitor_payload() -> dict: targets = load_targets() workers = max(1, min(PROBE_MAX_WORKERS, len(targets) or 1)) with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor: items = list(executor.map(probe_target, targets)) ok_count = sum(1 for item in items if item["ok"]) payload = { "updatedAt": utc_now(), "summary": { "total": len(items), "ok": ok_count, "down": len(items) - ok_count, }, "items": items, } return payload def get_monitor_payload() -> dict: global _CACHE_PAYLOAD, _CACHE_AT now = time.time() with _CACHE_LOCK: if _CACHE_PAYLOAD is not None and now - _CACHE_AT < PROBE_CACHE_TTL_SECONDS: return _CACHE_PAYLOAD payload = build_monitor_payload() with _CACHE_LOCK: _CACHE_PAYLOAD = payload _CACHE_AT = now return payload def response_json(payload: dict | list) -> bytes: return (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8") class MonitorHandler(BaseHTTPRequestHandler): server_version = "HyAppMonitor/1.0" def log_message(self, format: str, *args) -> None: # noqa: A003 return def do_GET(self) -> None: # noqa: N802 parsed = urlparse(self.path) if parsed.path in {"/", "/index.html"}: return self.serve_static("index.html", "text/html; charset=utf-8") if parsed.path == "/app.css": return self.serve_static("app.css", "text/css; charset=utf-8") if parsed.path == "/app.js": return self.serve_static("app.js", "application/javascript; charset=utf-8") if parsed.path == "/api/health": return self.send_payload(HTTPStatus.OK, response_json({"status": "ok", "time": utc_now()}), "application/json; charset=utf-8") if parsed.path == "/api/monitor/services": payload = get_monitor_payload() return self.send_payload(HTTPStatus.OK, response_json(payload), "application/json; charset=utf-8") if parsed.path == "/favicon.ico": return self.send_payload(HTTPStatus.NO_CONTENT, b"", "image/x-icon") return self.send_error_payload(HTTPStatus.NOT_FOUND, "not found") def serve_static(self, name: str, content_type: str) -> None: path = STATIC_DIR / name if not path.exists(): return self.send_error_payload(HTTPStatus.NOT_FOUND, "static file missing") self.send_payload(HTTPStatus.OK, path.read_bytes(), content_type) def send_error_payload(self, status: HTTPStatus, message: str) -> None: self.send_payload( status, response_json({"status": int(status), "error": message}), "application/json; charset=utf-8", ) def send_payload(self, status: HTTPStatus, body: bytes, content_type: str) -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.end_headers() if body: self.wfile.write(body) def main() -> None: server = ThreadingHTTPServer((APP_BIND, APP_PORT), MonitorHandler) print(f"hy-app-monitor listening on http://{APP_BIND}:{APP_PORT}", flush=True) server.serve_forever() if __name__ == "__main__": main()