853 lines
31 KiB
Python
853 lines
31 KiB
Python
#!/usr/bin/env python3
|
||
|
||
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()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 作为脚本执行时启动主服务。
|
||
main()
|