diff --git a/foms/__init__.py b/foms/__init__.py index d4d6580..13ad8a8 100644 --- a/foms/__init__.py +++ b/foms/__init__.py @@ -1,8 +1,12 @@ from .init import ( - create_cluster, update_cluster, delete_cluster, get_cluster_list, - create_node, update_node, delete_node, get_node_list, - create_sync_db, update_sync_db, delete_sync_db, get_sync_db_list, - create_sync_dir, update_sync_dir, delete_sync_dir, get_sync_dir_list, - agent_heartbeat, agent_report_health, trigger_switchover, trigger_switchback, + create_cluster, update_cluster, delete_cluster, + create_node, update_node, delete_node, + create_sync_db, update_sync_db, delete_sync_db, + create_sync_dir, update_sync_dir, delete_sync_dir, + agent_heartbeat, agent_report_health, + trigger_switchover, trigger_switchback, get_dashboard_stats, + agent_report_metrics, agent_push_logs, + get_node_detail, get_node_metrics_history, + exec_remote_command, agent_poll_commands, agent_report_command_result, ) diff --git a/foms/init.py b/foms/init.py index f6c7900..39d3059 100644 --- a/foms/init.py +++ b/foms/init.py @@ -321,6 +321,191 @@ async def get_dashboard_stats(request): } +# ═══════════════════════════════════════════ +# Metrics & Log Collection +# ═══════════════════════════════════════════ + +async def agent_report_metrics(request, params_kw): + """Agent上报系统指标（CPU/内存/磁盘/负载/网络）""" + env = ServerEnv() + node_id = params_kw.get('node_id', '') if hasattr(params_kw, 'get') else '' + metrics = params_kw.get('metrics', {}) if hasattr(params_kw, 'get') else {} + if isinstance(metrics, str): + metrics = json.loads(metrics) + async with get_sor_context(env, DBNAME) as sor: + await sor.C('node_metrics', { + 'id': getID(), + 'node_id': node_id, + 'cpu_percent': metrics.get('cpu_percent'), + 'mem_percent': metrics.get('mem_percent'), + 'mem_used_mb': metrics.get('mem_used_mb'), + 'mem_total_mb': metrics.get('mem_total_mb'), + 'disk_percent': metrics.get('disk_percent'), + 'disk_used_gb': metrics.get('disk_used_gb'), + 'disk_path': metrics.get('disk_path', '/'), + 'load_1m': metrics.get('load_1m'), + 'load_5m': metrics.get('load_5m'), + 'load_15m': metrics.get('load_15m'), + 'net_rx_mb': metrics.get('net_rx_mb'), + 'net_tx_mb': metrics.get('net_tx_mb'), + 'process_count': metrics.get('process_count'), + 'collected_at': curDateString(), + }) + return {'status': 'ok'} + + +async def agent_push_logs(request, params_kw): + """Agent推送日志条目到管理中心""" + env = ServerEnv() + node_id = params_kw.get('node_id', '') if hasattr(params_kw, 'get') else '' + logs = params_kw.get('logs', []) if hasattr(params_kw, 'get') else [] + if isinstance(logs, str): + logs = json.loads(logs) + if not logs: + return {'status': 'ok', 'count': 0} + async with get_sor_context(env, DBNAME) as sor: + count = 0 + for entry in logs[:100]: # max 100 per batch + await sor.C('node_logs', { + 'id': getID(), + 'node_id': node_id, + 'log_source': entry.get('source', 'syslog'), + 'log_level': entry.get('level', 'INFO'), + 'message': entry.get('message', '')[:2000], + 'logged_at': entry.get('time', curDateString()), + }) + count += 1 + return {'status': 'ok', 'count': count} + + +async def get_node_detail(request, params_kw): + """获取节点详情（实时指标 + 最新日志 + 基本信息）""" + env = ServerEnv() + node_id = params_kw.get('node_id', '') if hasattr(params_kw, 'get') else '' + async with get_sor_context(env, DBNAME) as sor: + node = await sor.R('nodes', {'id': node_id}) + metrics = await sor.sqlExe( + 'SELECT * FROM node_metrics WHERE node_id=${nid}$ ORDER BY collected_at DESC LIMIT 1', + {'nid': node_id} + ) + logs = await sor.sqlExe( + 'SELECT log_source, log_level, message, logged_at FROM node_logs WHERE node_id=${nid}$ ORDER BY logged_at DESC LIMIT 50', + {'nid': node_id} + ) + node_info = {} + if node: + n = node[0] + node_info = { + 'id': getattr(n, 'id', ''), + 'host': getattr(n, 'host', ''), + 'name': getattr(n, 'name', ''), + 'ip': getattr(n, 'ip', ''), + 'role': getattr(n, 'role', ''), + 'status': getattr(n, 'status', ''), + 'agent_version': getattr(n, 'agent_version', ''), + 'last_heartbeat': str(getattr(n, 'last_heartbeat', '')), + } + return { + 'node': node_info, + 'latest_metrics': dict(metrics[0]) if metrics else {}, + 'recent_logs': [{ + 'source': getattr(l, 'log_source', ''), + 'level': getattr(l, 'log_level', ''), + 'message': getattr(l, 'message', ''), + 'time': str(getattr(l, 'logged_at', '')), + } for l in (logs or [])], + } + + +async def get_node_metrics_history(request, params_kw): + """获取节点历史指标（用于图表）""" + env = ServerEnv() + node_id = params_kw.get('node_id', '') if hasattr(params_kw, 'get') else '' + hours = int(params_kw.get('hours', 1)) if hasattr(params_kw, 'get') else 1 + async with get_sor_context(env, DBNAME) as sor: + rows = await sor.sqlExe( + 'SELECT cpu_percent, mem_percent, disk_percent, load_1m, collected_at ' + 'FROM node_metrics WHERE node_id=${nid}$ AND collected_at > DATE_SUB(NOW(), INTERVAL ${h}$ HOUR) ' + 'ORDER BY collected_at ASC LIMIT 200', + {'nid': node_id, 'h': hours} + ) + return {'metrics': [{ + 'cpu': getattr(r, 'cpu_percent', 0), + 'mem': getattr(r, 'mem_percent', 0), + 'disk': getattr(r, 'disk_percent', 0), + 'load': getattr(r, 'load_1m', 0), + 'time': str(getattr(r, 'collected_at', '')), + } for r in (rows or [])]} + + +# ═══════════════════════════════════════════ +# Remote Command Execution +# ═══════════════════════════════════════════ + +async def exec_remote_command(request, params_kw): + """向Agent下发远程命令，Agent执行后返回结果""" + # Commands are pushed to agent via its next heartbeat poll, or via a pending queue + # For simplicity, this stores the command; agent reads it via poll_commands + env = ServerEnv() + node_id = params_kw.get('node_id', '') if hasattr(params_kw, 'get') else '' + command = params_kw.get('command', '') if hasattr(params_kw, 'get') else '' + if not node_id or not command: + return {'status': 'error', 'message': 'node_id and command required'} + # Store pending command (simple approach: use sync_databases or a new table) + # For MVP: just validate and return that command was queued + # Agent checks for commands via agent_poll_commands below + cmd_id = getID() + now = curDateString() + async with get_sor_context(env, DBNAME) as sor: + await sor.C('node_logs', { + 'id': cmd_id, + 'node_id': node_id, + 'log_source': 'foms_cmd', + 'log_level': 'INFO', + 'message': f'[PENDING] {command}', + 'logged_at': now, + }) + return { + 'status': 'ok', + 'message': '命令已下发，等待Agent执行', + 'cmd_id': cmd_id, + } + + +async def agent_poll_commands(request, params_kw): + """Agent轮询获取待执行命令""" + env = ServerEnv() + node_id = params_kw.get('node_id', '') if hasattr(params_kw, 'get') else '' + async with get_sor_context(env, DBNAME) as sor: + rows = await sor.sqlExe( + "SELECT id, message FROM node_logs WHERE node_id=${nid}$ AND log_source='foms_cmd' AND message LIKE '[PENDING]%' ORDER BY logged_at ASC LIMIT 5", + {'nid': node_id} + ) + commands = [] + for r in (rows or []): + msg = getattr(r, 'message', '') + cmd = msg.replace('[PENDING] ', '') if '[PENDING]' in msg else msg + commands.append({'id': getattr(r, 'id', ''), 'command': cmd}) + return {'commands': commands} + + +async def agent_report_command_result(request, params_kw): + """Agent上报命令执行结果""" + env = ServerEnv() + cmd_id = params_kw.get('cmd_id', '') if hasattr(params_kw, 'get') else '' + node_id = params_kw.get('node_id', '') if hasattr(params_kw, 'get') else '' + output = params_kw.get('output', '') if hasattr(params_kw, 'get') else '' + exit_code = params_kw.get('exit_code', -1) if hasattr(params_kw, 'get') else -1 + async with get_sor_context(env, DBNAME) as sor: + await sor.U('node_logs', { + 'id': cmd_id, + 'log_source': 'foms_cmd', + 'log_level': 'INFO' if exit_code == 0 else 'ERROR', + 'message': f'[EXIT:{exit_code}] {output[:2000]}', + }) + return {'status': 'ok'} + + # ═══════════════════════════════════════════ # ServerEnv Registration # ═══════════════════════════════════════════ @@ -363,4 +548,13 @@ def load_foms(): env.trigger_switchback = trigger_switchback # Dashboard env.get_dashboard_stats = get_dashboard_stats + # Metrics & Logs + env.agent_report_metrics = agent_report_metrics + env.agent_push_logs = agent_push_logs + env.get_node_detail = get_node_detail + env.get_node_metrics_history = get_node_metrics_history + # Remote Command + env.exec_remote_command = exec_remote_command + env.agent_poll_commands = agent_poll_commands + env.agent_report_command_result = agent_report_command_result return True diff --git a/init/data.yaml b/init/data.yaml index e4c2b43..1cff890 100644 --- a/init/data.yaml +++ b/init/data.yaml @@ -29,6 +29,9 @@ appcodes: - id: user_status name: 用户状态 hierarchy_flg: 0 +- id: log_level + name: 日志级别 + hierarchy_flg: 0 appcodes_kv: - id: cluster_status_running @@ -150,3 +153,19 @@ appcodes_kv: parentid: user_status k: 1 v: 停用 +- id: log_level_info + parentid: log_level + k: INFO + v: 信息 +- id: log_level_warn + parentid: log_level + k: WARN + v: 警告 +- id: log_level_error + parentid: log_level + k: ERROR + v: 错误 +- id: log_level_crit + parentid: log_level + k: CRIT + v: 严重 diff --git a/models/node_logs.json b/models/node_logs.json new file mode 100644 index 0000000..b6fe840 --- /dev/null +++ b/models/node_logs.json @@ -0,0 +1,26 @@ +{ + "summary": [ + { + "name": "node_logs", + "title": "节点日志收集", + "primary": ["id"], + "catelog": "entity" + } + ], + "fields": [ + {"name": "id", "title": "日志ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "node_id", "title": "节点ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "log_source", "title": "日志来源", "type": "str", "length": 64}, + {"name": "log_level", "title": "日志级别", "type": "str", "length": 16}, + {"name": "message", "title": "日志内容", "type": "text"}, + {"name": "logged_at", "title": "日志时间", "type": "datetime", "nullable": "no"} + ], + "indexes": [ + {"name": "idx_nl_node", "idxtype": "index", "idxfields": ["node_id"]}, + {"name": "idx_nl_time", "idxtype": "index", "idxfields": ["logged_at"]}, + {"name": "idx_nl_level", "idxtype": "index", "idxfields": ["log_level"]} + ], + "codes": [ + {"field": "log_level", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='log_level'"} + ] +} diff --git a/models/node_metrics.json b/models/node_metrics.json new file mode 100644 index 0000000..37c99e8 --- /dev/null +++ b/models/node_metrics.json @@ -0,0 +1,32 @@ +{ + "summary": [ + { + "name": "node_metrics", + "title": "节点系统指标", + "primary": ["id"], + "catelog": "indication" + } + ], + "fields": [ + {"name": "id", "title": "记录ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "node_id", "title": "节点ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "cpu_percent", "title": "CPU使用率(%)", "type": "float", "length": 5, "dec": 1}, + {"name": "mem_percent", "title": "内存使用率(%)", "type": "float", "length": 5, "dec": 1}, + {"name": "mem_used_mb", "title": "已用内存(MB)", "type": "int"}, + {"name": "mem_total_mb", "title": "总内存(MB)", "type": "int"}, + {"name": "disk_percent", "title": "磁盘使用率(%)", "type": "float", "length": 5, "dec": 1}, + {"name": "disk_used_gb", "title": "已用磁盘(GB)", "type": "float", "length": 8, "dec": 1}, + {"name": "disk_path", "title": "磁盘挂载点", "type": "str", "length": 128}, + {"name": "load_1m", "title": "1分钟负载", "type": "float", "length": 5, "dec": 2}, + {"name": "load_5m", "title": "5分钟负载", "type": "float", "length": 5, "dec": 2}, + {"name": "load_15m", "title": "15分钟负载", "type": "float", "length": 5, "dec": 2}, + {"name": "net_rx_mb", "title": "网络接收(MB)", "type": "float", "length": 10, "dec": 1}, + {"name": "net_tx_mb", "title": "网络发送(MB)", "type": "float", "length": 10, "dec": 1}, + {"name": "process_count", "title": "进程数", "type": "int"}, + {"name": "collected_at", "title": "采集时间", "type": "datetime", "nullable": "no"} + ], + "indexes": [ + {"name": "idx_nm_node", "idxtype": "index", "idxfields": ["node_id"]}, + {"name": "idx_nm_time", "idxtype": "index", "idxfields": ["collected_at"]} + ] +} diff --git a/scripts/foms-agent.py b/scripts/foms-agent.py index 29ee6c7..331ba61 100644 --- a/scripts/foms-agent.py +++ b/scripts/foms-agent.py @@ -8,6 +8,9 @@ FOMS Agent — 部署在主机(A)和备机(B)上的守护进程 3. MySQL主从复制状态监控 4. 文件rsync同步 5. 响应切换/切回指令 + 6. 系统指标采集 (CPU/内存/磁盘/网络) + 7. 系统日志推送到管理中心 + 8. 远程命令执行 """ import os import sys @@ -19,6 +22,12 @@ import urllib.request import urllib.error from datetime import datetime +try: + import psutil + HAS_PSUTIL = True +except ImportError: + HAS_PSUTIL = False + # ═══ Configuration ═══ MANAGEMENT_URL = os.environ.get('FOMS_MANAGEMENT_URL', 'http://localhost:9080') NODE_ID = os.environ.get('FOMS_NODE_ID', '') @@ -228,21 +237,38 @@ class FomsAgent: # ═══ Main Loop ═══ def run(self): log.info(f"FOMS Agent v{self.version} starting...") + if HAS_PSUTIL: + log.info(" psutil: available (system metrics enabled)") + else: + log.warning(" psutil: NOT installed — pip install psutil") self.fetch_config() last_health_report = 0 - HEALTH_REPORT_INTERVAL = 60 # report health every 60s + last_metrics_report = 0 + last_command_poll = 0 + HEALTH_REPORT_INTERVAL = 60 + METRICS_INTERVAL = 30 while True: try: self.heartbeat() - - # Report health (includes MySQL status + rsync) now = time.time() + + # Report health (MySQL status + rsync) if now - last_health_report > HEALTH_REPORT_INTERVAL: self.report_health() last_health_report = now + # Collect & report system metrics + if now - last_metrics_report > METRICS_INTERVAL: + self.report_metrics() + last_metrics_report = now + + # Poll for remote commands + if now - last_command_poll > 15: + self.poll_and_execute_commands() + last_command_poll = now + time.sleep(HEARTBEAT_INTERVAL) except KeyboardInterrupt: log.info("Agent stopped") @@ -251,6 +277,80 @@ class FomsAgent: log.error(f"Loop error: {e}") time.sleep(HEARTBEAT_INTERVAL) + # ═══ System Metrics ═══ + def collect_metrics(self): + """Collect system metrics via psutil""" + if not HAS_PSUTIL: + return {} + try: + cpu = psutil.cpu_percent(interval=0.5) + mem = psutil.virtual_memory() + disk = psutil.disk_usage('/') + load = os.getloadavg() if hasattr(os, 'getloadavg') else (0, 0, 0) + net = psutil.net_io_counters() + uptime = int(time.time() - psutil.boot_time()) + return { + 'cpu_percent': round(cpu, 1), + 'mem_percent': round(mem.percent, 1), + 'mem_used_mb': round(mem.used / 1024 / 1024), + 'mem_total_mb': round(mem.total / 1024 / 1024), + 'disk_percent': round(disk.percent, 1), + 'disk_used_gb': round(disk.used / 1024 / 1024 / 1024, 1), + 'disk_path': '/', + 'load_1m': round(load[0], 2), + 'load_5m': round(load[1], 2), + 'load_15m': round(load[2], 2), + 'net_rx_mb': round(net.bytes_recv / 1024 / 1024, 1), + 'net_tx_mb': round(net.bytes_sent / 1024 / 1024, 1), + 'process_count': len(psutil.pids()), + } + except Exception as e: + log.error(f"collect_metrics error: {e}") + return {} + + def report_metrics(self): + """Send metrics to management""" + metrics = self.collect_metrics() + if not metrics: + return + data = {'node_id': self.node_id, 'metrics': metrics} + result = self._api_post('agent_report_metrics.dspy', data) + if result and result.get('status') == 'ok': + log.debug(f"Metrics reported: CPU={metrics.get('cpu_percent')}% MEM={metrics.get('mem_percent')}%") + else: + log.warning(f"Metrics report failed: {result}") + + # ═══ Remote Command Execution ═══ + def poll_and_execute_commands(self): + """Poll management for pending commands, execute, report results""" + result = self._api_get('agent_poll_commands.dspy?node_id=' + self.node_id) + if not result or not result.get('commands'): + return + for cmd in result['commands']: + cmd_id = cmd.get('id', '') + command = cmd.get('command', '') + log.info(f"Executing command [{cmd_id}]: {command}") + try: + proc = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30) + output = (proc.stdout + proc.stderr)[:2000] + self._api_post('agent_report_command_result.dspy', { + 'node_id': self.node_id, + 'cmd_id': cmd_id, + 'output': output, + 'exit_code': proc.returncode, + }) + log.info(f"Command [{cmd_id}] done: exit={proc.returncode}") + except subprocess.TimeoutExpired: + self._api_post('agent_report_command_result.dspy', { + 'node_id': self.node_id, 'cmd_id': cmd_id, + 'output': 'TIMEOUT (30s)', 'exit_code': -1, + }) + except Exception as e: + self._api_post('agent_report_command_result.dspy', { + 'node_id': self.node_id, 'cmd_id': cmd_id, + 'output': str(e), 'exit_code': -1, + }) + if __name__ == '__main__': if not NODE_ID: diff --git a/scripts/load_path.py b/scripts/load_path.py index 1d3ef7d..1b97e8c 100644 --- a/scripts/load_path.py +++ b/scripts/load_path.py @@ -68,11 +68,24 @@ PATHS_LOGINED = [ "/sync_directories_list/add_sync_directories_list.dspy", "/sync_directories_list/update_sync_directories_list.dspy", "/sync_directories_list/delete_sync_directories_list.dspy", + # Metrics & logs + "/api/agent_report_metrics.dspy", + "/api/agent_push_logs.dspy", + "/api/node_detail.dspy", + "/api/node_metrics_history.dspy", + # Remote command + "/api/exec_remote_command.dspy", + "/api/agent_poll_commands.dspy", + "/api/agent_report_command_result.dspy", ] # Agent heartbeat endpoint: no auth (agents use token) PATHS_ANY.append("/api/agent_heartbeat.dspy") PATHS_ANY.append("/api/agent_report_health.dspy") +PATHS_ANY.append("/api/agent_report_metrics.dspy") +PATHS_ANY.append("/api/agent_push_logs.dspy") +PATHS_ANY.append("/api/agent_poll_commands.dspy") +PATHS_ANY.append("/api/agent_report_command_result.dspy") if __name__ == '__main__': for p in PATHS_ANY: diff --git a/wwwroot/api/agent_poll_commands.dspy b/wwwroot/api/agent_poll_commands.dspy new file mode 100644 index 0000000..258f715 --- /dev/null +++ b/wwwroot/api/agent_poll_commands.dspy @@ -0,0 +1,2 @@ +result = await agent_poll_commands(request, params_kw) +return json.dumps(result, ensure_ascii=False) diff --git a/wwwroot/api/agent_push_logs.dspy b/wwwroot/api/agent_push_logs.dspy new file mode 100644 index 0000000..38a480a --- /dev/null +++ b/wwwroot/api/agent_push_logs.dspy @@ -0,0 +1,2 @@ +result = await agent_push_logs(request, params_kw) +return json.dumps(result, ensure_ascii=False) diff --git a/wwwroot/api/agent_report_command_result.dspy b/wwwroot/api/agent_report_command_result.dspy new file mode 100644 index 0000000..5d5a300 --- /dev/null +++ b/wwwroot/api/agent_report_command_result.dspy @@ -0,0 +1,2 @@ +result = await agent_report_command_result(request, params_kw) +return json.dumps(result, ensure_ascii=False) diff --git a/wwwroot/api/agent_report_metrics.dspy b/wwwroot/api/agent_report_metrics.dspy new file mode 100644 index 0000000..2a76301 --- /dev/null +++ b/wwwroot/api/agent_report_metrics.dspy @@ -0,0 +1,2 @@ +result = await agent_report_metrics(request, params_kw) +return json.dumps(result, ensure_ascii=False) diff --git a/wwwroot/api/exec_remote_command.dspy b/wwwroot/api/exec_remote_command.dspy new file mode 100644 index 0000000..8ace0e9 --- /dev/null +++ b/wwwroot/api/exec_remote_command.dspy @@ -0,0 +1,2 @@ +result = await exec_remote_command(request, params_kw) +return json.dumps(result, ensure_ascii=False) diff --git a/wwwroot/api/node_detail.dspy b/wwwroot/api/node_detail.dspy new file mode 100644 index 0000000..4970bc3 --- /dev/null +++ b/wwwroot/api/node_detail.dspy @@ -0,0 +1,2 @@ +result = await get_node_detail(request, params_kw) +return json.dumps(result, ensure_ascii=False, default=str) diff --git a/wwwroot/api/node_metrics_history.dspy b/wwwroot/api/node_metrics_history.dspy new file mode 100644 index 0000000..ab16f77 --- /dev/null +++ b/wwwroot/api/node_metrics_history.dspy @@ -0,0 +1,2 @@ +result = await get_node_metrics_history(request, params_kw) +return json.dumps(result, ensure_ascii=False, default=str)