354 lines
13 KiB
Python
354 lines
13 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
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)
|