462 lines
15 KiB
Python
462 lines
15 KiB
Python
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)
|
||
|