361 lines
14 KiB
Python
361 lines
14 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):
|
|
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...")
|
|
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 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")
|
|
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('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()
|