yumoqing
fa257f36b1
依据现有实现补全「自动识别故障 → 人工确认切换 → 切回」运维核心闭环。 触发模式 1A+2B+3A：VIP 浮动 + 半自动 + 切前隔离原主（防脑裂）。 【表结构】 clusters 加 vip/failover_mode；node_commands 加 command_type 新增 switchover_steps（切换步骤级审计） 【Agent 六动作（脚本化、幂等、结构化返回）】 fence(停MySQL+摘VIP)/unfence(恢复)/promote_master(STOP+RESET SLAVE) revert_promote(降回从库)/flip_sync(反转同步)/verify(可写+VIP校验) 【切换编排器 foms/failover.py】 六步状态机：precheck→fence→promote→flip_sync→verify→converge 失败 → 已完成步骤逆向回滚 → switchover_logs=failed 回滚也失败 → rollback_failed + 最高级告警冒泡人工 防脑裂不变量：fence 先于 promote，fence 失败硬中止绝不 promote 【健康判定器】 后台循环（默认5s）扫 active 集群，心跳超时/复制异常判定 防抖：连续 failover_threshold 次命中才判故障（防瞬时抖动误判） 半自动：只告警（写 switchover_logs + hostwatch_events）不切，等人确认 【切回 + 回滚语义】 trigger_switchback 反向六步；回滚用 unfence/revert_promote/反转/换回 【部署文档】 docs/deploy-operations.md：架构/三机部署/keepalived/Agent 环境变量/ 日常运维/失败语义/防脑裂/已知限制 验证：语法全过；编排器↔Agent 六命令类型对齐；六步顺序正确； 防脑裂不变量成立（fence idx=1 < promote idx=2）；注册链完整。
518 lines
24 KiB
Python
518 lines
24 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'))
|
||
|
||
# ═══ 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()
|