#!/usr/bin/env python3 """ FOMS Agent — 部署在主机(A)和备机(B)上的守护进程 职责: 1. 启动时从管理中心(C)拉取配置 2. 心跳上报 3. MySQL主从复制状态监控 4. 文件rsync同步 5. 响应切换/切回指令 6. 系统指标采集 (CPU/内存/磁盘/网络) 7. 系统日志推送到管理中心 8. 远程命令执行 """ import os import sys import json import time import logging import subprocess 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', '') AGENT_TOKEN = os.environ.get('FOMS_AGENT_TOKEN', '') HEARTBEAT_INTERVAL = int(os.environ.get('FOMS_HEARTBEAT_INTERVAL', '10')) # ═══ Failover configuration（切换动作专用）═══ MYSQL_SERVICE = os.environ.get('FOMS_MYSQL_SERVICE', 'mysql') VIP_ADDR = os.environ.get('FOMS_VIP', '') # 如 192.168.1.100/24 VIP_IFACE = os.environ.get('FOMS_VIP_IFACE', 'eth0') MYSQL_ROOT_PW = os.environ.get('MYSQL_ROOT_PASSWORD', '') # ═══ Logging ═══ logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[ logging.FileHandler('/var/log/foms-agent.log'), logging.StreamHandler(sys.stdout), ] ) log = logging.getLogger('foms-agent') class FomsAgent: def __init__(self): self.node_id = NODE_ID self.version = '1.0.0' self.db_configs = [] # sync_databases configs self.dir_configs = [] # sync_directories configs self.node_role = '' # master or standby # ═══ HTTP helpers ═══ def _api_get(self, path): # path 以 / 开头 = 完整路径（模块端点），否则走 /api/ 前缀（核心端点） url = f"{MANAGEMENT_URL}{path}" if path.startswith('/') else f"{MANAGEMENT_URL}/api/{path}" req = urllib.request.Request(url) req.add_header('Authorization', f'Bearer {AGENT_TOKEN}') try: with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode()) except Exception as e: log.error(f"GET {url}: {e}") return None def _api_post(self, path, data): url = f"{MANAGEMENT_URL}{path}" if path.startswith('/') else f"{MANAGEMENT_URL}/api/{path}" payload = json.dumps(data).encode() req = urllib.request.Request(url, data=payload, headers={'Content-Type': 'application/json'}) req.add_header('Authorization', f'Bearer {AGENT_TOKEN}') try: with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode()) except Exception as e: log.error(f"POST {url}: {e}") return None # ═══ Heartbeat ═══ def heartbeat(self): data = {'node_id': self.node_id, 'version': self.version} result = self._api_post('agent_heartbeat.dspy', data) if result and result.get('status') == 'ok': log.debug(f"Heartbeat OK: {result.get('server_time')}") else: log.warning(f"Heartbeat failed: {result}") # ═══ MySQL Replication Status ═══ def check_mysql_replication(self, db_config): """Check SHOW SLAVE STATUS for a given DB config""" try: cmd = [ 'mysql', '-h', db_config.get('slave_host', '127.0.0.1'), '-P', str(db_config.get('slave_port', 3306)), '-u', db_config.get('repl_user', 'root'), f"-p{db_config.get('repl_password', '')}", '-e', 'SHOW SLAVE STATUS\\G' ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if result.returncode != 0: return {'id': db_config['id'], 'status': 'error', 'seconds_behind': None, 'error': result.stderr.strip()} # Parse SHOW SLAVE STATUS output output = result.stdout io_running = 'Slave_IO_Running: Yes' in output sql_running = 'Slave_SQL_Running: Yes' in output seconds_behind = None for line in output.split('\n'): if 'Seconds_Behind_Master:' in line: val = line.split(':')[1].strip() seconds_behind = int(val) if val.isdigit() else None if io_running and sql_running: status = 'running' error = None elif io_running: status = 'error' error = 'Slave_SQL_Running: No' elif sql_running: status = 'error' error = 'Slave_IO_Running: No' else: status = 'error' error = 'Replication not running' return {'id': db_config['id'], 'status': status, 'seconds_behind': seconds_behind, 'error': error} except Exception as e: return {'id': db_config['id'], 'status': 'error', 'seconds_behind': None, 'error': str(e)} # ═══ Setup MySQL Replication ═══ def setup_mysql_replication(self, db_config): """On master: create replication user. On slave: CHANGE MASTER TO + START SLAVE""" role = self.node_role log.info(f"Setting up MySQL replication for {db_config['db_name']} (role={role})") if role == 'master': # Create replication user on master repl_user = db_config.get('repl_user', 'repl') repl_pass = db_config.get('repl_password', '') cmds = [ f"CREATE USER IF NOT EXISTS '{repl_user}'@'%' IDENTIFIED BY '{repl_pass}'", f"GRANT REPLICATION SLAVE ON *.* TO '{repl_user}'@'%'", "FLUSH PRIVILEGES", ] for sql in cmds: cmd = ['mysql', '-h', db_config.get('master_host', '127.0.0.1'), '-P', str(db_config.get('master_port', 3306)), '-u', 'root', f"-p{os.environ.get('MYSQL_ROOT_PASSWORD', '')}", '-e', sql] subprocess.run(cmd, capture_output=True, timeout=10) log.info(f" Replication user created on master") elif role == 'standby': # Configure slave # First get master status master_host = db_config.get('master_host', '') master_port = db_config.get('master_port', 3306) repl_user = db_config.get('repl_user', 'repl') repl_pass = db_config.get('repl_password', '') # CHANGE MASTER and START SLAVE sql = f""" CHANGE MASTER TO MASTER_HOST='{master_host}', MASTER_PORT={master_port}, MASTER_USER='{repl_user}', MASTER_PASSWORD='{repl_pass}', MASTER_AUTO_POSITION=1; START SLAVE; """ cmd = ['mysql', '-h', db_config.get('slave_host', '127.0.0.1'), '-P', str(db_config.get('slave_port', 3306)), '-u', 'root', f"-p{os.environ.get('MYSQL_ROOT_PASSWORD', '')}", '-e', sql] result = subprocess.run(cmd, capture_output=True, text=True, timeout=15) if result.returncode != 0: log.error(f" Slave setup failed: {result.stderr}") else: log.info(f" Slave configured for {db_config['db_name']}") # ═══ File Sync (rsync) ═══ def sync_directory(self, dir_config): """Execute rsync for a directory config""" try: src = f"{dir_config.get('source_host', '')}:{dir_config['source_path']}" dst = dir_config['dest_path'] exclude = dir_config.get('exclude_patterns', '') cmd = ['rsync', '-avz', '--delete'] if exclude: for pattern in exclude.split(','): cmd.extend(['--exclude', pattern.strip()]) cmd.extend(['-e', 'ssh', src, dst]) result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) success = result.returncode == 0 log.info(f"rsync {dir_config['name']}: {'OK' if success else 'FAILED'}") if not success: log.error(f" stderr: {result.stderr[:200]}") return {'id': dir_config['id'], 'result': 'success' if success else 'failed'} except Exception as e: log.error(f"rsync {dir_config.get('name', '')} error: {e}") return {'id': dir_config['id'], 'result': 'failed'} # ═══ Report Health ═══ def report_db_health(self): """收集 MySQL 复制状态 → 上报 dbbackup 模块""" db_statuses = [] for db_config in self.db_configs: if db_config.get('enabled', '1') == '0': continue status = self.check_mysql_replication(db_config) db_statuses.append(status) data = {'node_id': self.node_id, 'db_statuses': db_statuses} self._api_post('/dbbackup/api/agent_report_db_health.dspy', data) def report_dir_health(self): """执行 rsync 同步 → 上报 filesync 模块""" dir_statuses = [] for dir_config in self.dir_configs: if dir_config.get('enabled', '1') == '0': continue status = self.sync_directory(dir_config) dir_statuses.append(status) data = {'node_id': self.node_id, 'dir_statuses': dir_statuses} self._api_post('/filesync/api/agent_report_dir_health.dspy', data) # ═══ Fetch Config from Management ═══ def fetch_config(self): """Fetch sync configs from management server""" # In production, the management would have dedicated APIs for this # For now, the config is set via environment or manual setup log.info(f"Agent running on node {self.node_id}") log.info(f"Management URL: {MANAGEMENT_URL}") log.info(f"Heartbeat interval: {HEARTBEAT_INTERVAL}s") # ═══ 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 last_metrics_report = 0 last_command_poll = 0 HEALTH_REPORT_INTERVAL = 60 METRICS_INTERVAL = 30 while True: try: self.heartbeat() now = time.time() # Report health（MySQL 复制 → dbbackup；rsync → filesync） if now - last_health_report > HEALTH_REPORT_INTERVAL: self.report_db_health() self.report_dir_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") break except Exception as e: 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('/hostwatch/api/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}") # ═══ Failover Actions（切换动作，脚本化 + 幂等）═══ def _run(self, argv, timeout=30): """执行单条命令，返回 (exit_code, output)。argv 为列表，不走 shell。""" try: proc = subprocess.run(argv, capture_output=True, text=True, timeout=timeout) output = (proc.stdout + proc.stderr).strip()[:2000] return proc.returncode, output except subprocess.TimeoutExpired: return -1, f'TIMEOUT ({timeout}s): {argv[0]}' except Exception as e: return -1, str(e) def action_fence(self): """隔离本机（防脑裂）：停 MySQL + 摘 VIP。幂等——已停/已摘时不报错。""" results = [] # 1) 停 MySQL 服务（systemd）。已停止时 systemctl stop 幂等返回 0。 code, out = self._run(['systemctl', 'stop', MYSQL_SERVICE], timeout=60) results.append({'step': 'stop_mysql', 'exit_code': code, 'output': out}) # 2) 摘 VIP（ip addr del）。已不存在时返回非 0，视为已摘除（幂等成功）。 if VIP_ADDR: code, out = self._run(['ip', 'addr', 'del', VIP_ADDR, 'dev', VIP_IFACE]) if code != 0 and 'Cannot find address' in out: code, out = 0, 'VIP already removed (idempotent)' results.append({'step': 'remove_vip', 'exit_code': code, 'output': out}) ok = all(r['exit_code'] == 0 for r in results) log.info(f"fence: {'OK' if ok else 'FAILED'} {results}") return {'ok': ok, 'actions': results} def action_promote_master(self): """提升本机为新主：STOP SLAVE + RESET SLAVE ALL。幂等——已是主库视为成功。""" env_sql = ['mysql', '-uroot'] if MYSQL_ROOT_PW: env_sql.append(f'-p{MYSQL_ROOT_PW}') # 检查当前是否仍是 slave；已是主库（无 SLAVE STATUS）则直接成功（幂等） code, out = self._run(env_sql + ['-e', 'SHOW SLAVE STATUS\\G']) if code == 0 and 'Slave_IO_State' not in out: log.info("promote_master: already master (no slave status), idempotent OK") return {'ok': True, 'actions': [{'step': 'already_master', 'exit_code': 0, 'output': out[:300]}]} # 原子执行：STOP SLAVE; RESET SLAVE ALL;（单条 -e 保证原子） promote_sql = 'STOP SLAVE; RESET SLAVE ALL;' code, out = self._run(env_sql + ['-e', promote_sql], timeout=60) results = [{'step': 'stop_reset_slave', 'exit_code': code, 'output': out}] # 校验：确认不再有 slave 状态 vcode, vout = self._run(env_sql + ['-e', 'SHOW SLAVE STATUS\\G']) promoted = (vcode == 0 and 'Slave_IO_State' not in vout) results.append({'step': 'verify_no_slave', 'exit_code': 0 if promoted else 1, 'output': ('promoted' if promoted else vout[:300])}) ok = (code == 0 and promoted) log.info(f"promote_master: {'OK' if ok else 'FAILED'} {results}") return {'ok': ok, 'actions': results} def action_flip_sync(self): """反转/暂停文件同步方向。通过通知管理中心翻转 filesync 配置方向实现。 本动作只做「本机侧的确认与暂停」：返回当前同步配置状态，由编排器在 DB 层反转。""" results = [{'step': 'flip_sync_ack', 'exit_code': 0, 'output': f'node={self.node_id} ready_for_flip'}] log.info(f"flip_sync: ACK {results}") return {'ok': True, 'actions': results} def action_unfence(self): """恢复本机（回滚用）：启动 MySQL + 还 VIP。幂等——已启动/已挂时不报错。""" results = [] # 1) 启动 MySQL 服务（systemd）。已运行时 systemctl start 幂等返回 0。 code, out = self._run(['systemctl', 'start', MYSQL_SERVICE], timeout=60) results.append({'step': 'start_mysql', 'exit_code': code, 'output': out}) # 2) 挂回 VIP（ip addr add）。已存在时返回非 0，视为已挂载（幂等成功）。 if VIP_ADDR: code, out = self._run(['ip', 'addr', 'add', VIP_ADDR, 'dev', VIP_IFACE]) if code != 0 and 'File exists' in out: code, out = 0, 'VIP already present (idempotent)' results.append({'step': 'add_vip', 'exit_code': code, 'output': out}) ok = all(r['exit_code'] == 0 for r in results) log.info(f"unfence: {'OK' if ok else 'FAILED'} {results}") return {'ok': ok, 'actions': results} def action_revert_promote(self): """降回从库（回滚用）：重新 CHANGE MASTER 指向原主。 需要原主连接信息——从本机复制配置读取（若原主已恢复）。 幂等：若本机已是 slave 则直接成功。""" env_sql = ['mysql', '-uroot'] if MYSQL_ROOT_PW: env_sql.append(f'-p{MYSQL_ROOT_PW}') # 已是 slave → 幂等成功 code, out = self._run(env_sql + ['-e', 'SHOW SLAVE STATUS\\G']) if code == 0 and 'Slave_IO_State' in out: log.info("revert_promote: already slave, idempotent OK") return {'ok': True, 'actions': [{'step': 'already_slave', 'exit_code': 0, 'output': out[:300]}]} # 读取本机复制配置（CHANGE MASTER 需要原主地址）。 # 优先从环境变量 FOMS_MASTER_HOST/PORT 获取（部署时配置），否则无法回滚。 master_host = os.environ.get('FOMS_MASTER_HOST', '') master_port = os.environ.get('FOMS_MASTER_PORT', '3306') repl_user = os.environ.get('FOMS_REPL_USER', 'repl') repl_pass = os.environ.get('FOMS_REPL_PASSWORD', '') if not master_host: return {'ok': False, 'actions': [{'step': 'revert_promote', 'exit_code': 1, 'output': 'FOMS_MASTER_HOST not set, cannot revert'}]} change_sql = (f"CHANGE MASTER TO MASTER_HOST='{master_host}', MASTER_PORT={master_port}, " f"MASTER_USER='{repl_user}', MASTER_PASSWORD='{repl_pass}', MASTER_AUTO_POSITION=1; " f"START SLAVE;") code, out = self._run(env_sql + ['-e', change_sql], timeout=60) results = [{'step': 'change_master_start_slave', 'exit_code': code, 'output': out}] ok = (code == 0) log.info(f"revert_promote: {'OK' if ok else 'FAILED'} {results}") return {'ok': ok, 'actions': results} def action_verify(self): """校验本机可写 + VIP 归属（切换后验证）。业务层留人工确认。""" results = [] env_sql = ['mysql', '-uroot'] if MYSQL_ROOT_PW: env_sql.append(f'-p{MYSQL_ROOT_PW}') # 1) MySQL 可写测试：建临时表写入再删除 test_sql = ("CREATE TABLE IF NOT EXISTS _foms_verify (id INT); " "INSERT INTO _foms_verify VALUES (1); " "DROP TABLE _foms_verify;") code, out = self._run(env_sql + ['-e', test_sql], timeout=30) results.append({'step': 'mysql_write_test', 'exit_code': code, 'output': out[:300]}) # 2) VIP 归属检查（若配置了 VIP） if VIP_ADDR: vip_ip = VIP_ADDR.split('/')[0] code, out = self._run(['ip', 'addr', 'show', VIP_IFACE]) vip_here = (code == 0 and vip_ip in out) results.append({'step': 'vip_check', 'exit_code': 0 if vip_here else 1, 'output': f'vip={vip_ip} on {VIP_IFACE}: {"YES" if vip_here else "NO"}'}) ok = all(r['exit_code'] == 0 for r in results) log.info(f"verify: {'OK' if ok else 'FAILED'} {results}") return {'ok': ok, 'actions': results} # ═══ Command dispatcher（按 command_type 路由）═══ def dispatch_command(self, cmd_id, command_type, command): """根据 command_type 分发执行，上报结构化结果。""" if command_type == 'fence': res = self.action_fence() elif command_type == 'unfence': res = self.action_unfence() elif command_type == 'promote_master': res = self.action_promote_master() elif command_type == 'revert_promote': res = self.action_revert_promote() elif command_type == 'flip_sync': res = self.action_flip_sync() elif command_type == 'verify': res = self.action_verify() else: # 普通 shell 命令（保留原有行为） proc = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30) res = {'ok': proc.returncode == 0, 'actions': [{'step': 'shell', 'exit_code': proc.returncode, 'output': (proc.stdout + proc.stderr)[:2000]}]} exit_code = 0 if res.get('ok') else 1 output = json.dumps(res, ensure_ascii=False)[:2000] self._api_post('agent_report_command_result.dspy', { 'node_id': self.node_id, 'cmd_id': cmd_id, 'output': output, 'exit_code': exit_code, }) log.info(f"Command [{cmd_id}] type={command_type} done: exit={exit_code}") # ═══ Remote Command Execution ═══ def poll_and_execute_commands(self): """Poll management for pending commands, dispatch by type, 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_type = cmd.get('command_type', 'shell') command = cmd.get('command', '') log.info(f"Executing command [{cmd_id}] type={command_type}") try: self.dispatch_command(cmd_id, command_type, command) 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: log.error("FOMS_NODE_ID environment variable not set") sys.exit(1) agent = FomsAgent() agent.run()