#!/usr/bin/env python3 """🛡️ Sentinel — 多节点 SenSu 集群性能监控插件""" import logging import asyncio import os import time 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 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", ""), "enabled": n.get("enabled", True), "online": c.get("online", False), "system": c.get("system", {}), "last_seen": c.get("last_seen", 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(), "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 = await self._fetch_system(node) _NODE_CACHE[node.get("id", "")] = { "online": online, "system": sys_info or {}, "last_seen": time.time(), "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: 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: if resp.status == 200: return True, _json.loads(resp.read().decode()), "" return False, None, f"HTTP {resp.status}" except Exception as e: return False, None, str(e)[:120] 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", ""), "enabled": n.get("enabled", True), "online": c.get("online", False), "system": c.get("system", {}), "last_seen": c.get("last_seen", 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 已关闭")