# coding: utf-8 import json import os import re import sqlite3 import sys import time from typing import Any, Dict, List, Optional, Tuple if "/www/server/panel/class" not in sys.path: sys.path.insert(0, "/www/server/panel/class") import public DEFAULT_SNAPSHOT_RETENTION_DAYS = 7 MIN_SNAPSHOT_RETENTION_DAYS = 1 MAX_SNAPSHOT_RETENTION_DAYS = 3650 def _json_dumps(data: Any) -> str: try: return json.dumps(data) except Exception: return "{}" def _json_loads(data: Any, default: Any) -> Any: if isinstance(data, (dict, list)): return data if not isinstance(data, str) or not data: return default try: return json.loads(data) except Exception: return default def _to_float(value: Any, default: float = 0.0) -> float: try: if isinstance(value, str): value = value.strip().replace("%", "") return float(value) except Exception: return default def _to_int(value: Any, default: int = 0) -> int: try: return int(float(value)) except Exception: return default def _to_bytes(value: Any) -> int: if isinstance(value, (int, float)): return int(value) if not isinstance(value, str): return 0 value = value.strip().replace(",", "") if not value: return 0 match = re.match(r"^([0-9]+(?:\.[0-9]+)?)\s*([KMGTPE]?)(?:I?B)?$", value, re.I) if not match: return _to_int(value) number = _to_float(match.group(1)) unit = match.group(2).upper() units = { "": 1, "K": 1024, "M": 1024 ** 2, "G": 1024 ** 3, "T": 1024 ** 4, "P": 1024 ** 5, "E": 1024 ** 6, } return int(number * units.get(unit, 1)) def _to_bytes_from_panel_mem(value: Any) -> int: """面板获取网络内存数值单位通常为 MB,Shell 探针返回单位可能为字节,统一单位""" number = _to_float(value) if number <= 0: return 0 if number < 1024 * 1024: return int(number * 1024 * 1024) return int(number) def _percent(value: Any) -> float: return round(_to_float(value), 2) def _normalize_retention_days(value: Any) -> int: days = _to_int(value, DEFAULT_SNAPSHOT_RETENTION_DAYS) if days < MIN_SNAPSHOT_RETENTION_DAYS: return MIN_SNAPSHOT_RETENTION_DAYS if days > MAX_SNAPSHOT_RETENTION_DAYS: return MAX_SNAPSHOT_RETENTION_DAYS return days TABLE_SCHEMAS = { # 保存每个节点的监控总开关、采集周期(默认采集,在线状态检测,详细指标采集 间隔)、费用(金额,币种代码/符号,费用周期)、到期时间 "node_monitor_setting": { "columns": [ ("id", "INTEGER PRIMARY KEY AUTOINCREMENT"), ("node_id", "INTEGER NOT NULL"), ("enabled", "INTEGER NOT NULL DEFAULT 1"), ("collect_interval", "INTEGER NOT NULL DEFAULT 60"), ("status_interval", "INTEGER NOT NULL DEFAULT 60"), ("detail_interval", "INTEGER NOT NULL DEFAULT 60"), ("snapshot_retention_days", "INTEGER NOT NULL DEFAULT 7"), ("cost_amount", "REAL NOT NULL DEFAULT 0"), ("cost_currency", "TEXT NOT NULL DEFAULT 'CNY'"), ("cost_symbol", "TEXT NOT NULL DEFAULT ''"), ("cost_period", "TEXT NOT NULL DEFAULT 'month'"), ("expire_at", "INTEGER NOT NULL DEFAULT 0"), ("remark", "TEXT NOT NULL DEFAULT ''"), ("created_at", "INTEGER NOT NULL DEFAULT 0"), ("updated_at", "INTEGER NOT NULL DEFAULT 0"), ], "indexes": [ ("idx_node_monitor_setting_node", True, ("node_id",)), ], }, # 保存节点选择监控哪个磁盘、哪个网卡、哪些服务 target_type:disk,nic,service target_key:对象唯一标识 extra:JSON 扩展字段 is_primary:标记展示的磁盘网卡 "node_monitor_target": { "columns": [ ("id", "INTEGER PRIMARY KEY AUTOINCREMENT"), ("node_id", "INTEGER NOT NULL"), ("target_type", "TEXT NOT NULL"), ("target_key", "TEXT NOT NULL"), ("target_name", "TEXT NOT NULL DEFAULT ''"), ("enabled", "INTEGER NOT NULL DEFAULT 1"), ("is_primary", "INTEGER NOT NULL DEFAULT 0"), ("sort", "INTEGER NOT NULL DEFAULT 0"), ("extra", "TEXT NOT NULL DEFAULT '{}'"), ("discovered_at", "INTEGER NOT NULL DEFAULT 0"), ("updated_at", "INTEGER NOT NULL DEFAULT 0"), ], "indexes": [ ("idx_node_monitor_target_unique", True, ("node_id", "target_type", "target_key")), ("idx_node_monitor_target_node_type", False, ("node_id", "target_type")), ], }, # 保存每个节点最新监控快照 每分钟覆盖写 status:-1在线 -0离线 -2采集失败但未判定离线 -4重启等待中 {}_json:当前节点全部磁盘网卡服务数据 "node_monitor_latest": { "columns": [ ("node_id", "INTEGER PRIMARY KEY"), ("ts", "INTEGER NOT NULL DEFAULT 0"), ("status", "INTEGER NOT NULL DEFAULT 0"), ("error_msg", "TEXT NOT NULL DEFAULT ''"), ("cpu_usage", "REAL NOT NULL DEFAULT 0"), ("mem_usage", "REAL NOT NULL DEFAULT 0"), ("mem_used", "INTEGER NOT NULL DEFAULT 0"), ("mem_total", "INTEGER NOT NULL DEFAULT 0"), ("load1", "REAL NOT NULL DEFAULT 0"), ("load5", "REAL NOT NULL DEFAULT 0"), ("load15", "REAL NOT NULL DEFAULT 0"), ("disk_json", "TEXT NOT NULL DEFAULT '[]'"), ("nic_json", "TEXT NOT NULL DEFAULT '[]'"), ("service_json", "TEXT NOT NULL DEFAULT '[]'"), ("raw_json", "TEXT NOT NULL DEFAULT '{}'"), ], "indexes": [ ("idx_node_monitor_latest_ts", False, ("ts",)), ], }, # 保存历史监控数据，用于趋势图、平均值告警、持续时间判断 # 每分钟一条，一台节点一天 1440 条，100 台节点一天 14.4 万条。 # 保留策略： # 分钟级数据保留 7 天 # 5 分钟聚合数据保留 30 天 # 超期数据定时清理 "node_monitor_snapshot": { "columns": [ ("id", "INTEGER PRIMARY KEY AUTOINCREMENT"), ("node_id", "INTEGER NOT NULL"), ("ts", "INTEGER NOT NULL"), ("status", "INTEGER NOT NULL DEFAULT 0"), ("error_msg", "TEXT NOT NULL DEFAULT ''"), ("cpu_usage", "REAL NOT NULL DEFAULT 0"), ("mem_usage", "REAL NOT NULL DEFAULT 0"), ("mem_used", "INTEGER NOT NULL DEFAULT 0"), ("mem_total", "INTEGER NOT NULL DEFAULT 0"), ("load1", "REAL NOT NULL DEFAULT 0"), ("load5", "REAL NOT NULL DEFAULT 0"), ("load15", "REAL NOT NULL DEFAULT 0"), ("disk_json", "TEXT NOT NULL DEFAULT '[]'"), ("nic_json", "TEXT NOT NULL DEFAULT '[]'"), ("service_json", "TEXT NOT NULL DEFAULT '[]'"), ("raw_json", "TEXT NOT NULL DEFAULT '{}'"), ], "indexes": [ ("idx_node_monitor_snapshot_node_ts", False, ("node_id", "ts")), ("idx_node_monitor_snapshot_ts", False, ("ts",)), ], }, # 节点告警设置 通用设置 告警通道 告警最小间隔 允许告警时间范围 "node_alert_setting": { "columns": [ ("id", "INTEGER PRIMARY KEY AUTOINCREMENT"), ("node_id", "INTEGER NOT NULL DEFAULT 0"), ("sender_ids", "TEXT NOT NULL DEFAULT '[]'"), ("send_interval_enabled", "INTEGER NOT NULL DEFAULT 1"), ("send_interval", "INTEGER NOT NULL DEFAULT 1800"), ("time_range_enabled", "INTEGER NOT NULL DEFAULT 0"), ("time_range_start", "INTEGER NOT NULL DEFAULT 0"), ("time_range_end", "INTEGER NOT NULL DEFAULT 86400"), ("created_at", "INTEGER NOT NULL DEFAULT 0"), ("updated_at", "INTEGER NOT NULL DEFAULT 0"), ], "indexes": [ ("idx_node_alert_setting_node", True, ("node_id",)), ], }, # 固定监控项配置 每个节点或全局(node_id=0) 每个metric监控指标 一条 # target_key：磁盘路径或网卡名. CPU/内存/离线/到期为空 # duration_minutes：持续多少分钟触发 # threshold 阈值 xx 超过xx% xxMB/s xxGB xxMB # threshold_unit % MB/s GB "node_alert_item": { "columns": [ ("id", "INTEGER PRIMARY KEY AUTOINCREMENT"), ("node_id", "INTEGER NOT NULL DEFAULT 0"), ("metric", "TEXT NOT NULL"), ("enabled", "INTEGER NOT NULL DEFAULT 0"), ("operator", "TEXT NOT NULL DEFAULT '>='"), ("threshold", "REAL NOT NULL DEFAULT 0"), ("threshold_unit", "TEXT NOT NULL DEFAULT ''"), ("duration_minutes", "INTEGER NOT NULL DEFAULT 1"), ("target_key", "TEXT NOT NULL DEFAULT ''"), ("target_name", "TEXT NOT NULL DEFAULT ''"), ("created_at", "INTEGER NOT NULL DEFAULT 0"), ("updated_at", "INTEGER NOT NULL DEFAULT 0"), ], "indexes": [ ("idx_node_alert_item_unique", True, ("node_id", "metric", "target_key")), ("idx_node_alert_item_node", False, ("node_id",)), ("idx_node_alert_item_enabled", False, ("enabled",)), ], }, # 保存当前告警状态 避免重复发送 # status：normal / alert。 # first_trigger_at：用于持续时间判断 # last_sent_at：用于最小发送间隔 # send_count：告警次数 后续可做次数限制 "node_alert_state": { "columns": [ ("id", "INTEGER PRIMARY KEY AUTOINCREMENT"), ("node_id", "INTEGER NOT NULL"), ("item_id", "INTEGER NOT NULL DEFAULT 0"), ("metric", "TEXT NOT NULL DEFAULT ''"), ("target_key", "TEXT NOT NULL DEFAULT ''"), ("status", "TEXT NOT NULL DEFAULT 'normal'"), ("last_value", "REAL NOT NULL DEFAULT 0"), ("last_message", "TEXT NOT NULL DEFAULT ''"), ("first_trigger_at", "INTEGER NOT NULL DEFAULT 0"), ("last_checked_at", "INTEGER NOT NULL DEFAULT 0"), ("last_sent_at", "INTEGER NOT NULL DEFAULT 0"), ("send_count", "INTEGER NOT NULL DEFAULT 0"), ("updated_at", "INTEGER NOT NULL DEFAULT 0"), ], "indexes": [ ("idx_node_alert_state_node_metric_target", True, ("node_id", "metric", "target_key")), ("idx_node_alert_state_v2_node", False, ("node_id",)), ("idx_node_alert_state_v2_status", False, ("status",)), ], }, # 告警历史 "node_alert_history": { "columns": [ ("id", "INTEGER PRIMARY KEY AUTOINCREMENT"), ("node_id", "INTEGER NOT NULL"), ("item_id", "INTEGER NOT NULL DEFAULT 0"), ("metric", "TEXT NOT NULL DEFAULT ''"), ("target_key", "TEXT NOT NULL DEFAULT ''"), ("trigger_value", "REAL NOT NULL DEFAULT 0"), ("operator", "TEXT NOT NULL DEFAULT ''"), ("threshold", "REAL NOT NULL DEFAULT 0"), ("threshold_unit", "TEXT NOT NULL DEFAULT ''"), ("action", "TEXT NOT NULL DEFAULT ''"), ("message", "TEXT NOT NULL DEFAULT ''"), ("sender_ids", "TEXT NOT NULL DEFAULT '[]'"), ("send_result", "TEXT NOT NULL DEFAULT '{}'"), ("created_at", "INTEGER NOT NULL DEFAULT 0"), ], "indexes": [ ("idx_node_alert_history_v2_node_time", False, ("node_id", "created_at")), ("idx_node_alert_history_v2_metric_time", False, ("metric", "created_at")), ("idx_node_alert_history_v2_time", False, ("created_at",)), ], }, } class NodeMonitorDB: _DB_FILE = public.get_panel_path() + "/data/db/node_monitor.db" def __init__(self): pass def close(self): pass def __enter__(self): return self def __exit__(self, exc_type, exc_value, exc_traceback): self.close() def __del__(self): self.close() def table(self, table_name: str): return public.S(table_name, self._DB_FILE) @classmethod def _column_map(cls, table_name: str) -> Dict[str, str]: schema = TABLE_SCHEMAS[table_name] return {name: definition for name, definition in schema["columns"]} @staticmethod def _create_table_sql(table_name: str) -> str: columns = ["`{}` {}".format(name, definition) for name, definition in TABLE_SCHEMAS[table_name]["columns"]] return "CREATE TABLE IF NOT EXISTS `{}` (\n {}\n);".format(table_name, ",\n ".join(columns)) @staticmethod def _create_index_sql(table_name: str, index_name: str, unique: bool, columns: Tuple[str, ...]) -> str: unique_sql = "UNIQUE " if unique else "" col_sql = ", ".join("`{}`".format(item) for item in columns) return "CREATE {}INDEX IF NOT EXISTS `{}` ON `{}` ({});".format(unique_sql, index_name, table_name, col_sql) @staticmethod def _sqlite_table_exists(conn: sqlite3.Connection, table_name: str) -> bool: row = conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,), ).fetchone() return bool(row) @staticmethod def _sqlite_columns(conn: sqlite3.Connection, table_name: str) -> List[str]: if not NodeMonitorDB._sqlite_table_exists(conn, table_name): return [] return [row[1] for row in conn.execute("PRAGMA table_info(`{}`)".format(table_name)).fetchall()] @staticmethod def _archive_sqlite_table(conn: sqlite3.Connection, table_name: str, suffix: int) -> None: if not NodeMonitorDB._sqlite_table_exists(conn, table_name): return new_name = "{}_legacy_{}".format(table_name, suffix) while NodeMonitorDB._sqlite_table_exists(conn, new_name): suffix += 1 new_name = "{}_legacy_{}".format(table_name, suffix) conn.execute("ALTER TABLE `{}` RENAME TO `{}`".format(table_name, new_name)) @staticmethod def _archive_legacy_alert_tables(conn: sqlite3.Connection) -> None: suffix = int(time.time()) if NodeMonitorDB._sqlite_table_exists(conn, "node_alert_rule"): NodeMonitorDB._archive_sqlite_table(conn, "node_alert_rule", suffix) state_columns = NodeMonitorDB._sqlite_columns(conn, "node_alert_state") if "rule_id" in state_columns: NodeMonitorDB._archive_sqlite_table(conn, "node_alert_state", suffix) history_columns = NodeMonitorDB._sqlite_columns(conn, "node_alert_history") if "rule_id" in history_columns or "value" in history_columns: NodeMonitorDB._archive_sqlite_table(conn, "node_alert_history", suffix) # 初始化数据库 支持新增字段(在表定义中添加 def init_db(self): db_dir = os.path.dirname(self._DB_FILE) if not os.path.exists(db_dir): os.makedirs(db_dir) if not os.path.exists(self._DB_FILE): public.writeFile(self._DB_FILE, "") conn = sqlite3.connect(self._DB_FILE) try: conn.execute("PRAGMA busy_timeout=5000;") try: conn.execute("PRAGMA journal_mode=WAL;") except Exception: pass # 不用兼容旧数据 # self._archive_legacy_alert_tables(conn) cur = conn.cursor() for table_name in TABLE_SCHEMAS: cur.execute(self._create_table_sql(table_name)) cur.execute("PRAGMA table_info(`{}`)".format(table_name)) existing_cols = {row[1] for row in cur.fetchall()} for col_name, col_def in TABLE_SCHEMAS[table_name]["columns"]: if col_name in existing_cols: continue if "PRIMARY KEY" in col_def.upper(): continue cur.execute("ALTER TABLE `{}` ADD COLUMN `{}` {}".format(table_name, col_name, col_def)) for index_name, unique, columns in TABLE_SCHEMAS[table_name].get("indexes", []): cur.execute(self._create_index_sql(table_name, index_name, unique, columns)) conn.commit() finally: conn.close() def get_setting(self, node_id: int, create: bool = True) -> Dict[str, Any]: node_id = int(node_id) data = self.table("node_monitor_setting").where("node_id=?", (node_id,)).find() if isinstance(data, dict) and data: return data if not create: return {} now = int(time.time()) row = { "node_id": node_id, "enabled": 1, "collect_interval": 60, "status_interval": 60, "detail_interval": 60, "snapshot_retention_days": DEFAULT_SNAPSHOT_RETENTION_DAYS, "cost_amount": 0, "cost_currency": "CNY", "cost_symbol": "", "cost_period": "month", "expire_at": 0, "remark": "", "created_at": now, "updated_at": now, } self.table("node_monitor_setting").insert(row) return self.table("node_monitor_setting").where("node_id=?", (node_id,)).find() or row def save_setting(self, node_id: int, data: Dict[str, Any]) -> str: node_id = int(node_id) allowed = { "enabled", "collect_interval", "status_interval", "detail_interval", "snapshot_retention_days", "cost_amount", "cost_currency", "cost_symbol", "cost_period", "expire_at", "remark", } row = {} for key, value in data.items(): if key not in allowed: continue if key == "snapshot_retention_days": value = _normalize_retention_days(value) row[key] = value if not row: return "" row["updated_at"] = int(time.time()) if not self.get_setting(node_id, create=False): self.get_setting(node_id, create=True) res = self.table("node_monitor_setting").where("node_id=?", (node_id,)).update(row) return res if isinstance(res, str) else "" def get_targets(self, node_id: int, target_type: str = "") -> List[Dict[str, Any]]: query = self.table("node_monitor_target") if target_type: data = query.where("node_id=? AND target_type=?", (int(node_id), target_type)).order("sort", "ASC").order("id", "ASC").select() else: data = query.where("node_id=?", (int(node_id),)).order("target_type", "ASC").order("sort", "ASC").order("id", "ASC").select() if isinstance(data, list): for item in data: item["extra"] = _json_loads(item.get("extra"), {}) return data return [] def get_enabled_service_targets(self, node_id: int) -> List[Dict[str, Any]]: return [item for item in self.get_targets(node_id, "service") if int(item.get("enabled", 0)) == 1] def upsert_target(self, node_id: int, target_type: str, target_key: str, data: Dict[str, Any]) -> str: node_id = int(node_id) target_type = str(target_type) target_key = str(target_key) now = int(time.time()) row = { "target_name": data.get("target_name", ""), "enabled": int(data.get("enabled", 1)), "is_primary": int(data.get("is_primary", 0)), "sort": int(data.get("sort", 0)), "extra": _json_dumps(data.get("extra", {})), "updated_at": now, } old = self.table("node_monitor_target").where( "node_id=? AND target_type=? AND target_key=?", (node_id, target_type, target_key) ).find() try: if isinstance(old, dict) and old: self.table("node_monitor_target").where("id=?", (old["id"],)).update(row) else: row.update({ "node_id": node_id, "target_type": target_type, "target_key": target_key, "discovered_at": now, }) self.table("node_monitor_target").insert(row) except Exception as e: return str(e) return "" def set_primary_target(self, node_id: int, target_type: str, target_key: str) -> str: if target_type not in ("disk", "nic"): return "Only disk or nic can be set as the primary target" node_id = int(node_id) target_key = str(target_key) self.table("node_monitor_target").where( "node_id=? AND target_type=?", (node_id, target_type) ).update({"is_primary": 0, "updated_at": int(time.time())}) old = self.table("node_monitor_target").where( "node_id=? AND target_type=? AND target_key=?", (node_id, target_type, target_key) ).find() if not isinstance(old, dict) or not old: err = self.upsert_target(node_id, target_type, target_key, { "target_name": target_key, "enabled": 1, "is_primary": 1, }) return err res = self.table("node_monitor_target").where("id=?", (old["id"],)).update({ "enabled": 1, "is_primary": 1, "updated_at": int(time.time()), }) return res if isinstance(res, str) else "" def save_service_targets(self, node_id: int, services: List[Dict[str, Any]]) -> str: node_id = int(node_id) old = self.get_targets(node_id, "service") old_map = {item["target_key"]: item for item in old} seen = set() for idx, item in enumerate(services): key = str(item.get("target_key") or item.get("name") or "").strip() if not key: continue seen.add(key) err = self.upsert_target(node_id, "service", key, { "target_name": item.get("target_name", key), "enabled": int(item.get("enabled", 1)), "is_primary": 0, "sort": int(item.get("sort", idx)), "extra": item.get("extra", {}), }) if err: return err now = int(time.time()) for key, item in old_map.items(): if key not in seen: self.table("node_monitor_target").where("id=?", (item["id"],)).update({ "enabled": 0, "updated_at": now, }) return "" def sync_discovered_targets(self, node_id: int, disks: List[Dict[str, Any]], nics: List[Dict[str, Any]]) -> None: self._sync_discovered_by_type(node_id, "disk", disks, "path") self._sync_discovered_by_type(node_id, "nic", nics, "name") def _sync_discovered_by_type(self, node_id: int, target_type: str, items: List[Dict[str, Any]], key_name: str) -> None: if not items: return current = self.get_targets(node_id, target_type) current_keys = {item["target_key"] for item in current} has_primary = any(int(item.get("is_primary", 0)) == 1 for item in current) for idx, item in enumerate(items): key = str(item.get(key_name, "")).strip() if not key: continue is_primary = 1 if (not has_primary and idx == 0) else 0 has_primary = has_primary or bool(is_primary) if key in current_keys: old = next((row for row in current if row["target_key"] == key), None) if old: self.table("node_monitor_target").where("id=?", (old["id"],)).update({ "target_name": item.get("name", key), "extra": _json_dumps(item), "updated_at": int(time.time()), }) continue self.upsert_target(node_id, target_type, key, { "target_name": item.get("name", key), "enabled": 1, "is_primary": is_primary, "sort": idx, "extra": item, }) def normalize_monitor_data(self, raw_data: Dict[str, Any], status: int = 1, error_msg: str = "") -> Dict[str, Any]: now = int(time.time()) cpu_usage = 0.0 cpu_data = raw_data.get("cpu", []) if isinstance(cpu_data, (list, tuple)) and cpu_data: cpu_usage = _percent(cpu_data[0]) elif isinstance(cpu_data, dict): cpu_usage = _percent(cpu_data.get("usage", cpu_data.get("used", 0))) mem = raw_data.get("mem", {}) if isinstance(raw_data.get("mem", {}), dict) else {} mem_used = _to_bytes_from_panel_mem(mem.get("memRealUsed", mem.get("used", 0))) mem_total = _to_bytes_from_panel_mem(mem.get("memTotal", mem.get("total", 0))) mem_usage = round(mem_used / mem_total * 100, 2) if mem_total else 0.0 load = raw_data.get("load", {}) if isinstance(raw_data.get("load", {}), dict) else {} disks = self._normalize_disks(raw_data) nics = self._normalize_nics(raw_data) services = raw_data.get("services", []) if not isinstance(services, list): services = [] else: services = self._normalize_services(services) normalized = { "ts": now, "status": int(status), "error_msg": str(error_msg or ""), "cpu_usage": cpu_usage, "mem_usage": mem_usage, "mem_used": mem_used, "mem_total": mem_total, "load1": _to_float(load.get("one", load.get("load1", 0))), "load5": _to_float(load.get("five", load.get("load5", 0))), "load15": _to_float(load.get("fifteen", load.get("load15", 0))), "disk_json": _json_dumps(disks), "nic_json": _json_dumps(nics), "service_json": _json_dumps(services), "raw_json": _json_dumps(raw_data), } return normalized @staticmethod def _normalize_disks(raw_data: Dict[str, Any]) -> List[Dict[str, Any]]: disk_data = raw_data.get("disk", raw_data.get("disks", [])) if not isinstance(disk_data, list): return [] result = [] for item in disk_data: if not isinstance(item, dict): continue byte_size = item.get("byte_size", []) total = used = available = 0 if isinstance(byte_size, (list, tuple)) and len(byte_size) >= 3: total, used, available = _to_int(byte_size[0]), _to_int(byte_size[1]), _to_int(byte_size[2]) elif isinstance(item.get("size"), (list, tuple)) and len(item.get("size")) >= 3: size_data = item.get("size") total, used, available = _to_bytes(size_data[0]), _to_bytes(size_data[1]), _to_bytes(size_data[2]) else: total = _to_int(item.get("total", item.get("size_total", 0))) used = _to_int(item.get("used", item.get("size_used", 0))) available = _to_int(item.get("available", item.get("free", 0))) usage = _percent(item.get("usage", item.get("d_size", 0))) if not usage and total: usage = round(used / total * 100, 2) inode_usage = 0.0 inodes = item.get("inodes", []) if isinstance(inodes, (list, tuple)) and len(inodes) >= 4: inode_usage = _percent(inodes[3]) else: inode_usage = _percent(item.get("inode_usage", 0)) result.append({ "path": item.get("path", item.get("mountpoint", "")), "filesystem": item.get("filesystem", item.get("device", "")), "type": item.get("type", item.get("types", "")), "total": total, "used": used, "available": available, "usage": usage, "inode_usage": inode_usage, }) return [item for item in result if item["path"]] @staticmethod def _normalize_nics(raw_data: Dict[str, Any]) -> List[Dict[str, Any]]: network = raw_data.get("network", raw_data.get("nics", {})) result = [] if isinstance(network, list): items = [(item.get("name", ""), item) for item in network if isinstance(item, dict)] elif isinstance(network, dict): items = [(name, data) for name, data in network.items() if isinstance(data, dict)] else: items = [] for name, item in items: if not name or name == "lo": continue result.append({ "name": name, "up_kbs": _to_float(item.get("up", item.get("up_kbs", 0))), "down_kbs": _to_float(item.get("down", item.get("down_kbs", 0))), "up_total": _to_int(item.get("upTotal", item.get("up_total", 0))), "down_total": _to_int(item.get("downTotal", item.get("down_total", 0))), "up_packets": _to_int(item.get("upPackets", item.get("up_packets", 0))), "down_packets": _to_int(item.get("downPackets", item.get("down_packets", 0))), }) return result @staticmethod def _normalize_services(services: List[Dict[str, Any]]) -> List[Dict[str, Any]]: result = [] for item in services: if not isinstance(item, dict): continue name = str(item.get("name") or item.get("target_key") or "").strip() if not name: continue status = _to_int(item.get("status", 1 if item.get("running") else 0)) result.append({ "name": name, "target_key": str(item.get("target_key") or name), "target_name": str(item.get("target_name") or item.get("name") or name), "status": status, "running": bool(item.get("running", status == 1)), "state": str(item.get("state") or ("active" if status == 1 else "inactive")), "error_msg": str(item.get("error_msg") or ""), "ts": _to_int(item.get("ts"), int(time.time())), }) return result def save_node_snapshot(self, node_id: int, raw_data: Dict[str, Any], status: int = 1, error_msg: str = "") -> str: node_id = int(node_id) if not isinstance(raw_data, dict): raw_data = {} row = self.normalize_monitor_data(raw_data, status=status, error_msg=error_msg) latest_row = {"node_id": node_id} latest_row.update(row) snapshot_row = {"node_id": node_id} snapshot_row.update(row) try: exists = self.table("node_monitor_latest").where("node_id=?", (node_id,)).count() > 0 if exists: data = latest_row.copy() data.pop("node_id", None) self.table("node_monitor_latest").where("node_id=?", (node_id,)).update(data) else: self.table("node_monitor_latest").insert(latest_row) self.table("node_monitor_snapshot").insert(snapshot_row) self.sync_discovered_targets( node_id, _json_loads(row["disk_json"], []), _json_loads(row["nic_json"], []), ) except Exception as e: return str(e) return "" def save_latest_status(self, node_id: int, status: int, error_msg: str = "") -> str: node_id = int(node_id) now = int(time.time()) row = { "node_id": node_id, "ts": now, "status": int(status), "error_msg": str(error_msg or ""), } try: exists = self.table("node_monitor_latest").where("node_id=?", (node_id,)).count() > 0 if exists: data = row.copy() data.pop("node_id", None) self.table("node_monitor_latest").where("node_id=?", (node_id,)).update(data) else: self.table("node_monitor_latest").insert(row) except Exception as e: return str(e) return "" def save_latest_services(self, node_id: int, services: List[Dict[str, Any]], error_msg: str = "") -> str: node_id = int(node_id) service_list = self._normalize_services(services if isinstance(services, list) else []) try: latest = self.get_latest(node_id) raw_json = latest.get("raw_json", {}) if latest else {} if not isinstance(raw_json, dict): raw_json = {} raw_json["services"] = service_list row = { "service_json": _json_dumps(service_list), "raw_json": _json_dumps(raw_json), } if latest: self.table("node_monitor_latest").where("node_id=?", (node_id,)).update(row) else: row.update({ "node_id": node_id, "ts": int(time.time()), "status": 2, "error_msg": str(error_msg or ""), }) self.table("node_monitor_latest").insert(row) except Exception as e: return str(e) return "" def get_latest(self, node_id: int) -> Dict[str, Any]: row = self.table("node_monitor_latest").where("node_id=?", (int(node_id),)).find() if not isinstance(row, dict) or not row: return {} return self._decode_latest_row(row) def _decode_latest_row(self, row: Dict[str, Any]) -> Dict[str, Any]: row = dict(row) row["disk_json"] = _json_loads(row.get("disk_json"), []) row["nic_json"] = _json_loads(row.get("nic_json"), []) row["service_json"] = _json_loads(row.get("service_json"), []) row["raw_json"] = _json_loads(row.get("raw_json"), {}) return row def get_latest_view(self, node_id: int) -> Dict[str, Any]: latest = self.get_latest(node_id) if not latest: return {} targets = self.get_targets(node_id) target_map = {} for item in targets: target_map.setdefault(item["target_type"], []).append(item) latest["selected_disk"] = self._select_primary(latest["disk_json"], target_map.get("disk", []), "path") latest["selected_nic"] = self._select_primary(latest["nic_json"], target_map.get("nic", []), "name") service_targets = [item["target_key"] for item in target_map.get("service", []) if int(item.get("enabled", 0)) == 1] if service_targets: latest["service_json"] = [ item for item in latest["service_json"] if isinstance(item, dict) and (item.get("target_key") in service_targets or item.get("name") in service_targets) ] return latest def get_list_monitor_map(self, node_ids: List[int]) -> Dict[int, Dict[str, Any]]: node_ids = self._normalize_node_ids(node_ids) if not node_ids: return {} setting_list = [] latest_list = [] target_list = [] for chunk_ids in self._chunk_node_ids(node_ids): placeholders = ",".join(["?"] * len(chunk_ids)) params = tuple(chunk_ids) settings = self.table("node_monitor_setting").where( "node_id IN ({})".format(placeholders), params ).select() latest = self.table("node_monitor_latest").where( "node_id IN ({})".format(placeholders), params ).select() targets = self.table("node_monitor_target").where( "node_id IN ({})".format(placeholders), params ).order("node_id", "ASC").order("target_type", "ASC").order("sort", "ASC").order("id", "ASC").select() if isinstance(settings, list): setting_list.extend(settings) if isinstance(latest, list): latest_list.extend(latest) if isinstance(targets, list): target_list.extend(targets) setting_map = {} if isinstance(setting_list, list): setting_map = {int(item["node_id"]): item for item in setting_list if isinstance(item, dict)} latest_map = {} if isinstance(latest_list, list): latest_map = { int(item["node_id"]): self._decode_latest_row(item) for item in latest_list if isinstance(item, dict) } target_map = {node_id: {"disk": [], "nic": [], "service": []} for node_id in node_ids} if isinstance(target_list, list): for item in target_list: if not isinstance(item, dict): continue node_id = int(item.get("node_id", 0)) target_type = item.get("target_type") if node_id not in target_map or target_type not in target_map[node_id]: continue item = dict(item) item["extra"] = _json_loads(item.get("extra"), {}) target_map[node_id][target_type].append(item) return { node_id: self._build_list_monitor_item( node_id, setting_map.get(node_id) or self._default_setting_row(node_id), latest_map.get(node_id) or {}, target_map.get(node_id) or {"disk": [], "nic": [], "service": []}, ) for node_id in node_ids } @staticmethod def _normalize_node_ids(node_ids: List[int]) -> List[int]: result = [] seen = set() for node_id in node_ids: try: node_id = int(node_id) except Exception: continue if node_id <= 0 or node_id in seen: continue seen.add(node_id) result.append(node_id) return result @staticmethod def _chunk_node_ids(node_ids: List[int], chunk_size: int = 500) -> List[List[int]]: return [node_ids[idx:idx + chunk_size] for idx in range(0, len(node_ids), chunk_size)] @staticmethod def _default_setting_row(node_id: int) -> Dict[str, Any]: now = int(time.time()) return { "id": 0, "node_id": int(node_id), "enabled": 1, "collect_interval": 60, "status_interval": 60, "detail_interval": 60, "snapshot_retention_days": DEFAULT_SNAPSHOT_RETENTION_DAYS, "cost_amount": 0, "cost_currency": "CNY", "cost_symbol": "", "cost_period": "month", "expire_at": 0, "remark": "", "created_at": now, "updated_at": now, } def _build_list_monitor_item( self, node_id: int, setting: Dict[str, Any], latest: Dict[str, Any], targets: Dict[str, List[Dict[str, Any]]], ) -> Dict[str, Any]: disks = latest.get("disk_json", []) if isinstance(latest, dict) else [] nics = latest.get("nic_json", []) if isinstance(latest, dict) else [] services = latest.get("service_json", []) if isinstance(latest, dict) else [] if not isinstance(disks, list): disks = [] if not isinstance(nics, list): nics = [] if not isinstance(services, list): services = [] selected_disk = self._select_primary(disks, targets.get("disk", []), "path") selected_nic = self._select_primary(nics, targets.get("nic", []), "name") service_targets = [ item.get("target_key") for item in targets.get("service", []) if int(item.get("enabled", 0)) == 1 and item.get("target_key") ] if service_targets: service_key_set = set(service_targets) services = [ item for item in services if isinstance(item, dict) and (item.get("target_key") in service_key_set or item.get("name") in service_key_set) ] else: services = [] summary = { "ts": _to_int(latest.get("ts", 0)) if latest else 0, "status": _to_int(latest.get("status", 0)) if latest else 0, "error_msg": str(latest.get("error_msg", "")) if latest else "", "cpu_usage": _percent(latest.get("cpu_usage", 0)) if latest else 0, "mem_usage": _percent(latest.get("mem_usage", 0)) if latest else 0, "mem_used": _to_int(latest.get("mem_used", 0)) if latest else 0, "mem_total": _to_int(latest.get("mem_total", 0)) if latest else 0, "load1": _to_float(latest.get("load1", 0)) if latest else 0, "load5": _to_float(latest.get("load5", 0)) if latest else 0, "load15": _to_float(latest.get("load15", 0)) if latest else 0, } latest_view = dict(summary) latest_view.update({ "selected_disk": selected_disk, "selected_nic": selected_nic, "services": services, }) return { "enabled": int(setting.get("enabled", 1)), "setting": setting, "summary": summary, "latest": latest_view, "selected_disk": selected_disk, "selected_nic": selected_nic, "disks": disks, "nics": nics, "services": services, "targets": targets, "cost": { "amount": _to_float(setting.get("cost_amount", 0)), "currency": str(setting.get("cost_currency", "")), "symbol": str(setting.get("cost_symbol", "")), "period": str(setting.get("cost_period", "")), }, "expire": self._build_expire_info(setting.get("expire_at", 0)), "has_latest": bool(latest), } @staticmethod def _build_expire_info(expire_at: Any) -> Dict[str, Any]: expire_at = _to_int(expire_at) if expire_at <= 0: return { "expire_at": 0, "days_left": None, "status": "none", "expired": False, "is_soon": False, } now = int(time.time()) seconds_left = expire_at - now if seconds_left >= 0: days_left = seconds_left // 86400 if seconds_left % 86400: days_left += 1 else: days_left = -((-seconds_left) // 86400) if seconds_left % 86400: days_left -= 1 expired = seconds_left < 0 is_soon = (not expired) and seconds_left <= 7 * 86400 return { "expire_at": expire_at, "days_left": int(days_left), "status": "expired" if expired else ("soon" if is_soon else "normal"), "expired": expired, "is_soon": is_soon, } def get_alert_setting(self, node_id: int, create: bool = True) -> Dict[str, Any]: node_id = max(_to_int(node_id), 0) row = self.table("node_alert_setting").where("node_id=?", (node_id,)).find() if isinstance(row, dict) and row: return self._decode_alert_setting(row) if not create: return {} now = int(time.time()) row = self._default_alert_setting_row(node_id) row.update({"created_at": now, "updated_at": now}) self.table("node_alert_setting").insert({ "node_id": row["node_id"], "sender_ids": _json_dumps(row["sender_ids"]), "send_interval_enabled": row["send_interval_enabled"], "send_interval": row["send_interval"], "time_range_enabled": row["time_range_enabled"], "time_range_start": row["time_range_start"], "time_range_end": row["time_range_end"], "created_at": row["created_at"], "updated_at": row["updated_at"], }) data = self.table("node_alert_setting").where("node_id=?", (node_id,)).find() return self._decode_alert_setting(data) if isinstance(data, dict) and data else row def save_alert_setting(self, node_id: int, data: Dict[str, Any]) -> str: node_id = max(_to_int(node_id), 0) row, err = self._normalize_alert_setting_data(data) if err: return err if not row: return "" now = int(time.time()) row["updated_at"] = now try: old = self.get_alert_setting(node_id, create=False) if old: self.table("node_alert_setting").where("node_id=?", (node_id,)).update(row) else: default_row = self._default_alert_setting_row(node_id) insert_row = { "node_id": node_id, "sender_ids": _json_dumps(default_row["sender_ids"]), "send_interval_enabled": default_row["send_interval_enabled"], "send_interval": default_row["send_interval"], "time_range_enabled": default_row["time_range_enabled"], "time_range_start": default_row["time_range_start"], "time_range_end": default_row["time_range_end"], "created_at": now, "updated_at": now, } insert_row.update(row) self.table("node_alert_setting").insert(insert_row) except Exception as e: return str(e) return "" @staticmethod def _default_alert_setting_row(node_id: int) -> Dict[str, Any]: return { "id": 0, "node_id": max(_to_int(node_id), 0), "sender_ids": [], "send_interval_enabled": 1, "send_interval": 1800, "time_range_enabled": 0, "time_range_start": 0, "time_range_end": 86400, "created_at": 0, "updated_at": 0, "scope": "global" if max(_to_int(node_id), 0) == 0 else "node", } @staticmethod def _decode_alert_setting(row: Dict[str, Any]) -> Dict[str, Any]: row = dict(row) row["sender_ids"] = NodeMonitorDB._normalize_sender_ids(row.get("sender_ids", [])) row["scope"] = "global" if _to_int(row.get("node_id", 0)) == 0 else "node" return row @staticmethod def _normalize_sender_ids(value: Any) -> List[str]: if isinstance(value, str): value = _json_loads(value, []) if not isinstance(value, list): return [] result = [] seen = set() for item in value: sender_id = str(item).strip() if not sender_id or sender_id in seen: continue seen.add(sender_id) result.append(sender_id) return result @staticmethod def _normalize_alert_setting_data(data: Dict[str, Any]) -> Tuple[Dict[str, Any], str]: row = {} if "sender_ids" in data: row["sender_ids"] = _json_dumps(NodeMonitorDB._normalize_sender_ids(data.get("sender_ids"))) if "send_interval_enabled" in data: row["send_interval_enabled"] = 1 if _to_int(data.get("send_interval_enabled")) else 0 if "send_interval" in data: row["send_interval"] = max(_to_int(data.get("send_interval"), 1800), 60) if "time_range_enabled" in data: row["time_range_enabled"] = 1 if _to_int(data.get("time_range_enabled")) else 0 has_start = "time_range_start" in data has_end = "time_range_end" in data if has_start: row["time_range_start"] = _to_int(data.get("time_range_start"), 0) if has_end: row["time_range_end"] = _to_int(data.get("time_range_end"), 86400) start = row.get("time_range_start", _to_int(data.get("time_range_start", 0))) end = row.get("time_range_end", _to_int(data.get("time_range_end", 86400))) if has_start or has_end: if not (0 <= start <= 86400 and 0 <= end <= 86400): return {}, "time range format error" if _to_int(data.get("time_range_enabled", row.get("time_range_enabled", 0))) and start == end: return {}, "time range format error" return row, "" def get_alert_items(self, node_id: int = -1, include_global: bool = False, enabled_only: bool = False) -> List[Dict[str, Any]]: where = [] params = [] node_id = _to_int(node_id, -1) if node_id >= 0: if include_global and node_id > 0: where.append("(node_id=0 OR node_id=?)") params.append(node_id) else: where.append("node_id=?") params.append(node_id) if enabled_only: where.append("enabled=1") query = self.table("node_alert_item") if where: query = query.where(" AND ".join(where), tuple(params)) data = query.order("node_id", "ASC").order("metric", "ASC").order("target_key", "ASC").order("id", "ASC").select() if not isinstance(data, list): return [] return [self._decode_alert_item(item) for item in data if isinstance(item, dict)] def get_alert_item(self, item_id: int) -> Dict[str, Any]: row = self.table("node_alert_item").where("id=?", (_to_int(item_id),)).find() if not isinstance(row, dict) or not row: return {} return self._decode_alert_item(row) def save_alert_item(self, data: Dict[str, Any]) -> Tuple[int, str]: row, err = self._normalize_alert_item_data(data) if err: return 0, err item_id = _to_int(data.get("id", data.get("item_id", 0))) now = int(time.time()) row["updated_at"] = now try: if item_id: old = self.get_alert_item(item_id) if not old: return 0, "Alert item does not exist" self.table("node_alert_item").where("id=?", (item_id,)).update(row) return item_id, "" old = self.table("node_alert_item").where( "node_id=? AND metric=? AND target_key=?", (row["node_id"], row["metric"], row["target_key"]), ).find() if isinstance(old, dict) and old: self.table("node_alert_item").where("id=?", (old["id"],)).update(row) return _to_int(old.get("id")), "" row["created_at"] = now new_id = self.table("node_alert_item").insert(row) return _to_int(new_id, 0), "" if not isinstance(new_id, str) else new_id except Exception as e: return 0, str(e) def save_alert_items(self, node_id: int, items: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], str]: node_id = max(_to_int(node_id), 0) if not isinstance(items, list): return [], "items format error" result = [] for item in items: if not isinstance(item, dict): continue data = dict(item) data["node_id"] = node_id item_id, err = self.save_alert_item(data) if err: return result, err result.append(self.get_alert_item(item_id)) return result, "" def set_alert_item_status(self, item_id: int, enabled: int) -> str: item_id = _to_int(item_id) if item_id <= 0: return "item_id cannot be empty" try: self.table("node_alert_item").where("id=?", (item_id,)).update({ "enabled": 1 if _to_int(enabled) else 0, "updated_at": int(time.time()), }) except Exception as e: return str(e) return "" def delete_alert_item(self, item_id: int) -> str: item_id = _to_int(item_id) if item_id <= 0: return "item_id cannot be empty" try: self.table("node_alert_item").where("id=?", (item_id,)).delete() self.table("node_alert_state").where("item_id=?", (item_id,)).delete() except Exception as e: return str(e) return "" @staticmethod def _decode_alert_item(row: Dict[str, Any]) -> Dict[str, Any]: row = dict(row) row["scope"] = "global" if _to_int(row.get("node_id", 0)) == 0 else "node" return row @staticmethod def _normalize_alert_item_data(data: Dict[str, Any]) -> Tuple[Dict[str, Any], str]: metric = str(data.get("metric", "")).strip()[:64] if not metric: return {}, "metric cannot be empty" operator = str(data.get("operator", ">=")).strip() if operator not in (">", ">=", "<", "<=", "==", "!="): return {}, "Unsupported alert operator" row = { "node_id": max(_to_int(data.get("node_id", 0)), 0), "metric": metric, "enabled": 1 if _to_int(data.get("enabled", 0)) else 0, "operator": operator, "threshold": _to_float(data.get("threshold", 0)), "threshold_unit": str(data.get("threshold_unit", "") or "").strip()[:32], "duration_minutes": max(_to_int(data.get("duration_minutes", 1)), 0), "target_key": str(data.get("target_key", "") or "").strip()[:255], "target_name": str(data.get("target_name", "") or "").strip()[:255], } return row, "" def get_alert_state(self, node_id: int, metric: str, target_key: str = "") -> Dict[str, Any]: row = self.table("node_alert_state").where( "node_id=? AND metric=? AND target_key=?", (_to_int(node_id), str(metric or ""), str(target_key or "")), ).find() if not isinstance(row, dict) or not row: return {} return row def upsert_alert_state(self, node_id: int, metric: str, target_key: str, data: Dict[str, Any]) -> str: node_id = _to_int(node_id) metric = str(metric or "") target_key = str(target_key or "") now = int(time.time()) row = { "item_id": _to_int(data.get("item_id", 0)), "status": str(data.get("status", "normal")), "last_value": _to_float(data.get("last_value", 0)), "last_message": str(data.get("last_message", ""))[:1000], "first_trigger_at": _to_int(data.get("first_trigger_at", 0)), "last_checked_at": _to_int(data.get("last_checked_at", now)), "last_sent_at": _to_int(data.get("last_sent_at", 0)), "send_count": _to_int(data.get("send_count", 0)), "updated_at": now, } try: old = self.get_alert_state(node_id, metric, target_key) if old: self.table("node_alert_state").where("id=?", (old["id"],)).update(row) else: row.update({ "node_id": node_id, "metric": metric, "target_key": target_key, }) self.table("node_alert_state").insert(row) except Exception as e: return str(e) return "" def get_alert_states(self, node_id: int = 0, metric: str = "", status: str = "", item_id: int = 0, limit: int = 1000) -> List[Dict[str, Any]]: where = [] params = [] if _to_int(node_id): where.append("node_id=?") params.append(_to_int(node_id)) if metric: where.append("metric=?") params.append(str(metric)) if status: where.append("status=?") params.append(str(status)) if _to_int(item_id): where.append("item_id=?") params.append(_to_int(item_id)) query = self.table("node_alert_state") if where: query = query.where(" AND ".join(where), tuple(params)) data = query.order("updated_at", "DESC").limit(max(1, min(_to_int(limit, 1000), 5000))).select() return data if isinstance(data, list) else [] def add_alert_history(self, data: Dict[str, Any]) -> str: row = { "node_id": _to_int(data.get("node_id", 0)), "item_id": _to_int(data.get("item_id", 0)), "metric": str(data.get("metric", "")), "target_key": str(data.get("target_key", "")), "trigger_value": _to_float(data.get("trigger_value", 0)), "operator": str(data.get("operator", "")), "threshold": _to_float(data.get("threshold", 0)), "threshold_unit": str(data.get("threshold_unit", "") or "").strip()[:32], "action": str(data.get("action", "")), "message": str(data.get("message", ""))[:2000], "sender_ids": _json_dumps(self._normalize_sender_ids(data.get("sender_ids", []))), "send_result": _json_dumps(data.get("send_result", {})), "created_at": _to_int(data.get("created_at", int(time.time()))), } try: self.table("node_alert_history").insert(row) except Exception as e: return str(e) return "" def get_alert_history(self, node_id: int = 0, metric: str = "", item_id: int = 0, limit: int = 100) -> List[Dict[str, Any]]: where = [] params = [] if _to_int(node_id): where.append("node_id=?") params.append(_to_int(node_id)) if metric: where.append("metric=?") params.append(str(metric)) if _to_int(item_id): where.append("item_id=?") params.append(_to_int(item_id)) query = self.table("node_alert_history") if where: query = query.where(" AND ".join(where), tuple(params)) data = query.order("created_at", "DESC").limit(max(1, min(_to_int(limit, 100), 1000))).select() if not isinstance(data, list): return [] for item in data: if isinstance(item, dict): item["sender_ids"] = _json_loads(item.get("sender_ids"), []) item["send_result"] = _json_loads(item.get("send_result"), {}) return data @staticmethod def _select_primary(items: List[Dict[str, Any]], targets: List[Dict[str, Any]], item_key: str) -> Dict[str, Any]: if not isinstance(items, list) or not items: return {} primary_key = "" for target in targets: if int(target.get("is_primary", 0)) == 1: primary_key = target.get("target_key", "") break if primary_key: for item in items: if isinstance(item, dict) and item.get(item_key) == primary_key: return item return items[0] if isinstance(items[0], dict) else {} def get_snapshots(self, node_id: int, start_ts: int = 0, end_ts: int = 0, limit: int = 1440) -> List[Dict[str, Any]]: where = ["node_id=?"] params = [int(node_id)] if start_ts: where.append("ts>=?") params.append(int(start_ts)) if end_ts: where.append("ts<=?") params.append(int(end_ts)) limit = max(1, min(int(limit), 10080)) data = self.table("node_monitor_snapshot").where(" AND ".join(where), tuple(params)).order("ts", "DESC").limit(limit).select() if not isinstance(data, list): return [] result = [self._decode_latest_row(item) for item in data if isinstance(item, dict)] result.reverse() return result def cleanup_snapshots(self, before_ts: int) -> str: try: self.table("node_monitor_snapshot").where("ts str: now = int(time.time()) try: settings = self.table("node_monitor_setting").field("node_id,snapshot_retention_days").select() if not isinstance(settings, list) or not settings: return self.cleanup_snapshots(now - DEFAULT_SNAPSHOT_RETENTION_DAYS * 86400) cleaned_nodes = set() for item in settings: if not isinstance(item, dict): continue node_id = _to_int(item.get("node_id", 0)) if node_id <= 0 or node_id in cleaned_nodes: continue cleaned_nodes.add(node_id) retention_days = _normalize_retention_days(item.get("snapshot_retention_days")) before_ts = now - retention_days * 86400 self.table("node_monitor_snapshot").where("node_id=? AND ts