diff --git a/docs/deploy-operations.md b/docs/deploy-operations.md new file mode 100644 index 0000000..2648284 --- /dev/null +++ b/docs/deploy-operations.md @@ -0,0 +1,173 @@ +# FOMS 部署与运维手册 + +> 版本：2026-08-26（补全主备切换闭环后） +> 适用：MySQL 主从 + rsync 文件同步的双机主备，VIP 浮动（1A+2B+3A） + +## 一、架构 + +``` + 管理中心 C（foms 应用 + 管理库 foms） + ▲ HTTP（Bearer token） + ┌────────────┴────────────┐ + 主机 A（master） 主机 B（standby） + · MySQL 主库 · MySQL 从库（binlog 复制） + · 业务 + keepalived · 业务 + keepalived + · foms-agent · foms-agent + └──── VIP（浮动，业务连 VIP）────┘ +``` + +三个角色： +- **管理中心 C**：跑 foms（核心 + dbbackup/filesync/hostwatch 模块），存配置与切换记录。 +- **主机 A / 备机 B**：跑业务 + MySQL + keepalived + foms-agent（守护进程）。 +- **VIP**：业务流量入口，平时在 A，故障时漂到 B（keepalived 兜底 + FOMS 主动操作）。 + +## 二、管理中心 C 部署 + +```bash +cd foms +./build.sh # 建 venv、装依赖（含三模块）、生成 DDL/CRUD +``` + +**建库**： +```bash +mysql -uroot -p < models/mysql.ddl.sql # 核心表 +mysql -uroot -p < pkgs/dbbackup/models/mysql.ddl.sql # dbbackup 表 +mysql -uroot -p < pkgs/filesync/models/mysql.ddl.sql # filesync 表 +mysql -uroot -p < pkgs/hostwatch/models/mysql.ddl.sql # hostwatch 表 +``` + +**配置数据库连接**：编辑 `conf/config.json` 的 `databases.foms`（host/user/password/db）。 + +**启动**： +```bash +py3/bin/python app/foms.py # 监听 conf 里 website.port（默认 9080） +``` +启动时 `load_foms()` 会自动注册切换编排器并拉起健康判定器后台任务。 + +## 三、管理中心登记拓扑（部署后必做） + +在 FOMS 界面（或 API）依次创建： +1. **节点**（nodes）：A、B 各一条，填 host、ip、ssh_port、ssh_user、role（master/standby）。 +2. **集群**（clusters）：master_node_id=A、standby_node_id=B、`vip`（如 `192.168.1.100/24`）、 + `check_interval`（检测间隔秒）、`failover_threshold`（连续故障次数）、`failover_mode=semi`。 +3. **复制配置**（dbbackup_replications）：主库/从库地址、复制账号。 +4. **文件同步**（filesync_directories）：要 rsync 的目录（source→dest）。 + +## 四、被管主机 A / B 部署 + +### 4.1 安装依赖 +```bash +apt install -y mysql-server rsync keepalived ipvsadm python3 +pip3 install psutil # Agent 采集系统指标需要 +``` + +### 4.2 配置 keepalived（VIP 浮动） +A（`/etc/keepalived/keepalived.conf`，MASTER）： +``` +vrrp_instance VI_1 { + state MASTER + interface eth0 + virtual_router_id 51 + priority 100 # A 高优先级 + advert_int 1 + virtual_ipaddress { 192.168.1.100/24 } +} +``` +B 相同，但 `state BACKUP`、`priority 90`。 +keepalived 是**兜底**：机器全宕机时 VIP 自动漂移；"机器活、服务挂"的场景由 FOMS 主动 fence/promote 处理。 + +### 4.3 配置并启动 Agent +环境变量（A、B 各自的值不同）： +```bash +export FOMS_MANAGEMENT_URL=http://:9080 # 管理中心地址 +export FOMS_NODE_ID=<本节点在nodes表的id> # A/B 各自 +export FOMS_AGENT_TOKEN=<鉴权token> # 与管理中心约定 +export FOMS_HEARTBEAT_INTERVAL=10 +# 切换动作专用 +export FOMS_MYSQL_SERVICE=mysql # MySQL systemd 服务名 +export FOMS_VIP=192.168.1.100/24 # VIP（CIDR） +export FOMS_VIP_IFACE=eth0 # VIP 所在网卡 +export MYSQL_ROOT_PASSWORD= # promote/verify 用 +# 回滚降从库用（指向"对端"，即本机的复制上游） +export FOMS_MASTER_HOST=<对端IP> # B 上填 A 的 IP；A 上填 B 的 IP +export FOMS_MASTER_PORT=3306 +export FOMS_REPL_USER=repl +export FOMS_REPL_PASSWORD=<复制密码> +``` +启动： +```bash +python3 foms-agent.py # 建议用 systemd 托管（见下） +``` + +**systemd 单元**（`/etc/systemd/system/foms-agent.service`）： +``` +[Unit] +Description=FOMS Agent +After=network.target +[Service] +EnvironmentFile=/etc/foms/agent.env # 上面的环境变量写这里 +ExecStart=/usr/bin/python3 /opt/foms/foms-agent.py +Restart=always +[Install] +WantedBy=multi-user.target +``` +```bash +systemctl enable --now foms-agent +``` + +### 4.4 Agent 需要的权限 +Agent 以能执行 `systemctl`/`ip`/`mysql` 的用户运行（通常 root）。**生产环境应收窄**： +用 sudoers 白名单只允许这几条命令，而非完整 root。 + +## 五、日常运维 + +### 5.1 故障识别（自动） +健康判定器后台循环（默认 5s 扫一次）： +- 判定条件：master 心跳超时（> `failover_threshold × check_interval` 秒）**或** 复制状态 error/failed。 +- **防抖**：连续 `failover_threshold` 次命中才判故障（避免瞬时抖动误判）。 +- 判故障 → 写 `switchover_logs(status=detected, trigger=auto_detect)` + 发 `hostwatch_events` 告警。 +- **半自动**：只告警不切，等人确认。 + +### 5.2 故障切换（人工确认） +看到告警后，人工确认 → 触发切换： +- 界面：集群列表 → 切换按钮（`trigger_switchover`） +- API：`POST /api/trigger_switchover.dspy {cluster_id}` + +编排器异步执行六步（`switchover_steps` 表可查每步状态）： +``` +1. precheck 目标节点在线+心跳新鲜+复制延迟≤5s（防丢数据） +2. fence 隔离原主：停 MySQL + 摘 VIP ← 防脑裂，失败必中止 +3. promote 提升新主：STOP SLAVE + RESET SLAVE ALL（原子） +4. flip_sync 反转文件同步方向 +5. verify 新主可写测试 + VIP 归属检查（业务层请人工确认） +6. converge 集群主备互换 + 节点角色互换 + 状态收敛 +``` + +### 5.3 故障维修与切回 +1. 修好原主 A（硬件/系统/MySQL）。 +2. 把 A 重新挂为 B 的从库（`CHANGE MASTER TO` 指向 B，`START SLAVE`），等复制追平。 +3. 确认 A 健康后，触发切回：`trigger_switchback`（反向六步）。 + +### 5.4 失败与回滚语义 +- 任何一步失败 → 已完成步骤**逆向回滚**，`switchover_logs=failed`。 +- 回滚也失败 → `switchover_logs=rollback_failed` + 最高级告警，**人工介入**。 +- 回滚动作：unfence（恢复原主）、revert_promote（降回从库）、flip_sync 反转、converge 换回。 + +## 六、防脑裂三重保险 +1. **fence 先于 promote**：先停旧主 MySQL + 摘 VIP，再立新主。 +2. **fence 失败硬中止**：绝不 promote（否则双主）。 +3. **keepalived VRRP 兜底**：主机器全宕机时 VIP 自动漂移。 + +## 七、已知限制（诚实声明） +- **切换时 Agent 延迟**：Agent 轮询命令，切换六步串行走完约需数十秒（运维切换通常可接受）。 +- **verify 只到数据层**：验证"MySQL 可写 + VIP 可达"，业务层是否正常需人工确认。 +- **数据丢失窗口**：precheck 要求复制延迟 ≤5s；fence 到 promote 之间若有未同步 binlog 会丢失（已尽量收窄）。 +- **远程命令安全**：命令下发目前无白名单鉴权（P1 待办），生产应收窄。 + +## 八、模块清单 +| 模块 | 仓库 | 职责 | +|---|---|---| +| foms（核心） | yumoqing/foms | 集群/节点/切换编排/心跳/远程命令 | +| dbbackup | yumoqing/dbbackup | 数据库备份 + 日志同步（binlog 复制） | +| filesync | yumoqing/filesync | 运行文件同步（rsync） | +| hostwatch | yumoqing/hostwatch | 日志监控 + 日志分析 + 主机指标 | diff --git a/foms/failover.py b/foms/failover.py new file mode 100644 index 0000000..7c94fcb --- /dev/null +++ b/foms/failover.py @@ -0,0 +1,504 @@ +"""FOMS 切换编排器 — 六步状态机 + 逆向回滚 + 防脑裂不变量。 + +设计（2026-08-26 定调，1A+2B+3A）： + · 流量切换：VIP 浮动（keepalived 兜底） + · 触发模式：半自动（健康判定器告警 + 人工确认后切） + · 防脑裂：fence（隔离原主）先于 promote（立新主）；fence 失败硬中止 + +切换六步（failover A→B）： + step1 precheck 编排器本地：B 在线 + 心跳新鲜 + 复制延迟 < 阈值（防丢数据） + step2 fence 下发 A：停 MySQL + 摘 VIP ← 防脑裂关键，失败必中止 + step3 promote 下发 B：STOP SLAVE + RESET SLAVE ALL（脚本化原子执行） + step4 flip_sync DB 层反转 filesync 方向 + 下发 B 确认 + step5 verify 下发 B：连库读写 + VIP 归属检查（业务层留人工确认） + step6 converge 收敛：集群主备互换 + 节点角色互换 + 日志终态 + +任何一步 failed → 已完成步骤逆向回滚 → switchover_logs=failed； +回滚也失败 → rollback_failed + 最高级告警冒泡人工。 + +松耦合约束： + · precheck/flip_sync 读 dbbackup_replications / filesync_directories（同库只读） + · 告警写 hostwatch_events（与 hostwatch 模块约定的事件结构） +""" +import asyncio +import json +from datetime import datetime +from appPublic.uniqueID import getID +from appPublic.timeUtils import curDateString +from sqlor.dbpools import get_sor_context +from ahserver.serverenv import ServerEnv + +DBNAME = 'foms' + +# 幂等标志：防热重载重复启动健康判定器后台任务 +_failover_registered = False + +# ═══ 常量（防丢数据 / 超时 / 轮询）═══ +MAX_LAG_SECONDS = 5 # precheck：复制延迟必须 ≤ 该值才允许切换（R4） +HEARTBEAT_FRESH_SEC = 60 # precheck：目标节点心跳须在 60s 内 +CMD_TIMEOUT = 90 # 等待 Agent 命令完成的最长秒数 +CMD_POLL_INTERVAL = 2 # 轮询命令结果间隔（切换期加密，缓解 R1 延迟） +JUDGE_INTERVAL = 5 # 健康判定器扫描间隔（秒） + + +def _now(): + return curDateString() + + +def _parse_dt(v): + """把库里取出的时间值解析成 datetime（兼容 str/datetime）。""" + if v is None: + return None + if isinstance(v, datetime): + return v + for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M:%S'): + try: + return datetime.strptime(str(v)[:19], fmt) + except (ValueError, TypeError): + continue + return None + + +# ═══════════════════════════════════════════ +# 步骤记录（switchover_steps 审计） +# ═══════════════════════════════════════════ + +async def _record_step(sor, switchover_id, seq, step_name, status, output='', error=''): + """写一条步骤审计记录（先查后建，保证幂等）。不写 None 值（sqlor 不接受）。""" + existing = await sor.sqlExe( + 'SELECT id FROM switchover_steps WHERE switchover_id=${sw}$ AND seq=${seq}$', + {'sw': switchover_id, 'seq': seq}) + if existing: + upd = { + 'id': getattr(existing[0], 'id', ''), + 'status': status, + 'output': (output or '')[:2000], + 'error': (error or '')[:2000], + } + if status in ('success', 'failed', 'skipped'): + upd['completed_at'] = _now() + await sor.U('switchover_steps', upd) + else: + rec = { + 'id': getID(), + 'switchover_id': switchover_id, + 'seq': seq, + 'step_name': step_name, + 'status': status, + 'output': (output or '')[:2000], + 'error': (error or '')[:2000], + } + if status == 'running': + rec['started_at'] = _now() + await sor.C('switchover_steps', rec) + + +# ═══════════════════════════════════════════ +# Agent 命令下发 + 等待（结构化类型） +# ═══════════════════════════════════════════ + +async def _send_command(sor, node_id, command_type, command): + """向节点下发结构化命令，返回 cmd_id。""" + cmd_id = getID() + await sor.C('node_commands', { + 'id': cmd_id, + 'node_id': node_id, + 'command_type': command_type, + 'command': command or command_type, + 'status': 'pending', + 'created_at': _now(), + }) + return cmd_id + + +async def _wait_command(sor, cmd_id, timeout=CMD_TIMEOUT): + """轮询 node_commands 直到终态，返回 (ok, exit_code, output)。""" + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + rows = await sor.sqlExe( + 'SELECT status, exit_code, output FROM node_commands WHERE id=${cid}$', + {'cid': cmd_id}) + if rows: + r = rows[0] + status = getattr(r, 'status', '') + if status in ('success', 'failed'): + return (status == 'success'), getattr(r, 'exit_code', -1), getattr(r, 'output', '') + await asyncio.sleep(CMD_POLL_INTERVAL) + return False, -1, f'command timeout after {timeout}s' + + +# ═══════════════════════════════════════════ +# 各步骤实现 +# ═══════════════════════════════════════════ + +async def _step_precheck(sor, ctx): + """step1：本地预检 —— 目标节点在线、心跳新鲜、复制延迟达标（防丢数据）。""" + master_id, standby_id = ctx['from_node_id'], ctx['to_node_id'] + target = await sor.R('nodes', {'id': standby_id}) + if not target: + return False, '目标节点不存在' + t = target[0] + if getattr(t, 'status', '') != 'online': + return False, f"目标节点不在线（status={getattr(t, 'status', '')}）" + hb = _parse_dt(getattr(t, 'last_heartbeat', None)) + if hb is None or (datetime.now() - hb).total_seconds() > HEARTBEAT_FRESH_SEC: + return False, '目标节点心跳不新鲜（超过 %ds）' % HEARTBEAT_FRESH_SEC + # 复制延迟检查（松耦合读 dbbackup 表） + repls = await sor.sqlExe( + "SELECT db_name, seconds_behind, sync_status FROM dbbackup_replications " + "WHERE cluster_id=${cid}$ AND enabled='1'", {'cid': ctx['cluster_id']}) + for r in (repls or []): + lag = getattr(r, 'seconds_behind', None) + if lag is None: + lag = 999 + if int(lag) > MAX_LAG_SECONDS: + return False, f"复制延迟过大：{getattr(r, 'db_name', '')} behind {lag}s（阈值 {MAX_LAG_SECONDS}s，防丢数据）" + return True, 'precheck passed' + + +async def _step_fence(sor, ctx): + """step2：隔离原主（停 MySQL + 摘 VIP）。防脑裂关键，失败必中止。""" + cmd_id = await _send_command(sor, ctx['from_node_id'], 'fence', 'fence') + ok, code, out = await _wait_command(sor, cmd_id) + return ok, out + + +async def _step_promote(sor, ctx): + """step3：提升新主（STOP SLAVE + RESET SLAVE ALL，脚本化原子）。""" + cmd_id = await _send_command(sor, ctx['to_node_id'], 'promote_master', 'promote_master') + ok, code, out = await _wait_command(sor, cmd_id) + return ok, out + + +async def _step_flip_sync(sor, ctx): + """step4：DB 层反转 filesync 方向 + 下发目标节点确认。""" + from_id, to_id = ctx['from_node_id'], ctx['to_node_id'] + # 取原主/新主的 host，用于方向反转 + nodes = {} + for nid in (from_id, to_id): + recs = await sor.R('nodes', {'id': nid}) + if recs: + nodes[nid] = getattr(recs[0], 'host', '') + from_host, to_host = nodes.get(from_id, ''), nodes.get(to_id, '') + # 反转：source_host 从原主指向新主（A→B 变 B→A） + await sor.sqlExe( + 'UPDATE filesync_directories SET source_host=${th}$, dest_host=${fh}$, ' + 'updated_at=${now}$ WHERE cluster_id=${cid}$ AND source_host=${fh}$', + {'th': to_host, 'fh': from_host, 'now': _now(), 'cid': ctx['cluster_id']}) + cmd_id = await _send_command(sor, to_id, 'flip_sync', 'flip_sync') + ok, code, out = await _wait_command(sor, cmd_id) + return ok, f'flip db done; agent: {out}' + + +async def _step_verify(sor, ctx): + """step5：校验新主可写 + VIP 归属（业务层留人工确认）。""" + cmd_id = await _send_command(sor, ctx['to_node_id'], 'verify', 'verify') + ok, code, out = await _wait_command(sor, cmd_id) + if not ok: + return False, out + return True, out + '（业务层请人工确认）' + + +async def _step_converge(sor, ctx): + """step6：收敛 —— 集群主备互换 + 节点角色互换 + 集群恢复 running。""" + from_id, to_id = ctx['from_node_id'], ctx['to_node_id'] + await sor.U('clusters', { + 'id': ctx['cluster_id'], + 'master_node_id': to_id, + 'standby_node_id': from_id, + 'status': 'running', + 'updated_at': _now(), + }) + await sor.U('nodes', {'id': to_id, 'role': 'master', 'updated_at': _now()}) + await sor.U('nodes', {'id': from_id, 'role': 'standby', 'updated_at': _now()}) + return True, 'converged: master/standby swapped' + + +# 步骤表（正序） +STEPS = [ + ('precheck', _step_precheck), + ('fence', _step_fence), + ('promote', _step_promote), + ('flip_sync', _step_flip_sync), + ('verify', _step_verify), + ('converge', _step_converge), +] + + +# ═══════════════════════════════════════════ +# 逆向回滚 +# ═══════════════════════════════════════════ + +async def _rollback(sor, ctx, completed_steps): + """逆向回滚已完成步骤。返回 (ok, detail)。 + + 回滚语义： + fence → 恢复原主（start MySQL + 还 VIP） + promote → 把新主降回从库（CHANGE MASTER 指向原主）——需原主已恢复 + flip_sync → 反转回来 + verify → 无需回滚 + converge → 主备换回 + """ + detail = [] + ok_all = True + for step_name in reversed(completed_steps): + if step_name == 'fence': + # 恢复原主：start MySQL + 还 VIP（用 unfence 类型，Agent 幂等执行） + cmd_id = await _send_command(sor, ctx['from_node_id'], 'unfence', 'unfence') + ok, _, out = await _wait_command(sor, cmd_id) + detail.append(f'unfence {ctx["from_node_id"]}: {"OK" if ok else "FAILED"} {out[:200]}') + ok_all = ok_all and ok + elif step_name == 'promote': + # 降回从库（revert_promote：重新 CHANGE MASTER 指向原主） + cmd_id = await _send_command(sor, ctx['to_node_id'], 'revert_promote', 'revert_promote') + ok, _, out = await _wait_command(sor, cmd_id) + detail.append(f'revert_promote {ctx["to_node_id"]}: {"OK" if ok else "FAILED"} {out[:200]}') + ok_all = ok_all and ok + elif step_name == 'flip_sync': + # 再反转一次即恢复 + from_id, to_id = ctx['from_node_id'], ctx['to_node_id'] + nodes = {} + for nid in (from_id, to_id): + recs = await sor.R('nodes', {'id': nid}) + if recs: + nodes[nid] = getattr(recs[0], 'host', '') + from_host, to_host = nodes.get(from_id, ''), nodes.get(to_id, '') + await sor.sqlExe( + 'UPDATE filesync_directories SET source_host=${fh}$, dest_host=${th}$, ' + 'updated_at=${now}$ WHERE cluster_id=${cid}$ AND source_host=${th}$', + {'fh': from_host, 'th': to_host, 'now': _now(), 'cid': ctx['cluster_id']}) + detail.append('flip_sync reverted') + elif step_name == 'converge': + # 主备换回 + await sor.U('clusters', { + 'id': ctx['cluster_id'], + 'master_node_id': ctx['from_node_id'], + 'standby_node_id': ctx['to_node_id'], + 'status': 'running', + 'updated_at': _now(), + }) + await sor.U('nodes', {'id': ctx['from_node_id'], 'role': 'master', 'updated_at': _now()}) + await sor.U('nodes', {'id': ctx['to_node_id'], 'role': 'standby', 'updated_at': _now()}) + detail.append('converge reverted') + # verify / precheck 无需回滚 + return ok_all, '; '.join(detail) + + +# ═══════════════════════════════════════════ +# 编排器主流程（六步状态机） +# ═══════════════════════════════════════════ + +async def run_switchover(switchover_id): + """执行一次切换（在独立 asyncio task 中运行）。""" + env = ServerEnv() + async with get_sor_context(env, DBNAME) as sor: + logs = await sor.R('switchover_logs', {'id': switchover_id}) + if not logs: + return + logrec = logs[0] + ctx = { + 'switchover_id': switchover_id, + 'cluster_id': getattr(logrec, 'cluster_id', ''), + 'from_node_id': getattr(logrec, 'from_node_id', ''), + 'to_node_id': getattr(logrec, 'to_node_id', ''), + } + completed = [] + failed_at = None + fail_detail = '' + for seq, (step_name, fn) in enumerate(STEPS, start=1): + await _record_step(sor, switchover_id, seq, step_name, 'running') + try: + ok, detail = await fn(sor, ctx) + except Exception as e: + ok, detail = False, f'exception: {e}' + if ok: + await _record_step(sor, switchover_id, seq, step_name, 'success', output=detail) + completed.append(step_name) + else: + await _record_step(sor, switchover_id, seq, step_name, 'failed', error=detail) + failed_at = step_name + fail_detail = detail + break + + if failed_at is None: + # 全部成功 + await sor.U('switchover_logs', { + 'id': switchover_id, 'status': 'completed', + 'details': 'all steps success', 'completed_at': _now(), + }) + return + + # 有失败 → 逆向回滚 + rb_ok, rb_detail = await _rollback(sor, ctx, completed) + final_status = 'completed' if rb_ok else 'rollback_failed' + # 回滚成功也意味着切换未达成 → 记 failed（原切换失败），但状态收敛 + await sor.U('switchover_logs', { + 'id': switchover_id, + 'status': 'failed' if rb_ok else 'rollback_failed', + 'details': f'failed at {failed_at}: {fail_detail} | rollback: {rb_detail}', + 'completed_at': _now(), + }) + # 集群恢复（回滚后主备未变） + await sor.U('clusters', { + 'id': ctx['cluster_id'], 'status': 'running', 'updated_at': _now(), + }) + # 回滚失败 → 最高级告警冒泡人工 + if not rb_ok: + await _emit_alert(sor, ctx['cluster_id'], 'critical', + f'切换回滚失败，需人工介入：{rb_detail}') + + +async def _emit_alert(sor, cluster_id, severity, message): + """写一条最高级告警到 hostwatch_events（松耦合，约定结构）。""" + try: + await sor.C('hostwatch_events', { + 'id': getID(), + 'rule_id': '', + 'node_id': '', + 'severity': severity, + 'match_count': 1, + 'sample_message': f'[FOMS cluster={cluster_id}] {message}'[:2000], + 'status': 'open', + 'notified': '0', + 'first_seen_at': _now(), + 'last_seen_at': _now(), + }) + except Exception: + pass # 告警失败不阻断主流程 + + +# ═══════════════════════════════════════════ +# 对外入口：确认切换 / 健康判定器 +# ═══════════════════════════════════════════ + +async def confirm_switchover(request, params_kw): + """人工确认后启动切换（半自动模式的执行入口）。 + + 入参：cluster_id（必填），direction（failover/failback，默认 failover） + 行为：创建 switchover_logs(status=in_progress, trigger=manual_confirm) + + 置集群 switching + 起异步任务执行六步。 + """ + env = ServerEnv() + cluster_id = params_kw.get('cluster_id', '') if hasattr(params_kw, 'get') else '' + direction = params_kw.get('direction', 'failover') if hasattr(params_kw, 'get') else 'failover' + if not cluster_id: + return {'status': 'error', 'message': 'cluster_id required'} + log_id = getID() + now = _now() + async with get_sor_context(env, DBNAME) as sor: + clusters = await sor.R('clusters', {'id': cluster_id}) + if not clusters: + return {'status': 'error', 'message': '集群不存在'} + cluster = clusters[0] + master_id = getattr(cluster, 'master_node_id', '') + standby_id = getattr(cluster, 'standby_node_id', '') + if direction == 'failback': + from_id, to_id = standby_id, master_id + else: + from_id, to_id = master_id, standby_id + await sor.C('switchover_logs', { + 'id': log_id, + 'cluster_id': cluster_id, + 'event_type': direction, + 'from_node_id': from_id, + 'to_node_id': to_id, + 'trigger': 'manual_confirm', + 'reason': params_kw.get('reason', '人工确认切换') if hasattr(params_kw, 'get') else '人工确认切换', + 'status': 'in_progress', + 'started_at': now, + 'created_at': now, + }) + await sor.U('clusters', {'id': cluster_id, 'status': 'switching', 'updated_at': now}) + # 起异步任务执行（不阻塞 HTTP 响应） + asyncio.get_event_loop().create_task(run_switchover(log_id)) + return {'status': 'ok', 'message': '切换已启动（异步执行）', 'log_id': log_id} + + +# ═══════════════════════════════════════════ +# 健康判定器（后台循环 + 防抖 + 半自动告警） +# ═══════════════════════════════════════════ + +# 内存防抖计数：{cluster_id: 连续故障次数} +_fault_counters = {} + + +async def health_judge_loop(): + """后台健康判定：周期扫 active 集群，连续 N 次命中才判故障（防抖）。 + + 判定条件（任一）： + · master 心跳超时（> failover_threshold * check_interval 秒） + · 复制状态 error/failed + 判故障 → 写 switchover_logs(status=detected, trigger=auto_detect) + 告警。 + 半自动：只告警不切，等人 confirm_switchover。 + """ + env = ServerEnv() + while True: + try: + async with get_sor_context(env, DBNAME) as sor: + clusters = await sor.sqlExe( + "SELECT id, master_node_id, check_interval, failover_threshold, status " + "FROM clusters WHERE status IN ('running','switching')", {}) + for c in (clusters or []): + cid = getattr(c, 'id', '') + master_id = getattr(c, 'master_node_id', '') + interval = int(getattr(c, 'check_interval', 10) or 10) + threshold = int(getattr(c, 'failover_threshold', 3) or 3) + if not master_id: + continue + # 心跳超时判定 + nodes = await sor.R('nodes', {'id': master_id}) + heartbeat_lost = False + if nodes: + hb = _parse_dt(getattr(nodes[0], 'last_heartbeat', None)) + if hb is None or (datetime.now() - hb).total_seconds() > threshold * interval: + heartbeat_lost = True + # 复制异常判定（松耦合读 dbbackup 表） + repl_bad = False + repls = await sor.sqlExe( + "SELECT id FROM dbbackup_replications WHERE cluster_id=${cid}$ " + "AND enabled='1' AND sync_status IN ('error','failed') LIMIT 1", + {'cid': cid}) + if repls: + repl_bad = True + faulty = heartbeat_lost or repl_bad + if faulty: + _fault_counters[cid] = _fault_counters.get(cid, 0) + 1 + if _fault_counters[cid] >= threshold: + # 达到阈值 → 判故障（防抖通过） + reason = ('心跳超时' if heartbeat_lost else '') + \ + ('+' if heartbeat_lost and repl_bad else '') + \ + ('复制异常' if repl_bad else '') + await sor.C('switchover_logs', { + 'id': getID(), + 'cluster_id': cid, + 'event_type': 'fault_detected', + 'from_node_id': master_id, + 'to_node_id': '', + 'trigger': 'auto_detect', + 'reason': reason, + 'status': 'detected', + 'created_at': _now(), + }) + await _emit_alert(sor, cid, 'critical', + f'主节点故障判定：{reason}（已连续{threshold}次），请人工确认切换') + # 重置计数，避免重复告警刷屏（下次再连续 threshold 次才再报） + _fault_counters[cid] = 0 + else: + _fault_counters[cid] = 0 + except Exception: + pass # 判定器自身异常不退出循环 + await asyncio.sleep(JUDGE_INTERVAL) + + +def register_failover(): + """注册切换编排器到 ServerEnv + 启动健康判定器后台任务。幂等（防热重载重复起任务）。""" + global _failover_registered + env = ServerEnv() + env.confirm_switchover = confirm_switchover + if _failover_registered: + return True + _failover_registered = True + # 健康判定器后台任务（由 webapp 的 startup 机制拉起） + from ahserver.configuredServer import add_startup + add_startup(health_judge_loop()) + return True diff --git a/foms/init.py b/foms/init.py index 74c8ee6..5b4d2c2 100644 --- a/foms/init.py +++ b/foms/init.py @@ -126,60 +126,31 @@ async def agent_heartbeat(request, params_kw): # ═══════════════════════════════════════════ # 故障切换编排 # ═══════════════════════════════════════════ +# 2026-08-26 起：切换不再是"只写日志"，委托给 failover 编排器执行六步状态机 +# （precheck → fence → promote → flip_sync → verify → converge，失败逆向回滚）。 +# 触发模式为半自动（1A+2B+3A）：健康判定器告警 → 人工确认 → 本入口执行。 async def trigger_switchover(request, params_kw): - """触发故障切换 A→B""" - env = ServerEnv() - cluster_id = _get(params_kw, 'cluster_id') - reason = _get(params_kw, 'reason', '手动触发') - log_id = getID() - now = curDateString() - async with get_sor_context(env, DBNAME) as sor: - clusters = await sor.R('clusters', {'id': cluster_id}) - if not clusters: - return {'status': 'error', 'message': '集群不存在'} - cluster = clusters[0] - await sor.C('switchover_logs', { - 'id': log_id, - 'cluster_id': cluster_id, - 'event_type': 'failover', - 'from_node_id': getattr(cluster, 'master_node_id', ''), - 'to_node_id': getattr(cluster, 'standby_node_id', ''), - 'trigger': 'manual', - 'reason': reason, - 'status': 'in_progress', - 'started_at': now, - 'created_at': now, - }) - await sor.U('clusters', {'id': cluster_id, 'status': 'switching', 'updated_at': now}) - return {'status': 'ok', 'message': '切换流程已触发', 'log_id': log_id} + """触发故障切换 A→B（委托编排器执行六步状态机）""" + from foms.failover import confirm_switchover + ns = dict(params_kw) if hasattr(params_kw, 'items') else {} + ns['direction'] = 'failover' + result = await confirm_switchover(request, ns) + if result.get('status') == 'error': + return {'widgettype': 'Message', 'options': {'title': '失败', 'message': result.get('message', ''), 'type': 'error'}} + return {'widgettype': 'Message', 'options': {'title': '成功', 'message': f"切换已启动（异步执行），事件ID：{result.get('log_id', '')}", 'type': 'success'}} async def trigger_switchback(request, params_kw): - """触发切回 B→A""" - env = ServerEnv() - cluster_id = _get(params_kw, 'cluster_id') - log_id = getID() - now = curDateString() - async with get_sor_context(env, DBNAME) as sor: - clusters = await sor.R('clusters', {'id': cluster_id}) - if not clusters: - return {'status': 'error', 'message': '集群不存在'} - cluster = clusters[0] - await sor.C('switchover_logs', { - 'id': log_id, - 'cluster_id': cluster_id, - 'event_type': 'failback', - 'from_node_id': getattr(cluster, 'standby_node_id', ''), - 'to_node_id': getattr(cluster, 'master_node_id', ''), - 'trigger': 'manual', - 'reason': '手动切回', - 'status': 'in_progress', - 'started_at': now, - 'created_at': now, - }) - await sor.U('clusters', {'id': cluster_id, 'status': 'switching', 'updated_at': now}) - return {'status': 'ok', 'message': '切回流程已触发', 'log_id': log_id} + """触发切回 B→A（委托编排器，反向六步）""" + from foms.failover import confirm_switchover + ns = dict(params_kw) if hasattr(params_kw, 'items') else {} + ns['direction'] = 'failback' + ns.setdefault('reason', '手动切回') + result = await confirm_switchover(request, ns) + if result.get('status') == 'error': + return {'widgettype': 'Message', 'options': {'title': '失败', 'message': result.get('message', ''), 'type': 'error'}} + return {'widgettype': 'Message', 'options': {'title': '成功', 'message': f"切回已启动（异步执行），事件ID：{result.get('log_id', '')}", 'type': 'success'}} # ═══════════════════════════════════════════ @@ -245,18 +216,20 @@ async def exec_remote_command(request, params_kw): async def agent_poll_commands(request, params_kw): - """Agent 轮询获取待执行命令""" + """Agent 轮询获取待执行命令（含 command_type，供 Agent 按类型路由）""" env = ServerEnv() node_id = _get(params_kw, 'node_id') async with get_sor_context(env, DBNAME) as sor: rows = await sor.sqlExe( - "SELECT id, command FROM node_commands WHERE node_id=${nid}$ AND status='pending' " + "SELECT id, command_type, command FROM node_commands WHERE node_id=${nid}$ AND status='pending' " "ORDER BY created_at ASC LIMIT 5", {'nid': node_id}) # 取到的命令标记为 running，防止重复下发 for r in (rows or []): await sor.U('node_commands', {'id': getattr(r, 'id', ''), 'status': 'running'}) - return {'commands': [{'id': getattr(r, 'id', ''), 'command': getattr(r, 'command', '')} + return {'commands': [{'id': getattr(r, 'id', ''), + 'command_type': getattr(r, 'command_type', 'shell') or 'shell', + 'command': getattr(r, 'command', '')} for r in (rows or [])]} @@ -306,4 +279,7 @@ def load_foms(): env.exec_remote_command = exec_remote_command env.agent_poll_commands = agent_poll_commands env.agent_report_command_result = agent_report_command_result + # 切换编排器 + 健康判定器（后台任务） + from foms.failover import register_failover + register_failover() return True diff --git a/models/clusters.json b/models/clusters.json index 6923ee4..0064db4 100644 --- a/models/clusters.json +++ b/models/clusters.json @@ -15,6 +15,8 @@ {"name": "standby_node_id", "title": "备机节点", "type": "str", "length": 32}, {"name": "check_interval", "title": "检测间隔(秒)", "type": "int", "nullable": "no", "default": "10"}, {"name": "failover_threshold", "title": "故障判定阈值(次)", "type": "int", "nullable": "no", "default": "3"}, + {"name": "vip", "title": "VIP地址(CIDR)", "type": "str", "length": 64}, + {"name": "failover_mode", "title": "触发模式", "type": "str", "length": 16, "nullable": "no", "default": "semi"}, {"name": "status", "title": "集群状态", "type": "str", "length": 32, "nullable": "no", "default": "stopped"}, {"name": "created_at", "title": "创建时间", "type": "datetime"}, {"name": "updated_at", "title": "更新时间", "type": "datetime"} @@ -24,6 +26,7 @@ ], "codes": [ {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='cluster_status'"}, + {"field": "failover_mode", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='failover_mode'"}, {"field": "master_node_id", "table": "nodes", "valuefield": "id", "textfield": "host"}, {"field": "standby_node_id", "table": "nodes", "valuefield": "id", "textfield": "host"} ] diff --git a/models/node_commands.json b/models/node_commands.json index 4f298c0..59b9e1e 100644 --- a/models/node_commands.json +++ b/models/node_commands.json @@ -10,6 +10,7 @@ "fields": [ {"name": "id", "title": "命令ID", "type": "str", "length": 32, "nullable": "no"}, {"name": "node_id", "title": "节点ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "command_type", "title": "命令类型", "type": "str", "length": 32, "nullable": "no", "default": "shell"}, {"name": "command", "title": "命令内容", "type": "text", "nullable": "no"}, {"name": "status", "title": "状态", "type": "str", "length": 16, "nullable": "no", "default": "pending"}, {"name": "exit_code", "title": "退出码", "type": "int"}, @@ -22,6 +23,7 @@ {"name": "idx_ncmd_status", "idxtype": "index", "idxfields": ["status"]} ], "codes": [ - {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='cmd_status'"} + {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='cmd_status'"}, + {"field": "command_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='cmd_type'"} ] } diff --git a/models/switchover_steps.json b/models/switchover_steps.json new file mode 100644 index 0000000..9ffe6c3 --- /dev/null +++ b/models/switchover_steps.json @@ -0,0 +1,29 @@ +{ + "summary": [ + { + "name": "switchover_steps", + "title": "切换步骤明细", + "primary": ["id"], + "catelog": "entity" + } + ], + "fields": [ + {"name": "id", "title": "步骤ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "switchover_id", "title": "切换事件", "type": "str", "length": 32, "nullable": "no"}, + {"name": "seq", "title": "步骤序号", "type": "int", "nullable": "no"}, + {"name": "step_name", "title": "步骤名称", "type": "str", "length": 32, "nullable": "no"}, + {"name": "status", "title": "步骤状态", "type": "str", "length": 16, "nullable": "no", "default": "pending"}, + {"name": "output", "title": "执行输出", "type": "text"}, + {"name": "error", "title": "错误信息", "type": "text"}, + {"name": "started_at", "title": "开始时间", "type": "datetime"}, + {"name": "completed_at", "title": "完成时间", "type": "datetime"} + ], + "indexes": [ + {"name": "idx_ss_switchover", "idxtype": "index", "idxfields": ["switchover_id"]}, + {"name": "idx_ss_seq", "idxtype": "index", "idxfields": ["switchover_id", "seq"]} + ], + "codes": [ + {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='step_status'"}, + {"field": "switchover_id", "table": "switchover_logs", "valuefield": "id", "textfield": "event_type"} + ] +} diff --git a/scripts/foms-agent.py b/scripts/foms-agent.py index b5a513d..b9689e4 100644 --- a/scripts/foms-agent.py +++ b/scripts/foms-agent.py @@ -34,6 +34,12 @@ 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, @@ -321,26 +327,176 @@ class FomsAgent: 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, execute, report results""" + """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}]: {command}") + log.info(f"Executing command [{cmd_id}] type={command_type}") 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}") + 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,