yumoqing
e1be45e940
按产线规范将 FOMS 单体拆分为 4 部分： 【foms 核心（保留在本仓库）】主备切换本身 表：clusters、nodes、switchover_logs、node_commands(新增) 功能：集群/节点 CRUD、Agent 心跳、故障切换/切回、仪表盘、远程命令 远程命令改用专用 node_commands 表（原借用 node_logs 的 MVP 方案，顺手修 P1 隐患） 仪表盘跨模块只读聚合（同库松耦合 SQL） 【dbbackup 模块（独立仓库）】数据库全量备份 + 日志同步(binlog复制)管理 表 dbbackup_replications(原 sync_databases)/jobs(备份任务)/records(备份记录) 新建备份执行/记录/清理逻辑 【filesync 模块（独立仓库）】运行文件同步(rsync) 表 filesync_directories(原 sync_directories) 【hostwatch 模块（独立仓库）】日志监控+日志分析+主机指标 表 hostwatch_logs(原 node_logs)/metrics(原 node_metrics)/rules/events 新建日志归类、规则匹配告警、聚合分析逻辑 设计（松耦合方案 A）： · 拆出模块表加模块名前缀，cluster_id/node_id 仅存字符串不建外键 · 拆出模块不 import foms 核心代码，可独立部署测试 · 单 Agent 多端点：心跳→foms、指标日志→hostwatch、复制→dbbackup、rsync→filesync · 模块 wwwroot 软链到主 wwwroot/<模块名>（URL 前缀隔离） Agent 端点拆分：report_health 拆为 report_db_health/report_dir_health， metrics 上报改 /hostwatch/api/，_api_post/_api_get 支持完整路径。
362 lines
15 KiB
Python
362 lines
15 KiB
Python
#!/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'))
|
||
|
||
# ═══ 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}")
|
||
|
||
# ═══ 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:
|
||
log.error("FOMS_NODE_ID environment variable not set")
|
||
sys.exit(1)
|
||
agent = FomsAgent()
|
||
agent.run()
|