From 52182a13ad40769ce40f2c6e44007ec13c346cad Mon Sep 17 00:00:00 2001 From: ZuoZuo <68836346+Mrz-sakura@users.noreply.github.com> Date: Wed, 22 Apr 2026 01:33:32 +0800 Subject: [PATCH] feat: modularize monitor and add nacos mongo views --- .env.example | 22 + AGENTS.md | 3 + README.md | 16 + config/targets.json | 21 + monitor/__init__.py | 2 + monitor/cache.py | 55 +++ monitor/config.py | 187 +++++++++ monitor/dashboard.py | 148 +++++++ monitor/env.py | 87 ++++ monitor/host_metrics.py | 167 ++++++++ monitor/http_app.py | 129 ++++++ monitor/http_probe.py | 176 ++++++++ monitor/mongo_metrics.py | 357 +++++++++++++++++ monitor/nacos.py | 461 +++++++++++++++++++++ monitor/tencent_tat.py | 199 +++++++++ requirements-ops.txt | 1 + server.py | 847 +-------------------------------------- static/app.css | 246 +++++++++--- static/app.js | 190 ++++++++- static/index.html | 312 +++++++++++++- 20 files changed, 2688 insertions(+), 938 deletions(-) create mode 100644 monitor/__init__.py create mode 100644 monitor/cache.py create mode 100644 monitor/config.py create mode 100644 monitor/dashboard.py create mode 100644 monitor/env.py create mode 100644 monitor/host_metrics.py create mode 100644 monitor/http_app.py create mode 100644 monitor/http_probe.py create mode 100644 monitor/mongo_metrics.py create mode 100644 monitor/nacos.py create mode 100644 monitor/tencent_tat.py diff --git a/.env.example b/.env.example index 8f331d5..3c08947 100644 --- a/.env.example +++ b/.env.example @@ -1,11 +1,33 @@ APP_BIND=0.0.0.0 APP_PORT=2026 +DEFAULT_REFRESH_INTERVAL_SECONDS=10 +REFRESH_INTERVAL_OPTIONS=5,10,15,30,60 PROBE_TIMEOUT_SECONDS=3 PROBE_CACHE_TTL_SECONDS=5 PROBE_MAX_WORKERS=12 HOST_METRIC_CACHE_TTL_SECONDS=25 TAT_METRIC_TIMEOUT_SECONDS=30 TAT_CPU_SAMPLE_SECONDS=1 +NACOS_BASE_URL=http://10.2.1.17:8848 +NACOS_NAMESPACE_ID=f7a8594e-090c-4830-a3ad-b5c0f9afcc07 +NACOS_NAMESPACE_NAME=RedCircleServiceProd +NACOS_DISCOVERY_GROUP=DEFAULT_GROUP +NACOS_DISCOVERY_CLUSTER_NAME=DEFAULT +NACOS_CACHE_TTL_SECONDS=15 +MONGO_URI= +MONGO_DB_NAME=tarab_all +MONGO_INSTANCE_ID=ins-o9mmob66 +MONGO_CACHE_TTL_SECONDS=20 +CPU_WARN_PERCENT=60 +CPU_BAD_PERCENT=85 +MEMORY_WARN_PERCENT=70 +MEMORY_BAD_PERCENT=90 +DISK_WARN_PERCENT=75 +DISK_BAD_PERCENT=90 +PROCESS_WARN_COUNT=180 +PROCESS_BAD_COUNT=260 +SERVICE_LATENCY_WARN_MS=1000 +SERVICE_LATENCY_BAD_MS=3000 DEPLOY_REGION=me-saudi-arabia TENCENT_SECRET_ID= TENCENT_SECRET_KEY= diff --git a/AGENTS.md b/AGENTS.md index 9eb0b07..c5782c4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,6 +6,7 @@ - 页面端口固定走 `2026`，不要再改回 `6666` 一类浏览器 unsafe port。 - 服务健康检查直接走各业务机器的 HTTP 接口。 - 主机 `CPU / 内存 / 磁盘 / 进程数` 统一走 `TAT` 远程采集，不额外安装 agent。 +- `Nacos` 和 `MongoDB` 指标优先走 deploy 机直连，不需要额外 agent。 ## 配置约定 @@ -13,6 +14,7 @@ - `.env` 必须保留在本地和远端项目目录里，但不能提交到 Git。 - `.env.example` 只作为模板，不能写真实密钥。 - `TENCENT_SECRET_ID` / `TENCENT_SECRET_KEY` / `DEPLOY_REGION` 从 `.env` 读取。 +- `NACOS_BASE_URL` / `NACOS_NAMESPACE_ID` / `MONGO_URI` 也从 `.env` 读取。 - `config/targets.json` 里的 `instanceId` 不能删，`TAT` 主机采集依赖它。 ## 运行约定 @@ -29,6 +31,7 @@ - 该脚本需要把本地项目根目录 `.env` 同步到远端项目根目录 `.env`。 - 远端 Python 运行时优先使用项目内 `.venv`。 - `requirements-ops.txt` 里的依赖既服务于部署脚本，也服务于运行时 `TAT` 指标采集。 +- Python 代码按模块维护，核心目录是 `monitor/`，`server.py` 只保留入口。 ## 修改原则 diff --git a/README.md b/README.md index bcd4f3c..9549384 100644 --- a/README.md +++ b/README.md @@ -6,10 +6,13 @@ - 页面入口：`http://43.164.75.199:2026/` - 服务健康：直接探测各业务服务健康接口 - 主机资源：通过 `TAT` 批量采集 `CPU / 内存 / 磁盘 / 进程数` +- Nacos：直连 Nacos OpenAPI，列出命名空间、服务和注册实例 +- MongoDB：直连 Mongo，采集 `serverStatus / listDatabases / replSetGetStatus` ## 目录 - `server.py`：后端入口 +- `monitor/`：Python 模块实现 - `config/targets.json`：监控目标和实例 ID - `static/`：前端页面 - `scripts/`：启停、自启安装、前台运行脚本 @@ -26,6 +29,11 @@ ```bash APP_BIND=0.0.0.0 APP_PORT=2026 +DEFAULT_REFRESH_INTERVAL_SECONDS=10 +REFRESH_INTERVAL_OPTIONS=5,10,15,30,60 +NACOS_BASE_URL=http://10.2.1.17:8848 +NACOS_NAMESPACE_ID=... +MONGO_URI=mongodb://... DEPLOY_REGION=me-saudi-arabia TENCENT_SECRET_ID=*** TENCENT_SECRET_KEY=*** @@ -92,3 +100,11 @@ python3 ops/deploy_via_tat.py 4. 安装 `requirements-ops.txt` 5. 安装并启用 `systemd` 6. 重启服务并检查 `2026` 监听 + +## 页面内容 + +- 主机：`CPU / 内存 / 磁盘 / 进程数`，带阈值颜色分级 +- 服务：Java / Golang 健康检查和耗时 +- Nacos：命名空间、服务、注册实例、健康状态 +- MongoDB：连接、内存、网络、操作计数、复制集、数据库统计 +- 自动刷新：前端可切换刷新间隔，并保存在浏览器本地 diff --git a/config/targets.json b/config/targets.json index 974f3e8..04c010d 100644 --- a/config/targets.json +++ b/config/targets.json @@ -1,5 +1,26 @@ { "hosts": [ + { + "name": "nacos-1", + "ip": "10.2.22.2", + "group": "nacos", + "instanceId": "ins-3yelqb7y", + "services": [] + }, + { + "name": "nacos-2", + "ip": "10.2.22.10", + "group": "nacos", + "instanceId": "ins-0oxyvn58", + "services": [] + }, + { + "name": "mongo-1", + "ip": "10.2.21.9", + "group": "mongo", + "instanceId": "ins-o9mmob66", + "services": [] + }, { "name": "gateway-1", "ip": "10.2.1.8", diff --git a/monitor/__init__.py b/monitor/__init__.py new file mode 100644 index 0000000..e791fb5 --- /dev/null +++ b/monitor/__init__.py @@ -0,0 +1,2 @@ +"""hy-app-monitor package.""" + diff --git a/monitor/cache.py b/monitor/cache.py new file mode 100644 index 0000000..c36d482 --- /dev/null +++ b/monitor/cache.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +# 线程锁用于保护缓存读写。 +import threading +# 时间模块用于判断缓存是否过期。 +import time +# 类型提示用于表达构建函数和返回值。 +from typing import Callable, Generic, TypeVar + + +# 泛型参数表示缓存中实际保存的数据类型。 +T = TypeVar("T") + + +class TTLCache(Generic[T]): + # 初始化一个简单的 TTL 缓存。 + def __init__(self, ttl_seconds: float) -> None: + # 保存缓存 TTL。 + self.ttl_seconds = ttl_seconds + # 用锁保护并发访问。 + self._lock = threading.Lock() + # 初始时没有缓存内容。 + self._payload: T | None = None + # 初始构建时间为 0。 + self._built_at = 0.0 + + # 按需获取缓存，过期时重新构建。 + def get_or_build(self, builder: Callable[[], T]) -> T: + # 记录当前时间。 + now = time.time() + + # 先在锁内尝试读取缓存。 + with self._lock: + # 有缓存且未过期时直接返回。 + if self._payload is not None and now - self._built_at < self.ttl_seconds: + return self._payload + + # 锁外构建新数据，避免长时间占锁。 + payload = builder() + + # 再次进入锁，把新结果写回去。 + with self._lock: + self._payload = payload + self._built_at = now + + # 返回新构建的结果。 + return payload + + # 主动清空缓存。 + def clear(self) -> None: + # 在锁内重置缓存状态。 + with self._lock: + self._payload = None + self._built_at = 0.0 + diff --git a/monitor/config.py b/monitor/config.py new file mode 100644 index 0000000..60a8063 --- /dev/null +++ b/monitor/config.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +# 读取 targets.json 需要 JSON 模块。 +import json +# Path 用来稳定定位项目目录和配置文件。 +from pathlib import Path +# 类型提示只用于提升可读性。 +from typing import Any + +# 统一从这里读取 .env 和环境变量。 +from .env import env_float, env_int, env_int_list, env_str, load_env_file + + +# 项目根目录。 +ROOT = Path(__file__).resolve().parents[1] +# 静态资源目录。 +STATIC_DIR = ROOT / "static" +# 默认 .env 路径。 +DEFAULT_ENV_PATH = ROOT / ".env" + +# 先加载项目根目录的 .env。 +load_env_file(Path(env_str("APP_ENV_FILE", str(DEFAULT_ENV_PATH))).expanduser()) + +# 监控目标配置文件路径。 +CONFIG_PATH = Path(env_str("TARGETS_FILE", str(ROOT / "config" / "targets.json"))).expanduser() +# HTTP 服务监听地址。 +APP_BIND = env_str("APP_BIND", "0.0.0.0") or "0.0.0.0" +# HTTP 服务监听端口。 +APP_PORT = env_int("APP_PORT", 2026) + +# 服务健康检查超时时间。 +PROBE_TIMEOUT_SECONDS = env_float("PROBE_TIMEOUT_SECONDS", 3.0) +# 服务健康检查缓存秒数。 +PROBE_CACHE_TTL_SECONDS = env_float("PROBE_CACHE_TTL_SECONDS", 5.0) +# 服务健康检查并发数。 +PROBE_MAX_WORKERS = env_int("PROBE_MAX_WORKERS", 12) + +# TAT 主机指标缓存秒数。 +HOST_METRIC_CACHE_TTL_SECONDS = env_float("HOST_METRIC_CACHE_TTL_SECONDS", 25.0) +# TAT 远端命令超时时间。 +TAT_METRIC_TIMEOUT_SECONDS = env_int("TAT_METRIC_TIMEOUT_SECONDS", 30) +# TAT CPU 采样间隔。 +TAT_CPU_SAMPLE_SECONDS = env_float("TAT_CPU_SAMPLE_SECONDS", 1.0) + +# TAT 所在区域。 +DEPLOY_REGION = env_str("DEPLOY_REGION", "me-saudi-arabia") +# 腾讯云 SecretId。 +TENCENT_SECRET_ID = env_str("TENCENT_SECRET_ID", "") +# 腾讯云 SecretKey。 +TENCENT_SECRET_KEY = env_str("TENCENT_SECRET_KEY", "") + +# 页面默认刷新间隔。 +DEFAULT_REFRESH_INTERVAL_SECONDS = env_int("DEFAULT_REFRESH_INTERVAL_SECONDS", 10) +# 页面可选刷新间隔。 +REFRESH_INTERVAL_OPTIONS = env_int_list("REFRESH_INTERVAL_OPTIONS", [5, 10, 15, 30, 60]) + +# Nacos API 基础地址。 +NACOS_BASE_URL = env_str("NACOS_BASE_URL", "http://10.2.1.17:8848").removesuffix("/nacos").rstrip("/") +# Nacos 目标命名空间 ID。 +NACOS_NAMESPACE_ID = env_str("NACOS_NAMESPACE_ID", "f7a8594e-090c-4830-a3ad-b5c0f9afcc07") +# Nacos 目标命名空间名。 +NACOS_NAMESPACE_NAME = env_str("NACOS_NAMESPACE_NAME", "RedCircleServiceProd") +# Nacos 默认注册分组。 +NACOS_DISCOVERY_GROUP = env_str("NACOS_DISCOVERY_GROUP", "DEFAULT_GROUP") +# Nacos 默认注册集群。 +NACOS_DISCOVERY_CLUSTER_NAME = env_str("NACOS_DISCOVERY_CLUSTER_NAME", "DEFAULT") +# Nacos 请求超时。 +NACOS_TIMEOUT_SECONDS = env_float("NACOS_TIMEOUT_SECONDS", 6.0) +# Nacos 结果缓存秒数。 +NACOS_CACHE_TTL_SECONDS = env_float("NACOS_CACHE_TTL_SECONDS", 15.0) +# Nacos service list 单页大小。 +NACOS_PAGE_SIZE = env_int("NACOS_PAGE_SIZE", 200) +# Nacos 可选用户名。 +NACOS_USERNAME = env_str("NACOS_USERNAME", "") +# Nacos 可选密码。 +NACOS_PASSWORD = env_str("NACOS_PASSWORD", "") + +# Mongo 连接串。 +MONGO_URI = env_str("MONGO_URI", "") +# Mongo 默认关注的业务库名。 +MONGO_DB_NAME = env_str("MONGO_DB_NAME", "tarab_all") +# Mongo 机器实例 ID，主要用于和主机指标联动显示。 +MONGO_INSTANCE_ID = env_str("MONGO_INSTANCE_ID", "ins-o9mmob66") +# Mongo 结果缓存秒数。 +MONGO_CACHE_TTL_SECONDS = env_float("MONGO_CACHE_TTL_SECONDS", 20.0) +# Mongo 建连超时。 +MONGO_CONNECT_TIMEOUT_MS = env_int("MONGO_CONNECT_TIMEOUT_MS", 3000) +# Mongo server selection 超时。 +MONGO_SERVER_SELECTION_TIMEOUT_MS = env_int("MONGO_SERVER_SELECTION_TIMEOUT_MS", 3000) +# Mongo socket 超时。 +MONGO_SOCKET_TIMEOUT_MS = env_int("MONGO_SOCKET_TIMEOUT_MS", 5000) + +# CPU 告警阈值。 +CPU_WARN_PERCENT = env_float("CPU_WARN_PERCENT", 60.0) +# CPU 严重阈值。 +CPU_BAD_PERCENT = env_float("CPU_BAD_PERCENT", 85.0) +# 内存告警阈值。 +MEMORY_WARN_PERCENT = env_float("MEMORY_WARN_PERCENT", 70.0) +# 内存严重阈值。 +MEMORY_BAD_PERCENT = env_float("MEMORY_BAD_PERCENT", 90.0) +# 磁盘告警阈值。 +DISK_WARN_PERCENT = env_float("DISK_WARN_PERCENT", 75.0) +# 磁盘严重阈值。 +DISK_BAD_PERCENT = env_float("DISK_BAD_PERCENT", 90.0) +# 进程数告警阈值。 +PROCESS_WARN_COUNT = env_int("PROCESS_WARN_COUNT", 180) +# 进程数严重阈值。 +PROCESS_BAD_COUNT = env_int("PROCESS_BAD_COUNT", 260) +# 服务耗时告警阈值。 +SERVICE_LATENCY_WARN_MS = env_int("SERVICE_LATENCY_WARN_MS", 1000) +# 服务耗时严重阈值。 +SERVICE_LATENCY_BAD_MS = env_int("SERVICE_LATENCY_BAD_MS", 3000) + +# 统一暴露给前端的阈值定义。 +THRESHOLDS = { + "cpuPercent": {"warn": CPU_WARN_PERCENT, "bad": CPU_BAD_PERCENT}, + "memoryPercent": {"warn": MEMORY_WARN_PERCENT, "bad": MEMORY_BAD_PERCENT}, + "diskPercent": {"warn": DISK_WARN_PERCENT, "bad": DISK_BAD_PERCENT}, + "processCount": {"warn": PROCESS_WARN_COUNT, "bad": PROCESS_BAD_COUNT}, + "latencyMs": {"warn": SERVICE_LATENCY_WARN_MS, "bad": SERVICE_LATENCY_BAD_MS}, +} + + +def utc_now() -> str: + # 时间函数延迟导入，避免这里堆太多全局依赖。 + from datetime import datetime, timezone + + # 统一输出无微秒的 UTC ISO 字符串。 + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +def settings_payload() -> dict[str, Any]: + # 把页面需要的动态设置统一打包返回。 + return { + "refreshIntervalSeconds": DEFAULT_REFRESH_INTERVAL_SECONDS, + "refreshIntervalOptions": list(REFRESH_INTERVAL_OPTIONS), + "thresholds": THRESHOLDS, + } + + +def load_hosts() -> list[dict[str, Any]]: + # 读取监控目标配置。 + payload = json.loads(CONFIG_PATH.read_text(encoding="utf-8")) + # 没有 hosts 时回退为空列表。 + hosts = payload.get("hosts") or [] + # 规范化后的主机列表。 + normalized: list[dict[str, Any]] = [] + + # 逐台主机做结构清洗。 + for host in hosts: + normalized.append( + { + "host": str(host.get("name") or "").strip(), + "group": str(host.get("group") or "").strip(), + "ip": str(host.get("ip") or "").strip(), + "instanceId": str(host.get("instanceId") or "").strip(), + "services": host.get("services") or [], + } + ) + + # 返回可供各模块复用的主机列表。 + return normalized + + +def flatten_services(hosts: list[dict[str, Any]]) -> list[dict[str, Any]]: + # 这里把主机结构拍平成服务列表，方便并发探测。 + services: list[dict[str, Any]] = [] + + # 逐台主机展开服务。 + for host in hosts: + for service in host.get("services") or []: + services.append( + { + "host": host["host"], + "group": host["group"], + "ip": host["ip"], + "instanceId": host["instanceId"], + "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 services + diff --git a/monitor/dashboard.py b/monitor/dashboard.py new file mode 100644 index 0000000..6e208b1 --- /dev/null +++ b/monitor/dashboard.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +# 类型提示只用于表达返回结构。 +from typing import Any + +# 聚合各模块监控结果。 +from .config import THRESHOLDS, load_hosts, settings_payload, utc_now +from .host_metrics import get_host_metrics +from .http_probe import get_service_payload +from .mongo_metrics import get_mongo_payload +from .nacos import get_nacos_payload + + +def classify_threshold(value: float | int | None, thresholds: dict[str, float | int]) -> str: + # 没值时返回 neutral。 + if value is None: + return "neutral" + + # 读取 warn 阈值。 + warn = float(thresholds["warn"]) + # 读取 bad 阈值。 + bad = float(thresholds["bad"]) + # 统一转成浮点数比较。 + numeric = float(value) + + # 超过 bad 阈值记为 bad。 + if numeric >= bad: + return "bad" + + # 超过 warn 阈值记为 warn。 + if numeric >= warn: + return "warn" + + # 其余情况记为 ok。 + return "ok" + + +def annotate_metric_levels(metrics: dict[str, Any]) -> dict[str, Any]: + # 复制一份，避免直接修改缓存原对象。 + payload = dict(metrics) + + # 指标采集失败时统一按 bad 标记。 + if not payload.get("ok"): + payload["cpuLevel"] = "bad" + payload["memoryLevel"] = "bad" + payload["diskLevel"] = "bad" + payload["processLevel"] = "bad" + return payload + + # 给 CPU 指标增加级别。 + payload["cpuLevel"] = classify_threshold(payload.get("cpuPercent"), THRESHOLDS["cpuPercent"]) + # 给内存指标增加级别。 + payload["memoryLevel"] = classify_threshold(payload.get("memoryPercent"), THRESHOLDS["memoryPercent"]) + # 给磁盘指标增加级别。 + payload["diskLevel"] = classify_threshold(payload.get("diskPercent"), THRESHOLDS["diskPercent"]) + # 给进程数增加级别。 + payload["processLevel"] = classify_threshold(payload.get("processCount"), THRESHOLDS["processCount"]) + + # 返回加完等级的结构。 + return payload + + +def build_host_payload( + hosts: list[dict[str, Any]], + service_payload: dict[str, Any], + metrics_by_host: dict[str, dict[str, Any]], +) -> list[dict[str, Any]]: + # 先把服务结果按 host 分组。 + services_by_host: dict[str, list[dict[str, Any]]] = {} + + # 遍历所有服务结果。 + for item in service_payload.get("items") or []: + services_by_host.setdefault(item["host"], []).append(item) + + # 最终的主机维度结果。 + grouped_hosts: list[dict[str, Any]] = [] + + # 按配置顺序输出主机。 + for host in hosts: + # 当前主机的服务结果列表。 + items = services_by_host.get(host["host"], []) + # 当前主机健康服务数量。 + ok_count = sum(1 for item in items if item["ok"]) + # 当前主机原始指标。 + raw_metrics = metrics_by_host.get(host["host"], {"ok": False, "error": "metrics unavailable"}) + # 加上指标阈值等级。 + metrics = annotate_metric_levels(raw_metrics) + + # 追加主机结构。 + grouped_hosts.append( + { + "host": host["host"], + "ip": host["ip"], + "group": host["group"], + "instanceId": host["instanceId"], + "metrics": metrics, + "serviceSummary": { + "total": len(items), + "ok": ok_count, + "down": len(items) - ok_count, + }, + "items": items, + } + ) + + # 返回主机分组结果。 + return grouped_hosts + + +def build_monitor_payload() -> dict[str, Any]: + # 读取当前监控目标。 + hosts = load_hosts() + # 获取服务探测结果。 + service_payload = get_service_payload(hosts, THRESHOLDS["latencyMs"]) + # 获取主机资源指标。 + metrics_by_host = get_host_metrics(hosts) + # 获取 Nacos 注册信息。 + nacos_payload = get_nacos_payload() + # 获取 Mongo 指标。 + mongo_payload = get_mongo_payload() + # 组装主机维度视图。 + grouped_hosts = build_host_payload(hosts, service_payload, metrics_by_host) + # 统计主机指标成功数。 + host_metric_ok = sum(1 for host in grouped_hosts if host["metrics"].get("ok")) + + # 返回完整的页面 API 载荷。 + return { + "updatedAt": utc_now(), + "settings": settings_payload(), + "summary": { + **service_payload["summary"], + "hostMetricOk": host_metric_ok, + "hostMetricDown": len(grouped_hosts) - host_metric_ok, + "nacosServiceTotal": int(nacos_payload.get("summary", {}).get("serviceTotal", 0)), + "nacosInstanceTotal": int(nacos_payload.get("summary", {}).get("instanceTotal", 0)), + "nacosUnhealthyInstanceTotal": int( + nacos_payload.get("summary", {}).get("unhealthyInstanceTotal", 0) + ), + "mongoDatabaseTotal": int(mongo_payload.get("summary", {}).get("databaseTotal", 0)), + "mongoCollectionTotal": int(mongo_payload.get("summary", {}).get("collectionTotal", 0)), + "mongoOk": 1 if mongo_payload.get("ok") else 0, + }, + "hosts": grouped_hosts, + "items": service_payload["items"], + "nacos": nacos_payload, + "mongo": mongo_payload, + } + diff --git a/monitor/env.py b/monitor/env.py new file mode 100644 index 0000000..48478d9 --- /dev/null +++ b/monitor/env.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +# 读取环境变量时要处理项目根目录的 .env。 +import os +# Path 用来稳定处理 .env 路径。 +from pathlib import Path + + +def load_env_file(path: Path) -> None: + # .env 文件不存在时允许继续使用外部环境变量。 + if not path.exists(): + return + + # 逐行读取 .env 内容。 + for raw_line in path.read_text(encoding="utf-8").splitlines(): + # 先去掉两端空白。 + line = raw_line.strip() + + # 空行不处理。 + if not line: + continue + + # 注释不处理。 + if line.startswith("#"): + continue + + # 不符合 KEY=VALUE 结构的行直接跳过。 + if "=" not in line: + continue + + # 只按第一个等号拆分，避免 value 里包含额外等号时被误切。 + key, value = line.split("=", 1) + + # 清理 key 和 value 两端的空白。 + key = key.strip() + value = value.strip() + + # 去掉成对双引号。 + if value.startswith('"') and value.endswith('"'): + value = value[1:-1] + + # 去掉成对单引号。 + if value.startswith("'") and value.endswith("'"): + value = value[1:-1] + + # 只在外部环境没有显式传入时才写入。 + os.environ.setdefault(key, value) + + +def env_str(name: str, default: str = "") -> str: + # 读取字符串环境变量。 + value = os.environ.get(name, default) + + # 返回去掉两端空白后的结果。 + return value.strip() + + +def env_int(name: str, default: int) -> int: + # 先按字符串读取。 + raw_value = env_str(name, str(default)) + + # 转成整数返回。 + return int(raw_value) + + +def env_float(name: str, default: float) -> float: + # 先按字符串读取。 + raw_value = env_str(name, str(default)) + + # 转成浮点数返回。 + return float(raw_value) + + +def env_int_list(name: str, default: list[int]) -> tuple[int, ...]: + # 先读取原始文本。 + raw_value = env_str(name, "") + + # 没配置时直接返回默认值。 + if not raw_value: + return tuple(default) + + # 逐项切分并过滤空项。 + values = [item.strip() for item in raw_value.split(",") if item.strip()] + + # 转成整数元组返回。 + return tuple(int(item) for item in values) + diff --git a/monitor/host_metrics.py b/monitor/host_metrics.py new file mode 100644 index 0000000..f7502e9 --- /dev/null +++ b/monitor/host_metrics.py @@ -0,0 +1,167 @@ +from __future__ import annotations + +# JSON 用来解析远端脚本输出。 +import json +# 类型提示提升可读性。 +from typing import Any + +# 读取主机指标相关配置。 +from .cache import TTLCache +from .config import HOST_METRIC_CACHE_TTL_SECONDS, TAT_CPU_SAMPLE_SECONDS, TAT_METRIC_TIMEOUT_SECONDS, utc_now +from .tencent_tat import run_shell_command + + +# 主机指标单独缓存，避免页面频繁刷新时不断调用 TAT。 +HOST_METRIC_CACHE = TTLCache[dict[str, dict[str, Any]]](HOST_METRIC_CACHE_TTL_SECONDS) + + +def build_metrics_script() -> str: + # 远端脚本只依赖 Python 标准库，避免每台机器额外装 agent。 + return f"""set -euo pipefail +python3 - <<'PY' +import json +import os +import shutil +import time + +# 读取 /proc/stat 的 CPU 统计值。 +def read_cpu(): + with open('/proc/stat', 'r', encoding='utf-8') as handle: + parts = handle.readline().split()[1:8] + values = [int(item) for item in parts] + idle = values[3] + values[4] + total = sum(values) + return idle, total + +# 第一次采样。 +idle_1, total_1 = read_cpu() +# 等待一个很短的采样窗口。 +time.sleep({TAT_CPU_SAMPLE_SECONDS}) +# 第二次采样。 +idle_2, total_2 = read_cpu() + +# 计算 CPU 使用率。 +total_diff = max(total_2 - total_1, 1) +idle_diff = max(idle_2 - idle_1, 0) +cpu_percent = round((1 - idle_diff / total_diff) * 100, 2) + +# 读取内存信息。 +memory_info = {{}} +with open('/proc/meminfo', 'r', encoding='utf-8') as handle: + for line in handle: + key, value = line.split(':', 1) + memory_info[key] = int(value.strip().split()[0]) + +# 计算内存用量。 +memory_total_gb = round(memory_info.get('MemTotal', 0) / 1024 / 1024, 2) +memory_available_gb = round(memory_info.get('MemAvailable', 0) / 1024 / 1024, 2) +memory_used_gb = round(max(memory_total_gb - memory_available_gb, 0), 2) +memory_percent = round((memory_used_gb / memory_total_gb) * 100, 2) if memory_total_gb else 0.0 + +# 读取根分区磁盘占用。 +disk = shutil.disk_usage('/') +disk_total_gb = round(disk.total / 1024 / 1024 / 1024, 2) +disk_used_gb = round(disk.used / 1024 / 1024 / 1024, 2) +disk_percent = round((disk.used / disk.total) * 100, 2) if disk.total else 0.0 + +# 通过 /proc 估算当前进程数。 +process_count = len([name for name in os.listdir('/proc') if name.isdigit()]) + +# 输出统一 JSON。 +print(json.dumps({{ + 'ok': True, + 'hostname': os.uname().nodename, + 'cpuPercent': cpu_percent, + 'memoryPercent': memory_percent, + 'memoryUsedGb': memory_used_gb, + 'memoryTotalGb': memory_total_gb, + 'diskPercent': disk_percent, + 'diskUsedGb': disk_used_gb, + 'diskTotalGb': disk_total_gb, + 'processCount': process_count +}})) +PY +""" + + +def metrics_error_payload(error: str, instance_id: str = "") -> dict[str, Any]: + # 返回统一的主机指标错误结构。 + return { + "ok": False, + "error": error, + "updatedAt": utc_now(), + "instanceId": instance_id, + } + + +def collect_host_metrics(hosts: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: + # 建立实例 ID 到 host 的映射。 + host_by_instance = {host["instanceId"]: host for host in hosts if host.get("instanceId")} + # 去重后的实例 ID 列表。 + instance_ids = list(host_by_instance.keys()) + + # 没有实例 ID 时直接返回错误。 + if not instance_ids: + return { + host["host"]: metrics_error_payload("missing instanceId in config/targets.json") + for host in hosts + } + + # 通过 TAT 批量执行主机指标采集脚本。 + task_results = run_shell_command( + instance_ids, + build_metrics_script(), + command_name="hy-app-monitor-host-metrics", + timeout_seconds=TAT_METRIC_TIMEOUT_SECONDS, + ) + + # 这里按 host name 回填最终结果。 + payload_by_host: dict[str, dict[str, Any]] = {} + + # 逐台主机解析结果。 + for host in hosts: + # 读取实例 ID。 + instance_id = host.get("instanceId", "") + # 读取该实例的远端执行结果。 + task = task_results.get(instance_id, {}) + # 读取执行状态。 + status = task.get("status", "") + # 读取远端输出。 + output = task.get("output", "") + + # 成功时尝试解析 JSON。 + if status == "SUCCESS": + try: + # 把 JSON 输出反序列化成 Python 结构。 + payload = json.loads(output or "{}") + # 补齐统一字段。 + payload["updatedAt"] = utc_now() + payload["instanceId"] = instance_id + payload_by_host[host["host"]] = payload + continue + except Exception as exc: # noqa: BLE001 + payload_by_host[host["host"]] = metrics_error_payload(f"metrics parse failed: {exc}", instance_id) + continue + + # 失败时保留错误文本或状态码。 + payload_by_host[host["host"]] = metrics_error_payload(output.strip() or status or "metrics failed", instance_id) + + # 返回按 host 索引的主机指标。 + return payload_by_host + + +def get_host_metrics(hosts: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: + # 定义真正的采集构建函数。 + def builder() -> dict[str, dict[str, Any]]: + try: + # 优先返回真实采集结果。 + return collect_host_metrics(hosts) + except Exception as exc: # noqa: BLE001 + # 任意异常都降级成错误占位，避免拖垮整个页面。 + return { + host["host"]: metrics_error_payload(f"metrics collection failed: {exc}", host.get("instanceId", "")) + for host in hosts + } + + # 通过缓存保护 TAT 调用。 + return HOST_METRIC_CACHE.get_or_build(builder) diff --git a/monitor/http_app.py b/monitor/http_app.py new file mode 100644 index 0000000..5af483f --- /dev/null +++ b/monitor/http_app.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +# JSON 用于构造统一接口响应。 +import json +# HTTP 状态码常量。 +from http import HTTPStatus +# 标准库 HTTP Server 足够承载这个轻量页面。 +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +# 类型提示用于方法签名。 +from typing import Any +# urlparse 只解析 path。 +from urllib.parse import urlparse + +# 读取应用配置和静态目录。 +from .config import APP_BIND, APP_PORT, STATIC_DIR, utc_now +# 统一构造监控页 API 载荷。 +from .dashboard import build_monitor_payload + + +def response_json(payload: dict[str, Any] | list[Any]) -> bytes: + # 统一把 JSON 编码成 UTF-8 文本。 + return (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8") + + +class MonitorHandler(BaseHTTPRequestHandler): + # 自定义服务名，便于 curl 时识别来源。 + server_version = "HyAppMonitor/2.0" + + def log_message(self, format: str, *args: Any) -> None: # noqa: A003 + # 关闭默认访问日志，避免轮询刷屏。 + return + + def do_GET(self) -> None: # noqa: N802 + # GET 请求需要响应头和响应体。 + self.dispatch(send_body=True) + + def do_HEAD(self) -> None: # noqa: N802 + # HEAD 请求只返回响应头。 + self.dispatch(send_body=False) + + def dispatch(self, send_body: bool) -> None: + # 只解析 URL 的 path 段。 + parsed = urlparse(self.path) + + # 首页返回静态 HTML。 + if parsed.path in {"/", "/index.html"}: + return self.serve_static("index.html", "text/html; charset=utf-8", send_body=send_body) + + # CSS 文件。 + if parsed.path == "/app.css": + return self.serve_static("app.css", "text/css; charset=utf-8", send_body=send_body) + + # 前端业务 JS。 + if parsed.path == "/app.js": + return self.serve_static("app.js", "application/javascript; charset=utf-8", send_body=send_body) + + # 本地 Vue 运行时。 + 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": + return self.send_payload( + HTTPStatus.OK, + response_json(build_monitor_payload()), + "application/json; charset=utf-8", + send_body=send_body, + ) + + # favicon 直接返回空。 + if parsed.path == "/favicon.ico": + return self.send_payload(HTTPStatus.NO_CONTENT, b"", "image/x-icon", send_body=send_body) + + # 其他路径统一 404。 + 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 + + # 文件不存在时返回 JSON 错误。 + 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: + # 错误响应也统一走 JSON。 + 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") + # 写入长度，便于 HEAD 请求工作正常。 + self.send_header("Content-Length", str(len(body))) + # 结束响应头。 + self.end_headers() + + # 需要响应体时再写内容。 + if send_body and body: + self.wfile.write(body) + + +def main() -> None: + # 创建线程型 HTTP 服务。 + server = ThreadingHTTPServer((APP_BIND, APP_PORT), MonitorHandler) + # 打印监听地址便于排查。 + print(f"hy-app-monitor listening on http://{APP_BIND}:{APP_PORT}", flush=True) + # 进入永久监听。 + server.serve_forever() diff --git a/monitor/http_probe.py b/monitor/http_probe.py new file mode 100644 index 0000000..633e64c --- /dev/null +++ b/monitor/http_probe.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +# 并发探测服务使用线程池即可。 +import concurrent.futures +# JSON 只用于类型结构兼容和调试。 +from typing import Any +# urllib 直接探测内网 HTTP 健康接口。 +import urllib.error +import urllib.request +# 计时用于计算耗时。 +import time + +# 读取服务配置和运行参数。 +from .cache import TTLCache +from .config import PROBE_CACHE_TTL_SECONDS, PROBE_MAX_WORKERS, PROBE_TIMEOUT_SECONDS, flatten_services, utc_now + + +# 服务探测使用独立缓存，避免页面轮询直接把内网接口打爆。 +SERVICE_CACHE = TTLCache[dict[str, Any]](PROBE_CACHE_TTL_SECONDS) + + +def healthy_from_body(service_kind: str, status_code: int, body_text: str) -> bool: + # 先把响应体转成小写，后续判断更简单。 + lowered = body_text.lower() + + # Golang 服务按现有 /health 的返回结构判断。 + if service_kind == "golang": + return status_code == 200 and ( + '"code":"ok"' in lowered + or '"code": "ok"' in lowered + or '"status":"ok"' in lowered + or '"status": "ok"' in lowered + ) + + # 某些 Java 健康接口虽然返回 401/403，但说明进程本身活着。 + if status_code in {401, 403}: + return True + + # 除此之外，HTTP 非 200 直接判失败。 + if status_code != 200: + return False + + # Spring Boot actuator 常见格式。 + 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 classify_latency(latency_ms: int, ok: bool, thresholds: dict[str, float | int]) -> str: + # 服务本身不健康时直接标成 bad。 + if not ok: + return "bad" + + # 拿到 warn 阈值。 + warn = float(thresholds["warn"]) + # 拿到 bad 阈值。 + bad = float(thresholds["bad"]) + + # 超过 bad 阈值记为 bad。 + if latency_ms >= bad: + return "bad" + + # 超过 warn 阈值记为 warn。 + if latency_ms >= warn: + return "warn" + + # 其余都记为 ok。 + return "ok" + + +def probe_service(service: dict[str, Any], latency_thresholds: dict[str, float | int]) -> dict[str, Any]: + # 拼出健康检查地址。 + url = f"http://{service['ip']}:{service['port']}{service['path']}" + # 记录开始时间。 + started = time.perf_counter() + # 构造 HTTP 请求。 + request = urllib.request.Request(url, headers={"User-Agent": "hy-app-monitor/1.0"}) + + # 准备统一的基础返回结构。 + base = { + **service, + "url": url, + "statusCode": 0, + "latencyMs": 0, + "ok": False, + "level": "bad", + "detail": "", + } + + try: + # 发起 HTTP 探测。 + 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(service["kind"], response.status, body_text) + # 计算当前服务的颜色级别。 + level = classify_latency(latency_ms, ok, latency_thresholds) + + # 返回成功探测结果。 + return { + **base, + "statusCode": int(response.status), + "latencyMs": latency_ms, + "ok": ok, + "level": level, + "detail": body_text[:220].replace("\n", " ").strip(), + } + except urllib.error.HTTPError as exc: + # HTTPError 里仍然尽量读取响应体。 + body_text = exc.read().decode("utf-8", errors="replace") + # 计算失败请求耗时。 + latency_ms = int((time.perf_counter() - started) * 1000) + # 继续按状态码和响应体判断服务活性。 + ok = healthy_from_body(service["kind"], exc.code, body_text) + # 按最终健康度和耗时标记级别。 + level = classify_latency(latency_ms, ok, latency_thresholds) + + # 返回 HTTP 错误场景结果。 + return { + **base, + "statusCode": int(exc.code), + "latencyMs": latency_ms, + "ok": ok, + "level": level, + "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_service_payload(hosts: list[dict[str, Any]], latency_thresholds: dict[str, float | int]) -> dict[str, Any]: + # 先把 hosts 展开成 services。 + services = flatten_services(hosts) + # 服务数量可能为 0，这里仍然保证线程池至少 1 个 worker。 + workers = max(1, min(PROBE_MAX_WORKERS, len(services) or 1)) + + # 并发探测所有服务。 + with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor: + items = list(executor.map(lambda service: probe_service(service, latency_thresholds), services)) + + # 统计健康服务数量。 + ok_count = sum(1 for item in items if item["ok"]) + + # 返回服务维度监控结果。 + return { + "updatedAt": utc_now(), + "summary": { + "total": len(items), + "ok": ok_count, + "down": len(items) - ok_count, + "hostTotal": len(hosts), + }, + "items": items, + } + + +def get_service_payload(hosts: list[dict[str, Any]], latency_thresholds: dict[str, float | int]) -> dict[str, Any]: + # 通过缓存保护服务探测。 + return SERVICE_CACHE.get_or_build(lambda: build_service_payload(hosts, latency_thresholds)) diff --git a/monitor/mongo_metrics.py b/monitor/mongo_metrics.py new file mode 100644 index 0000000..2d6edba --- /dev/null +++ b/monitor/mongo_metrics.py @@ -0,0 +1,357 @@ +from __future__ import annotations + +# datetime 用于复制集 optime 时间格式化。 +from datetime import datetime, timezone +# 类型提示只用于表达返回结构。 +from typing import Any + +# 读取 Mongo 配置。 +from .cache import TTLCache +from .config import ( + MONGO_CACHE_TTL_SECONDS, + MONGO_CONNECT_TIMEOUT_MS, + MONGO_DB_NAME, + MONGO_SERVER_SELECTION_TIMEOUT_MS, + MONGO_SOCKET_TIMEOUT_MS, + MONGO_URI, + utc_now, +) + + +# Mongo 结果缓存。 +MONGO_CACHE = TTLCache[dict[str, Any]](MONGO_CACHE_TTL_SECONDS) + + +def mongo_error_payload(error: str) -> dict[str, Any]: + # 统一返回 Mongo 错误结构。 + return { + "ok": False, + "updatedAt": utc_now(), + "summary": { + "databaseTotal": 0, + "collectionTotal": 0, + "objectTotal": 0, + "storageSizeMb": 0.0, + "dataSizeMb": 0.0, + "connectionCurrent": 0, + "connectionAvailable": 0, + }, + "info": {}, + "connections": {}, + "memory": {}, + "network": {}, + "opcounters": {}, + "wiredTigerCache": {}, + "replicaSet": {"configured": False, "members": []}, + "databases": [], + "error": error, + } + + +def iso_datetime(value: Any) -> str: + # datetime 对象转成 UTC ISO 字符串。 + if isinstance(value, datetime): + return value.astimezone(timezone.utc).replace(microsecond=0).isoformat() + + # 其余情况直接转字符串。 + return str(value or "") + + +def build_mongo_client() -> tuple[Any | None, str]: + # 没配置 URI 时直接报错。 + if not MONGO_URI: + return None, "missing MONGO_URI in .env" + + try: + # 运行时导入 PyMongo，避免缺依赖时整个服务起不来。 + from pymongo import MongoClient + except Exception as exc: # noqa: BLE001 + # 依赖不存在时返回错误文本。 + return None, f"pymongo unavailable: {exc}" + + # 创建 MongoClient。 + client = MongoClient( + MONGO_URI, + appname="hy-app-monitor", + connectTimeoutMS=MONGO_CONNECT_TIMEOUT_MS, + serverSelectionTimeoutMS=MONGO_SERVER_SELECTION_TIMEOUT_MS, + socketTimeoutMS=MONGO_SOCKET_TIMEOUT_MS, + ) + + # 返回连接对象。 + return client, "" + + +def safe_get(mapping: dict[str, Any], key: str, default: float = 0.0) -> float: + # 读取数值型字段。 + value = mapping.get(key, default) + + try: + # 尽量转成 float 返回。 + return float(value) + except Exception: # noqa: BLE001 + # 转换失败时回退默认值。 + return default + + +def build_replica_payload(client: Any) -> dict[str, Any]: + try: + # 读取复制集状态。 + payload = client.admin.command("replSetGetStatus") + except Exception as exc: # noqa: BLE001 + # 未配置复制集或权限不足时给出降级信息。 + return { + "configured": False, + "error": str(exc), + "name": "", + "members": [], + "summary": {"memberTotal": 0, "healthyMemberTotal": 0, "primary": ""}, + } + + # 原始成员列表。 + members = payload.get("members") or [] + # 尝试定位 primary 成员。 + primary_member = next((item for item in members if str(item.get("stateStr") or "") == "PRIMARY"), None) + # 拿到 primary optime 时间。 + primary_optime = primary_member.get("optimeDate") if primary_member else None + # 归一化后的成员列表。 + normalized_members: list[dict[str, Any]] = [] + # 健康成员数量。 + healthy_member_total = 0 + + # 逐个成员整理结构。 + for item in members: + # 当前成员 optime 时间。 + optime_date = item.get("optimeDate") + # 默认 lag 为空。 + lag_seconds: int | None = None + + # 主节点和当前成员都有 optime 时间时计算 lag。 + if isinstance(primary_optime, datetime) and isinstance(optime_date, datetime): + lag_seconds = max(int((primary_optime - optime_date).total_seconds()), 0) + + # 成员 health 为 1 视为健康。 + healthy = float(item.get("health") or 0) >= 1.0 + + # 统计健康成员数。 + if healthy: + healthy_member_total += 1 + + # 追加成员结构。 + normalized_members.append( + { + "name": str(item.get("name") or "").strip(), + "state": int(item.get("state") or 0), + "stateStr": str(item.get("stateStr") or "").strip(), + "health": float(item.get("health") or 0.0), + "uptime": int(item.get("uptime") or 0), + "optimeDate": iso_datetime(optime_date), + "lagSeconds": lag_seconds, + "self": bool(item.get("self")), + } + ) + + # 返回复制集结构。 + return { + "configured": True, + "error": "", + "name": str(payload.get("set") or "").strip(), + "members": normalized_members, + "summary": { + "memberTotal": len(normalized_members), + "healthyMemberTotal": healthy_member_total, + "primary": str(primary_member.get("name") or "").strip() if primary_member else "", + }, + } + + +def build_database_payloads(client: Any, databases: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], dict[str, Any]]: + # 所有数据库的详情列表。 + payloads: list[dict[str, Any]] = [] + # 汇总统计。 + summary = { + "databaseTotal": 0, + "collectionTotal": 0, + "objectTotal": 0, + "storageSizeMb": 0.0, + "dataSizeMb": 0.0, + } + + # 逐个数据库补充 dbStats。 + for item in databases: + # 读取数据库名。 + name = str(item.get("name") or "").strip() + + # 没名字时跳过。 + if not name: + continue + + try: + # 获取 dbStats，并把 scale 设成 1，方便后端自己换算。 + stats = client[name].command("dbStats", 1, scale=1) + except Exception as exc: # noqa: BLE001 + # 获取失败时仍保留基础信息。 + payloads.append( + { + "name": name, + "empty": bool(item.get("empty")), + "sizeOnDiskMb": round(safe_get(item, "sizeOnDisk") / 1024 / 1024, 2), + "error": str(exc), + } + ) + continue + + # 读取 collection 总数。 + collections = int(stats.get("collections") or 0) + # 读取 object 总数。 + objects = int(stats.get("objects") or 0) + # 读取 dataSize。 + data_size_mb = round(safe_get(stats, "dataSize") / 1024 / 1024, 2) + # 读取 storageSize。 + storage_size_mb = round(safe_get(stats, "storageSize") / 1024 / 1024, 2) + # 读取 index 数。 + indexes = int(stats.get("indexes") or 0) + # 读取 indexSize。 + index_size_mb = round(safe_get(stats, "indexSize") / 1024 / 1024, 2) + # 读取 avgObjSize。 + avg_obj_size_b = round(safe_get(stats, "avgObjSize"), 2) + + # 汇总数据库级统计。 + summary["databaseTotal"] += 1 + summary["collectionTotal"] += collections + summary["objectTotal"] += objects + summary["storageSizeMb"] += storage_size_mb + summary["dataSizeMb"] += data_size_mb + + # 追加当前数据库详情。 + payloads.append( + { + "name": name, + "empty": bool(item.get("empty")), + "collections": collections, + "views": int(stats.get("views") or 0), + "objects": objects, + "dataSizeMb": data_size_mb, + "storageSizeMb": storage_size_mb, + "indexCount": indexes, + "indexSizeMb": index_size_mb, + "avgObjSizeBytes": avg_obj_size_b, + "sizeOnDiskMb": round(safe_get(item, "sizeOnDisk") / 1024 / 1024, 2), + } + ) + + # 数据库列表按 storageSize 倒序显示。 + payloads.sort(key=lambda item: float(item.get("storageSizeMb") or 0.0), reverse=True) + + # 汇总值保留两位小数。 + summary["storageSizeMb"] = round(summary["storageSizeMb"], 2) + summary["dataSizeMb"] = round(summary["dataSizeMb"], 2) + + # 返回数据库列表和汇总信息。 + return payloads, summary + + +def build_mongo_payload() -> dict[str, Any]: + # 先构造客户端。 + client, error = build_mongo_client() + + # 客户端不可用时直接返回错误结构。 + if client is None: + return mongo_error_payload(error) + + try: + # 先做一次 ping，保证连接可用。 + client.admin.command("ping") + # 读取 serverStatus，全局指标都在这里。 + server_status = client.admin.command("serverStatus", recordStats=0) + # 读取数据库列表。 + database_list = client.admin.command("listDatabases", 1) + # 读取复制集状态。 + replica_payload = build_replica_payload(client) + + # 拿到数据库项数组。 + database_items = list(database_list.get("databases") or []) + # 构建数据库详情和汇总。 + database_payloads, database_summary = build_database_payloads(client, database_items) + + # 读取连接数信息。 + connections = server_status.get("connections") or {} + # 读取内存信息。 + memory = server_status.get("mem") or {} + # 读取网络信息。 + network = server_status.get("network") or {} + # 读取操作计数器。 + opcounters = server_status.get("opcounters") or {} + # 读取 WiredTiger 指标。 + wired_tiger = server_status.get("wiredTiger") or {} + # 读取 cache 指标。 + wired_tiger_cache = wired_tiger.get("cache") or {} + + # 组装最终返回结构。 + return { + "ok": True, + "updatedAt": utc_now(), + "summary": { + **database_summary, + "connectionCurrent": int(connections.get("current") or 0), + "connectionAvailable": int(connections.get("available") or 0), + }, + "info": { + "host": str(server_status.get("host") or "").strip(), + "version": str(server_status.get("version") or "").strip(), + "process": str(server_status.get("process") or "").strip(), + "uptimeSeconds": int(server_status.get("uptime") or 0), + "localTime": iso_datetime(server_status.get("localTime")), + "focusedDatabase": MONGO_DB_NAME, + }, + "connections": { + "current": int(connections.get("current") or 0), + "available": int(connections.get("available") or 0), + "totalCreated": int(connections.get("totalCreated") or 0), + "active": int(connections.get("active") or 0), + "threaded": int(connections.get("threaded") or 0), + }, + "memory": { + "residentMb": safe_get(memory, "resident"), + "virtualMb": safe_get(memory, "virtual"), + "mappedMb": safe_get(memory, "mapped"), + "mappedWithJournalMb": safe_get(memory, "mappedWithJournal"), + "bits": int(memory.get("bits") or 0), + }, + "network": { + "bytesIn": int(network.get("bytesIn") or 0), + "bytesOut": int(network.get("bytesOut") or 0), + "numRequests": int(network.get("numRequests") or 0), + }, + "opcounters": { + "insert": int(opcounters.get("insert") or 0), + "query": int(opcounters.get("query") or 0), + "update": int(opcounters.get("update") or 0), + "delete": int(opcounters.get("delete") or 0), + "getmore": int(opcounters.get("getmore") or 0), + "command": int(opcounters.get("command") or 0), + }, + "wiredTigerCache": { + "bytesCurrentlyInCache": int(wired_tiger_cache.get("bytes currently in the cache") or 0), + "trackedDirtyBytesInCache": int( + wired_tiger_cache.get("tracked dirty bytes in the cache") or 0 + ), + "maximumBytesConfigured": int(wired_tiger_cache.get("maximum bytes configured") or 0), + "pagesReadIntoCache": int(wired_tiger_cache.get("pages read into cache") or 0), + "pagesWrittenFromCache": int(wired_tiger_cache.get("pages written from cache") or 0), + }, + "replicaSet": replica_payload, + "databases": database_payloads, + } + except Exception as exc: # noqa: BLE001 + # 任意异常都降级成统一错误结构。 + return mongo_error_payload(str(exc)) + finally: + # 连接用完就关闭。 + client.close() + + +def get_mongo_payload() -> dict[str, Any]: + # 使用缓存保护 Mongo 采集。 + return MONGO_CACHE.get_or_build(build_mongo_payload) + diff --git a/monitor/nacos.py b/monitor/nacos.py new file mode 100644 index 0000000..f317a2e --- /dev/null +++ b/monitor/nacos.py @@ -0,0 +1,461 @@ +from __future__ import annotations + +# JSON 用来解析 Nacos 响应。 +import json +# 线程锁用于保护登录 token 缓存。 +import threading +# token 过期和缓存使用时间模块。 +import time +# urllib 负责直接请求 Nacos OpenAPI。 +import urllib.error +import urllib.parse +import urllib.request +# 类型提示只用于表达返回结构。 +from typing import Any + +# 读取 Nacos 配置。 +from .cache import TTLCache +from .config import ( + NACOS_BASE_URL, + NACOS_CACHE_TTL_SECONDS, + NACOS_DISCOVERY_CLUSTER_NAME, + NACOS_DISCOVERY_GROUP, + NACOS_NAMESPACE_ID, + NACOS_NAMESPACE_NAME, + NACOS_PAGE_SIZE, + NACOS_PASSWORD, + NACOS_TIMEOUT_SECONDS, + NACOS_USERNAME, + utc_now, +) + + +# Nacos 监控结果缓存。 +NACOS_CACHE = TTLCache[dict[str, Any]](NACOS_CACHE_TTL_SECONDS) +# Nacos access token 缓存锁。 +TOKEN_LOCK = threading.Lock() +# Nacos access token 文本。 +TOKEN_VALUE = "" +# Nacos access token 失效时间。 +TOKEN_EXPIRES_AT = 0.0 + + +def nacos_truthy(value: Any) -> bool: + # 布尔值直接返回。 + if isinstance(value, bool): + return value + + # 字符串则按 true/false 判断。 + if isinstance(value, str): + return value.strip().lower() == "true" + + # 其余值按 Python truthy 规则处理。 + return bool(value) + + +def parse_json_payload(payload: str) -> dict[str, Any]: + # 空响应视为空对象。 + if not payload.strip(): + return {} + + # 尝试按 JSON 解析。 + try: + return json.loads(payload) + except json.JSONDecodeError: + # 非 JSON 文本也保留 raw 字段，方便排查。 + return {"raw": payload.strip()} + + +def request_url(path: str, params: dict[str, Any] | None = None) -> str: + # 先拼基础 URL。 + base = f"{NACOS_BASE_URL}{path}" + + # 没 query 参数时直接返回。 + if not params: + return base + + # 去掉值为 None 的参数。 + cleaned = {key: value for key, value in params.items() if value is not None} + # 编码 query string。 + query = urllib.parse.urlencode(cleaned, quote_via=urllib.parse.quote) + + # 拼回完整 URL。 + return f"{base}?{query}" if query else base + + +def request_text( + method: str, + path: str, + *, + params: dict[str, Any] | None = None, + data: dict[str, Any] | None = None, +) -> str: + # 先构造 URL。 + url = request_url(path, params) + # 默认没有请求体。 + body: bytes | None = None + + # 有请求体时按 form-urlencoded 提交。 + if data is not None: + body = urllib.parse.urlencode(data, quote_via=urllib.parse.quote).encode("utf-8") + + # 构造 HTTP 请求对象。 + request = urllib.request.Request(url, data=body, method=method, headers={"User-Agent": "hy-app-monitor/1.0"}) + + try: + # 发起请求并读取文本响应。 + with urllib.request.urlopen(request, timeout=NACOS_TIMEOUT_SECONDS) as response: + return response.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as exc: + # HTTP 错误时尽量把响应体保留下来。 + detail = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"nacos {method} {path} failed: {exc.code} {detail}") from exc + except urllib.error.URLError as exc: + # 网络错误也转成统一异常。 + raise RuntimeError(f"nacos {method} {path} failed: {exc}") from exc + + +def request_json( + method: str, + path: str, + *, + params: dict[str, Any] | None = None, + data: dict[str, Any] | None = None, +) -> dict[str, Any]: + # 读取文本响应。 + payload = request_text(method, path, params=params, data=data) + # 解析成 JSON 结构。 + return parse_json_payload(payload) + + +def fetch_access_token() -> str: + # 没配置用户名密码时直接不登录。 + if not NACOS_USERNAME or not NACOS_PASSWORD: + return "" + + # 优先复用缓存 token。 + global TOKEN_VALUE, TOKEN_EXPIRES_AT + + # 先在锁内检查 token 是否还有效。 + with TOKEN_LOCK: + if TOKEN_VALUE and time.time() < TOKEN_EXPIRES_AT: + return TOKEN_VALUE + + # 依次尝试常见的两个登录端点。 + for path in ("/nacos/v1/auth/users/login", "/nacos/v1/auth/login"): + try: + # 发起登录请求。 + payload = request_json("POST", path, data={"username": NACOS_USERNAME, "password": NACOS_PASSWORD}) + except RuntimeError: + # 这个端点失败时继续尝试下一个。 + continue + + # 读取 accessToken。 + token = str(payload.get("accessToken") or "").strip() + + # 没拿到 token 时继续尝试。 + if not token: + continue + + # 读取 token TTL，单位秒。 + token_ttl = int(payload.get("tokenTtl") or 18000) + + # 在锁内写回缓存。 + with TOKEN_LOCK: + TOKEN_VALUE = token + TOKEN_EXPIRES_AT = time.time() + max(token_ttl - 30, 30) + + # 返回有效 token。 + return token + + # 所有端点都没拿到 token 时返回空串。 + return "" + + +def nacos_request(method: str, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: + # 复制一份参数，避免污染外部调用方。 + merged_params = dict(params or {}) + # 如果登录拿到了 token，就把 accessToken 带上。 + access_token = fetch_access_token() + if access_token: + merged_params["accessToken"] = access_token + + # 发起请求并解析 JSON。 + return request_json(method, path, params=merged_params) + + +def list_namespaces() -> list[dict[str, Any]]: + # 读取 Nacos 命名空间列表。 + payload = nacos_request("GET", "/nacos/v1/console/namespaces") + # 兼容 data 字段和空列表。 + return list(payload.get("data") or []) + + +def resolve_namespace(namespaces: list[dict[str, Any]]) -> tuple[str, str]: + # 优先按显式 namespace id 匹配。 + if NACOS_NAMESPACE_ID: + for item in namespaces: + if str(item.get("namespace") or item.get("namespaceId") or "").strip() == NACOS_NAMESPACE_ID: + return NACOS_NAMESPACE_ID, str(item.get("namespaceShowName") or item.get("namespace") or "").strip() + + # 其次按命名空间名称匹配。 + if NACOS_NAMESPACE_NAME: + for item in namespaces: + show_name = str(item.get("namespaceShowName") or "").strip() + if show_name == NACOS_NAMESPACE_NAME: + return str(item.get("namespace") or item.get("namespaceId") or "").strip(), show_name + + # 再退回配置值。 + return NACOS_NAMESPACE_ID, NACOS_NAMESPACE_NAME + + +def split_grouped_service_name(raw_name: str) -> tuple[str, str]: + # service list 可能返回 DEFAULT_GROUP@@serviceName。 + if "@@" in raw_name: + group_name, service_name = raw_name.split("@@", 1) + return group_name, service_name + + # 没有分组前缀时使用默认分组。 + return NACOS_DISCOVERY_GROUP, raw_name + + +def list_services(namespace_id: str) -> list[str]: + # 这里逐页拉取 service list。 + doms: list[str] = [] + # 初始页码从 1 开始。 + page_no = 1 + + while True: + # 请求当前页服务列表。 + payload = nacos_request( + "GET", + "/nacos/v1/ns/service/list", + { + "pageNo": str(page_no), + "pageSize": str(NACOS_PAGE_SIZE), + "namespaceId": namespace_id, + }, + ) + + # 读取当前页服务名列表。 + page_items = list(payload.get("doms") or []) + + # 没有数据时结束。 + if not page_items: + break + + # 追加本页数据。 + doms.extend(str(item).strip() for item in page_items if str(item).strip()) + + # 读取总数。 + total = int(payload.get("count") or len(doms)) + + # 收齐所有服务后结束。 + if len(doms) >= total: + break + + # 继续下一页。 + page_no += 1 + + # 去重并排序后返回。 + return sorted(dict.fromkeys(doms)) + + +def list_instances(namespace_id: str, service_name: str, group_name: str) -> list[dict[str, Any]]: + # 调用 instance list 拉取服务实例。 + payload = nacos_request( + "GET", + "/nacos/v1/ns/instance/list", + { + "serviceName": service_name, + "groupName": group_name, + "namespaceId": namespace_id, + "healthyOnly": "false", + }, + ) + + # 返回实例数组。 + return list(payload.get("hosts") or []) + + +def build_service_level(instance_total: int, healthy_total: int, enabled_total: int) -> str: + # 没有实例时直接标 bad。 + if instance_total == 0: + return "bad" + + # 没有可用健康实例时也标 bad。 + if healthy_total == 0 or enabled_total == 0: + return "bad" + + # 不是全健康时标 warn。 + if healthy_total < instance_total or enabled_total < instance_total: + return "warn" + + # 其余标 ok。 + return "ok" + + +def build_nacos_payload() -> dict[str, Any]: + # 先取命名空间列表。 + namespaces = list_namespaces() + # 解析目标 namespace。 + namespace_id, namespace_name = resolve_namespace(namespaces) + # 拉取目标 namespace 下所有服务。 + raw_services = list_services(namespace_id) + + # 归一化后的服务列表。 + services: list[dict[str, Any]] = [] + # 统计实例总量。 + instance_total = 0 + # 统计健康实例总量。 + healthy_instance_total = 0 + # 统计启用实例总量。 + enabled_instance_total = 0 + # 统计全健康服务数量。 + healthy_service_total = 0 + + # 逐个服务拉实例详情。 + for raw_name in raw_services: + # 解析 group 和真正的 serviceName。 + group_name, service_name = split_grouped_service_name(raw_name) + # 拉取当前服务实例。 + instance_items = list_instances(namespace_id, service_name, group_name) + # 归一化实例结构。 + normalized_instances: list[dict[str, Any]] = [] + + # 当前服务健康实例数。 + healthy_total = 0 + # 当前服务启用实例数。 + enabled_total = 0 + + # 逐个实例整理字段。 + for item in instance_items: + # 读取 healthy 标记。 + healthy = nacos_truthy(item.get("healthy", False)) + # 读取 enabled 标记。 + enabled = nacos_truthy(item.get("enabled", True)) + + # 统计健康实例数。 + if healthy: + healthy_total += 1 + + # 统计启用实例数。 + if enabled: + enabled_total += 1 + + # 追加归一化实例结构。 + normalized_instances.append( + { + "ip": str(item.get("ip") or "").strip(), + "port": int(item.get("port") or 0), + "healthy": healthy, + "enabled": enabled, + "ephemeral": nacos_truthy(item.get("ephemeral", True)), + "weight": float(item.get("weight") or 0.0), + "clusterName": str(item.get("clusterName") or NACOS_DISCOVERY_CLUSTER_NAME).strip(), + "metadata": item.get("metadata") or {}, + } + ) + + # 当前服务实例总数。 + current_total = len(normalized_instances) + # 汇总到全局实例总数。 + instance_total += current_total + # 汇总到全局健康实例总数。 + healthy_instance_total += healthy_total + # 汇总到全局启用实例总数。 + enabled_instance_total += enabled_total + + # 计算当前服务级别。 + level = build_service_level(current_total, healthy_total, enabled_total) + + # 全健康服务才记入 healthy_service_total。 + if level == "ok": + healthy_service_total += 1 + + # 追加当前服务结构。 + services.append( + { + "rawServiceName": raw_name, + "serviceName": service_name, + "groupName": group_name, + "level": level, + "instanceTotal": current_total, + "healthyTotal": healthy_total, + "unhealthyTotal": current_total - healthy_total, + "enabledTotal": enabled_total, + "disabledTotal": current_total - enabled_total, + "instances": normalized_instances, + } + ) + + # 命名空间列表也做一层归一化。 + normalized_namespaces = [ + { + "namespaceId": str(item.get("namespace") or item.get("namespaceId") or "").strip(), + "name": str(item.get("namespaceShowName") or item.get("namespace") or "").strip(), + "configCount": int(item.get("configCount") or 0), + "type": int(item.get("type") or 0), + } + for item in namespaces + ] + + # 返回完整 Nacos 监控结构。 + return { + "ok": True, + "updatedAt": utc_now(), + "baseUrl": NACOS_BASE_URL, + "namespaceId": namespace_id, + "namespaceName": namespace_name, + "summary": { + "namespaceTotal": len(normalized_namespaces), + "serviceTotal": len(services), + "healthyServiceTotal": healthy_service_total, + "degradedServiceTotal": len(services) - healthy_service_total, + "instanceTotal": instance_total, + "healthyInstanceTotal": healthy_instance_total, + "unhealthyInstanceTotal": instance_total - healthy_instance_total, + "enabledInstanceTotal": enabled_instance_total, + "disabledInstanceTotal": instance_total - enabled_instance_total, + }, + "namespaces": normalized_namespaces, + "services": services, + } + + +def nacos_error_payload(error: str) -> dict[str, Any]: + # 返回统一错误结构，前端可以直接消费。 + return { + "ok": False, + "updatedAt": utc_now(), + "baseUrl": NACOS_BASE_URL, + "namespaceId": NACOS_NAMESPACE_ID, + "namespaceName": NACOS_NAMESPACE_NAME, + "summary": { + "namespaceTotal": 0, + "serviceTotal": 0, + "healthyServiceTotal": 0, + "degradedServiceTotal": 0, + "instanceTotal": 0, + "healthyInstanceTotal": 0, + "unhealthyInstanceTotal": 0, + "enabledInstanceTotal": 0, + "disabledInstanceTotal": 0, + }, + "namespaces": [], + "services": [], + "error": error, + } + + +def get_nacos_payload() -> dict[str, Any]: + # 使用缓存保护 Nacos API。 + def builder() -> dict[str, Any]: + try: + return build_nacos_payload() + except Exception as exc: # noqa: BLE001 + return nacos_error_payload(str(exc)) + + # 返回缓存或最新构建结果。 + return NACOS_CACHE.get_or_build(builder) + diff --git a/monitor/tencent_tat.py b/monitor/tencent_tat.py new file mode 100644 index 0000000..2fc1a38 --- /dev/null +++ b/monitor/tencent_tat.py @@ -0,0 +1,199 @@ +from __future__ import annotations + +# base64 用来传输 shell 脚本和结果。 +import base64 +# JSON 用来构造和解析腾讯云 SDK 请求。 +import json +# 轮询结果需要 sleep 和超时判断。 +import time +# 类型提示只用于表达结构。 +from typing import Any + +# 读取 TAT 连接配置。 +from .config import DEPLOY_REGION, TAT_METRIC_TIMEOUT_SECONDS, TENCENT_SECRET_ID, TENCENT_SECRET_KEY, utc_now + + +def decode_task_output(output: str) -> str: + # 没有输出时直接返回空串。 + if not output: + return "" + + # 把 TAT 返回的 base64 文本解码成普通字符串。 + return base64.b64decode(output).decode("utf-8", errors="replace") + + +def build_tat_client() -> tuple[Any | None, str]: + # TAT 运行需要完整的腾讯云密钥和区域。 + if not TENCENT_SECRET_ID or not TENCENT_SECRET_KEY or not DEPLOY_REGION: + return None, "missing TAT credentials in .env" + + try: + # 运行时才导入 SDK，避免没装依赖时整个服务启动失败。 + from tencentcloud.common import credential + # 腾讯云异常类型用于重试。 + from tencentcloud.common.profile.client_profile import ClientProfile + # 指定 TAT API endpoint。 + from tencentcloud.common.profile.http_profile import HttpProfile + # TAT 客户端。 + from tencentcloud.tat.v20201028 import tat_client + except Exception as exc: # noqa: BLE001 + # SDK 不存在时直接返回错误。 + return None, f"tencentcloud sdk unavailable: {exc}" + + # 创建凭据对象。 + cred = credential.Credential(TENCENT_SECRET_ID, TENCENT_SECRET_KEY) + + # 创建 TAT 客户端。 + client = tat_client.TatClient( + cred, + DEPLOY_REGION, + ClientProfile(httpProfile=HttpProfile(endpoint="tat.tencentcloudapi.com")), + ) + + # 成功时返回客户端。 + return client, "" + + +def invoke_tat(action: Any, label: str, retries: int = 3, delay_seconds: int = 2) -> Any: + # 这里延迟导入异常类型，避免 SDK 缺失时模块导入失败。 + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + + # 记录最后一次异常。 + last_error: Exception | None = None + + # 简单重试几次，防止瞬时 API 抖动。 + for attempt in range(1, retries + 1): + try: + # 成功时直接返回结果。 + return action() + except TencentCloudSDKException as exc: + # 保存最后一次异常。 + last_error = exc + + # 最后一次重试失败时不再等待。 + if attempt == retries: + break + + # 做一个线性退避。 + time.sleep(delay_seconds * attempt) + + # 有异常时抛出最后一次异常。 + if last_error is not None: + raise last_error + + # 理论上不会走到这里，这里只是兜底。 + raise RuntimeError(f"{label} failed without error") + + +def run_shell_command( + instance_ids: list[str], + script: str, + *, + command_name: str, + timeout_seconds: int = TAT_METRIC_TIMEOUT_SECONDS, + working_directory: str = "/root", +) -> dict[str, dict[str, Any]]: + # 先初始化 TAT 客户端。 + client, error = build_tat_client() + + # 客户端不可用时直接抛错。 + if client is None: + raise RuntimeError(error) + + # 这里延迟导入 TAT models。 + from tencentcloud.tat.v20201028 import models as tat_models + + # 创建命令执行请求。 + req = tat_models.RunCommandRequest() + + # 组装批量执行命令。 + req.from_json_string( + json.dumps( + { + "CommandName": command_name, + "CommandType": "SHELL", + "Content": base64.b64encode(script.encode("utf-8")).decode("ascii"), + "InstanceIds": instance_ids, + "Timeout": timeout_seconds, + "WorkingDirectory": working_directory, + } + ) + ) + + # 发起远端命令执行。 + response = json.loads(invoke_tat(lambda: client.RunCommand(req), "RunCommand").to_json_string()) + # 拿到 invocation id，后面轮询执行结果。 + invocation_id = response["InvocationId"] + # 设置轮询截止时间。 + deadline = time.time() + timeout_seconds + 60 + # 这里保存各实例的最终结果。 + results: dict[str, dict[str, Any]] = {} + + # 开始轮询任务结果。 + while time.time() < deadline: + # 轮询间隔不必太短。 + time.sleep(2) + + # 创建查询结果请求。 + query = tat_models.DescribeInvocationTasksRequest() + # 按 invocation id 查询。 + query.from_json_string( + json.dumps( + { + "HideOutput": False, + "Limit": 50, + "Filters": [{"Name": "invocation-id", "Values": [invocation_id]}], + } + ) + ) + + # 拉取当前命令所有实例的执行状态。 + payload = json.loads( + invoke_tat(lambda: client.DescribeInvocationTasks(query), "DescribeInvocationTasks").to_json_string() + ) + # 任务列表可能暂时为空。 + tasks = payload.get("InvocationTaskSet") or [] + + # 没任务时继续等待。 + if not tasks: + continue + + # 遍历当前拿到的实例结果。 + for task in tasks: + # 读取实例 ID。 + instance_id = task.get("InstanceId", "") + # 读取任务状态。 + status = task.get("TaskStatus", "") + + # 只处理终态。 + if status not in {"SUCCESS", "FAILED", "TIMEOUT", "CANCELLED", "START_FAILED"}: + continue + + # 解码远端输出。 + output = decode_task_output(task.get("TaskResult", {}).get("Output", "")) + + # 统一存成结构化结果。 + results[instance_id] = { + "status": status, + "output": output, + "updatedAt": utc_now(), + } + + # 收齐全部实例结果后结束轮询。 + if len(results) >= len(instance_ids): + break + + # 为缺失结果的实例补超时占位。 + for instance_id in instance_ids: + results.setdefault( + instance_id, + { + "status": "TIMEOUT", + "output": "", + "updatedAt": utc_now(), + }, + ) + + # 返回按实例 ID 索引的结果。 + return results + diff --git a/requirements-ops.txt b/requirements-ops.txt index 938940a..1ec6d22 100644 --- a/requirements-ops.txt +++ b/requirements-ops.txt @@ -1 +1,2 @@ tencentcloud-sdk-python>=3.0.0 +pymongo>=4.8.0 diff --git a/server.py b/server.py index 1ad3d2b..1e0bfc1 100644 --- a/server.py +++ b/server.py @@ -2,851 +2,10 @@ from __future__ import annotations -# 标准库并发工具，用来并行探测各个服务健康状态。 -import concurrent.futures -# 标准 JSON 编解码，用来读取配置和输出接口响应。 -import json -# 进程环境变量读取，用来接收 .env 和 systemd 注入的运行参数。 -import os -# 线程锁用来保护缓存，避免并发请求重复打 TAT 和健康检查。 -import threading -# 时间模块用于缓存过期判断和 CPU 采样间隔。 -import time -# urllib 用于直接探测各台机器上的 HTTP 健康检查接口。 -import urllib.error -import urllib.request -# 时间戳统一输出成 UTC ISO 字符串，前端展示更稳定。 -from datetime import datetime, timezone -# HTTP 状态码常量，避免在代码里写魔法数字。 -from http import HTTPStatus -# 使用标准库 HTTP 服务器承载监控页和 JSON API。 -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -# 路径对象比字符串拼接更安全。 -from pathlib import Path -# 类型提示用于提升代码可读性。 -from typing import Any -# 解析请求路径时只关心 path 段。 -from urllib.parse import urlparse - - -# 项目根目录，后续所有相对路径都从这里展开。 -ROOT = Path(__file__).resolve().parent -# 静态资源目录，存放 HTML、CSS、JS、Vue 文件。 -STATIC_DIR = ROOT / "static" -# 默认 .env 文件放在项目根目录，并且不入库。 -DEFAULT_ENV_PATH = ROOT / ".env" - - -def load_env_file(path: Path) -> None: - # 如果 .env 不存在就直接跳过，允许只靠外部环境变量运行。 - if not path.exists(): - return - - # 逐行读取 .env，使用 shell 兼容的 KEY=VALUE 结构。 - for raw_line in path.read_text(encoding="utf-8").splitlines(): - # 去掉首尾空白，便于判断空行和注释。 - line = raw_line.strip() - # 空行直接跳过。 - if not line: - continue - # 注释行直接跳过。 - if line.startswith("#"): - continue - # 没有等号的无效行直接跳过，避免把异常内容灌进环境变量。 - if "=" not in line: - continue - - # 只按第一个等号拆分，避免 value 里出现额外等号时被截断。 - key, value = line.split("=", 1) - # 去掉 key 的两端空格。 - key = key.strip() - # 去掉 value 的两端空格。 - value = value.strip() - - # 去掉成对的双引号，兼容常见 .env 写法。 - if value.startswith('"') and value.endswith('"'): - value = value[1:-1] - - # 去掉成对的单引号，兼容常见 .env 写法。 - if value.startswith("'") and value.endswith("'"): - value = value[1:-1] - - # 只在环境变量不存在时写入，保证外部显式传入的值优先级更高。 - os.environ.setdefault(key, value) - - -# 先加载项目根目录的 .env，后面常量都基于这个环境初始化。 -load_env_file(Path(os.environ.get("APP_ENV_FILE", DEFAULT_ENV_PATH)).expanduser()) - -# 配置文件路径默认指向 config/targets.json，也允许用环境变量覆盖。 -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" -# 页面服务默认监听 2026，避开浏览器 unsafe port。 -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")) -# 主机指标缓存时间比服务探测更长，因为 TAT 成本更高。 -HOST_METRIC_CACHE_TTL_SECONDS = float(os.environ.get("HOST_METRIC_CACHE_TTL_SECONDS", "25")) -# TAT 执行超时时间，主机采集会做一次 CPU 采样，所以时间略长一点。 -TAT_METRIC_TIMEOUT_SECONDS = int(os.environ.get("TAT_METRIC_TIMEOUT_SECONDS", "30")) -# 采样 CPU 时的两次读数间隔。 -TAT_CPU_SAMPLE_SECONDS = float(os.environ.get("TAT_CPU_SAMPLE_SECONDS", "1")) - -# 服务健康结果缓存锁。 -_SERVICE_CACHE_LOCK = threading.Lock() -# 服务健康结果缓存内容。 -_SERVICE_CACHE_PAYLOAD: dict[str, Any] | None = None -# 服务健康结果最后生成时间。 -_SERVICE_CACHE_AT = 0.0 - -# 主机指标缓存锁。 -_HOST_CACHE_LOCK = threading.Lock() -# 主机指标缓存内容。 -_HOST_CACHE_PAYLOAD: dict[str, dict[str, Any]] | None = None -# 主机指标最后生成时间。 -_HOST_CACHE_AT = 0.0 - - -def utc_now() -> str: - # 统一返回无微秒的 UTC ISO 时间，前后端更容易比较。 - return datetime.now(timezone.utc).replace(microsecond=0).isoformat() - - -def load_hosts() -> list[dict[str, Any]]: - # 读取监控配置文件。 - payload = json.loads(CONFIG_PATH.read_text(encoding="utf-8")) - # hosts 节点不存在时回退成空列表。 - hosts = payload.get("hosts") or [] - # 规范化后的主机列表。 - normalized: list[dict[str, Any]] = [] - - # 逐台主机提取可用字段。 - for host in hosts: - # 统一整理字段，避免前端和后端在不同地方重复判断默认值。 - normalized.append( - { - "host": str(host.get("name") or "").strip(), - "group": str(host.get("group") or "").strip(), - "ip": str(host.get("ip") or "").strip(), - "instanceId": str(host.get("instanceId") or "").strip(), - "services": host.get("services") or [], - } - ) - - # 返回主机列表给后续服务探测和 TAT 指标采集复用。 - return normalized - - -def flatten_services(hosts: list[dict[str, Any]]) -> list[dict[str, Any]]: - # 扁平化后的服务列表，便于并发探测。 - services: list[dict[str, Any]] = [] - - # 按主机展开它上面的所有服务。 - for host in hosts: - # 逐个服务转换为统一结构。 - for service in host.get("services") or []: - services.append( - { - "host": host["host"], - "group": host["group"], - "ip": host["ip"], - "instanceId": host["instanceId"], - "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(), - } - ) - - # 返回扁平服务列表供 HTTP 探测使用。 - return services - - -def healthy_from_body(service_kind: str, status_code: int, body_text: str) -> bool: - # 先把响应体统一转成小写，便于后续模糊判断。 - lowered = body_text.lower() - - # Golang 服务当前按项目里的 /health 返回体做判断。 - if service_kind == "golang": - return status_code == 200 and ( - '"code":"ok"' in lowered - or '"code": "ok"' in lowered - or '"status":"ok"' in lowered - or '"status": "ok"' in lowered - ) - - # 有些 Java 网关健康接口会返回 401/403，但服务本身是活的。 - if status_code in {401, 403}: - return True - - # 其余 Java 服务非 200 一律认为异常。 - if status_code != 200: - return False - - # Spring Boot actuator 常见格式是 status=UP。 - if '"status":"up"' in lowered or '"status": "up"' in lowered: - return True - - # 兼容只返回裸字符串 UP 的老接口。 - if body_text.strip() == "UP": - return True - - # 最后再做一次保守模糊判断。 - return "up" in lowered and "status" in lowered - - -def probe_service(service: dict[str, Any]) -> dict[str, Any]: - # 拼出服务健康检查地址。 - url = f"http://{service['ip']}:{service['port']}{service['path']}" - # 记录开始时间，用来计算耗时。 - started = time.perf_counter() - # 构造请求对象，带一个简单的 User-Agent 便于区分来源。 - request = urllib.request.Request(url, headers={"User-Agent": "hy-app-monitor/1.0"}) - - # 先准备公共返回结构。 - base = { - **service, - "url": url, - "statusCode": 0, - "latencyMs": 0, - "ok": False, - "detail": "", - } - - try: - # 发起 HTTP 请求探测服务。 - with urllib.request.urlopen(request, timeout=PROBE_TIMEOUT_SECONDS) as response: - # 读取响应体，并按 UTF-8 兜底解码。 - body_text = response.read().decode("utf-8", errors="replace") - # 计算请求耗时。 - latency_ms = int((time.perf_counter() - started) * 1000) - # 基于状态码和响应体判断服务是否健康。 - ok = healthy_from_body(service["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: - # 即使是 HTTP 错误，也尽量读取响应体做更准确的健康判断。 - body_text = exc.read().decode("utf-8", errors="replace") - # 计算失败请求耗时。 - latency_ms = int((time.perf_counter() - started) * 1000) - # 用 HTTP 错误码继续判断是否属于“服务活着但被拒绝访问”。 - ok = healthy_from_body(service["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_service_payload(hosts: list[dict[str, Any]]) -> dict[str, Any]: - # 把主机配置展开成服务配置。 - services = flatten_services(hosts) - # 并发数至少 1，且不超过配置值和服务总数。 - workers = max(1, min(PROBE_MAX_WORKERS, len(services) or 1)) - - # 并发探测所有服务。 - with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor: - items = list(executor.map(probe_service, services)) - - # 统计成功服务数量。 - ok_count = sum(1 for item in items if item["ok"]) - - # 返回服务层结果。 - return { - "updatedAt": utc_now(), - "summary": { - "total": len(items), - "ok": ok_count, - "down": len(items) - ok_count, - "hostTotal": len(hosts), - }, - "items": items, - } - - -def get_service_payload(hosts: list[dict[str, Any]]) -> dict[str, Any]: - # 使用独立缓存，减少频繁刷新带来的内网 HTTP 压力。 - global _SERVICE_CACHE_PAYLOAD, _SERVICE_CACHE_AT - - # 记录当前时间，用来判断缓存是否过期。 - now = time.time() - - # 先在锁内检查服务缓存是否还能复用。 - with _SERVICE_CACHE_LOCK: - if _SERVICE_CACHE_PAYLOAD is not None and now - _SERVICE_CACHE_AT < PROBE_CACHE_TTL_SECONDS: - return _SERVICE_CACHE_PAYLOAD - - # 缓存过期时重新构建服务探测结果。 - payload = build_service_payload(hosts) - - # 在锁内写回最新缓存。 - with _SERVICE_CACHE_LOCK: - _SERVICE_CACHE_PAYLOAD = payload - _SERVICE_CACHE_AT = now - - # 返回最新服务探测结果。 - return payload - - -def decode_task_output(output: str) -> str: - # TAT 返回的输出是 base64 文本，这里统一做解码。 - if not output: - return "" - - # 解码成 UTF-8 文本，异常字符用 replace 兜底。 - import base64 - - return base64.b64decode(output).decode("utf-8", errors="replace") - - -def build_tat_client() -> tuple[Any | None, str]: - # 如果没有腾讯云密钥，直接返回错误说明，前端就显示“无主机指标”。 - secret_id = os.environ.get("TENCENT_SECRET_ID", "").strip() - secret_key = os.environ.get("TENCENT_SECRET_KEY", "").strip() - region = os.environ.get("DEPLOY_REGION", "").strip() - - # 密钥缺失时不再继续初始化 SDK。 - if not secret_id or not secret_key or not region: - return None, "missing TAT credentials in .env" - - try: - # 运行时再动态导入 SDK，避免没装 SDK 时整个页面直接起不来。 - from tencentcloud.common import credential - from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException - from tencentcloud.common.profile.client_profile import ClientProfile - from tencentcloud.common.profile.http_profile import HttpProfile - from tencentcloud.tat.v20201028 import tat_client - except Exception as exc: # noqa: BLE001 - # SDK 缺失时返回错误信息给前端。 - return None, f"tencentcloud sdk unavailable: {exc}" - - # 创建凭据对象。 - cred = credential.Credential(secret_id, secret_key) - # 创建 TAT 客户端，后续所有主机指标都通过它拉取。 - client = tat_client.TatClient( - cred, - region, - ClientProfile(httpProfile=HttpProfile(endpoint="tat.tencentcloudapi.com")), - ) - - # 成功时返回客户端和空错误。 - return client, "" - - -def invoke_tat(action: Any, label: str, retries: int = 3, delay_seconds: int = 2) -> Any: - # 这里延迟导入腾讯云异常类型，避免 SDK 缺失时模块导入直接失败。 - from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException - - # 保存最后一次异常，便于最终抛出。 - last_error: Exception | None = None - - # 按给定次数做简单重试。 - for attempt in range(1, retries + 1): - try: - # 成功时直接返回调用结果。 - return action() - except TencentCloudSDKException as exc: - # 保存最后一次异常。 - last_error = exc - # 最后一次重试不再 sleep。 - if attempt == retries: - break - # 做一个简单退避，避免瞬时网络抖动。 - time.sleep(delay_seconds * attempt) - - # 有异常时直接抛出最后一次异常。 - if last_error is not None: - raise last_error - - # 理论上不会执行到这里，这里只是兜底。 - raise RuntimeError(f"{label} failed without error") - - -def build_metrics_script() -> str: - # 远端脚本本身只依赖 Python 标准库，避免每台机器再装额外 agent。 - return f"""set -euo pipefail -python3 - <<'PY' -import json -import os -import shutil -import time - -# 读取 /proc/stat 的第一行来采样 CPU。 -def read_cpu(): - with open('/proc/stat', 'r', encoding='utf-8') as handle: - parts = handle.readline().split()[1:8] - values = [int(item) for item in parts] - idle = values[3] + values[4] - total = sum(values) - return idle, total - -# 第一次 CPU 采样。 -idle_1, total_1 = read_cpu() -# 等待一小段时间做第二次采样。 -time.sleep({TAT_CPU_SAMPLE_SECONDS}) -# 第二次 CPU 采样。 -idle_2, total_2 = read_cpu() - -# 计算总差值，防止极端情况下出现 0。 -total_diff = max(total_2 - total_1, 1) -# 计算 idle 差值。 -idle_diff = max(idle_2 - idle_1, 0) -# 算出 CPU 使用率百分比。 -cpu_percent = round((1 - idle_diff / total_diff) * 100, 2) - -# 解析内存信息。 -memory_info = {{}} -with open('/proc/meminfo', 'r', encoding='utf-8') as handle: - for line in handle: - key, value = line.split(':', 1) - memory_info[key] = int(value.strip().split()[0]) - -# 总内存，单位换成 GB。 -memory_total_gb = round(memory_info.get('MemTotal', 0) / 1024 / 1024, 2) -# 可用内存，单位换成 GB。 -memory_available_gb = round(memory_info.get('MemAvailable', 0) / 1024 / 1024, 2) -# 已用内存。 -memory_used_gb = round(max(memory_total_gb - memory_available_gb, 0), 2) -# 内存使用率。 -memory_percent = round((memory_used_gb / memory_total_gb) * 100, 2) if memory_total_gb else 0.0 - -# 读取根分区磁盘占用。 -disk = shutil.disk_usage('/') -# 总磁盘，单位换成 GB。 -disk_total_gb = round(disk.total / 1024 / 1024 / 1024, 2) -# 已用磁盘，单位换成 GB。 -disk_used_gb = round(disk.used / 1024 / 1024 / 1024, 2) -# 磁盘使用率。 -disk_percent = round((disk.used / disk.total) * 100, 2) if disk.total else 0.0 - -# 通过 /proc 目录估算当前进程数。 -process_count = len([name for name in os.listdir('/proc') if name.isdigit()]) - -# 输出统一 JSON，方便服务端直接解析。 -print(json.dumps({{ - 'ok': True, - 'hostname': os.uname().nodename, - 'cpuPercent': cpu_percent, - 'memoryPercent': memory_percent, - 'memoryUsedGb': memory_used_gb, - 'memoryTotalGb': memory_total_gb, - 'diskPercent': disk_percent, - 'diskUsedGb': disk_used_gb, - 'diskTotalGb': disk_total_gb, - 'processCount': process_count, -}})) -PY -""" - - -def collect_host_metrics(hosts: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: - # 先把主机名和实例 ID 建立映射，后面 TAT 返回时要按实例回填。 - host_by_instance = { - host["instanceId"]: host - for host in hosts - if host.get("instanceId") - } - - # 去重后的实例 ID 列表。 - instance_ids = list(host_by_instance.keys()) - - # 如果配置里没有 instanceId，直接返回错误占位。 - if not instance_ids: - return { - host["host"]: { - "ok": False, - "error": "missing instanceId in config/targets.json", - "updatedAt": utc_now(), - } - for host in hosts - } - - # 初始化 TAT 客户端。 - client, error = build_tat_client() - - # 没拿到客户端时，给所有主机写统一错误。 - if client is None: - return { - host["host"]: { - "ok": False, - "error": error, - "updatedAt": utc_now(), - } - for host in hosts - } - - # 这里延迟导入 SDK model，避免上面客户端初始化失败时多余导入。 - from tencentcloud.tat.v20201028 import models as tat_models - - # 准备执行远端指标采集脚本。 - req = tat_models.RunCommandRequest() - # 批量下发到所有实例。 - req.from_json_string( - json.dumps( - { - "CommandName": "hy-app-monitor-host-metrics", - "CommandType": "SHELL", - "Content": __import__("base64").b64encode(build_metrics_script().encode("utf-8")).decode("ascii"), - "InstanceIds": instance_ids, - "Timeout": TAT_METRIC_TIMEOUT_SECONDS, - "WorkingDirectory": "/root", - } - ) - ) - - # 发起 TAT 命令。 - response = json.loads(invoke_tat(lambda: client.RunCommand(req), "RunCommand").to_json_string()) - # 拿到本次调用 ID，后面轮询它的执行结果。 - invocation_id = response["InvocationId"] - # 轮询截止时间比命令超时时间稍长一点。 - deadline = time.time() + TAT_METRIC_TIMEOUT_SECONDS + 60 - # 记录最终结果，key 是实例 ID。 - results: dict[str, dict[str, Any]] = {} - - # 开始轮询所有实例的执行结果。 - while time.time() < deadline: - # 轮询间隔不要太短，避免频繁打 API。 - time.sleep(2) - - # 构造查询任务结果的请求。 - query = tat_models.DescribeInvocationTasksRequest() - # 按 invocation id 过滤当前这次调用的所有实例结果。 - query.from_json_string( - json.dumps( - { - "HideOutput": False, - "Limit": 50, - "Filters": [{"Name": "invocation-id", "Values": [invocation_id]}], - } - ) - ) - - # 拉取本次调用的实例执行结果。 - data = json.loads(invoke_tat(lambda: client.DescribeInvocationTasks(query), "DescribeInvocationTasks").to_json_string()) - # 任务列表可能还没完全生成。 - tasks = data.get("InvocationTaskSet") or [] - - # 没有任务结果时继续等待。 - if not tasks: - continue - - # 遍历当前已返回的所有实例结果。 - for task in tasks: - # 拿到实例 ID。 - instance_id = task.get("InstanceId", "") - # 读取当前实例执行状态。 - status = task.get("TaskStatus", "") - # 只处理终态，未完成的继续等下一轮。 - if status not in {"SUCCESS", "FAILED", "TIMEOUT", "CANCELLED", "START_FAILED"}: - continue - - # 成功时解码输出并尝试解析成 JSON。 - if status == "SUCCESS": - try: - # 读取远端标准输出。 - output_text = decode_task_output(task.get("TaskResult", {}).get("Output", "")) - # 解析远端输出的 JSON。 - payload = json.loads(output_text or "{}") - # 补齐统一时间字段。 - payload["updatedAt"] = utc_now() - # 收下当前实例成功结果。 - results[instance_id] = payload - except Exception as exc: # noqa: BLE001 - # 输出格式异常时仍然要保留错误信息。 - results[instance_id] = { - "ok": False, - "error": f"metrics parse failed: {exc}", - "updatedAt": utc_now(), - } - continue - - # 失败时解码输出并记录错误。 - results[instance_id] = { - "ok": False, - "error": decode_task_output(task.get("TaskResult", {}).get("Output", "")) or status, - "updatedAt": utc_now(), - } - - # 如果已经收齐所有实例结果，就可以结束轮询。 - if len(results) >= len(instance_ids): - break - - # 按 host name 回填最终主机指标。 - payload_by_host: dict[str, dict[str, Any]] = {} - - # 对每个主机构造最终输出。 - for host in hosts: - # 读取主机对应实例 ID。 - instance_id = host.get("instanceId", "") - # 读取该实例的结果，没有就补超时错误。 - metrics = results.get( - instance_id, - { - "ok": False, - "error": "metrics timeout or no task result", - "updatedAt": utc_now(), - }, - ) - # 补上实例 ID 便于前端排查。 - metrics["instanceId"] = instance_id - # 以 host name 为 key 返回。 - payload_by_host[host["host"]] = metrics - - # 返回按 host name 索引的主机指标结果。 - return payload_by_host - - -def get_host_metrics(hosts: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: - # 使用独立缓存，减少页面刷新时反复打 TAT。 - global _HOST_CACHE_PAYLOAD, _HOST_CACHE_AT - - # 记录当前时间。 - now = time.time() - - # 在锁内优先读取缓存。 - with _HOST_CACHE_LOCK: - if _HOST_CACHE_PAYLOAD is not None and now - _HOST_CACHE_AT < HOST_METRIC_CACHE_TTL_SECONDS: - return _HOST_CACHE_PAYLOAD - - # 缓存失效时重新采集。 - try: - # 真正执行主机指标采集。 - payload = collect_host_metrics(hosts) - except Exception as exc: # noqa: BLE001 - # 指标采集失败时不能拖垮整页，统一降级成错误占位。 - payload = { - host["host"]: { - "ok": False, - "error": f"metrics collection failed: {exc}", - "updatedAt": utc_now(), - "instanceId": host.get("instanceId", ""), - } - for host in hosts - } - - # 把最新指标写回缓存。 - with _HOST_CACHE_LOCK: - _HOST_CACHE_PAYLOAD = payload - _HOST_CACHE_AT = now - - # 返回最新指标。 - return payload - - -def build_host_payload( - hosts: list[dict[str, Any]], - service_payload: dict[str, Any], - metrics_by_host: dict[str, dict[str, Any]], -) -> list[dict[str, Any]]: - # 把服务结果按 host 分组，便于构造主机视图。 - services_by_host: dict[str, list[dict[str, Any]]] = {} - - # 遍历所有服务结果。 - for item in service_payload.get("items") or []: - # 创建主机对应列表。 - services_by_host.setdefault(item["host"], []).append(item) - - # 最终主机视图列表。 - grouped_hosts: list[dict[str, Any]] = [] - - # 按配置顺序输出主机，页面顺序更稳定。 - for host in hosts: - # 读取该主机的服务列表。 - items = services_by_host.get(host["host"], []) - # 统计该主机健康服务数量。 - ok_count = sum(1 for item in items if item["ok"]) - # 读取主机指标。 - metrics = metrics_by_host.get( - host["host"], - { - "ok": False, - "error": "metrics unavailable", - "updatedAt": utc_now(), - "instanceId": host.get("instanceId", ""), - }, - ) - - # 构造主机对象。 - grouped_hosts.append( - { - "host": host["host"], - "ip": host["ip"], - "group": host["group"], - "instanceId": host["instanceId"], - "metrics": metrics, - "serviceSummary": { - "total": len(items), - "ok": ok_count, - "down": len(items) - ok_count, - }, - "items": items, - } - ) - - # 返回按主机分组后的完整结构。 - return grouped_hosts - - -def build_monitor_payload() -> dict[str, Any]: - # 先读取监控配置。 - hosts = load_hosts() - # 获取服务健康结果。 - service_payload = get_service_payload(hosts) - # 获取主机资源指标。 - metrics_by_host = get_host_metrics(hosts) - # 拼出主机维度视图。 - grouped_hosts = build_host_payload(hosts, service_payload, metrics_by_host) - # 统计主机指标成功数量。 - host_metric_ok = sum(1 for host in grouped_hosts if host["metrics"].get("ok")) - - # 返回给前端的统一结果。 - return { - "updatedAt": utc_now(), - "summary": { - **service_payload["summary"], - "hostMetricOk": host_metric_ok, - "hostMetricDown": len(grouped_hosts) - host_metric_ok, - }, - "hosts": grouped_hosts, - "items": service_payload["items"], - } - - -def response_json(payload: dict[str, Any] | list[Any]) -> bytes: - # 统一把 JSON 编码为 UTF-8 文本响应。 - return (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8") - - -class MonitorHandler(BaseHTTPRequestHandler): - # 自定义服务名，便于 curl 时识别当前响应来自哪套服务。 - server_version = "HyAppMonitor/1.0" - - def log_message(self, format: str, *args: Any) -> None: # noqa: A003 - # 关闭默认访问日志，避免轮询接口把日志刷爆。 - return - - def do_GET(self) -> None: # noqa: N802 - # GET 请求需要返回响应头和响应体。 - self.dispatch(send_body=True) - - def do_HEAD(self) -> None: # noqa: N802 - # HEAD 请求只返回响应头，不返回响应体。 - self.dispatch(send_body=False) - - def dispatch(self, send_body: bool) -> None: - # 只解析 URL 的 path 部分。 - parsed = urlparse(self.path) - - # 首页直接返回静态 HTML。 - if parsed.path in {"/", "/index.html"}: - return self.serve_static("index.html", "text/html; charset=utf-8", send_body=send_body) - - # CSS 静态资源。 - if parsed.path == "/app.css": - return self.serve_static("app.css", "text/css; charset=utf-8", send_body=send_body) - - # 前端业务 JS。 - if parsed.path == "/app.js": - return self.serve_static("app.js", "application/javascript; charset=utf-8", send_body=send_body) - - # 本地 Vue 运行时。 - 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": - return self.send_payload( - HTTPStatus.OK, - response_json(build_monitor_payload()), - "application/json; charset=utf-8", - send_body=send_body, - ) - - # favicon 直接返回空内容即可。 - if parsed.path == "/favicon.ico": - return self.send_payload(HTTPStatus.NO_CONTENT, b"", "image/x-icon", send_body=send_body) - - # 其他路径统一返回 404 JSON。 - 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 - - # 文件不存在时直接返回 JSON 错误。 - 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: - # 错误响应也走统一 JSON 结构，前端更容易处理。 - 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: - # 写入 HTTP 状态码。 - self.send_response(status) - # 写入内容类型。 - self.send_header("Content-Type", content_type) - # 页面和接口都不缓存，保证看到的是当前状态。 - self.send_header("Cache-Control", "no-store") - # 写入内容长度，便于浏览器正确处理 HEAD 请求。 - self.send_header("Content-Length", str(len(body))) - # 结束响应头。 - self.end_headers() - - # 只有 GET 这类需要响应体的请求才真正写 body。 - if send_body and body: - self.wfile.write(body) - - -def main() -> None: - # 创建线程型 HTTP 服务，允许多个请求并发处理。 - server = ThreadingHTTPServer((APP_BIND, APP_PORT), MonitorHandler) - # 启动时打印监听地址，方便排查。 - print(f"hy-app-monitor listening on http://{APP_BIND}:{APP_PORT}", flush=True) - # 进入永久监听循环。 - server.serve_forever() +# 入口文件只保留启动逻辑，具体实现放进 monitor/ 模块里。 +from monitor.http_app import main if __name__ == "__main__": - # 作为脚本执行时启动主服务。 + # 作为脚本执行时启动 HTTP 服务。 main() diff --git a/static/app.css b/static/app.css index ae9f30f..561108b 100644 --- a/static/app.css +++ b/static/app.css @@ -8,6 +8,7 @@ --muted: oklch(76% 0.02 248); --accent: oklch(70% 0.13 206); --ok: oklch(76% 0.16 151); + --warn: oklch(80% 0.14 84); --bad: oklch(68% 0.18 24); --shadow: 0 24px 70px rgba(5, 10, 24, 0.35); } @@ -36,7 +37,7 @@ body { } .shell { - max-width: 1360px; + max-width: 1440px; margin: 0 auto; padding: 28px 18px 48px; } @@ -49,7 +50,7 @@ body { .hero { display: grid; - grid-template-columns: minmax(0, 1.2fr) minmax(280px, 0.8fr); + grid-template-columns: minmax(0, 1.1fr) minmax(360px, 0.9fr); gap: 18px; margin-bottom: 18px; } @@ -57,7 +58,8 @@ body { .hero-copy, .hero-side, .toolbar, -.host-panel { +.host-panel, +.infra-panel { border: 1px solid var(--line); border-radius: 24px; background: var(--panel); @@ -80,42 +82,48 @@ body { .hero-copy h1 { margin: 0; font-family: "Space Grotesk", "IBM Plex Sans", sans-serif; - font-size: clamp(2rem, 4vw, 3.7rem); + font-size: clamp(2rem, 4vw, 3.4rem); line-height: 0.96; letter-spacing: -0.05em; } .hero-side { display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 12px; padding: 18px; } -.stat-block { +.stat-block, +.summary-card { padding: 16px 18px; border-radius: 18px; background: var(--panel-strong); border: 1px solid var(--line); } -.stat-block span { +.stat-block span, +.summary-card small { display: block; color: var(--muted); font-size: 13px; } -.stat-block strong { +.stat-block strong, +.summary-card strong { display: block; margin-top: 8px; - font-size: 2rem; + font-size: 1.55rem; letter-spacing: -0.04em; } -.stat-block.ok strong { +.stat-block.ok strong, +.ok-text { color: var(--ok); } -.stat-block.bad strong { +.stat-block.bad strong, +.bad-text { color: var(--bad); } @@ -140,13 +148,19 @@ body { } .filter-chip, -.refresh { +.refresh, +.interval-select, +.search { border: 1px solid var(--line); background: var(--panel-strong); color: var(--text); + font: inherit; +} + +.filter-chip, +.refresh { border-radius: 999px; padding: 10px 14px; - font: inherit; cursor: pointer; transition: transform 180ms ease, background 180ms ease, border-color 180ms ease; } @@ -163,37 +177,60 @@ body { .search { min-width: 260px; - border: 1px solid var(--line); - background: var(--panel-strong); - color: var(--text); border-radius: 999px; padding: 10px 14px; - font: inherit; +} + +.interval-field { + display: inline-flex; + align-items: center; + gap: 8px; + color: var(--muted); + font-size: 13px; +} + +.interval-select { + min-width: 88px; + padding: 10px 12px; + border-radius: 999px; } .meta-strip { display: flex; justify-content: space-between; gap: 12px; + flex-wrap: wrap; padding: 2px 4px 16px; color: var(--muted); font-size: 13px; } -.bad-text { - color: var(--bad); +.section-head { + margin: 10px 0 12px; } -.hosts { +.section-head h2 { + margin: 0; + font-size: 1rem; + letter-spacing: 0.03em; + text-transform: uppercase; + color: var(--muted); +} + +.hosts, +.nacos-service-grid { display: grid; gap: 16px; } -.host-panel { +.host-panel, +.infra-panel { padding: 18px; } -.host-header { +.host-header, +.panel-head, +.nacos-service-head { display: flex; justify-content: space-between; align-items: flex-start; @@ -201,14 +238,26 @@ body { margin-bottom: 16px; } -.host-header h2 { +.host-header h2, +.panel-head h3, +.nacos-service-head h4 { margin: 0; font-family: "Space Grotesk", "IBM Plex Sans", sans-serif; - font-size: 1.45rem; letter-spacing: -0.03em; } -.host-header p { +.host-header h2 { + font-size: 1.45rem; +} + +.panel-head h3, +.nacos-service-head h4 { + font-size: 1.1rem; +} + +.host-header p, +.panel-head p, +.nacos-service-head p { margin: 6px 0 0; color: var(--muted); font-size: 13px; @@ -220,7 +269,9 @@ body { flex-wrap: wrap; } -.host-metrics { +.host-metrics, +.summary-grid, +.metric-grid { display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 12px; @@ -283,6 +334,11 @@ body { border-color: color-mix(in oklab, var(--ok) 50%, var(--line)); } +.badge.warn { + color: var(--warn); + border-color: color-mix(in oklab, var(--warn) 50%, var(--line)); +} + .badge.bad { color: var(--bad); border-color: color-mix(in oklab, var(--bad) 50%, var(--line)); @@ -294,7 +350,8 @@ body { gap: 14px; } -.service-card { +.service-card, +.nacos-service-card { padding: 16px; border-radius: 20px; border: 1px solid var(--line); @@ -321,96 +378,153 @@ body { color: var(--muted); font-size: 12px; text-transform: uppercase; - letter-spacing: 0.12em; + letter-spacing: 0.08em; } .status-pill { display: inline-flex; align-items: center; gap: 8px; - min-height: 32px; - padding: 0 12px; + padding: 8px 12px; border-radius: 999px; + border: 1px solid var(--line); font-size: 12px; - font-weight: 700; -} - -.status-pill.ok { - color: var(--ok); - background: color-mix(in oklab, var(--ok) 14%, transparent); -} - -.status-pill.bad { - color: var(--bad); - background: color-mix(in oklab, var(--bad) 14%, transparent); } .dot { width: 8px; height: 8px; - border-radius: 50%; + border-radius: 999px; background: currentColor; } .service-data { display: grid; grid-template-columns: repeat(3, minmax(0, 1fr)); - gap: 12px; - margin-top: 16px; + gap: 10px; + margin: 16px 0 14px; } -.service-data small, -.service-url, -.service-detail { +.service-data small { + display: block; color: var(--muted); + font-size: 12px; } .service-data strong { display: block; - margin-top: 4px; + margin-top: 5px; font-size: 1rem; } -.service-url { - margin-top: 16px; +.service-url, +.service-detail, +.mono { + font-family: "IBM Plex Mono", "SFMono-Regular", monospace; +} + +.service-url, +.service-detail { + color: var(--muted); font-size: 12px; word-break: break-all; } .service-detail { - margin-top: 8px; - min-height: 40px; - font-size: 12px; - line-height: 1.55; + margin-top: 10px; + line-height: 1.45; } -@media (max-width: 1080px) { +.table-wrap { + overflow-x: auto; + margin-top: 12px; +} + +.table-wrap.compact { + margin-top: 8px; +} + +.table { + width: 100%; + border-collapse: collapse; + font-size: 13px; +} + +.table th, +.table td { + padding: 10px 12px; + border-bottom: 1px solid var(--line); + text-align: left; + vertical-align: top; +} + +.table th { + color: var(--muted); + font-weight: 600; +} + +.table.compact th, +.table.compact td { + padding: 8px 10px; + font-size: 12px; +} + +.tone-ok { + color: var(--ok); + border-color: color-mix(in oklab, var(--ok) 55%, var(--line)); +} + +.tone-warn { + color: var(--warn); + border-color: color-mix(in oklab, var(--warn) 55%, var(--line)); +} + +.tone-bad { + color: var(--bad); + border-color: color-mix(in oklab, var(--bad) 55%, var(--line)); +} + +.tone-neutral { + color: var(--text); +} + +@media (max-width: 1180px) { .hero { grid-template-columns: 1fr; } - .host-metrics, .service-grid { grid-template-columns: repeat(2, minmax(0, 1fr)); } + + .host-metrics, + .summary-grid, + .metric-grid { + grid-template-columns: repeat(2, minmax(0, 1fr)); + } } -@media (max-width: 760px) { +@media (max-width: 720px) { + .shell { + padding: 18px 12px 32px; + } + .toolbar, .host-header, - .meta-strip { + .panel-head, + .nacos-service-head { flex-direction: column; - align-items: stretch; + } + + .hero-side, + .service-grid, + .host-metrics, + .summary-grid, + .metric-grid { + grid-template-columns: 1fr; } .search { - min-width: 0; - width: 100%; - } - - .host-metrics, - .service-grid, - .service-data { - grid-template-columns: 1fr; + min-width: 100%; } } diff --git a/static/app.js b/static/app.js index ef4e668..489772c 100644 --- a/static/app.js +++ b/static/app.js @@ -1,5 +1,7 @@ const { createApp } = Vue; +const STORAGE_KEY = "hy-app-monitor-refresh-interval"; + createApp({ data() { return { @@ -7,8 +9,15 @@ createApp({ error: "", filter: "all", keyword: "", + refreshIntervalSeconds: 10, + refreshIntervalOptions: [5, 10, 15, 30, 60], payload: { updatedAt: "", + settings: { + refreshIntervalSeconds: 10, + refreshIntervalOptions: [5, 10, 15, 30, 60], + thresholds: {}, + }, summary: { total: 0, ok: 0, @@ -16,36 +25,54 @@ createApp({ hostTotal: 0, hostMetricOk: 0, hostMetricDown: 0, + nacosServiceTotal: 0, + nacosInstanceTotal: 0, + nacosUnhealthyInstanceTotal: 0, + mongoDatabaseTotal: 0, + mongoCollectionTotal: 0, + mongoOk: 0, }, hosts: [], + nacos: { + ok: false, + summary: {}, + services: [], + namespaces: [], + }, + mongo: { + ok: false, + summary: {}, + databases: [], + replicaSet: { + configured: false, + members: [], + }, + }, }, timer: null, }; }, computed: { summary() { - return this.payload.summary || { - total: 0, - ok: 0, - down: 0, - hostTotal: 0, - hostMetricOk: 0, - hostMetricDown: 0, - }; + return this.payload.summary || {}; + }, + thresholds() { + return this.payload.settings?.thresholds || {}; }, formattedUpdatedAt() { - if (!this.payload.updatedAt) { - return "-"; - } - const date = new Date(this.payload.updatedAt); - return Number.isNaN(date.getTime()) ? this.payload.updatedAt : date.toLocaleString(); + return this.formatDateTime(this.payload.updatedAt); }, groupedHosts() { const keyword = this.keyword.trim().toLowerCase(); return (this.payload.hosts || []) .map((host) => { - const items = (host.items || []).filter((item) => { + const hostMatched = [host.host, host.ip, host.group, host.instanceId] + .join(" ") + .toLowerCase() + .includes(keyword); + + const filteredItems = (host.items || []).filter((item) => { if (this.filter === "down" && item.ok) { return false; } @@ -58,20 +85,54 @@ createApp({ if (!keyword) { return true; } - return [item.host, item.ip, item.service, item.kind, item.group] + return [item.host, item.ip, item.service, item.kind, item.group, item.url, item.detail] .join(" ") .toLowerCase() .includes(keyword); }); + const shouldShowBecauseDown = this.filter === "down" && !(host.metrics && host.metrics.ok); + const shouldShowBecauseKeyword = keyword ? hostMatched || filteredItems.length > 0 : true; + const shouldShow = + shouldShowBecauseKeyword && + (this.filter === "all" || filteredItems.length > 0 || shouldShowBecauseDown || hostMatched); + return { ...host, groupLabel: this.groupLabel(host.group), - items, - downCount: items.filter((item) => !item.ok).length, + items: filteredItems, + downCount: (host.items || []).filter((item) => !item.ok).length, + shouldShow, }; }) - .filter((host) => host.items.length > 0 || !keyword); + .filter((host) => host.shouldShow); + }, + filteredNacosServices() { + const keyword = this.keyword.trim().toLowerCase(); + return (this.payload.nacos?.services || []).filter((service) => { + if (!keyword) { + return true; + } + const serviceMatched = [service.serviceName, service.groupName, service.rawServiceName] + .join(" ") + .toLowerCase() + .includes(keyword); + if (serviceMatched) { + return true; + } + return (service.instances || []).some((instance) => + [instance.ip, instance.port, instance.clusterName] + .join(" ") + .toLowerCase() + .includes(keyword) + ); + }); + }, + }, + watch: { + refreshIntervalSeconds() { + window.localStorage.setItem(STORAGE_KEY, String(this.refreshIntervalSeconds)); + this.applyRefreshTimer(); }, }, methods: { @@ -80,8 +141,16 @@ createApp({ app: "应用节点", pay: "支付节点", gateway: "网关节点", + nacos: "Nacos 节点", + mongo: "MongoDB 节点", }[group] || group; }, + toneClass(level) { + return `tone-${level || "neutral"}`; + }, + boolTone(ok) { + return ok ? "tone-ok" : "tone-bad"; + }, formatPercent(value) { if (value == null || Number.isNaN(Number(value))) { return "-"; @@ -94,6 +163,89 @@ createApp({ } return `${Number(used).toFixed(2)} / ${Number(total).toFixed(2)} GB`; }, + formatMb(value) { + if (value == null || Number.isNaN(Number(value))) { + return "-"; + } + return `${Number(value).toFixed(2)} MB`; + }, + formatBytes(value) { + if (value == null || Number.isNaN(Number(value))) { + return "-"; + } + const units = ["B", "KB", "MB", "GB", "TB"]; + let amount = Number(value); + let index = 0; + while (amount >= 1024 && index < units.length - 1) { + amount /= 1024; + index += 1; + } + return `${amount.toFixed(2)} ${units[index]}`; + }, + formatNumber(value) { + if (value == null || Number.isNaN(Number(value))) { + return "-"; + } + return Number(value).toLocaleString(); + }, + formatDateTime(value) { + if (!value) { + return "-"; + } + const date = new Date(value); + return Number.isNaN(date.getTime()) ? String(value) : date.toLocaleString(); + }, + formatDuration(seconds) { + if (seconds == null || Number.isNaN(Number(seconds))) { + return "-"; + } + const total = Math.max(0, Number(seconds)); + const days = Math.floor(total / 86400); + const hours = Math.floor((total % 86400) / 3600); + const minutes = Math.floor((total % 3600) / 60); + const remainSeconds = Math.floor(total % 60); + if (days > 0) { + return `${days}d ${hours}h ${minutes}m`; + } + if (hours > 0) { + return `${hours}h ${minutes}m ${remainSeconds}s`; + } + if (minutes > 0) { + return `${minutes}m ${remainSeconds}s`; + } + return `${remainSeconds}s`; + }, + syncRefreshSettings() { + const settings = this.payload.settings || {}; + const options = Array.isArray(settings.refreshIntervalOptions) + ? settings.refreshIntervalOptions.map((item) => Number(item)).filter((item) => !Number.isNaN(item) && item > 0) + : this.refreshIntervalOptions; + + if (options.length > 0) { + this.refreshIntervalOptions = options; + } + + const stored = Number(window.localStorage.getItem(STORAGE_KEY) || 0); + const configured = Number(settings.refreshIntervalSeconds || 0); + const fallback = this.refreshIntervalOptions[0] || 10; + const candidate = this.refreshIntervalOptions.includes(stored) + ? stored + : this.refreshIntervalOptions.includes(configured) + ? configured + : fallback; + + if (candidate !== this.refreshIntervalSeconds) { + this.refreshIntervalSeconds = candidate; + } else { + this.applyRefreshTimer(); + } + }, + applyRefreshTimer() { + if (this.timer) { + window.clearInterval(this.timer); + } + this.timer = window.setInterval(() => this.refresh(), this.refreshIntervalSeconds * 1000); + }, async refresh() { this.loading = true; this.error = ""; @@ -103,6 +255,7 @@ createApp({ throw new Error(`HTTP ${response.status}`); } this.payload = await response.json(); + this.syncRefreshSettings(); } catch (error) { this.error = error instanceof Error ? error.message : String(error); } finally { @@ -120,7 +273,6 @@ createApp({ bootIndicator.hidden = true; } this.refresh(); - this.timer = window.setInterval(() => this.refresh(), 10000); }, beforeUnmount() { if (this.timer) { diff --git a/static/index.html b/static/index.html index 85647f8..a61a377 100644 --- a/static/index.html +++ b/static/index.html @@ -20,7 +20,7 @@

HY APP MONITOR

-

服务与主机监控

+

服务 / 主机 / Nacos / MongoDB

@@ -39,18 +39,40 @@ 主机数 {{ summary.hostTotal }}
+
+ Nacos 服务 + {{ summary.nacosServiceTotal }} +
+
+ Nacos 异常实例 + {{ summary.nacosUnhealthyInstanceTotal }} +
+
+ Mongo 状态 + {{ payload.mongo && payload.mongo.ok ? 'OK' : 'ERR' }} +
+
+ Mongo 集合 + {{ summary.mongoCollectionTotal }} +
- +
- + +
@@ -58,9 +80,14 @@
最后刷新：{{ formattedUpdatedAt }} 主机指标正常：{{ summary.hostMetricOk }} / {{ summary.hostTotal }} + Nacos 实例：{{ summary.nacosInstanceTotal }} + Mongo 数据库：{{ summary.mongoDatabaseTotal }} 加载失败：{{ error }}
+
+

主机与服务

+
@@ -80,21 +107,21 @@
-
+
CPU {{ formatPercent(group.metrics.cpuPercent) }}
-
+
内存 {{ formatPercent(group.metrics.memoryPercent) }} {{ formatGb(group.metrics.memoryUsedGb, group.metrics.memoryTotalGb) }}
-
+
磁盘 {{ formatPercent(group.metrics.diskPercent) }} {{ formatGb(group.metrics.diskUsedGb, group.metrics.diskTotalGb) }}
-
+
进程数 {{ group.metrics.processCount ?? '-' }} {{ group.metrics.hostname || group.instanceId }} @@ -103,14 +130,14 @@

{{ group.metrics.error }}

-
+
{{ item.service }}
{{ item.kind }}
-
+
{{ item.ok ? 'UP' : 'DOWN' }}
@@ -137,6 +164,273 @@
+ +
+

Nacos 注册

+
+
+
+
+

Nacos

+

{{ payload.nacos.baseUrl }} · {{ payload.nacos.namespaceName }} · {{ payload.nacos.namespaceId }}

+
+ + {{ payload.nacos && payload.nacos.ok ? '连接正常' : '连接异常' }} + +
+ +
+
+ 命名空间 + {{ payload.nacos.summary.namespaceTotal || 0 }} +
+
+ 服务数 + {{ payload.nacos.summary.serviceTotal || 0 }} +
+
+ 实例数 + {{ payload.nacos.summary.instanceTotal || 0 }} +
+
+ 异常实例 + {{ payload.nacos.summary.unhealthyInstanceTotal || 0 }} +
+
+ +

{{ payload.nacos.error }}

+ +
+ + + + + + + + + + + + + + + + + +
命名空间ID配置数类型
{{ item.name }}{{ item.namespaceId }}{{ item.configCount }}{{ item.type }}
+
+ +
+
+
+
+

{{ service.serviceName }}

+

{{ service.groupName }}

+
+ + {{ service.instanceTotal }} / {{ service.healthyTotal }} + +
+
+ + + + + + + + + + + + + + + + + + + + + +
IP端口集群健康启用权重
{{ instance.ip }}{{ instance.port }}{{ instance.clusterName }}{{ instance.healthy ? 'Y' : 'N' }}{{ instance.enabled ? 'Y' : 'N' }}{{ instance.weight }}
+
+
+
+
+ +
+

MongoDB 指标

+
+
+
+
+

MongoDB

+

{{ payload.mongo.info.host || '-' }} · {{ payload.mongo.info.version || '-' }} · {{ payload.mongo.info.process || '-' }}

+
+ + {{ payload.mongo && payload.mongo.ok ? '连接正常' : '连接异常' }} + +
+ +
+
+ 运行时长 + {{ formatDuration(payload.mongo.info.uptimeSeconds) }} +
+
+ 当前连接 + {{ formatNumber(payload.mongo.connections.current) }} +
+
+ 可用连接 + {{ formatNumber(payload.mongo.connections.available) }} +
+
+ 数据库 + {{ formatNumber(payload.mongo.summary.databaseTotal) }} +
+
+ 集合 + {{ formatNumber(payload.mongo.summary.collectionTotal) }} +
+
+ 对象数 + {{ formatNumber(payload.mongo.summary.objectTotal) }} +
+
+ 数据量 + {{ formatMb(payload.mongo.summary.dataSizeMb) }} +
+
+ 存储量 + {{ formatMb(payload.mongo.summary.storageSizeMb) }} +
+
+ +

{{ payload.mongo.error }}

+ +
+
+ Resident + {{ formatMb(payload.mongo.memory.residentMb) }} +
+
+ Virtual + {{ formatMb(payload.mongo.memory.virtualMb) }} +
+
+ Mapped + {{ formatMb(payload.mongo.memory.mappedMb) }} +
+
+ Requests + {{ formatNumber(payload.mongo.network.numRequests) }} +
+
+ Bytes In + {{ formatBytes(payload.mongo.network.bytesIn) }} +
+
+ Bytes Out + {{ formatBytes(payload.mongo.network.bytesOut) }} +
+
+ WT Cache + {{ formatBytes(payload.mongo.wiredTigerCache.bytesCurrentlyInCache) }} +
+
+ WT Dirty + {{ formatBytes(payload.mongo.wiredTigerCache.trackedDirtyBytesInCache) }} +
+
+ +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
操作次数操作次数
insert{{ formatNumber(payload.mongo.opcounters.insert) }}query{{ formatNumber(payload.mongo.opcounters.query) }}
update{{ formatNumber(payload.mongo.opcounters.update) }}delete{{ formatNumber(payload.mongo.opcounters.delete) }}
getmore{{ formatNumber(payload.mongo.opcounters.getmore) }}command{{ formatNumber(payload.mongo.opcounters.command) }}
+
+ +
+ + + + + + + + + + + + + + + + + + + + + +
成员状态健康运行时长复制延迟最后 Optime
{{ member.name }}{{ member.stateStr }}{{ member.health }}{{ formatDuration(member.uptime) }}{{ member.lagSeconds == null ? '-' : `${member.lagSeconds}s` }}{{ formatDateTime(member.optimeDate) }}
+
+ +
+ + + + + + + + + + + + + + + + + + + + + + + +
数据库集合对象数据量存储量索引数索引量
{{ db.name }}{{ formatNumber(db.collections) }}{{ formatNumber(db.objects) }}{{ formatMb(db.dataSizeMb) }}{{ formatMb(db.storageSizeMb) }}{{ formatNumber(db.indexCount) }}{{ formatMb(db.indexSizeMb) }}
+
+