docs: 部署手册更新为 foms_app 纯伞化后的流程（clone foms_app + build + start.sh + URL 布局）

foms 只保留业务逻辑：
  · foms/ 包（init.py + failover.py 切换编排器）
  · models/（5 张表）、json/（CRUD 定义）、wwwroot/api/（端点）
  · scripts/（foms-agent 及部署脚本）
  · docs/deploy-operations.md

应用层迁出至 foms_app 纯伞仓：
  · app/（入口）、build.sh、start/stop.sh、conf/、init/data.yaml
  · scripts/init_rbac.py / load_path.py（重写为新 URL 布局）
  · wwwroot/index.ui（重写，指向各模块前缀路径）

同时修复 foms/__init__.py：移除对拆分时已删除函数的导出。

.gitignore

@@ -7,14 +7,6 @@ dist/
 *.pid
 logs/
 files/
-pkgs/
+models/mysql.ddl.sql
-bricks/
+.DS_Store
-py3/
 .env
-*.swp
-*.swo
-# CRUD-generated directories (xls2ui)
-wwwroot/clusters_list/
-wwwroot/nodes_list/
-wwwroot/sync_databases_list/
-wwwroot/sync_directories_list/

README.md

@@ -1,61 +1,41 @@
-# FOMS — Failover Management System
+# foms — 主备切换核心模块
 
-主备机切换管理系统。运行在管理中心 C 上，管理主机 A 和备机 B 之间的：
+FOMS 的**领域核心**（业务逻辑层），独立可安装模块。
+应用壳（构建/配置/启动/首页）在纯伞仓 **foms_app**。
 
-- **MySQL 主从复制** — 实时数据同步
+## 内容
-- **文件目录同步** — rsync 定期同步
-- **健康监控** — 节点心跳 + 复制状态
-- **故障切换** — 自动检测 A 故障 → 切换到 B
-- **故障切回** — A 修复后 → 切回 A
 
-## 架构
+| 部分 | 文件 | 说明 |
+|---|---|---|
+| 包入口 | `foms/init.py` | 集群/节点 CRUD、心跳、仪表盘、远程命令队列、`load_foms()` |
+| 切换编排器 | `foms/failover.py` | 六步状态机 + 逆向回滚 + 防脑裂 + 健康判定器 |
+| 表定义 | `models/*.json` | clusters、nodes、switchover_logs、switchover_steps、node_commands |
+| CRUD 定义 | `json/*.json` | clusters_list、nodes_list |
+| 端点 | `wwwroot/api/*.dspy` | 心跳、命令、切换、仪表盘等 |
+| Agent | `scripts/foms-agent.py` | 被管主机守护进程（独立部署，非本包代码） |
+| 运维文档 | `docs/deploy-operations.md` | 部署 + 运维手册 |
 
+## 切换闭环（2026-08-26 补全）
+
+健康判定器（防抖告警，半自动）→ 人工确认 → 六步状态机：
 ```
-管理中心 C (FOMS Web App)
+precheck → fence → promote → flip_sync → verify → converge
-    │
-    ├── 主机 A ←──MySQL复制/rsync──→ 备机 B
-    │
-    └── Agent 守护进程 (部署在 A & B)
 ```
+失败逆向回滚；防脑裂不变量：fence 先于 promote，fence 失败硬中止。
+详见 `docs/deploy-operations.md`。
 
-## 快速开始
+## 安装
 
 ```bash
-# 1. 构建
+pip install -e .
-./build.sh
-
-# 2. 配置 .env
-vim .env
-
-# 3. 初始化数据库 (手动执行 models/mysql.ddl.sql)
-mysql -u root < models/mysql.ddl.sql
-
-# 4. 注册RBAC权限
-cd py3/bin && python ../../scripts/load_path.py
-
-# 5. 启动
-./start.sh
 ```
+由 foms_app 的 build.sh 统一组装。`load_foms()` 时自动注册切换编排器
+并拉起健康判定器后台任务（幂等）。
 
-## 模块
+## 邻接模块（独立仓库）
 
-| 模块 | 路径 | 说明 |
+- dbbackup — 数据库备份 + 日志同步（binlog 复制）
-|------|------|------|
+- filesync — 运行文件同步（rsync）
-| 集群管理 | /clusters_list | 配置主备集群 |
+- hostwatch — 日志监控 + 日志分析 + 主机指标
-| 节点管理 | /nodes_list | 注册A/B节点(SSH信息) |
-| 数据库同步 | /sync_databases_list | MySQL主从复制配置 |
-| 文件同步 | /sync_directories_list | rsync目录同步配置 |
 
-## Agent部署
+仪表盘对这三模块的表做同库只读聚合（松耦合）。
-
-在主机A和备机B上：
-
-```bash
-# 配置环境变量
-export FOMS_MANAGEMENT_URL=http://C的IP:9080
-export FOMS_NODE_ID=<节点ID>
-export FOMS_AGENT_TOKEN=<安全令牌>
-
-# 运行Agent
-python3 scripts/foms-agent.py
-```

app/foms.py

@@ -1,32 +0,0 @@
-#!/usr/bin/env python3
-"""FOMS — Failover Management System 主入口"""
-import os, sys
-
-# Add root so sibling modules are importable
-root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-sys.path.insert(0, root_dir)
-
-from bricks_for_python.init import load_pybricks
-from ahserver.webapp import webapp
-from ahserver.serverenv import ServerEnv
-
-# Foundation modules
-from rbac.init import load_rbac
-from appbase.init import load_appbase
-
-# Business modules
-from foms.init import load_foms
-
-from app.global_func import set_globalvariable
-
-
-def init():
-    set_globalvariable()
-    load_pybricks()
-    load_appbase()
-    load_rbac()
-    load_foms()
-
-
-if __name__ == '__main__':
-    webapp(init)

app/global_func.py

@@ -1,12 +0,0 @@
-"""FOMS global functions — injected into dspy context"""
-from ahserver.serverenv import ServerEnv
-
-
-def get_module_dbname(mname):
-    """All FOMS modules share one database."""
-    return 'foms'
-
-
-def set_globalvariable():
-    env = ServerEnv()
-    env.get_module_dbname = get_module_dbname

build.sh

@@ -1,51 +0,0 @@
-#!/usr/bin/env bash
-set -e
-cdir=$(pwd)
-echo "=== FOMS Build ==="
-
-# 1. Create venv
-[ ! -d py3 ] && python3 -m venv py3
-source py3/bin/activate
-
-# 2. Install foundation packages
-mkdir -p pkgs
-for m in apppublic sqlor ahserver bricks-for-python xls2ddl; do
-    cd $cdir/pkgs
-    [ ! -d "$m" ] && git clone https://git.opencomputing.cn/yumoqing/$m
-    cd $m && $cdir/py3/bin/pip install -e .
-done
-
-# 3. Build bricks frontend
-cd $cdir/pkgs
-[ ! -d "bricks" ] && git clone https://git.opencomputing.cn/yumoqing/bricks
-cd bricks/bricks && ./build.sh
-ln -sf $cdir/pkgs/bricks/dist $cdir/bricks
-
-# 4. Install auth modules
-for m in appbase rbac; do
-    cd $cdir/pkgs
-    [ ! -d "$m" ] && git clone https://git.opencomputing.cn/yumoqing/$m
-    cd $m && $cdir/py3/bin/pip install -e .
-done
-
-# 5. Install FOMS
-cd $cdir && $cdir/py3/bin/pip install -e .
-
-# 6. Generate DDL
-cd $cdir
-[ -d models ] && json2ddl mysql models/ > $cdir/models/mysql.ddl.sql 2>/dev/null || echo "  skip ddl"
-
-# 7. Generate CRUD UI (xls2ui for json/)
-cd $cdir
-if [ -d json ]; then
-    cd json
-    for f in *.json; do
-        echo "  xls2ui $f"
-        xls2ui -m ../models -o ../wwwroot foms "$f" 2>/dev/null || echo "    skipped"
-    done
-fi
-
-# 8. Create runtime dirs
-mkdir -p $cdir/logs $cdir/files
-
-echo "=== FOMS Build Complete ==="

conf/config.json

@@ -1,42 +0,0 @@
-{
-    "password_key": "foms_secret_key_change_me",
-    "logger": {
-        "name": "foms",
-        "level": "INFO",
-        "file": "$[workdir]$/logs/foms.log"
-    },
-    "filesroot": "$[workdir]$/files",
-    "databases": {
-        "foms": {
-            "driver": "aiomysql",
-            "kwargs": {
-                "host": "127.0.0.1",
-                "port": 3306,
-                "user": "root",
-                "password": "",
-                "db": "foms",
-                "charset": "utf8mb4",
-                "autocommit": true
-            }
-        }
-    },
-    "website": {
-        "paths": [
-            ["$[workdir]$/wwwroot", ""],
-            ["$[workdir]$/bricks", "/bricks"]
-        ],
-        "port": 9080,
-        "indexes": ["index.ui"],
-        "session_max_time": 86400,
-        "session_issue_time": 3600,
-        "processors": [
-            [".dspy", "dspy"],
-            [".ui", "bui"],
-            [".tmpl", "tmpl"],
-            [".js", "staticfile"],
-            [".css", "staticfile"],
-            [".png", "staticfile"],
-            [".svg", "staticfile"]
-        ]
-    }
-}

docs/deploy-operations.md

@@ -0,0 +1,186 @@
+# FOMS 部署与运维手册
+
+> 版本：2026-08-27（foms_app 纯伞化之后）
+> 适用：MySQL 主从 + rsync 文件同步的双机主备，VIP 浮动（1A+2B+3A）
+>
+> 仓库分工：**foms_app（纯伞：构建/配置/启动）** + foms（核心业务）+ dbbackup/filesync/hostwatch（邻接模块）。
+> 部署以 foms_app 为入口。
+
+## 一、架构
+
+```
+                管理中心 C（foms_app 组装运行，管理库 foms）
+                     ▲  HTTP（Bearer token）
+        ┌────────────┴────────────┐
+   主机 A（master）          主机 B（standby）
+   · MySQL 主库              · MySQL 从库（binlog 复制）
+   · 业务 + keepalived       · 业务 + keepalived
+   · foms-agent              · foms-agent
+        └──── VIP（浮动，业务连 VIP）────┘
+```
+
+三个角色：
+- **管理中心 C**：跑 foms_app（组装核心 + dbbackup/filesync/hostwatch 模块），存配置与切换记录。
+- **主机 A / 备机 B**：跑业务 + MySQL + keepalived + foms-agent（守护进程）。
+- **VIP**：业务流量入口，平时在 A，故障时漂到 B（keepalived 兜底 + FOMS 主动操作）。
+
+## 二、管理中心 C 部署
+
+```bash
+git clone git@git.opencomputing.cn:yumoqing/foms_app.git
+cd foms_app
+./build.sh                    # 建 venv、clone+安装全部模块、生成 DDL/CRUD
+```
+
+**建库**（build 后各模块目录下生成了 DDL）：
+```bash
+mysql -uroot -p < pkgs/foms/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）。
+- 环境变量：`cp .env.example .env`，填 `FOMS_PORT` / `MYSQL_ROOT_PASSWORD` / `FOMS_AGENT_TOKEN`。
+
+**启动**：
+```bash
+./start.sh                    # 监听 FOMS_PORT（默认 9080）
+./stop.sh
+```
+启动时 `load_foms()` 会自动注册切换编排器并拉起健康判定器后台任务。
+
+**URL 布局**（conf/config.json website.paths）：
+- foms 核心：根路径（`/api/...`、`/clusters_list`、`/nodes_list`）
+- dbbackup：`/dbbackup/...`　filesync：`/filesync/...`　hostwatch：`/hostwatch/...`
+
+
+## 三、管理中心登记拓扑（部署后必做）
+
+在 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://<C的IP>: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=<root密码>            # 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_app | 纯伞（应用壳） | 组装入口、build、配置、启动、首页、RBAC 注册（无业务逻辑） |
+| foms | 业务模块（核心） | 集群/节点/切换编排/心跳/远程命令/Agent |
+| dbbackup | 业务模块 | 数据库备份 + 日志同步（binlog 复制） |
+| filesync | 业务模块 | 运行文件同步（rsync） |
+| hostwatch | 业务模块 | 日志监控 + 日志分析 + 主机指标 |

foms/__init__.py

@@ -1,8 +1,26 @@
+"""foms — 主备切换核心业务模块（FOMS 的领域核心）。
+
+独立可安装模块，由纯伞仓 foms_app 组装。
+
+内容：
+  · 表：clusters、nodes、switchover_logs、switchover_steps、node_commands
+  · 集群/节点 CRUD、Agent 心跳、仪表盘、远程命令队列
+  · 切换编排器（六步状态机 + 逆向回滚 + 防脑裂，见 failover.py）
+  · 健康判定器（后台循环 + 防抖 + 半自动告警）
+  · scripts/foms-agent.py：被管主机守护进程（非本包代码，独立部署）
+
+邻接能力已拆为独立模块（不在本包内）：
+  · dbbackup（数据库备份 + 日志同步）、filesync（文件同步）、hostwatch（日志监控分析）
+"""
 from .init import (
-    create_cluster, update_cluster, delete_cluster, get_cluster_list,
+    load_foms,
-    create_node, update_node, delete_node, get_node_list,
+    create_cluster, update_cluster, delete_cluster,
-    create_sync_db, update_sync_db, delete_sync_db, get_sync_db_list,
+    create_node, update_node, delete_node,
-    create_sync_dir, update_sync_dir, delete_sync_dir, get_sync_dir_list,
+    agent_heartbeat,
-    agent_heartbeat, agent_report_health, trigger_switchover, trigger_switchback,
+    trigger_switchover, trigger_switchback,
     get_dashboard_stats,
+    exec_remote_command, agent_poll_commands, agent_report_command_result,
+)
+from .failover import (
+    confirm_switchover, run_switchover, health_judge_loop, register_failover,
 )

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

foms/init.py

@@ -1,4 +1,20 @@
-"""FOMS business logic — CRUD operations + failover orchestration"""
+"""FOMS 核心业务 — 主备切换编排（集群/节点/切换/心跳/远程命令）。
+
+模块拆分后（2026-08-26），本模块只保留「主备切换」本身的核心概念与编排：
+  · 表：clusters、nodes、switchover_logs、node_commands
+  · 功能：集群/节点 CRUD、Agent 心跳、故障切换/切回、仪表盘、远程命令
+
+拆出的独立模块（各自独立 git 仓库，表加模块名前缀）：
+  · dbbackup  — 数据库全量备份 + 日志同步（binlog 复制）管理
+              表 dbbackup_replications/jobs/records
+  · filesync  — 运行文件同步（rsync）
+              表 filesync_directories
+  · hostwatch — 日志监控 + 日志分析 + 主机指标
+              表 hostwatch_logs/metrics/rules/events
+
+跨模块引用采用松耦合：本模块只通过 cluster_id/node_id 字符串关联，
+不 import 拆出模块的代码；仪表盘对拆出模块表做只读 SQL 聚合（同库）。
+"""
 import json
 from datetime import datetime
 from appPublic.uniqueID import getID
@@ -9,6 +25,18 @@ from ahserver.serverenv import ServerEnv
 DBNAME = 'foms'
 
 
+def _get(params_kw, key, default=''):
+    return params_kw.get(key, default) if hasattr(params_kw, 'get') else default
+
+
+def _ns(params_kw):
+    ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
+    for k in list(ns.keys()):
+        if k.endswith('_text'):
+            ns.pop(k)
+    return ns
+
+
 # ═══════════════════════════════════════════
 # Cluster CRUD
 # ═══════════════════════════════════════════
@@ -16,15 +44,11 @@ DBNAME = 'foms'
 async def create_cluster(request, params_kw):
     env = ServerEnv()
     async with get_sor_context(env, DBNAME) as sor:
-        ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
+        ns = _ns(params_kw)
         ns['id'] = getID()
         ns['status'] = 'stopped'
         ns['created_at'] = curDateString()
         ns['updated_at'] = curDateString()
-        # strip _text fields from Tabular
-        for k in list(ns.keys()):
-            if k.endswith('_text'):
-                ns.pop(k)
         await sor.C('clusters', ns)
     return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '集群创建成功', 'type': 'success'}}
 
@@ -32,11 +56,8 @@ async def create_cluster(request, params_kw):
 async def update_cluster(request, params_kw):
     env = ServerEnv()
     async with get_sor_context(env, DBNAME) as sor:
-        ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
+        ns = _ns(params_kw)
         ns['updated_at'] = curDateString()
-        for k in list(ns.keys()):
-            if k.endswith('_text'):
-                ns.pop(k)
         await sor.U('clusters', ns)
     return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '集群更新成功', 'type': 'success'}}
 
@@ -44,8 +65,7 @@ async def update_cluster(request, params_kw):
 async def delete_cluster(request, params_kw):
     env = ServerEnv()
     async with get_sor_context(env, DBNAME) as sor:
-        _id = params_kw.get('id', '') if hasattr(params_kw, 'get') else params_kw['id']
+        await sor.D('clusters', {'id': _get(params_kw, 'id')})
-        await sor.D('clusters', {'id': _id})
     return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '集群已删除', 'type': 'success'}}
 
 
@@ -56,14 +76,11 @@ async def delete_cluster(request, params_kw):
 async def create_node(request, params_kw):
     env = ServerEnv()
     async with get_sor_context(env, DBNAME) as sor:
-        ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
+        ns = _ns(params_kw)
         ns['id'] = getID()
         ns['status'] = 'offline'
         ns['created_at'] = curDateString()
         ns['updated_at'] = curDateString()
-        for k in list(ns.keys()):
-            if k.endswith('_text'):
-                ns.pop(k)
         await sor.C('nodes', ns)
     return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '节点创建成功', 'type': 'success'}}
 
@@ -71,11 +88,8 @@ async def create_node(request, params_kw):
 async def update_node(request, params_kw):
     env = ServerEnv()
     async with get_sor_context(env, DBNAME) as sor:
-        ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
+        ns = _ns(params_kw)
         ns['updated_at'] = curDateString()
-        for k in list(ns.keys()):
-            if k.endswith('_text'):
-                ns.pop(k)
         await sor.U('nodes', ns)
     return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '节点更新成功', 'type': 'success'}}
 
@@ -83,96 +97,18 @@ async def update_node(request, params_kw):
 async def delete_node(request, params_kw):
     env = ServerEnv()
     async with get_sor_context(env, DBNAME) as sor:
-        _id = params_kw.get('id', '') if hasattr(params_kw, 'get') else params_kw['id']
+        await sor.D('nodes', {'id': _get(params_kw, 'id')})
-        await sor.D('nodes', {'id': _id})
     return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '节点已删除', 'type': 'success'}}
 
 
 # ═══════════════════════════════════════════
-# Sync Database CRUD
+# Agent 心跳
-# ═══════════════════════════════════════════
-
-async def create_sync_db(request, params_kw):
-    env = ServerEnv()
-    async with get_sor_context(env, DBNAME) as sor:
-        ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
-        ns['id'] = getID()
-        ns['sync_status'] = 'not_configured'
-        ns['created_at'] = curDateString()
-        ns['updated_at'] = curDateString()
-        for k in list(ns.keys()):
-            if k.endswith('_text'):
-                ns.pop(k)
-        await sor.C('sync_databases', ns)
-    return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '数据库同步配置创建成功', 'type': 'success'}}
-
-
-async def update_sync_db(request, params_kw):
-    env = ServerEnv()
-    async with get_sor_context(env, DBNAME) as sor:
-        ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
-        ns['updated_at'] = curDateString()
-        for k in list(ns.keys()):
-            if k.endswith('_text'):
-                ns.pop(k)
-        await sor.U('sync_databases', ns)
-    return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '数据库同步配置更新成功', 'type': 'success'}}
-
-
-async def delete_sync_db(request, params_kw):
-    env = ServerEnv()
-    async with get_sor_context(env, DBNAME) as sor:
-        _id = params_kw.get('id', '') if hasattr(params_kw, 'get') else params_kw['id']
-        await sor.D('sync_databases', {'id': _id})
-    return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '数据库同步配置已删除', 'type': 'success'}}
-
-
-# ═══════════════════════════════════════════
-# Sync Directory CRUD
-# ═══════════════════════════════════════════
-
-async def create_sync_dir(request, params_kw):
-    env = ServerEnv()
-    async with get_sor_context(env, DBNAME) as sor:
-        ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
-        ns['id'] = getID()
-        ns['created_at'] = curDateString()
-        ns['updated_at'] = curDateString()
-        for k in list(ns.keys()):
-            if k.endswith('_text'):
-                ns.pop(k)
-        await sor.C('sync_directories', ns)
-    return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '目录同步配置创建成功', 'type': 'success'}}
-
-
-async def update_sync_dir(request, params_kw):
-    env = ServerEnv()
-    async with get_sor_context(env, DBNAME) as sor:
-        ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
-        ns['updated_at'] = curDateString()
-        for k in list(ns.keys()):
-            if k.endswith('_text'):
-                ns.pop(k)
-        await sor.U('sync_directories', ns)
-    return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '目录同步配置更新成功', 'type': 'success'}}
-
-
-async def delete_sync_dir(request, params_kw):
-    env = ServerEnv()
-    async with get_sor_context(env, DBNAME) as sor:
-        _id = params_kw.get('id', '') if hasattr(params_kw, 'get') else params_kw['id']
-        await sor.D('sync_directories', {'id': _id})
-    return {'widgettype': 'Message', 'options': {'title': '成功', 'message': '目录同步配置已删除', 'type': 'success'}}
-
-
-# ═══════════════════════════════════════════
-# Agent Communication
 # ═══════════════════════════════════════════
 
 async def agent_heartbeat(request, params_kw):
     """Agent 心跳上报"""
-    node_id = params_kw.get('node_id', '') if hasattr(params_kw, 'get') else ''
+    node_id = _get(params_kw, 'node_id')
-    version = params_kw.get('version', '') if hasattr(params_kw, 'get') else ''
+    version = _get(params_kw, 'version')
     if not node_id:
         return {'status': 'error', 'message': 'node_id required'}
     env = ServerEnv()
@@ -187,124 +123,56 @@ async def agent_heartbeat(request, params_kw):
     return {'status': 'ok', 'server_time': curDateString()}
 
 
-async def agent_report_health(request, params_kw):
-    """Agent健康报告（包含同步状态）"""
-    env = ServerEnv()
-    node_id = params_kw.get('node_id', '') if hasattr(params_kw, 'get') else ''
-    async with get_sor_context(env, DBNAME) as sor:
-        # Update sync_database statuses
-        db_statuses = params_kw.get('db_statuses', []) if hasattr(params_kw, 'get') else []
-        if isinstance(db_statuses, str):
-            db_statuses = json.loads(db_statuses)
-        for db_stat in db_statuses:
-            await sor.U('sync_databases', {
-                'id': db_stat.get('id'),
-                'sync_status': db_stat.get('status'),
-                'seconds_behind': db_stat.get('seconds_behind'),
-                'last_error': db_stat.get('error'),
-                'updated_at': curDateString(),
-            })
-        # Update sync_directory statuses
-        dir_statuses = params_kw.get('dir_statuses', []) if hasattr(params_kw, 'get') else []
-        if isinstance(dir_statuses, str):
-            dir_statuses = json.loads(dir_statuses)
-        for dir_stat in dir_statuses:
-            await sor.U('sync_directories', {
-                'id': dir_stat.get('id'),
-                'last_sync_at': curDateString(),
-                'last_sync_result': dir_stat.get('result'),
-                'updated_at': curDateString(),
-            })
-    return {'status': 'ok'}
-
-
 # ═══════════════════════════════════════════
-# Failover Orchestration
+# 故障切换编排
 # ═══════════════════════════════════════════
+# 2026-08-26 起：切换不再是"只写日志"，委托给 failover 编排器执行六步状态机
+# （precheck → fence → promote → flip_sync → verify → converge，失败逆向回滚）。
+# 触发模式为半自动（1A+2B+3A）：健康判定器告警 → 人工确认 → 本入口执行。
 
 async def trigger_switchover(request, params_kw):
-    """触发故障切换 A→B"""
+    """触发故障切换 A→B（委托编排器执行六步状态机）"""
-    env = ServerEnv()
+    from foms.failover import confirm_switchover
-    cluster_id = params_kw.get('cluster_id', '') if hasattr(params_kw, 'get') else ''
+    ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
-    reason = params_kw.get('reason', '手动触发') if hasattr(params_kw, 'get') else '手动触发'
+    ns['direction'] = 'failover'
-    log_id = getID()
+    result = await confirm_switchover(request, ns)
-    now = curDateString()
+    if result.get('status') == 'error':
-    async with get_sor_context(env, DBNAME) as sor:
+        return {'widgettype': 'Message', 'options': {'title': '失败', 'message': result.get('message', ''), 'type': 'error'}}
-        # Get cluster info
+    return {'widgettype': 'Message', 'options': {'title': '成功', 'message': f"切换已启动（异步执行），事件ID：{result.get('log_id', '')}", 'type': 'success'}}
-        clusters = await sor.R('clusters', {'id': cluster_id})
-        if not clusters:
-            return {'status': 'error', 'message': '集群不存在'}
-        cluster = clusters[0]
-        # Log the switchover
-        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,
-        })
-        # Update cluster status
-        await sor.U('clusters', {'id': cluster_id, 'status': 'switching', 'updated_at': now})
-    return {
-        'status': 'ok',
-        'message': '切换流程已触发',
-        'log_id': log_id,
-    }
 
 
 async def trigger_switchback(request, params_kw):
-    """触发切回 B→A"""
+    """触发切回 B→A（委托编排器，反向六步）"""
-    env = ServerEnv()
+    from foms.failover import confirm_switchover
-    cluster_id = params_kw.get('cluster_id', '') if hasattr(params_kw, 'get') else ''
+    ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
-    log_id = getID()
+    ns['direction'] = 'failback'
-    now = curDateString()
+    ns.setdefault('reason', '手动切回')
-    async with get_sor_context(env, DBNAME) as sor:
+    result = await confirm_switchover(request, ns)
-        clusters = await sor.R('clusters', {'id': cluster_id})
+    if result.get('status') == 'error':
-        if not clusters:
+        return {'widgettype': 'Message', 'options': {'title': '失败', 'message': result.get('message', ''), 'type': 'error'}}
-            return {'status': 'error', 'message': '集群不存在'}
+    return {'widgettype': 'Message', 'options': {'title': '成功', 'message': f"切回已启动（异步执行），事件ID：{result.get('log_id', '')}", 'type': 'success'}}
-        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,
-    }
 
 
 # ═══════════════════════════════════════════
-# Dashboard
+# 仪表盘（跨模块只读聚合，同库 SQL）
 # ═══════════════════════════════════════════
 
 async def get_dashboard_stats(request):
-    """仪表盘统计数据"""
+    """仪表盘统计：核心（集群/节点/切换）+ 拆出模块（复制/同步）"""
     env = ServerEnv()
     async with get_sor_context(env, DBNAME) as sor:
         clusters = await sor.sqlExe('SELECT count(*) as cnt FROM clusters', {})
         nodes_online = await sor.sqlExe("SELECT count(*) as cnt FROM nodes WHERE status='online'", {})
         nodes_offline = await sor.sqlExe("SELECT count(*) as cnt FROM nodes WHERE status='offline'", {})
-        dbs_synced = await sor.sqlExe("SELECT count(*) as cnt FROM sync_databases WHERE sync_status='running'", {})
+        # 拆出模块的表（同库松耦合只读）
-        dbs_error = await sor.sqlExe("SELECT count(*) as cnt FROM sync_databases WHERE sync_status IN ('error','failed')", {})
+        dbs_synced = await sor.sqlExe(
-        dirs_synced = await sor.sqlExe("SELECT count(*) as cnt FROM sync_directories WHERE last_sync_result='success'", {})
+            "SELECT count(*) as cnt FROM dbbackup_replications WHERE sync_status='running'", {})
+        dbs_error = await sor.sqlExe(
+            "SELECT count(*) as cnt FROM dbbackup_replications WHERE sync_status IN ('error','failed')", {})
+        dirs_synced = await sor.sqlExe(
+            "SELECT count(*) as cnt FROM filesync_directories WHERE last_sync_result='success'", {})
         recent_logs = await sor.sqlExe(
-            'SELECT event_type, status, reason, created_at FROM switchover_logs ORDER BY created_at DESC LIMIT 10', {}
+            'SELECT event_type, status, reason, created_at FROM switchover_logs ORDER BY created_at DESC LIMIT 10', {})
-        )
     return {
         'total_clusters': clusters[0].cnt if clusters else 0,
         'nodes_online': nodes_online[0].cnt if nodes_online else 0,
@@ -322,7 +190,66 @@ async def get_dashboard_stats(request):
 
 
 # ═══════════════════════════════════════════
-# ServerEnv Registration
+# 远程命令（专用 node_commands 表，不再借用日志表）
+# ═══════════════════════════════════════════
+# ⚠️ 安全待办：下发命令无鉴权白名单。管理中心被攻破 = 所有被管主机 root shell。
+# 后续独立任务应加：命令白名单 + RBAC 鉴权（仅 operator 以上可下发危险命令）。
+
+async def exec_remote_command(request, params_kw):
+    """下发远程命令，存入 node_commands 队列，Agent 轮询执行"""
+    env = ServerEnv()
+    node_id = _get(params_kw, 'node_id')
+    command = _get(params_kw, 'command')
+    if not node_id or not command:
+        return {'status': 'error', 'message': 'node_id and command required'}
+    cmd_id = getID()
+    now = curDateString()
+    async with get_sor_context(env, DBNAME) as sor:
+        await sor.C('node_commands', {
+            'id': cmd_id,
+            'node_id': node_id,
+            'command': command,
+            'status': 'pending',
+            'created_at': now,
+        })
+    return {'status': 'ok', 'message': '命令已下发，等待Agent执行', 'cmd_id': cmd_id}
+
+
+async def agent_poll_commands(request, params_kw):
+    """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_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_type': getattr(r, 'command_type', 'shell') or 'shell',
+                          'command': getattr(r, 'command', '')}
+                         for r in (rows or [])]}
+
+
+async def agent_report_command_result(request, params_kw):
+    """Agent 上报命令执行结果"""
+    env = ServerEnv()
+    cmd_id = _get(params_kw, 'cmd_id')
+    async with get_sor_context(env, DBNAME) as sor:
+        await sor.U('node_commands', {
+            'id': cmd_id,
+            'exit_code': _get(params_kw, 'exit_code', -1),
+            'output': (_get(params_kw, 'output', '') or '')[:2000],
+            'status': 'success' if int(_get(params_kw, 'exit_code', -1) or -1) == 0 else 'failed',
+            'executed_at': curDateString(),
+        })
+    return {'status': 'ok'}
+
+
+# ═══════════════════════════════════════════
+# 注册
 # ═══════════════════════════════════════════
 
 def load_foms():
@@ -341,26 +268,18 @@ def load_foms():
     env.update_nodes = update_node
     env.delete_node = delete_node
     env.delete_nodes = delete_node
-    # Sync DB CRUD
+    # Agent 心跳
-    env.create_sync_db = create_sync_db
-    env.create_sync_databases = create_sync_db
-    env.update_sync_db = update_sync_db
-    env.update_sync_databases = update_sync_db
-    env.delete_sync_db = delete_sync_db
-    env.delete_sync_databases = delete_sync_db
-    # Sync Dir CRUD
-    env.create_sync_dir = create_sync_dir
-    env.create_sync_directories = create_sync_dir
-    env.update_sync_dir = update_sync_dir
-    env.update_sync_directories = update_sync_dir
-    env.delete_sync_dir = delete_sync_dir
-    env.delete_sync_directories = delete_sync_dir
-    # Agent
     env.agent_heartbeat = agent_heartbeat
-    env.agent_report_health = agent_report_health
+    # 故障切换
-    # Failover
     env.trigger_switchover = trigger_switchover
     env.trigger_switchback = trigger_switchback
-    # Dashboard
+    # 仪表盘
     env.get_dashboard_stats = get_dashboard_stats
+    # 远程命令
+    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

init/data.yaml

@@ -1,134 +0,0 @@
-appcodes:
-- id: cluster_status
-  name: 集群状态
-  hierarchy_flg: 0
-- id: node_role
-  name: 节点角色
-  hierarchy_flg: 0
-- id: node_status
-  name: 节点状态
-  hierarchy_flg: 0
-- id: sync_status
-  name: 同步状态
-  hierarchy_flg: 0
-- id: sync_result
-  name: 同步结果
-  hierarchy_flg: 0
-- id: switchover_event_type
-  name: 切换事件类型
-  hierarchy_flg: 0
-- id: switchover_trigger
-  name: 触发方式
-  hierarchy_flg: 0
-- id: switchover_status
-  name: 切换执行状态
-  hierarchy_flg: 0
-
-appcodes_kv:
-- id: cluster_status_running
-  parentid: cluster_status
-  k: running
-  v: 运行中
-- id: cluster_status_stopped
-  parentid: cluster_status
-  k: stopped
-  v: 已停止
-- id: cluster_status_switching
-  parentid: cluster_status
-  k: switching
-  v: 切换中
-- id: cluster_status_error
-  parentid: cluster_status
-  k: error
-  v: 异常
-
-- id: node_role_master
-  parentid: node_role
-  k: master
-  v: 主机
-- id: node_role_standby
-  parentid: node_role
-  k: standby
-  v: 备机
-
-- id: node_status_online
-  parentid: node_status
-  k: online
-  v: 在线
-- id: node_status_offline
-  parentid: node_status
-  k: offline
-  v: 离线
-- id: node_status_error
-  parentid: node_status
-  k: error
-  v: 异常
-
-- id: sync_status_not_configured
-  parentid: sync_status
-  k: not_configured
-  v: 未配置
-- id: sync_status_running
-  parentid: sync_status
-  k: running
-  v: 同步中
-- id: sync_status_error
-  parentid: sync_status
-  k: error
-  v: 异常
-- id: sync_status_failed
-  parentid: sync_status
-  k: failed
-  v: 失败
-
-- id: sync_result_success
-  parentid: sync_result
-  k: success
-  v: 成功
-- id: sync_result_failed
-  parentid: sync_result
-  k: failed
-  v: 失败
-- id: sync_result_running
-  parentid: sync_result
-  k: running
-  v: 同步中
-
-- id: switchover_event_failover
-  parentid: switchover_event_type
-  k: failover
-  v: 故障切换
-- id: switchover_event_failback
-  parentid: switchover_event_type
-  k: failback
-  v: 故障切回
-- id: switchover_event_manual
-  parentid: switchover_event_type
-  k: manual
-  v: 手动操作
-
-- id: switchover_trigger_auto
-  parentid: switchover_trigger
-  k: auto
-  v: 自动检测
-- id: switchover_trigger_manual
-  parentid: switchover_trigger
-  k: manual
-  v: 手动触发
-
-- id: switchover_status_pending
-  parentid: switchover_status
-  k: pending
-  v: 等待执行
-- id: switchover_status_in_progress
-  parentid: switchover_status
-  k: in_progress
-  v: 执行中
-- id: switchover_status_success
-  parentid: switchover_status
-  k: success
-  v: 成功
-- id: switchover_status_failed
-  parentid: switchover_status
-  k: failed
-  v: 失败

json/sync_databases_list.json

@@ -1,22 +0,0 @@
-{
-    "tblname": "sync_databases",
-    "alias": "sync_databases_list",
-    "title": "数据库同步配置",
-    "params": {
-        "sortby": ["updated_at desc"],
-        "browserfields": {
-            "exclouded": ["id", "repl_password"],
-            "alters": {
-                "sync_status": {"uitype": "code"},
-                "enabled": {"uitype": "code", "data": [{"value": "1", "text": "启用"}, {"value": "0", "text": "禁用"}]},
-                "cluster_id": {"uitype": "code", "valueField": "id", "textField": "name"}
-            }
-        },
-        "editexclouded": ["id", "created_at", "updated_at", "sync_status", "seconds_behind", "last_error"],
-        "editable": {
-            "new_data_url": "{{entire_url('api/sync_databases_create.dspy')}}",
-            "update_data_url": "{{entire_url('api/sync_databases_update.dspy')}}",
-            "delete_data_url": "{{entire_url('api/sync_databases_delete.dspy')}}"
-        }
-    }
-}

json/sync_directories_list.json

@@ -1,22 +0,0 @@
-{
-    "tblname": "sync_directories",
-    "alias": "sync_directories_list",
-    "title": "文件目录同步配置",
-    "params": {
-        "sortby": ["updated_at desc"],
-        "browserfields": {
-            "exclouded": ["id"],
-            "alters": {
-                "last_sync_result": {"uitype": "code"},
-                "enabled": {"uitype": "code", "data": [{"value": "1", "text": "启用"}, {"value": "0", "text": "禁用"}]},
-                "cluster_id": {"uitype": "code", "valueField": "id", "textField": "name"}
-            }
-        },
-        "editexclouded": ["id", "created_at", "updated_at", "last_sync_at", "last_sync_result"],
-        "editable": {
-            "new_data_url": "{{entire_url('api/sync_directories_create.dspy')}}",
-            "update_data_url": "{{entire_url('api/sync_directories_update.dspy')}}",
-            "delete_data_url": "{{entire_url('api/sync_directories_delete.dspy')}}"
-        }
-    }
-}

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"}
     ]

models/node_commands.json

@@ -0,0 +1,29 @@
+{
+    "summary": [
+        {
+            "name": "node_commands",
+            "title": "节点远程命令",
+            "primary": ["id"],
+            "catelog": "entity"
+        }
+    ],
+    "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"},
+        {"name": "output", "title": "执行输出", "type": "text"},
+        {"name": "created_at", "title": "下发时间", "type": "datetime", "nullable": "no"},
+        {"name": "executed_at", "title": "执行时间", "type": "datetime"}
+    ],
+    "indexes": [
+        {"name": "idx_ncmd_node", "idxtype": "index", "idxfields": ["node_id"]},
+        {"name": "idx_ncmd_status", "idxtype": "index", "idxfields": ["status"]}
+    ],
+    "codes": [
+        {"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'"}
+    ]
+}

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"}
+    ]
+}

models/sync_databases.json

@@ -1,35 +0,0 @@
-{
-    "summary": [
-        {
-            "name": "sync_databases",
-            "title": "数据库同步配置",
-            "primary": ["id"],
-            "catelog": "entity"
-        }
-    ],
-    "fields": [
-        {"name": "id", "title": "配置ID", "type": "str", "length": 32, "nullable": "no"},
-        {"name": "cluster_id", "title": "所属集群", "type": "str", "length": 32, "nullable": "no"},
-        {"name": "db_name", "title": "数据库名", "type": "str", "length": 128, "nullable": "no"},
-        {"name": "master_host", "title": "主库主机", "type": "str", "length": 255},
-        {"name": "master_port", "title": "主库端口", "type": "int", "default": "3306"},
-        {"name": "slave_host", "title": "从库主机", "type": "str", "length": 255},
-        {"name": "slave_port", "title": "从库端口", "type": "int", "default": "3306"},
-        {"name": "repl_user", "title": "复制用户", "type": "str", "length": 64},
-        {"name": "repl_password", "title": "复制密码", "type": "str", "length": 255},
-        {"name": "sync_status", "title": "同步状态", "type": "str", "length": 32, "default": "not_configured"},
-        {"name": "seconds_behind", "title": "延迟秒数", "type": "int"},
-        {"name": "last_error", "title": "最后错误", "type": "text"},
-        {"name": "enabled", "title": "启用", "type": "str", "length": 1, "nullable": "no", "default": "1"},
-        {"name": "created_at", "title": "创建时间", "type": "datetime"},
-        {"name": "updated_at", "title": "更新时间", "type": "datetime"}
-    ],
-    "indexes": [
-        {"name": "idx_sync_db_cluster", "idxtype": "index", "idxfields": ["cluster_id"]},
-        {"name": "idx_sync_db_name", "idxtype": "index", "idxfields": ["db_name"]}
-    ],
-    "codes": [
-        {"field": "sync_status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='sync_status'"},
-        {"field": "cluster_id", "table": "clusters", "valuefield": "id", "textfield": "name"}
-    ]
-}

models/sync_directories.json

@@ -1,33 +0,0 @@
-{
-    "summary": [
-        {
-            "name": "sync_directories",
-            "title": "文件目录同步配置",
-            "primary": ["id"],
-            "catelog": "entity"
-        }
-    ],
-    "fields": [
-        {"name": "id", "title": "配置ID", "type": "str", "length": 32, "nullable": "no"},
-        {"name": "cluster_id", "title": "所属集群", "type": "str", "length": 32, "nullable": "no"},
-        {"name": "name", "title": "同步任务名称", "type": "str", "length": 128, "nullable": "no"},
-        {"name": "source_host", "title": "源主机", "type": "str", "length": 255},
-        {"name": "source_path", "title": "源目录路径", "type": "str", "length": 512, "nullable": "no"},
-        {"name": "dest_host", "title": "目标主机", "type": "str", "length": 255},
-        {"name": "dest_path", "title": "目标目录路径", "type": "str", "length": 512, "nullable": "no"},
-        {"name": "sync_interval", "title": "同步间隔(秒)", "type": "int", "nullable": "no", "default": "60"},
-        {"name": "exclude_patterns", "title": "排除模式", "type": "str", "length": 512},
-        {"name": "last_sync_at", "title": "上次同步时间", "type": "datetime"},
-        {"name": "last_sync_result", "title": "上次同步结果", "type": "str", "length": 32},
-        {"name": "enabled", "title": "启用", "type": "str", "length": 1, "nullable": "no", "default": "1"},
-        {"name": "created_at", "title": "创建时间", "type": "datetime"},
-        {"name": "updated_at", "title": "更新时间", "type": "datetime"}
-    ],
-    "indexes": [
-        {"name": "idx_sync_dir_cluster", "idxtype": "index", "idxfields": ["cluster_id"]}
-    ],
-    "codes": [
-        {"field": "last_sync_result", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='sync_result'"},
-        {"field": "cluster_id", "table": "clusters", "valuefield": "id", "textfield": "name"}
-    ]
-}

scripts/deploy-agent.sh

@@ -0,0 +1,47 @@
+#!/usr/bin/env bash
+# FOMS Agent 一键部署 — 在 A/B 机器上执行
+set -e
+
+INSTALL_DIR=${1:-/opt/foms-agent}
+MANAGEMENT_URL=${2:-http://192.168.1.100:9080}
+NODE_ID=${3:-}
+AGENT_TOKEN=${4:-}
+
+if [ -z "$NODE_ID" ]; then
+    echo "用法: $0 <安装目录> <管理端URL> <节点ID> <Agent令牌>"
+    echo "示例: $0 /opt/foms-agent http://192.168.1.100:9080 node_a_001 mytoken123"
+    exit 1
+fi
+
+echo "=== FOMS Agent 部署 ==="
+echo "  安装目录: $INSTALL_DIR"
+echo "  管理端:   $MANAGEMENT_URL"
+echo "  节点ID:   $NODE_ID"
+
+# 1. 创建目录
+mkdir -p "$INSTALL_DIR"
+
+# 2. 复制 agent 脚本
+SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+cp "$SCRIPT_DIR/foms-agent.py" "$INSTALL_DIR/"
+
+# 3. 创建 .env
+cat > "$INSTALL_DIR/.env" << EOF
+FOMS_MANAGEMENT_URL=$MANAGEMENT_URL
+FOMS_NODE_ID=$NODE_ID
+FOMS_AGENT_TOKEN=$AGENT_TOKEN
+FOMS_HEARTBEAT_INTERVAL=10
+EOF
+
+# 4. 安装 systemd 服务
+cp "$SCRIPT_DIR/foms-agent.service" /etc/systemd/system/foms-agent.service
+sed -i "s|/opt/foms-agent|$INSTALL_DIR|g" /etc/systemd/system/foms-agent.service
+
+# 5. 启用并启动
+systemctl daemon-reload
+systemctl enable foms-agent
+systemctl restart foms-agent
+
+echo "=== 部署完成 ==="
+echo "  systemctl status foms-agent  # 查看状态"
+echo "  journalctl -u foms-agent -f  # 查看日志"

scripts/foms-agent.py

@@ -8,6 +8,9 @@ FOMS Agent — 部署在主机(A)和备机(B)上的守护进程
   3. MySQL主从复制状态监控
   4. 文件rsync同步
   5. 响应切换/切回指令
+  6. 系统指标采集 (CPU/内存/磁盘/网络)
+  7. 系统日志推送到管理中心
+  8. 远程命令执行
 """
 import os
 import sys
@@ -19,12 +22,24 @@ 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,
@@ -47,7 +62,8 @@ class FomsAgent:
 
     # ═══ HTTP helpers ═══
     def _api_get(self, path):
-        url = f"{MANAGEMENT_URL}/api/{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:
@@ -58,7 +74,7 @@ class FomsAgent:
             return None
 
     def _api_post(self, path, data):
-        url = f"{MANAGEMENT_URL}/api/{path}"
+        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}')
@@ -193,28 +209,27 @@ class FomsAgent:
             return {'id': dir_config['id'], 'result': 'failed'}
 
     # ═══ Report Health ═══
-    def report_health(self):
+    def report_db_health(self):
-        """Collect and report all health statuses"""
+        """收集 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}
-        data = {
+        self._api_post('/filesync/api/agent_report_dir_health.dspy', 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):
@@ -228,21 +243,39 @@ class FomsAgent:
     # ═══ 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
-        HEALTH_REPORT_INTERVAL = 60  # report health every 60s
+        last_metrics_report = 0
+        last_command_poll = 0
+        HEALTH_REPORT_INTERVAL = 60
+        METRICS_INTERVAL = 30
         
         while True:
             try:
                 self.heartbeat()
-                
-                # Report health (includes MySQL status + rsync)
                 now = time.time()
+                
+                # Report health（MySQL 复制 → dbbackup；rsync → filesync）
                 if now - last_health_report > HEALTH_REPORT_INTERVAL:
-                    self.report_health()
+                    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")
@@ -251,6 +284,230 @@ class FomsAgent:
                 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:

scripts/foms-agent.service

@@ -0,0 +1,21 @@
+[Unit]
+Description=FOMS Agent - Failover Management System Agent
+Documentation=https://git.opencomputing.cn/yumoqing/foms
+After=network.target mysqld.service
+
+[Service]
+Type=simple
+User=root
+EnvironmentFile=/opt/foms-agent/.env
+WorkingDirectory=/opt/foms-agent
+ExecStart=/usr/bin/python3 /opt/foms-agent/foms-agent.py
+Restart=always
+RestartSec=10
+StandardOutput=append:/var/log/foms-agent.log
+StandardError=append:/var/log/foms-agent.log
+# 防止无限重启耗尽资源
+StartLimitInterval=300
+StartLimitBurst=5
+
+[Install]
+WantedBy=multi-user.target

scripts/load_path.py

@@ -1,84 +0,0 @@
-#!/usr/bin/env python3
-"""FOMS RBAC permission registration"""
-import os, sys
-
-# Find Sage root for set_role_perm.py
-sage_root = None
-for c in [os.path.expanduser("~/repos/sage"), os.path.expanduser("~/test/sage")]:
-    if os.path.isdir(os.path.join(c, "py3", "bin")):
-        sage_root = c
-        break
-if not sage_root:
-    print("ERROR: Sage root not found")
-    sys.exit(1)
-
-sys.path.insert(0, os.path.join(sage_root, "py3", "bin"))
-
-from set_role_perm import set_path_perm
-
-# Paths accessible without login
-PATHS_ANY = [
-    "/",
-    "/index.ui",
-    "/api/login.dspy",
-]
-
-# Paths requiring authentication
-PATHS_LOGINED = [
-    "/api/clusters_create.dspy",
-    "/api/clusters_update.dspy",
-    "/api/clusters_delete.dspy",
-    "/api/nodes_create.dspy",
-    "/api/nodes_update.dspy",
-    "/api/nodes_delete.dspy",
-    "/api/sync_databases_create.dspy",
-    "/api/sync_databases_update.dspy",
-    "/api/sync_databases_delete.dspy",
-    "/api/sync_directories_create.dspy",
-    "/api/sync_directories_update.dspy",
-    "/api/sync_directories_delete.dspy",
-    "/api/agent_heartbeat.dspy",
-    "/api/agent_report_health.dspy",
-    "/api/trigger_switchover.dspy",
-    "/api/trigger_switchback.dspy",
-    "/api/dashboard_stats.dspy",
-    # CRUD list pages
-    "/clusters_list",
-    "/clusters_list/index.ui",
-    "/nodes_list",
-    "/nodes_list/index.ui",
-    "/sync_databases_list",
-    "/sync_databases_list/index.ui",
-    "/sync_directories_list",
-    "/sync_directories_list/index.ui",
-    # CRUD generated files
-    "/clusters_list/get_clusters_list.dspy",
-    "/clusters_list/add_clusters_list.dspy",
-    "/clusters_list/update_clusters_list.dspy",
-    "/clusters_list/delete_clusters_list.dspy",
-    "/nodes_list/get_nodes_list.dspy",
-    "/nodes_list/add_nodes_list.dspy",
-    "/nodes_list/update_nodes_list.dspy",
-    "/nodes_list/delete_nodes_list.dspy",
-    "/sync_databases_list/get_sync_databases_list.dspy",
-    "/sync_databases_list/add_sync_databases_list.dspy",
-    "/sync_databases_list/update_sync_databases_list.dspy",
-    "/sync_databases_list/delete_sync_databases_list.dspy",
-    "/sync_directories_list/get_sync_directories_list.dspy",
-    "/sync_directories_list/add_sync_directories_list.dspy",
-    "/sync_directories_list/update_sync_directories_list.dspy",
-    "/sync_directories_list/delete_sync_directories_list.dspy",
-]
-
-# Agent heartbeat endpoint: no auth (agents use token)
-PATHS_ANY.append("/api/agent_heartbeat.dspy")
-PATHS_ANY.append("/api/agent_report_health.dspy")
-
-if __name__ == '__main__':
-    for p in PATHS_ANY:
-        set_path_perm("any", p)
-        print(f"  any    {p}")
-    for p in PATHS_LOGINED:
-        set_path_perm("logined", p)
-        print(f"  logined {p}")
-    print("Done. Restart FOMS to apply.")

start.sh

@@ -1,6 +0,0 @@
-#!/usr/bin/env bash
-cd "$(dirname "$0")"
-[ -f .env ] && source .env
-nohup $PWD/py3/bin/python $PWD/app/foms.py -p ${FOMS_PORT:-9080} -w $PWD > $PWD/logs/foms.out 2>&1 &
-echo $! > foms.pid
-echo "FOMS started (PID: $(cat foms.pid)) on port ${FOMS_PORT:-9080}"

stop.sh

@@ -1,9 +0,0 @@
-#!/usr/bin/env bash
-cd "$(dirname "$0")"
-if [ -f foms.pid ]; then
-    pid=$(cat foms.pid)
-    kill $pid 2>/dev/null && echo "FOMS stopped (PID: $pid)" || echo "FOMS not running"
-    rm -f foms.pid
-else
-    echo "No PID file found"
-fi

wwwroot/api/agent_poll_commands.dspy

@@ -0,0 +1,2 @@
+result = await agent_poll_commands(request, params_kw)
+return json.dumps(result, ensure_ascii=False)

wwwroot/api/agent_report_command_result.dspy

@@ -0,0 +1,2 @@
+result = await agent_report_command_result(request, params_kw)
+return json.dumps(result, ensure_ascii=False)

wwwroot/api/agent_report_health.dspy

@@ -1,3 +0,0 @@
-# Agent health report endpoint (sync statuses)
-result = await agent_report_health(request, params_kw)
-return json.dumps(result, ensure_ascii=False)

wwwroot/api/exec_remote_command.dspy

@@ -0,0 +1,2 @@
+result = await exec_remote_command(request, params_kw)
+return json.dumps(result, ensure_ascii=False)

wwwroot/api/sync_databases_create.dspy

@@ -1,9 +0,0 @@
-# Cluster create wrapper
-from ahserver.serverenv import ServerEnv
-env = ServerEnv()
-create_func = getattr(env, 'create_sync_db', None)
-if create_func is None:
-    print(json.dumps({"widgettype": "Message", "options": {"title": "Error", "message": "create_sync_db not found", "type": "error"}}))
-else:
-    result = await create_func(request, params_kw)
-    print(result)

wwwroot/api/sync_databases_delete.dspy

@@ -1,9 +0,0 @@
-# Cluster create wrapper
-from ahserver.serverenv import ServerEnv
-env = ServerEnv()
-create_func = getattr(env, 'delete_sync_db', None)
-if create_func is None:
-    print(json.dumps({"widgettype": "Message", "options": {"title": "Error", "message": "delete_sync_db not found", "type": "error"}}))
-else:
-    result = await create_func(request, params_kw)
-    print(result)

wwwroot/api/sync_databases_update.dspy

@@ -1,9 +0,0 @@
-# Cluster create wrapper
-from ahserver.serverenv import ServerEnv
-env = ServerEnv()
-create_func = getattr(env, 'update_sync_db', None)
-if create_func is None:
-    print(json.dumps({"widgettype": "Message", "options": {"title": "Error", "message": "update_sync_db not found", "type": "error"}}))
-else:
-    result = await create_func(request, params_kw)
-    print(result)

wwwroot/api/sync_directories_create.dspy

@@ -1,9 +0,0 @@
-# Cluster create wrapper
-from ahserver.serverenv import ServerEnv
-env = ServerEnv()
-create_func = getattr(env, 'create_sync_dir', None)
-if create_func is None:
-    print(json.dumps({"widgettype": "Message", "options": {"title": "Error", "message": "create_sync_dir not found", "type": "error"}}))
-else:
-    result = await create_func(request, params_kw)
-    print(result)

wwwroot/api/sync_directories_delete.dspy

@@ -1,9 +0,0 @@
-# Cluster create wrapper
-from ahserver.serverenv import ServerEnv
-env = ServerEnv()
-create_func = getattr(env, 'delete_sync_dir', None)
-if create_func is None:
-    print(json.dumps({"widgettype": "Message", "options": {"title": "Error", "message": "delete_sync_dir not found", "type": "error"}}))
-else:
-    result = await create_func(request, params_kw)
-    print(result)

wwwroot/api/sync_directories_update.dspy

@@ -1,9 +0,0 @@
-# Cluster create wrapper
-from ahserver.serverenv import ServerEnv
-env = ServerEnv()
-create_func = getattr(env, 'update_sync_dir', None)
-if create_func is None:
-    print(json.dumps({"widgettype": "Message", "options": {"title": "Error", "message": "update_sync_dir not found", "type": "error"}}))
-else:
-    result = await create_func(request, params_kw)
-    print(result)

wwwroot/index.ui

@@ -1,87 +0,0 @@
-{
-    "widgettype": "VBox",
-    "options": {"width": "100%", "height": "100%", "padding": "24px", "backgroundColor": "#F5F7FA"},
-    "subwidgets": [
-        {
-            "widgettype": "Text",
-            "options": {"text": "FOMS — 主备切换管理系统", "fontSize": "28px", "fontWeight": "bold", "marginBottom": "8px"}
-        },
-        {
-            "widgettype": "Text",
-            "options": {"text": "Failover Management System · 实时监控 · 自动切换 · 数据同步", "fontSize": "14px", "color": "#666", "marginBottom": "24px"}
-        },
-        {
-            "widgettype": "ResponsableBox",
-            "options": {"gap": "16px", "minWidth": "220px", "marginBottom": "24px"},
-            "subwidgets": [
-                {
-                    "widgettype": "VBox",
-                    "options": {"backgroundColor": "#FFFFFF", "padding": "20px", "borderRadius": "8px", "cursor": "pointer"},
-                    "binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.foms_content", "options": {"url": "{{entire_url('clusters_list')}}"}, "mode": "replace"}],
-                    "subwidgets": [
-                        {"widgettype": "Text", "options": {"text": "集群管理", "fontSize": "16px", "fontWeight": "bold", "marginBottom": "8px"}},
-                        {"widgettype": "RefreshWidget", "options": {"interval": 30000, "data_url": "{{entire_url('api/dashboard_stats.dspy')}}"},
-                            "subwidgets": [
-                                {"widgettype": "Text", "options": {"text": "{{data.total_clusters}} 个集群", "fontSize": "14px", "color": "#666"}}
-                            ]
-                        }
-                    ]
-                },
-                {
-                    "widgettype": "VBox",
-                    "options": {"backgroundColor": "#FFFFFF", "padding": "20px", "borderRadius": "8px", "cursor": "pointer"},
-                    "binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.foms_content", "options": {"url": "{{entire_url('nodes_list')}}"}, "mode": "replace"}],
-                    "subwidgets": [
-                        {"widgettype": "Text", "options": {"text": "节点管理", "fontSize": "16px", "fontWeight": "bold", "marginBottom": "8px"}},
-                        {"widgettype": "RefreshWidget", "options": {"interval": 30000, "data_url": "{{entire_url('api/dashboard_stats.dspy')}}"},
-                            "subwidgets": [
-                                {"widgettype": "Text", "options": {"text": "{{data.nodes_online}} 在线 / {{data.nodes_offline}} 离线", "fontSize": "14px", "color": "#666"}}
-                            ]
-                        }
-                    ]
-                },
-                {
-                    "widgettype": "VBox",
-                    "options": {"backgroundColor": "#FFFFFF", "padding": "20px", "borderRadius": "8px", "cursor": "pointer"},
-                    "binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.foms_content", "options": {"url": "{{entire_url('sync_databases_list')}}"}, "mode": "replace"}],
-                    "subwidgets": [
-                        {"widgettype": "Text", "options": {"text": "数据库同步", "fontSize": "16px", "fontWeight": "bold", "marginBottom": "8px"}},
-                        {"widgettype": "RefreshWidget", "options": {"interval": 30000, "data_url": "{{entire_url('api/dashboard_stats.dspy')}}"},
-                            "subwidgets": [
-                                {"widgettype": "Text", "options": {"text": "{{data.dbs_synced}} 正常 / {{data.dbs_error}} 异常", "fontSize": "14px", "color": "#666"}}
-                            ]
-                        }
-                    ]
-                },
-                {
-                    "widgettype": "VBox",
-                    "options": {"backgroundColor": "#FFFFFF", "padding": "20px", "borderRadius": "8px", "cursor": "pointer"},
-                    "binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.foms_content", "options": {"url": "{{entire_url('sync_directories_list')}}"}, "mode": "replace"}],
-                    "subwidgets": [
-                        {"widgettype": "Text", "options": {"text": "文件同步", "fontSize": "16px", "fontWeight": "bold", "marginBottom": "8px"}},
-                        {"widgettype": "RefreshWidget", "options": {"interval": 30000, "data_url": "{{entire_url('api/dashboard_stats.dspy')}}"},
-                            "subwidgets": [
-                                {"widgettype": "Text", "options": {"text": "{{data.dirs_synced}} 同步正常", "fontSize": "14px", "color": "#666"}}
-                            ]
-                        }
-                    ]
-                }
-            ]
-        },
-        {
-            "widgettype": "VBox",
-            "id": "foms_content",
-            "options": {"width": "100%", "flex": "1", "backgroundColor": "#FFFFFF", "borderRadius": "8px", "padding": "20px", "minHeight": "400px"},
-            "subwidgets": [
-                {
-                    "widgettype": "Text",
-                    "options": {"text": "欢迎使用 FOMS 主备切换管理系统", "fontSize": "18px", "color": "#999", "textAlign": "center", "marginTop": "80px"}
-                },
-                {
-                    "widgettype": "Text",
-                    "options": {"text": "点击上方卡片进入各功能模块", "fontSize": "14px", "color": "#BBB", "textAlign": "center"}
-                }
-            ]
-        }
-    ]
-}