882 lines
29 KiB
Python
882 lines
29 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from typing import Any
|
||
|
||
from core.cache import TTLCache
|
||
from .config_validation import validate_config_content
|
||
from core.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,
|
||
load_hosts,
|
||
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
|
||
|
||
# 复用现有部署链路里的 Nacos 注册名映射。
|
||
NACOS_SERVICE_NAME_MAP = {
|
||
"auth": "rc-auth",
|
||
"gateway": "rc-gateway",
|
||
"external": "rc-service-external",
|
||
"console": "rc-service-console",
|
||
"other": "rc-service-other",
|
||
"live": "rc-service-live",
|
||
"wallet": "rc-service-wallet",
|
||
"order": "rc-service-order",
|
||
}
|
||
# 反向映射，便于把注册名转回页面上的业务名。
|
||
FRIENDLY_SERVICE_NAME_MAP = {value: key for key, value in NACOS_SERVICE_NAME_MAP.items()}
|
||
|
||
|
||
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:
|
||
cleaned_data = {key: value for key, value in data.items() if value is not None}
|
||
body = urllib.parse.urlencode(cleaned_data, quote_via=urllib.parse.quote).encode("utf-8")
|
||
|
||
# 表单提交时显式声明 content type，避免不同 Python 版本行为不一致。
|
||
headers = {"User-Agent": "hy-app-monitor/1.0"}
|
||
if data is not None:
|
||
headers["Content-Type"] = "application/x-www-form-urlencoded; charset=utf-8"
|
||
|
||
# 构造 HTTP 请求对象。
|
||
request = urllib.request.Request(url, data=body, method=method, headers=headers)
|
||
|
||
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 nacos_request_json(
|
||
method: str,
|
||
path: str,
|
||
*,
|
||
params: dict[str, Any] | None = None,
|
||
data: 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, data=data)
|
||
|
||
|
||
def nacos_request_text(
|
||
method: str,
|
||
path: str,
|
||
*,
|
||
params: dict[str, Any] | None = None,
|
||
data: dict[str, Any] | None = None,
|
||
) -> str:
|
||
# 复制一份参数，避免污染外部调用方。
|
||
merged_params = dict(params or {})
|
||
# 如果登录拿到了 token，就把 accessToken 带上。
|
||
access_token = fetch_access_token()
|
||
if access_token:
|
||
merged_params["accessToken"] = access_token
|
||
|
||
# 发起请求并直接返回原始文本。
|
||
return request_text(method, path, params=merged_params, data=data)
|
||
|
||
|
||
def current_namespace() -> tuple[str, str]:
|
||
# 优先使用显式配置的 namespace，未配置时回退 public。
|
||
namespace_id = NACOS_NAMESPACE_ID or "public"
|
||
namespace_name = NACOS_NAMESPACE_NAME or namespace_id
|
||
return namespace_id, namespace_name
|
||
|
||
|
||
def tenant_param(namespace_id: str) -> str | None:
|
||
# public 命名空间不需要显式传 tenant。
|
||
if not namespace_id or namespace_id == "public":
|
||
return None
|
||
return namespace_id
|
||
|
||
|
||
def infer_config_type(data_id: str, fallback: str = "text") -> str:
|
||
# 按文件后缀猜一个最接近的 Nacos type。
|
||
lowered = data_id.strip().lower()
|
||
if lowered.endswith((".yml", ".yaml")):
|
||
return "yaml"
|
||
if lowered.endswith(".json"):
|
||
return "json"
|
||
if lowered.endswith(".properties"):
|
||
return "properties"
|
||
if lowered.endswith(".xml"):
|
||
return "xml"
|
||
if lowered.endswith(".html"):
|
||
return "html"
|
||
if lowered.endswith(".txt"):
|
||
return "text"
|
||
return fallback
|
||
|
||
|
||
def normalize_config_entry(item: dict[str, Any], namespace_id: str) -> dict[str, Any]:
|
||
# 从列表接口里抽取前端真正需要的元数据。
|
||
data_id = str(item.get("dataId") or "").strip()
|
||
group_name = str(item.get("group") or item.get("groupName") or "").strip()
|
||
config_type = str(item.get("type") or "").strip() or infer_config_type(data_id)
|
||
return {
|
||
"dataId": data_id,
|
||
"group": group_name,
|
||
"appName": str(item.get("appName") or "").strip(),
|
||
"type": config_type,
|
||
"description": str(item.get("description") or item.get("desc") or "").strip(),
|
||
"namespaceId": str(item.get("tenant") or item.get("namespaceId") or namespace_id).strip() or namespace_id,
|
||
"md5": str(item.get("md5") or "").strip(),
|
||
"lastModifiedTime": str(item.get("lastModifiedTime") or item.get("gmtModified") or "").strip(),
|
||
}
|
||
|
||
|
||
def list_config_entries(namespace_id: str) -> list[dict[str, Any]]:
|
||
# 逐页拉取当前 namespace 下全部配置项。
|
||
page_no = 1
|
||
items: list[dict[str, Any]] = []
|
||
|
||
while True:
|
||
# Nacos v1 配置列表要求显式带 search 参数。
|
||
payload = nacos_request_json(
|
||
"GET",
|
||
"/nacos/v1/cs/configs",
|
||
params={
|
||
"search": "blur",
|
||
"pageNo": str(page_no),
|
||
"pageSize": str(NACOS_PAGE_SIZE),
|
||
"tenant": tenant_param(namespace_id),
|
||
"dataId": "",
|
||
"group": "",
|
||
"appName": "",
|
||
"config_tags": "",
|
||
},
|
||
)
|
||
|
||
# 兼容常见字段名。
|
||
page_items = list(payload.get("pageItems") or payload.get("data") or payload.get("items") or [])
|
||
if not page_items:
|
||
break
|
||
|
||
items.extend(normalize_config_entry(item, namespace_id) for item in page_items)
|
||
|
||
# totalCount 是配置列表接口的标准分页字段。
|
||
total_count = int(payload.get("totalCount") or payload.get("count") or len(items))
|
||
if len(items) >= total_count:
|
||
break
|
||
|
||
page_no += 1
|
||
|
||
# 去重并按 group/dataId 排序，避免翻页结果不稳定。
|
||
deduped: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for item in items:
|
||
deduped[(item["group"], item["dataId"])] = item
|
||
return sorted(deduped.values(), key=lambda item: (item["group"], item["dataId"]))
|
||
|
||
|
||
def get_config_content(namespace_id: str, group_name: str, data_id: str) -> str:
|
||
# 读取指定配置的原始文本内容。
|
||
return nacos_request_text(
|
||
"GET",
|
||
"/nacos/v1/cs/configs",
|
||
params={
|
||
"tenant": tenant_param(namespace_id),
|
||
"group": group_name,
|
||
"dataId": data_id,
|
||
},
|
||
)
|
||
|
||
|
||
def get_nacos_config_list_payload() -> dict[str, Any]:
|
||
# 使用当前监控配置里的 namespace。
|
||
namespace_id, namespace_name = current_namespace()
|
||
# 读取全部配置项。
|
||
items = list_config_entries(namespace_id)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"baseUrl": NACOS_BASE_URL,
|
||
"namespaceId": namespace_id,
|
||
"namespaceName": namespace_name,
|
||
"total": len(items),
|
||
"items": items,
|
||
}
|
||
|
||
|
||
def get_nacos_config_payload(group_name: str, data_id: str) -> dict[str, Any]:
|
||
# 使用当前监控配置里的 namespace。
|
||
namespace_id, namespace_name = current_namespace()
|
||
# 读取配置原文。
|
||
content = get_config_content(namespace_id, group_name, data_id)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"baseUrl": NACOS_BASE_URL,
|
||
"namespaceId": namespace_id,
|
||
"namespaceName": namespace_name,
|
||
"item": {
|
||
"dataId": data_id,
|
||
"group": group_name,
|
||
"type": infer_config_type(data_id),
|
||
},
|
||
"content": content,
|
||
}
|
||
|
||
|
||
def validate_nacos_config_payload(content: str, *, group_name: str = "", data_id: str = "", config_type: str | None = None) -> dict[str, Any]:
|
||
# 给页面单独暴露一个“先校验再保存”的能力。
|
||
result = validate_config_content(
|
||
content,
|
||
config_type=(config_type or "").strip(),
|
||
data_id=data_id,
|
||
)
|
||
return {
|
||
**result,
|
||
"updatedAt": utc_now(),
|
||
"item": {
|
||
"group": group_name,
|
||
"dataId": data_id,
|
||
"type": (config_type or infer_config_type(data_id)).strip(),
|
||
},
|
||
}
|
||
|
||
|
||
def save_nacos_config_payload(group_name: str, data_id: str, content: str, config_type: str | None = None) -> dict[str, Any]:
|
||
# 使用当前监控配置里的 namespace。
|
||
namespace_id, namespace_name = current_namespace()
|
||
# 保存前先做一次强校验，避免误把错误格式写进 Nacos。
|
||
validation = validate_config_content(
|
||
content,
|
||
config_type=(config_type or "").strip(),
|
||
data_id=data_id,
|
||
)
|
||
if not validation.get("ok"):
|
||
raise RuntimeError(str(validation.get("message") or f"invalid config content: {group_name}/{data_id}"))
|
||
|
||
# public 命名空间不显式传 tenant。
|
||
save_data: dict[str, Any] = {
|
||
"group": group_name,
|
||
"dataId": data_id,
|
||
"type": (config_type or infer_config_type(data_id)).strip(),
|
||
"content": content,
|
||
}
|
||
if tenant_param(namespace_id):
|
||
save_data["tenant"] = tenant_param(namespace_id)
|
||
|
||
# Nacos v1 配置保存统一走 POST /v1/cs/configs。
|
||
response_text = nacos_request_text(
|
||
"POST",
|
||
"/nacos/v1/cs/configs",
|
||
data=save_data,
|
||
).strip()
|
||
|
||
# 2xx 下如果返回 false，也按失败处理。
|
||
if response_text and response_text.lower() == "false":
|
||
raise RuntimeError(f"nacos save config failed: {group_name}/{data_id}")
|
||
|
||
# 配置数量等汇总依赖 Nacos 面板缓存，保存后主动清掉。
|
||
NACOS_CACHE.clear()
|
||
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"baseUrl": NACOS_BASE_URL,
|
||
"namespaceId": namespace_id,
|
||
"namespaceName": namespace_name,
|
||
"item": {
|
||
"dataId": data_id,
|
||
"group": group_name,
|
||
"type": (config_type or infer_config_type(data_id)).strip(),
|
||
},
|
||
"content": content,
|
||
}
|
||
|
||
|
||
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 monitored_service_queries() -> list[dict[str, str]]:
|
||
# 读取当前 targets.json 里实际存在的服务名。
|
||
service_names = sorted(
|
||
{
|
||
str(service.get("name") or "").strip()
|
||
for host in load_hosts()
|
||
for service in host.get("services") or []
|
||
if str(service.get("name") or "").strip()
|
||
}
|
||
)
|
||
|
||
# 这里构造要查的注册目标。
|
||
queries: list[dict[str, str]] = []
|
||
|
||
# 逐个业务服务转成 Nacos 注册名。
|
||
for display_name in service_names:
|
||
registry_name = NACOS_SERVICE_NAME_MAP.get(display_name)
|
||
|
||
# 没映射的服务先跳过，比如当前 golang 不走 Nacos 注册。
|
||
if not registry_name:
|
||
continue
|
||
|
||
# 追加查询目标。
|
||
queries.append(
|
||
{
|
||
"displayName": display_name,
|
||
"registryName": registry_name,
|
||
"groupName": NACOS_DISCOVERY_GROUP,
|
||
"rawServiceName": f"{NACOS_DISCOVERY_GROUP}@@{registry_name}",
|
||
}
|
||
)
|
||
|
||
# 返回需要主动查询的服务清单。
|
||
return queries
|
||
|
||
|
||
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 or None,
|
||
},
|
||
)
|
||
|
||
# 读取当前页服务名列表。
|
||
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 or None,
|
||
"healthyOnly": "false",
|
||
},
|
||
)
|
||
|
||
# 返回实例数组。
|
||
return list(payload.get("hosts") or [])
|
||
|
||
|
||
def catalog_instances(namespace_id: str, service_name: str, group_name: str) -> list[dict[str, Any]]:
|
||
try:
|
||
# 调用 catalog/instances 获取注册表视图。
|
||
payload = nacos_request(
|
||
"GET",
|
||
"/nacos/v1/ns/catalog/instances",
|
||
{
|
||
"serviceName": service_name,
|
||
"groupName": group_name,
|
||
"namespaceId": namespace_id or None,
|
||
"clusterName": NACOS_DISCOVERY_CLUSTER_NAME,
|
||
"pageNo": "1",
|
||
"pageSize": str(max(NACOS_PAGE_SIZE, 200)),
|
||
},
|
||
)
|
||
except RuntimeError as exc:
|
||
# 服务不存在时按空列表处理，不让单服务失败拖垮整块 Nacos 面板。
|
||
message = str(exc).lower()
|
||
if "is not found" in message or "service " in message and "not found" in message:
|
||
return []
|
||
raise
|
||
|
||
# 返回 list 字段。
|
||
return list(payload.get("list") or [])
|
||
|
||
|
||
def set_instance_enabled(
|
||
namespace_id: str,
|
||
service_name: str,
|
||
group_name: str,
|
||
*,
|
||
ip: str,
|
||
port: int,
|
||
enabled: bool,
|
||
ephemeral: bool | None = None,
|
||
) -> str:
|
||
# 发布链路用 Nacos enabled 做摘流；这里只改当前 ip:port 这个实例。
|
||
params: dict[str, Any] = {
|
||
"serviceName": service_name,
|
||
"groupName": group_name,
|
||
"namespaceId": namespace_id or None,
|
||
"clusterName": NACOS_DISCOVERY_CLUSTER_NAME or None,
|
||
"ip": ip,
|
||
"port": str(int(port)),
|
||
"enabled": "true" if enabled else "false",
|
||
}
|
||
if ephemeral is not None:
|
||
params["ephemeral"] = "true" if bool(ephemeral) else "false"
|
||
|
||
return nacos_request_text("PUT", "/nacos/v1/ns/instance", params=params).strip()
|
||
|
||
|
||
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)
|
||
|
||
# 先把命名空间归一化。
|
||
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),
|
||
"visible": True,
|
||
}
|
||
for item in namespaces
|
||
]
|
||
|
||
# 建立当前可见 namespace id 集合。
|
||
visible_ids = {item["namespaceId"] for item in normalized_namespaces}
|
||
# 建立当前可见 namespace 名称集合。
|
||
visible_names = {item["name"] for item in normalized_namespaces}
|
||
|
||
# 如果配置里的 namespace 不在可见列表里，就补一个兜底项。
|
||
if (namespace_id and namespace_id not in visible_ids) or (namespace_name and namespace_name not in visible_names):
|
||
normalized_namespaces.append(
|
||
{
|
||
"namespaceId": namespace_id,
|
||
"name": namespace_name,
|
||
"configCount": 0,
|
||
"type": 0,
|
||
"visible": False,
|
||
}
|
||
)
|
||
|
||
# 归一化后的服务列表。
|
||
services: list[dict[str, Any]] = []
|
||
# 统计实例总量。
|
||
instance_total = 0
|
||
# 统计健康实例总量。
|
||
healthy_instance_total = 0
|
||
# 统计启用实例总量。
|
||
enabled_instance_total = 0
|
||
# 统计全健康服务数量。
|
||
healthy_service_total = 0
|
||
|
||
# 只对当前目标 namespace 构建服务和实例视图，避免把 public 下的空查询混进来。
|
||
target_namespaces = [
|
||
namespace
|
||
for namespace in normalized_namespaces
|
||
if namespace["namespaceId"] == namespace_id or namespace["name"] == namespace_name
|
||
]
|
||
if not target_namespaces:
|
||
target_namespaces = [
|
||
{
|
||
"namespaceId": namespace_id,
|
||
"name": namespace_name,
|
||
"configCount": 0,
|
||
"type": 0,
|
||
"visible": False,
|
||
}
|
||
]
|
||
|
||
# 逐个目标 namespace 读取服务和实例。
|
||
for namespace in target_namespaces:
|
||
# 当前 namespace 的自动发现结果。
|
||
raw_services = list_services(namespace["namespaceId"])
|
||
# 这里保存当前 namespace 下所有待查询服务。
|
||
queries_by_key: dict[tuple[str, str], dict[str, str]] = {}
|
||
# 当前目标 namespace 即使 service/list 为空，也要把已知注册服务补查一遍。
|
||
allow_fallback_queries = True
|
||
|
||
# 先收 service/list 返回的服务。
|
||
for raw_name in raw_services:
|
||
# 解析 group 和真正的 serviceName。
|
||
group_name, registry_name = split_grouped_service_name(raw_name)
|
||
# 保存查询项。
|
||
queries_by_key[(group_name, registry_name)] = {
|
||
"displayName": FRIENDLY_SERVICE_NAME_MAP.get(registry_name, registry_name),
|
||
"registryName": registry_name,
|
||
"groupName": group_name,
|
||
"rawServiceName": raw_name,
|
||
}
|
||
|
||
# 再补上 targets.json 里已知的注册服务，避免 service/list 返回空时整页丢数据。
|
||
if allow_fallback_queries:
|
||
for item in monitored_service_queries():
|
||
queries_by_key.setdefault(
|
||
(item["groupName"], item["registryName"]),
|
||
item,
|
||
)
|
||
|
||
# 逐个服务拉实例详情。
|
||
for query in queries_by_key.values():
|
||
# 拉取当前服务实例。
|
||
instance_items = list_instances(namespace["namespaceId"], query["registryName"], query["groupName"])
|
||
|
||
# 常规实例接口为空时，再走 catalog 视图兜底。
|
||
if not instance_items:
|
||
instance_items = catalog_instances(
|
||
namespace["namespaceId"],
|
||
query["registryName"],
|
||
query["groupName"],
|
||
)
|
||
# 归一化实例结构。
|
||
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(
|
||
{
|
||
"namespaceId": namespace["namespaceId"],
|
||
"namespaceName": namespace["name"],
|
||
"rawServiceName": query["rawServiceName"],
|
||
"serviceName": query["displayName"],
|
||
"registryServiceName": query["registryName"],
|
||
"groupName": query["groupName"],
|
||
"level": level,
|
||
"instanceTotal": current_total,
|
||
"healthyTotal": healthy_total,
|
||
"unhealthyTotal": current_total - healthy_total,
|
||
"enabledTotal": enabled_total,
|
||
"disabledTotal": current_total - enabled_total,
|
||
"instances": normalized_instances,
|
||
}
|
||
)
|
||
|
||
# 返回完整 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)
|