226 lines
8.1 KiB
Python
226 lines
8.1 KiB
Python
#!/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", "2026"))
|
|
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
|
|
self._dispatch(send_body=True)
|
|
|
|
def do_HEAD(self) -> None: # noqa: N802
|
|
self._dispatch(send_body=False)
|
|
|
|
def _dispatch(self, send_body: bool) -> None:
|
|
parsed = urlparse(self.path)
|
|
if parsed.path in {"/", "/index.html"}:
|
|
return self.serve_static("index.html", "text/html; charset=utf-8", send_body=send_body)
|
|
if parsed.path == "/app.css":
|
|
return self.serve_static("app.css", "text/css; charset=utf-8", send_body=send_body)
|
|
if parsed.path == "/app.js":
|
|
return self.serve_static("app.js", "application/javascript; charset=utf-8", send_body=send_body)
|
|
if parsed.path == "/vue.js":
|
|
return self.serve_static("vue.js", "application/javascript; charset=utf-8", send_body=send_body)
|
|
if parsed.path == "/api/health":
|
|
return self.send_payload(
|
|
HTTPStatus.OK,
|
|
response_json({"status": "ok", "time": utc_now()}),
|
|
"application/json; charset=utf-8",
|
|
send_body=send_body,
|
|
)
|
|
if parsed.path == "/api/monitor/services":
|
|
payload = get_monitor_payload()
|
|
return self.send_payload(
|
|
HTTPStatus.OK,
|
|
response_json(payload),
|
|
"application/json; charset=utf-8",
|
|
send_body=send_body,
|
|
)
|
|
if parsed.path == "/favicon.ico":
|
|
return self.send_payload(HTTPStatus.NO_CONTENT, b"", "image/x-icon", send_body=send_body)
|
|
return self.send_error_payload(HTTPStatus.NOT_FOUND, "not found", send_body=send_body)
|
|
|
|
def serve_static(self, name: str, content_type: str, send_body: bool) -> None:
|
|
path = STATIC_DIR / name
|
|
if not path.exists():
|
|
return self.send_error_payload(HTTPStatus.NOT_FOUND, "static file missing", send_body=send_body)
|
|
self.send_payload(HTTPStatus.OK, path.read_bytes(), content_type, send_body=send_body)
|
|
|
|
def send_error_payload(self, status: HTTPStatus, message: str, send_body: bool) -> None:
|
|
self.send_payload(
|
|
status,
|
|
response_json({"status": int(status), "error": message}),
|
|
"application/json; charset=utf-8",
|
|
send_body=send_body,
|
|
)
|
|
|
|
def send_payload(self, status: HTTPStatus, body: bytes, content_type: str, send_body: bool) -> 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 send_body and 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()
|