#!/usr/bin/env python3 """ FOMS Agent — 部署在主机(A)和备机(B)上的守护进程 职责: 1. 启动时从管理中心(C)拉取配置 2. 心跳上报 3. MySQL主从复制状态监控 4. 文件rsync同步 5. 响应切换/切回指令 """ import os import sys import json import time import logging import subprocess import urllib.request import urllib.error from datetime import datetime # ═══ 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')) # ═══ 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): url = 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}/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_health(self): """Collect and report all health statuses""" 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) 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, 'db_statuses': db_statuses, 'dir_statuses': dir_statuses, } self._api_post('agent_report_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...") self.fetch_config() last_health_report = 0 HEALTH_REPORT_INTERVAL = 60 # report health every 60s while True: try: self.heartbeat() # Report health (includes MySQL status + rsync) now = time.time() if now - last_health_report > HEALTH_REPORT_INTERVAL: self.report_health() last_health_report = 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) if __name__ == '__main__': if not NODE_ID: log.error("FOMS_NODE_ID environment variable not set") sys.exit(1) agent = FomsAgent() agent.run()