qinglong
279 lines
12 KiB
Python
279 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""🛡️ Sentinel — 多节点 SenSu 集群性能监控插件"""
|
|
import logging
|
|
import asyncio
|
|
import os
|
|
import time
|
|
import shutil
|
|
import yaml
|
|
import urllib.request
|
|
import json as _json
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
try:
|
|
from sdk.plugin_command_decorator import plugin_command
|
|
from sdk.plugin_web import PluginWebMixin
|
|
except ImportError:
|
|
def plugin_command(n=None, d=None, p=None):
|
|
def deco(f): f._is_plugin_command = True; f._command_name = n or f.__name__; return f
|
|
return deco
|
|
class PluginWebMixin:
|
|
def register_web_page(self, *a, **k): pass
|
|
def push_sse_event(self, *a, **k): pass
|
|
async def handle_sse(self, r): pass
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_NODE_CACHE: Dict[str, dict] = {}
|
|
|
|
|
|
class Plugin(PluginWebMixin):
|
|
def __init__(self, plugin_name=None, config=None, bridge=None):
|
|
PluginWebMixin.__init__(self)
|
|
self.plugin_name = plugin_name or "sentinel"
|
|
self.config = config or {}
|
|
self.bridge = bridge
|
|
self.network_bridge = None
|
|
self._poll_task = None
|
|
self._running = False
|
|
self._config_path = os.path.join(os.path.dirname(__file__), "config.yaml")
|
|
|
|
async def initialize(self):
|
|
logger.info(f"🛡️ Sentinel 初始化: {self.plugin_name}")
|
|
try:
|
|
internet = self.bridge.service_manager.get_service("internet")
|
|
from bridges.plugin_network_bridge import PluginNetworkBridge
|
|
self.network_bridge = PluginNetworkBridge(self.plugin_name, internet, self.bridge)
|
|
except Exception as e:
|
|
logger.warning(f"Sentinel network skip: {e}")
|
|
|
|
# 注册 WebUI 页面 (key 必须匹配 plugin_name 才能在侧边栏访问)
|
|
dash_path = os.path.join(os.path.dirname(__file__), "dashboard.html")
|
|
if os.path.exists(dash_path):
|
|
with open(dash_path) as f:
|
|
self.register_web_page("sentinel", "🛡️ Sentinel", f.read(), icon="S")
|
|
|
|
# 注册 API 路由
|
|
if self.network_bridge:
|
|
for method, path, handler in [
|
|
("GET", "/api/nodes", self._handle_nodes),
|
|
("POST", "/api/nodes", self._handle_add_node),
|
|
("POST", "/api/nodes/test", self._handle_test_node),
|
|
("POST", "/api/nodes/delete", self._handle_delete_node),
|
|
]:
|
|
await self.network_bridge.register_http_route(
|
|
path, handler, methods=[method], require_auth=True
|
|
)
|
|
|
|
# 启动后台轮询
|
|
interval = self.config.get("settings", {}).get("poll_interval", 3)
|
|
self._running = True
|
|
self._poll_task = asyncio.create_task(self._poll_loop(interval))
|
|
logger.info(f"🛡️ Sentinel 就绪 (轮询间隔: {interval}s)")
|
|
|
|
# ── 配置读写 ──
|
|
|
|
def _read_config(self) -> dict:
|
|
try:
|
|
with open(self._config_path, "r") as f:
|
|
return yaml.safe_load(f) or {}
|
|
except FileNotFoundError:
|
|
# 从模板创建
|
|
template = self._config_path + ".example"
|
|
if os.path.exists(template):
|
|
import shutil
|
|
shutil.copy(template, self._config_path)
|
|
with open(self._config_path, "r") as f:
|
|
return yaml.safe_load(f) or {}
|
|
return {}
|
|
except Exception:
|
|
return {}
|
|
|
|
def _get_nodes(self) -> List[dict]:
|
|
return self._read_config().get("nodes", [])
|
|
|
|
def _save_nodes(self, nodes: List[dict]):
|
|
cfg = self._read_config()
|
|
cfg["nodes"] = nodes
|
|
with open(self._config_path, "w") as f:
|
|
yaml.dump(cfg, f, default_flow_style=False, allow_unicode=True)
|
|
|
|
# ── HTTP handlers ──
|
|
|
|
async def _handle_nodes(self, request):
|
|
from aiohttp import web
|
|
nodes = self._get_nodes()
|
|
result = []
|
|
for n in nodes:
|
|
c = _NODE_CACHE.get(n.get("id", ""), {})
|
|
result.append({
|
|
"id": n.get("id", ""), "name": n.get("name", ""), "url": n.get("url", ""),
|
|
"notes": n.get("notes", ""),
|
|
"base_path": n.get("base_path", "/SenSu"),
|
|
"enabled": n.get("enabled", True),
|
|
"online": c.get("online", False),
|
|
"system": c.get("system", {}),
|
|
"platform": c.get("platform", {}),
|
|
"last_seen": c.get("last_seen", 0),
|
|
"latency_ms": c.get("latency_ms", 0),
|
|
"error": c.get("error", ""),
|
|
})
|
|
return web.json_response({"nodes": result})
|
|
|
|
async def _handle_add_node(self, request):
|
|
from aiohttp import web
|
|
try:
|
|
import secrets as _secrets
|
|
data = await request.json()
|
|
nodes = self._get_nodes()
|
|
node_id = data.get("id") or f"node-{_secrets.token_hex(4)}"
|
|
node = {
|
|
"id": node_id,
|
|
"name": data.get("name", "").strip(),
|
|
"url": data.get("url", "").strip().rstrip("/"),
|
|
"api_key": data.get("api_key", "").strip(),
|
|
"base_path": data.get("base_path", "/SenSu").strip() or "/SenSu",
|
|
"notes": data.get("notes", "").strip(),
|
|
"enabled": data.get("enabled", True),
|
|
}
|
|
if not node["name"] or not node["url"]:
|
|
return web.json_response({"ok": False, "error": "名称和 URL 不能为空"}, status=400)
|
|
existing = [i for i, n in enumerate(nodes) if n.get("id") == node_id]
|
|
if existing:
|
|
old = nodes[existing[0]]
|
|
# 保留未填写的敏感字段
|
|
if not node["api_key"]:
|
|
node["api_key"] = old.get("api_key", "")
|
|
nodes[existing[0]] = node
|
|
else:
|
|
nodes.append(node)
|
|
self._save_nodes(nodes)
|
|
_NODE_CACHE.pop(node_id, None)
|
|
logger.info(f"🛡️ 节点已保存: {node['name']} ({node['url']})")
|
|
return web.json_response({"ok": True, "node": node})
|
|
except Exception as e:
|
|
return web.json_response({"ok": False, "error": str(e)}, status=500)
|
|
|
|
async def _handle_delete_node(self, request):
|
|
from aiohttp import web
|
|
try:
|
|
data = await request.json()
|
|
node_id = data.get("id", "")
|
|
nodes = [n for n in self._get_nodes() if n.get("id") != node_id]
|
|
self._save_nodes(nodes)
|
|
_NODE_CACHE.pop(node_id, None)
|
|
logger.info(f"🛡️ 节点已删除: {node_id}")
|
|
return web.json_response({"ok": True})
|
|
except Exception as e:
|
|
return web.json_response({"ok": False, "error": str(e)}, status=500)
|
|
|
|
async def _handle_test_node(self, request):
|
|
from aiohttp import web
|
|
try:
|
|
data = await request.json()
|
|
raw_url = data.get("url", "").strip()
|
|
if not raw_url:
|
|
return web.json_response({"ok": False, "error": "URL 为空"}, status=400)
|
|
# SSRF 防护: 仅允许 HTTP/HTTPS
|
|
from urllib.parse import urlparse
|
|
parsed = urlparse(raw_url)
|
|
if parsed.scheme not in ("http", "https"):
|
|
return web.json_response({"ok": False, "error": "仅允许 HTTP/HTTPS"}, status=403)
|
|
if not parsed.hostname:
|
|
return web.json_response({"ok": False, "error": "无效的 URL"}, status=400)
|
|
node = {"url": raw_url.rstrip("/"), "api_key": data.get("api_key", "").strip()}
|
|
ok, info, err = await self._fetch_system(node)
|
|
return web.json_response({"ok": ok, "system": info, "error": err})
|
|
except Exception as e:
|
|
return web.json_response({"ok": False, "error": str(e)}, status=500)
|
|
|
|
# ── 后台轮询 ──
|
|
|
|
async def _poll_loop(self, interval: int):
|
|
while self._running:
|
|
nodes = self._get_nodes()
|
|
for node in nodes:
|
|
if not node.get("enabled", True):
|
|
continue
|
|
online, sys_info, err, latency = await self._fetch_system(node)
|
|
_NODE_CACHE[node.get("id", "")] = {
|
|
"online": online,
|
|
"system": sys_info or {},
|
|
"platform": (sys_info or {}).get("platform", {}),
|
|
"last_seen": time.time(),
|
|
"latency_ms": latency,
|
|
"error": err if not online else "",
|
|
}
|
|
# 每次轮询都推送 SSE (确保实时性)
|
|
if nodes:
|
|
await self.push_sse_event("nodes_update", {"nodes": self._build_cache_list()})
|
|
await asyncio.sleep(interval)
|
|
|
|
async def _fetch_system(self, node: dict):
|
|
"""获取远程系统状态 — 用线程池避免同事件循环 HTTP 死锁"""
|
|
url = node.get("url", "")
|
|
api_key = node.get("api_key", "")
|
|
base = node.get("base_path", "/SenSu")
|
|
timeout = self.config.get("settings", {}).get("request_timeout", 5)
|
|
|
|
def _sync_fetch():
|
|
try:
|
|
t0 = time.time()
|
|
headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
|
|
req = urllib.request.Request(f"{url}{base}/api/system", headers=headers)
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
elapsed = round((time.time() - t0) * 1000) # ms
|
|
if resp.status == 200:
|
|
return True, _json.loads(resp.read().decode()), "", elapsed
|
|
return False, None, f"HTTP {resp.status}", elapsed
|
|
except Exception as e:
|
|
elapsed = round((time.time() - t0) * 1000)
|
|
return False, None, str(e)[:120], elapsed
|
|
|
|
return await asyncio.get_event_loop().run_in_executor(None, _sync_fetch)
|
|
|
|
def _build_cache_list(self) -> list:
|
|
nodes = self._get_nodes()
|
|
result = []
|
|
for n in nodes:
|
|
c = _NODE_CACHE.get(n.get("id", ""), {})
|
|
result.append({
|
|
"id": n.get("id", ""), "name": n.get("name", ""), "url": n.get("url", ""),
|
|
"notes": n.get("notes", ""),
|
|
"base_path": n.get("base_path", "/SenSu"),
|
|
"enabled": n.get("enabled", True),
|
|
"online": c.get("online", False),
|
|
"system": c.get("system", {}),
|
|
"platform": c.get("platform", {}),
|
|
"last_seen": c.get("last_seen", 0),
|
|
"latency_ms": c.get("latency_ms", 0),
|
|
"error": c.get("error", ""),
|
|
})
|
|
return result
|
|
|
|
async def handle_sse(self, request):
|
|
return await PluginWebMixin.handle_sse(self, request)
|
|
|
|
# ── TUI 命令 ──
|
|
|
|
@plugin_command(name="sentinel", description="Sentinel 集群状态")
|
|
async def cmd_sentinel(self, *args):
|
|
nodes = self._get_nodes()
|
|
if not nodes:
|
|
return "🛡️ Sentinel: 无配置节点"
|
|
lines = ["🛡️ Sentinel 集群状态:", "=" * 40]
|
|
for n in nodes:
|
|
c = _NODE_CACHE.get(n.get("id", ""), {})
|
|
icon = "🟢" if c.get("online") else "🔴"
|
|
cpu = c.get("system", {}).get("cpu", {}).get("percent", "?")
|
|
mem = c.get("system", {}).get("memory", {}).get("percent", "?")
|
|
lines.append(f" {icon} {n['name']} CPU:{cpu}% MEM:{mem}%")
|
|
return "\n".join(lines)
|
|
|
|
async def shutdown(self):
|
|
self._running = False
|
|
if self._poll_task:
|
|
self._poll_task.cancel()
|
|
logger.info("🛡️ Sentinel 已关闭")
|