提交 05d43be5 authored 作者: 陈泽健's avatar 陈泽健

docs(deployment): 更新ARM部署文档和自动化脚本的skill技能文件

- 更新ARM-QLV10-XTYBS技能文档中的服务器IP从192.168.9.83到192.168.9.70
- 更新授权文件路径从E:\自动化部署\ARM-9.83\license.zip到E:\自动化部署\ARM-9.70\license.zip
- 更新系统网址、接口URL和报告命名中的IP地址变更
- 优化CreateCMD技能文档中的Windows CMD窗口启动逻辑和注意事项
- 新增完整的授权流程自动化脚本auth_full_flow.py
- 添加中间件验证框架包括基础检查类、配置文件和确认器
- 实现EMQX中间件检查器和验证逻辑
- 配置新的中间件验证参数和修复策略
上级 e126f6fd
......@@ -34,12 +34,12 @@ ARM架构-麒麟V10服务器远程自动化部署技能。
- **部署脚本目录**: `AuxiliaryTool/ScriptTool/RemoteDeploy/`
- **主部署脚本**: `AuxiliaryTool/ScriptTool/RemoteDeploy/full_deploy.py`
- **辅助包装脚本**: `AuxiliaryTool/ScriptTool/RemoteDeploy/auto_deploy_wrapper.sh`
- **授权文件路径**: `E:\自动化部署\ARM-9.83\license.zip`
- **授权文件路径**: `E:\自动化部署\ARM-9.70\license.zip`
## 目标服务器信息
- **IP**: 192.168.9.83
- **用户**: root
- **IP**: 192.168.9.70
- **用户**: openkylin
- **密码**: Ubains@123
- **架构**: ARM (aarch64)
- **操作系统**: 麒麟V10(Kylin V10)
......@@ -59,9 +59,9 @@ ARM架构-麒麟V10服务器远程自动化部署技能。
## 系统网址
- 前台地址: `https://192.168.9.83/`
- 维护地址: `https://192.168.9.83/#/LoginConfig`
- 后台地址: `https://192.168.9.83/#/LoginAdmin`
- 前台地址: `https://192.168.9.70/`
- 维护地址: `https://192.168.9.70/#/LoginConfig`
- 后台地址: `https://192.168.9.70/#/LoginAdmin`
---
......@@ -87,7 +87,7 @@ ARM架构-麒麟V10服务器远程自动化部署技能。
#### 步骤1:SSH连接目标服务器
- 使用 paramiko 或 `full_deploy.py` 中的 `ssh_connect` 函数连接 192.168.9.83
- 使用 paramiko 或 `full_deploy.py` 中的 `ssh_connect` 函数连接 192.168.9.70
- 用户名: root,密码: Ubains@123
- 切换到 root 用户(已是 root 则跳过)
......@@ -156,7 +156,7 @@ ARM架构-麒麟V10服务器远程自动化部署技能。
- 等待10分钟让服务完全启动
- 调用对外接口验证服务可达性:
- 命令: `curl -k https://192.168.9.83/exapi/message/getMsgPageList`
- 命令: `curl -k https://192.168.9.70/exapi/message/getMsgPageList`
- 成功标志: 响应包含 `无效token` 或 `Full authentication`
- 按照重试机制执行(最多5次,每次间隔30秒)
- 记录结果
......@@ -169,7 +169,7 @@ ARM架构-麒麟V10服务器远程自动化部署技能。
#### 步骤1:登录维护平台
- 打开浏览器访问: `https://192.168.9.83/#/LoginConfig`
- 打开浏览器访问: `https://192.168.9.70/#/LoginConfig`
- 使用 Chrome DevTools MCP 工具进行浏览器自动化
- 账号: `superadmin`,密码: `Ubains@1357`
- 验证码: `csba`
......@@ -183,7 +183,7 @@ ARM架构-麒麟V10服务器远程自动化部署技能。
#### 步骤3:上传授权文件
- 点击"上传授权文件"
- 使用本地授权文件: `E:\自动化部署\ARM-9.83\license.zip`
- 使用本地授权文件: `E:\自动化部署\ARM-9.70\license.zip`
- **注意**: 弹出"校验身份"对话框时,输入密码 `Ubains@1357` 和验证码 `csba`
- 等待页面加载成功
......@@ -226,10 +226,10 @@ ARM架构-麒麟V10服务器远程自动化部署技能。
| 接口 | URL | 成功标志 | 失败标志 |
|------|-----|---------|---------|
| 预定对外接口 | `https://192.168.9.83/exapi/message/getMsgPageList` | 包含 `无效token` 或 `Full authentication` | 包含 `Error` HTML页面 |
| 预定系统接口 | `https://192.168.9.83/meetingV3/api/systemConfiguration/globalConfig?companyNumber=CN-SZ-00-0201` | 包含 `accessToken为空` | 包含 `内部服务器错误` |
| 运维集控接口 | `https://192.168.9.83/monitor/api2/api/servermonitor/` | 包含 `用户不存在` | 包含 `Error` HTML页面 |
| 讯飞转录接口 | `https://192.168.9.83/voice/api/iflytek/roommaster?company_id=1&user_id=8&company_secret=57d00f9f-020f-5f1f-b788-55fae843bceb&getall=1` | 包含 `缺少关键参数` | 包含 `Error` HTML页面 |
| 预定对外接口 | `https://192.168.9.70/exapi/message/getMsgPageList` | 包含 `无效token` 或 `Full authentication` | 包含 `Error` HTML页面 |
| 预定系统接口 | `https://192.168.9.70/meetingV3/api/systemConfiguration/globalConfig?companyNumber=CN-SZ-00-0201` | 包含 `accessToken为空` | 包含 `内部服务器错误` |
| 运维集控接口 | `https://192.168.9.70/monitor/api2/api/servermonitor/` | 包含 `用户不存在` | 包含 `Error` HTML页面 |
| 讯飞转录接口 | `https://192.168.9.70/voice/api/iflytek/roommaster?company_id=1&user_id=8&company_secret=57d00f9f-020f-5f1f-b788-55fae843bceb&getall=1` | 包含 `缺少关键参数` | 包含 `Error` HTML页面 |
---
......@@ -259,7 +259,7 @@ ARM架构-麒麟V10服务器远程自动化部署技能。
#### 报告命名
- 部署分析报告: `ARM_192.168.9.83_麒麟V10自动化部署分析报告_YYYYMMDD.md`
- 部署分析报告: `ARM_192.168.9.70_麒麟V10自动化部署分析报告_YYYYMMDD.md`
- 保存路径: `AuxiliaryTool/ScriptTool/RemoteDeploy/reports/`
---
......
此差异已折叠。
......@@ -26,17 +26,25 @@ description: 在当前目录打开指定数量的CMD窗口
**执行步骤:**
1. 如果用户未提供数量参数,使用 AskUserQuestion 询问用户需要打开几个 CMD 窗口(提供 1、2、3、4 作为常见选项,也可输入其他数量)
2. 获取当前工作目录路径
3. 使用命令打开指定数量的 CMD 窗口:
2. 获取当前工作目录路径(Windows 形式,如 `E:\GithubData\ubains-module-test`
3. **定位 claude 的完整路径**:在 bash 中运行 `which claude`,得到形如 `/e/nodejs/claude` 的路径,转换成 Windows 形式并补 `.cmd`,即 `E:\nodejs\claude.cmd`。原因是 `E:\nodejs` 不在 Windows PATH 里,新开的 cmd 窗口直接敲 `claude` 会报"不是内部或外部命令",必须用完整路径。
4. **生成一个临时批处理文件**(如 `E:\GithubData\ubains-module-test\.claude\skills\CreateCMD\_run_claude.bat`),内容分两行,避免 `&&` 复合命令在引号传递时被截断:
```bat
@echo off
cd /d <当前目录>
<claude 完整路径> --permission-mode bypassPermissions
```
# 打开单个 CMD 窗口,切换到当前目录并自动启动 claude
start cmd /k "cd /d <当前目录> && claude --permission-mode bypassPermissions"
5. **用真正的 Windows cmd `start` 打开 N 个窗口运行该 bat**。在 bash 中通过 `cmd.exe /c start` 调用(注意:bash 里直接写 `start` 会走 `/usr/bin/start` 这个 MSYS 封装脚本,它会把 `"cd ... && claude"` 这种复合命令的引号传坏,导致 claude 不执行):
```bash
for i in $(seq 1 N); do
MSYS_NO_PATHCONV=1 cmd.exe /c start "Claude $i" cmd /k "<bat 完整路径>"
done
```
循环执行 N 次,每次打开一个新的 CMD 窗口,窗口打开后会自动执行 `claude --permission-mode bypassPermissions`
4. 确认并告知用户已成功打开的窗口数量
**注意事项:**
- 使用 `start cmd /k "cd /d <path> && claude --permission-mode bypassPermissions"` 命令,`/k` 参数确保窗口执行完命令后保持打开
- 工作目录设为 Claude Code 的当前工作目录
- 每个窗口打开后会自动切换到当前目录并启动 `claude --permission-mode bypassPermissions`
- 数量上限建议不超过 10 个,避免系统资源占用过高
6. 确认并告知用户已成功打开的窗口数量
**注意事项(两个坑,务必遵守):**
- **不要用裸 `start`**:bash 里的 `start` 解析为 `/usr/bin/start`(MSYS 封装),会损坏 `&&` 复合命令的引号,表现为窗口打开、cd 成功但 claude 不执行。必须用 `cmd.exe /c start` 走 Windows 原生 `start`,并加 `MSYS_NO_PATHCONV=1` 防止 MSYS 路径转换。
- **不要用裸 `claude`**`E:\nodejs` 不在 Windows PATH,cmd 窗口里找不到。必须用 `which claude` 推导出的 `claude.cmd` 完整路径。
- **用 .bat 文件承载命令序列**,而不是 `"cd ... && claude ..."` 单行,可彻底规避复合命令引号丢失问题。
- `/k` 参数确保窗口执行完命令后保持打开;工作目录设为 Claude Code 的当前工作目录。
- 数量上限建议不超过 10 个,避免系统资源占用过高
@echo off
REM 切换到项目工作目录
cd /d E:\GithubData\ubains-module-test
REM 用完整路径启动 Claude Code(E:\nodejs 不在 Windows PATH 中)
E:\nodejs\claude.cmd --permission-mode bypassPermissions
---
name: MiddleVerify
description: 中间件升级后验证与修复 - 用 MiddlewareVerify 工具对 umysql/uredis/unacos/unginx/uemqx/utracker 六个中间件做三层验证,确认真失败才分级修复,绝不误判
---
中间件升级后验证 + 修复工具。每次镜像组件升级后执行,验证六个中间件(u 前缀)可用,失败时安全分级修复且**绝不误判**。独立于 `ServiceSelfInspection`,互不耦合。
## Usage
```
/MiddleVerify # 默认:连 9.76 验证 + 修复(按 config.yaml)
/MiddleVerify 只读 # 仅验证不修复(冒烟测试)
/MiddleVerify 换服务器 # 提示改 config.yaml,不改代码
```
## 核心位置
- **工具目录**`AuxiliaryTool/ScriptTool/MiddlewareVerify/`
- **入口**`python verify_middleware.py`(在该目录下执行)
- **配置**`config.yaml`(改服务器/凭据只改这里,代码不动)
- **报告**`reports/verify_<时间戳>.md`
**执行命令:**
```bash
cd AuxiliaryTool/ScriptTool/MiddlewareVerify
python verify_middleware.py # 验证 + 修复
python verify_middleware.py --no-repair # 只读模式(冒烟/排障)
```
## 六个中间件 + 依赖顺序
验证顺序 = 依赖顺序(前者挂了后者必然受影响,便于定位根因):
`umysql → uredis → uemqx → unacos → utracker → unginx`
| 容器 | 镜像预期 | 关键端口 | 备注 |
|------|----------|----------|------|
| umysql | mysql:8.0.46 | **8306**(非3306) | 库 nacos_mysql/devops |
| uredis | redis:8 | host 6379 | host 网络模式 |
| uemqx | emqx:5.8.7 | 1883 / 18083 | 禁匿名 sha256 |
| unacos | nacos:v2.5.2 | 8848 / 9848 | 认证开启 |
| utracker | ufastdfs:v2 | host 22122/23000/8888 | **FastDFS**,非链路追踪 |
| unginx | nginx:1.30.2 | **仅 443**(无80) | HTTPS |
## 防误判设计(核心,不可破坏)
**状态三分类**(base.py):
- `PASS` 正常
- `FAIL` 服务功能性失败 → 确认真失败后**可修复**
- `ERROR` 基础设施/脚本错误(容器不存在、docker 缺失、ARM 不兼容)→ **永不修复**
- `SKIP` 降级或前置不满足,跳过
**五重门控**(确认真失败才动手,五条任一不满足即跳过修复):
1. **连续 N 次失败确认**(默认3次,间隔 10/15/20s):任一次 PASS 判瞬态抖动不修;遇 ERROR 立即停止永不修;连续 N 次全 FAIL 才算真失败
2. **基础设施 ERROR 不修**`infra_error=True` 直接跳过,需人工排查
3. **凭据类失败不 restart**`CheckItem.repairable=False`(认证/凭据配置类)→ restart 无效,跳过自动修复,标"需人工核对凭据"
4. **同容器冷却期**(默认1800s/30min):远端状态文件 `/data/logs/.middleware_verify_state``last_repair_<容器>`,冷却内不重复修,防反复 restart
5. **修复前取证 + 修复后复验**:动作前采 `docker ps` + `logs --tail 50`,动作后等待就绪再用 Confirmer 复验(复验也走连续失败确认)
## 三层验证(L1/L2/L3)
- **L1 容器状态**`docker inspect` 容器存活;不存在→ERROR,非running→FAIL
- **L2 端口监听**`ss -tln` 查不到则 `nc -z` 兜底;都失败→FAIL
- **L3 功能验证**(真实操作,最强证据):
- umysql:真实登录 + 查 required_databases
- uredis:带密码 AUTH + SET/GET 读写
- uemqx:dashboard 登录取 token + 业务账号 MQTT 真实发布/订阅
- unacos:HTTP 200 + 配置中心读取
- utracker:FastDFS 真实上传一个测试文件
- unginx:HTTPS 443 握手 + 返回码
## 分级修复(L1/L2/L3,递进高危)
仅 Confirmer 确认"真失败"且非基础设施错误时执行:
- **L1 容器级**(所有中间件):running 但服务坏→`docker restart`;非running→`docker start`
- **L2 账号配置**(仅 uemqx):dashboard 登录取 token → 幂等重注册业务账号 `mqtt@cmdb`
- **L3 数据/连带**(⚠️高危,控制台+报告双重告警):
- uredis:清数据目录 + restart(**防删根**:路径深度校验 + `${RDP:?}` 空则中止,丢缓存数据)
- unacos:连带 `docker restart ujava2`(token 变更致 JWT 失效,业务短暂中断)
修复后按 `readiness_wait` 等待就绪(umysql30/unacos40/uemqx35/uredis10/默认15s),再复验。
## 退出码
- `0` 服务级全部可用(凭据异常算可用,但报告标 ⚠️)
- `1` 有服务异常(修复后仍失败 / 基础设施错误)
- `2` SSH 连接失败
## 代码结构
```
MiddlewareVerify/
verify_middleware.py # 入口:加载配置→SSH→逐中间件(验证→确认→修复→复验)→汇总+报告
config.yaml # 服务器/凭据/修复策略(唯一需改的文件)
ssh_exec.py # paramiko 封装:run / docker_exec / 状态文件读写;兼容 root 与 admin+sudo
checks/
base.py # 三层验证基类 + 状态枚举 + CheckItem/CheckOutcome
mysql/redis/emqx/nacos/nginx/fastdfs.py # 各中间件 verify()
confirmer.py # 连续失败确认 + 失败分类(PASS/ERROR 立即返回,FAIL 重试)
repair.py # 分级修复 + 取证 + 冷却期 + 复验
results.py # 记录结构(避免 reporter 与 repair 循环依赖)
reporter.py # 控制台彩色输出 + Markdown 报告
reports/ # 生成的报告
```
## 已知坑(实测,勿改勿当故障处理)
**9.76 SSH(2026-06-26 实测)**
- `root` 登录被禁(Authentication failed),实际用 **admin / Ubains@123** + `sudo -S`(sudo 密码同 Ubains@123)
- config.yaml 已配 `use_sudo: true`,命令自动经 sudo;密码经 stdin 传,不进进程列表
**9.76 中间件凭据(部署脚本/文档与实测不一致,以实测为准)**
- **umysql**:root / `dNrprU&2S`,对外端口 **8306**(非3306)→ 正常
- **uredis**:密码 `dNrprU&2S`,host 网络 6379 → 正常
- **uemqx Dashboard**:实测是 EMQX 默认 **admin/public**(部署脚本的 `Admin@2026Secure` 无效→BAD_USERNAME_OR_PWD,安全改密步骤未生效);MQTT 业务账号 `mqtt@cmdb` / `mqtt@webpassw0RD` 已注册
- **unacos 控制台登录失败 = 假异常(勿当服务故障)**
- **现象**:L3 `/v1/auth/login` 对任何密码返回 `user not found!`(连续3次确认失败),工具标 `⚠️凭据类·不自动修复`
- **根因**:9.76 "安全改密"破坏了认证链——mysql `nacos_mysql.users` 表里**有** nacos 用户,但登录校验认不出(改了密码但认证配置/哈希未同步)。纯部署侧缺陷,**restart 修不了**(根因不在容器)
- **判为假异常的证据**(2026-06-26 实测):① `/console/server/state` 返回 HTTP 200 + 完整 JSON(`auth_enabled`/`console_ui_enabled` 正常)→ 服务进程健康;② 容器日志近 200 行无真 ERROR(仅 Spring Security 初始化 INFO),`Up 3 weeks` 稳定;③ **依赖 nacos 的业务容器 `ujava2` Up 2 days** → 业务面(配置中心+注册中心)在用且可用,最强证据
- **结论**:nacos 对业务完全可用,仅控制台管理登录这条路径坏了。工具标 ⚠️ + 退出码 0 正确;**看到 ⚠️ unacos 不要当 nacos 挂了去 restart**,属部署待还的债
- **unginx**:HTTPS 443 → 正常
- **utracker**:FastDFS host 22122/23000/8888 → 正常
## 注意事项
- **改服务器/凭据只改 `config.yaml`**,不要动代码
- Windows 控制台默认 GBK,入口已强制 `sys.stdout.reconfigure(utf-8)`;报告文件始终 UTF-8
- 验证顺序 = 依赖顺序,定位根因时看**第一个失败**的中间件
- 冒烟/排障先用 `--no-repair`,确认无误判再开修复
- L3(Redis 清数据 / Nacos 连带重启)是高危动作,会丢缓存数据、业务短暂中断,执行前看控制台与报告的 ⚠️ 告警
- 报告里凭据类异常标 ⚠️ 但退出码算可用,这是**有意的**(凭据问题不是服务故障)
## 关联记忆
- `middleware-verify-tool` — 工具位置与五重门控速查
- `server-976-middleware-credentials` — 9.76 实测凭据全集
- `server-976-ssh-admin` — 9.76 用 admin+sudo(root 被禁)
此差异已折叠。
---
name: X86-UPDATE-CONTAINER
description: X86架构镜像组件版本升级 - 对192.168.5.52服务器的JDK/MySQL/Nginx/Redis/EMQX/Nacos进行版本升级,严格按PRD执行并验证
---
X86架构镜像组件版本升级,对 192.168.5.52 服务器执行 JDK/MySQL/Nginx/Redis/EMQX/Nacos 版本升级,严格按 PRD 需求文档和计划执行文档操作。升级完成后验证容器状态和服务可用性,打包新镜像替换离线包。
## Usage
```
/X86-UPDATE-CONTAINER [组件名]
```
可选组件名参数(不传则按顺序执行全部):
- `jdk` — 升级 JDK(容器内 + 宿主机)
- `mysql` — 升级 MySQL
- `nginx` — 升级 Nginx
- `redis` — 升级 Redis
- `emqx` — 升级 EMQX
- `nacos` — 升级 Nacos(当前已是 v2.5.2,通常无需升级)
## 核心文档
- **PRD需求文档**`Docs/PRD/镜像组件升级/_PRD_X86架构_远程升级镜像组件版本_需求文档.md`
- **计划执行文档**`Docs/PRD/镜像组件升级/_PRD_X86架构_远程升级镜像组件版本_计划执行.md`
## 目标服务器
| 架构 | IP | SSH凭据 | 部署包路径 |
|------|-----|----------|-----------|
| X86 | 192.168.5.52 | root / Ubains@123 | `/data/offline_auto_unifiedPlatform` |
| 镜像目录 | — | — | `/data/offline_auto_unifiedPlatform/data/temp` |
## 版本对照表(执行前确认)
| 组件 | 当前版本 | 升级目标 | 约束 |
|------|---------|----------|------|
| JDK(容器+宿主) | 1.8.0_201 | **1.8.0_492** | 只能升级 **1.8.X**,禁止非 1.8.X |
| MySQL | 8.0.28 | **8.0.46** | 只能升级 **8.x**,禁止非 8.x |
| Nginx | 1.29.3 | **1.30.2** | 官方 stable 最新 |
| Redis | 8.2.2 | **8.8.0** | 官方最新开源版 |
| EMQX | 5.8.7 | **6.0.0** | 官方最新主版本(Enterprise) |
| Nacos | v2.5.2 | — | 当前已是最新,禁止跨 **3.x**,只能 2.5.X |
| Docker | 29.1.3 | — | **仅记录最新版本,不执行升级** |
## 升级流程(顺序执行)
按依赖顺序:**JDK → MySQL → Nginx → Redis → EMQX → Nacos**
### 1. 升级 JDK(容器内 + 宿主机)
**步骤:**
1. 下载 `jdk-8u492-linux-x64.tar.gz`(Temurin/Adoptium,清华镜像源)
2. 备份容器启动参数:`docker inspect ujava2 > /tmp/ujava2_inspect.json`
3. 复制新 JDK 进容器:`docker cp jdk8-temurin.tar.gz ujava2:/tmp/`
4. 进入容器替换 JDK(路径:`/opt/deploy/java/jdk1.8.0_201` → 符号链接至 `jdk8u492-b09`
5. commit 新镜像:`docker commit ujava2 139.9.60.86:5000/ujava:v7`
6. 验证:`docker exec ujava2 java -version` → 输出 **1.8.0_492**
7. 检查服务日志无异常
8. 打包新镜像:`docker save ... | gzip > /data/.../data/temp/java_new.tar.gz`
9. 宿主机安装 JDK:`/usr/local/java/jdk8u492-b09`,环境变量 `/etc/profile`
**验证点:**
- 容器内 `java -version` → 1.8.0_492
- 预定服务日志:`/data/services/api/java-meeting/java-meeting-extapi/logs/ubains-INFO-AND-ERROR.log` 无异常
- 预定对内服务日志:`/data/services/api/java-meeting/java-meeting2.0/logs/ubains-INFO-AND-ERROR.log` 无异常
### 2. 升级 MySQL
**步骤:**
1. 确认当前版本:`docker exec umysql mysql --version` → 8.0.28
2. 备份数据:`mysqldump --result-file` + `docker cp`(密码:`dNrprU&2S`
3. 备份容器参数:`docker inspect umysql > /tmp/umysql_inspect.json`
4. 拉取新镜像:`docker pull mysql:8.0`(实际 8.0.46)
5. 停止旧容器:`docker stop umysql && docker rm umysql`
6. 用新镜像 + 原配置/数据卷启动新容器
7. 验证:`docker exec umysql mysql --version`**8.0.46**
8. 确认版本升级日志(Server upgrade from '80028' to '80046')
9. 验证 Nacos 等服务连接正常
10. 打包:`/data/.../data/temp/mysql-8.0.46.tar.gz`
**验证点:**
- 容器运行状态:`docker ps` → umysql **Up**
- 版本命令:`mysql --version` → 8.0.46
- Nacos/ujava2 等依赖服务正常连接
### 3. 升级 Nginx
**步骤:**
1. 备份配置:`docker cp unginx:/etc/nginx/nginx.conf /tmp/nginx.conf.bak`
2. 备份容器参数:`docker inspect unginx > /tmp/unginx_inspect.json`
3. 拉取新镜像:`docker pull nginx:1.30.2`
4. 停止旧容器:`docker stop unginx && docker rm unginx`
5. 用新镜像 + 原配置/挂载启动新容器
6. 验证:`docker exec unginx nginx -v`**1.30.2**
7. 测试代理功能:`curl -k https://127.0.0.1/` 正常
8. 打包:`/data/.../data/temp/nginx-1.30.2.tar.gz`
9. 旧包 `nginx-1.29.3.tar.gz` 保留
**验证点:**
- 版本命令:`nginx -v` → 1.30.2
- HTTPS 代理:`curl -k https://192.168.5.52/` → HTTP 200
### 4. 升级 Redis
**步骤:**
1. 备份配置
2. 备份容器参数:`docker inspect uredis > /tmp/uredis_inspect.json`
3. 拉取新镜像:`docker pull redis:8`(实际 8.8.0)
4. 停止旧容器:`docker stop uredis && docker rm uredis`
5. 用新镜像 + 原配置/挂载启动新容器(**需加 `--network host`**
6. 验证:`docker exec uredis redis-server --version`**8.8.0**
7. 测试连接:`docker exec uredis redis-cli -a <密码> ping`**PONG**
8. 确认依赖 Redis 的服务正常
9. 打包:`/data/.../data/temp/redis-8.8.0.tar.gz`
**验证点:**
- 版本命令:`redis-server --version` → 8.8.0
- 连接测试:`redis-cli ping` → PONG
- 运维服务日志:`/data/services/api/python-cmdb/log/uinfo.log` 无异常
### 5. 升级 EMQX
**步骤:**
1. 备份 EMQX 数据卷和配置
2. 备份容器参数:`docker inspect uemqx > /tmp/uemqx_inspect.json`
3. 拉取新镜像:`docker pull emqx/emqx:latest`(实际 6.0.0 Enterprise)
4. 停止旧容器:`docker stop uemqx && docker rm uemqx`
5. 用新镜像 + 原配置启动新容器(**配置需兼容 6.x**
6. 验证:`docker exec uemqx emqx ctl broker`**6.0.0**
7. 测试 MQTT 连接正常
8. 重启依赖 MQTT 的服务(ujava2 等),确认重连成功
9. 打包:`/data/.../data/temp/uemqx-6.0.0.tar.gz`
**验证点:**
- 版本命令:`emqx ctl broker` → 6.0.0
- MQTT 连接测试:业务账号 pub/sub 正常
- 讯飞服务日志:`/data/services/api/python-voice/log/uinfo.log` 无异常
### 6. 升级 Nacos
**当前已是 v2.5.2,通常无需升级。**
若需升级:
- **禁止跨越 3.x 版本**
- 只能升级到 **2.5.X** 版本
- 步骤同其他容器组件(备份配置 → 拉取新镜像 → 替换容器 → 验证)
## 每步验证清单(必做)
升级后必须执行以下验证:
1. **容器运行状态**`docker ps` 确认所有容器 **Up**
- X86:13 个容器
2. **版本确认**:对应版本命令输出新版本号
3. **容器日志检查**`docker logs <container> --tail 50`
- MQTT 重连日志属正常现象
4. **服务日志检查**
- 预定对外:`/data/services/api/java-meeting/java-meeting-extapi/logs/ubains-INFO-AND-ERROR.log`
- 预定对内:`/data/services/api/java-meeting/java-meeting2.0/logs/ubains-INFO-AND-ERROR.log`
- 运维服务:`/data/services/api/python-cmdb/log/uinfo.log`
- 讯飞服务:`/data/services/api/python-voice/log/uinfo.log`
5. **接口测试**
- `curl -k https://192.168.5.52/exapi/message/getMsgPageList` → 返回"无效 token"(接口存活)
- `curl -k https://192.168.5.52/monitor/api2/api/servermonitor/` → 正常响应
## 回滚方案
每个组件升级前必须做:
1. 导出容器完整参数:`docker inspect <container> > /tmp/<container>_inspect.json`
2. 备份配置文件到 `/tmp/`
3. **保留旧镜像不删除**
**异常时回滚:**
- 用旧镜像 + 备份配置重新启动容器
- 恢复数据(如有 mysqldump 备份)
## 风险点与缓解措施
| 风险 | 影响 | 缓解措施 |
|------|------|----------|
| EMQX 5.8.7→6.0.2 主版本升级 | MQTT 协议/配置不兼容 | 先在 X86 测试,确认后再升级 ARM |
| MySQL 8.0.x→8.x 最新 | 数据兼容性、Nacos 等服务连接 | **先 mysqldump 全量备份**,保留旧容器 |
| JDK 容器替换 | 可能影响 Java 服务 | **先 commit 备份**,失败可回滚 |
| Nginx 1.29→1.30 | 配置格式变化 | 备份原配置,逐步验证 |
| Redis 8.2→8.8 | 数据兼容性 | dump.rdb 通常向后兼容 |
## 输出物
升级完成后必须生成:
- **升级日志**`AuxiliaryTool/ScriptTool/RemoteDeploy/reports/镜像组件升级日志_<YYYYMMDD>.log`
- **升级分析报告**`AuxiliaryTool/ScriptTool/RemoteDeploy/reports/镜像组件升级报告_<YYYYMMDD>.md`
报告内容应包含:
- 各组件升级前/后版本
- 验证结果(容器状态、服务日志、接口测试)
- 异常情况及处理
- 打包镜像路径
## 注意事项
1. **禁止随意改变文件**:如需调整文件,需先询问用户确认
2. **升级过程需日志记录分析**:所有操作命令、输出、异常均记录
3. **Docker 不升级**:仅记录最新版本,当前 29.1.3
4. **Nacos 禁跨 3.x**:只能升级到 2.5.X 版本
5. **JDK 禁跨非 1.8.X**:只能升级 1.8.X 版本
6. **MySQL 禁跨非 8.x**:只能升级 8.x 版本
7. **Redis 需 host 网络**:启动时加 `--network host`
8. **EMQX 主版本升级**:5.x→6.x 需确认配置兼容,先在 X86 测试
9. **镜像打包路径**:所有新镜像 tar.gz 放到 `/data/offline_auto_unifiedPlatform/data/temp/`
10. **旧镜像包保留**:不删除旧版本 tar.gz,便于回滚
## 关联文档
- `Docs/PRD/镜像组件升级/_PRD_ARM架构_远程升级镜像组件版本_需求文档.md` — ARM 版本升级
- `Docs/PRD/镜像组件升级/_PRD_ARM架构_远程升级镜像组件版本_计划执行.md` — ARM 执行计划
\ No newline at end of file
---
name: XTY-GN-TEST
description: 新统一平台业务功能验证 - 阶段8-16 Web UI 全链路验证(建管理员→改密→权限绑定→用户/部门→会议授权码→会议室→建会议→运维人员/类型标签),强制 Selenium 页面操作
---
新统一平台(ARM/X86 Ubuntu)业务功能验证专项。覆盖部署、授权完成后的**阶段 8–16 Web UI 验证链**,验证系统核心业务可用性。可作为 `/ARM-UBUNTU-XTYBS` 的验证子集独立执行。
## Usage
`/XTY-GN-TEST [阶段参数]`(不传则按顺序执行全链 8→16)
- `8` 建公司管理员 + 前台首登改密
- `9` 建权限组 + 绑定(退出重登)
- `10` 新增用户
- `11` 部门管理(页面可访问即可)
- `12` 会议授权码批量启用
- `13` 添加会议室
- `14` 新建会议
- `15` 运维人员核实(补充验证)
- `16` 类型标签新增(补充验证)
## 参考文档
- PRD 需求:`Docs/PRD/远程自动化部署/_PRD_ARM_Ubuntu远程自动化部署_需求文档.md`
- 部署指导:`Docs/PRD/远程自动化部署/ARM架构_新统一平台自动化部署操作指导.md`
- 部署主 skill:`/ARM-UBUNTU-XTYBS`(本 skill 为其业务验证子集)
- 脚本目录:`AuxiliaryTool/ScriptTool/RemoteDeploy`(多 skill 共用,勿删他人脚本)
## 关键配置
| 项 | 值 |
|----|----|
| 服务器 | 192.168.9.76(ARM aarch64 / openEuler 24.03) |
| 前台 / 维护平台 / 后台 | `https://192.168.9.76/` · `/#/LoginConfig` · `/#/LoginAdmin` |
| 超管(后台) | superadmin / Ubains@1357 / 验证码 csba |
| 新建公司管理员 | admin / 首密码 Ubains@1357 → 改 Admin@2024ub |
| 前台公司编号 | CN-SZ-00-0201 |
## 🚨 硬性禁令(核心红线,不可违反)
**阶段 8–14 必须且只能用 Web 页面 UI 操作(Selenium 驱动页面点击/输入/提交)。严禁改数据库、严禁直接调后端接口绕过页面、严禁改配置伪造数据。页面走不通就停下报告,绝不靠非页面手段凑结果。**
## 业务功能验证链(阶段 8–16)
| 阶段 | 验证内容 | 脚本 | 成功判定要点 |
|------|---------|------|------|
| **8** | superadmin 建 admin(公司=新统一平台,角色=公司管理员)→ 前台首登改密 `Admin@2024ub` | `web_operations_phase8_14.py --phase=8``phase8_changepwd.py` | 密码框一律 send_keys;旧密码登录失败=改密成功 |
| **9** | 建权限组「管理员权限」→ 绑定 admin(角色=公司管理员)→ **退出重登** | `phase9_robust.py` | 绑定后必须重登刷新菜单,否则全 not found |
| **10** | 用户列表 → 新增 `admin@test` | `phase10_13_robust.py` | 列表出现新账号行即成功 |
| **11** | 部门管理页面可访问即可(编辑非必需,可跳过) | — | 页面能打开即可 |
| **12** | 会议授权 → 全选 → 批量启用 | `phase10_13_robust.py` | 状态文案「已激活」(非"已启用"),或按钮变「停用」 |
| **13** | 区域列表 → 添加「测试会议室」+ 第一个授权码 | `phase10_13_robust.py` | 先填名称→选授权码→保存 |
| **14** | 前台登录 → 新建会议 → 勾「测试会议室」行 → 确定创建 | `web_operations_phase8_14.py --phase=14` | `closest("tr")` 定位 checkbox;URL 含 `%2FHome` |
| **15** | admin 后台 → 运维设置→人员设置 → 查询 `admin@test` 是否存在(补充验证) | `ops_query_person.py` | 查询框 placeholder="请输入名称";列表 2→1 条精确命中即存在 |
| **16** | 运维设置→类型标签 → 新增「测试类型新增」→ 核实列表(补充验证) | `ops_add_typetag.py` | 名称框 placeholder="请输入类型标签名称";列表出新行 |
## 关键脚本清单(`AuxiliaryTool/ScriptTool/RemoteDeploy/`)
- `web_operations_phase8_14.py` 阶段 8/14(建管理员、建会议),支持 `--phase=8` / `--phase=14`
- `phase8_changepwd.py` 前台首登 admin → 改密 Admin@2024ub
- `phase9_robust.py` 阶段 9:建权限组 + 绑定(角色=公司管理员),只展开未展开菜单,每步验证
- `phase10_13_robust.py` 阶段 10-13:admin 重登后做 用户/部门/授权码/会议室
- `verify_room_authcode.py` 核实会议室/授权码(状态"已激活")
- `verify_meeting_front.py` 稳健前台登录(协议处理)+ 核实会议创建
- `ops_query_person.py` 阶段 15:运维设置→人员设置查询(展开→等 2s→点子项,后台 keep-alive)
- `ops_add_typetag.py` 阶段 16:类型标签新增(自动跳过 readonly/number 框,send_keys 填名称)
- `ssh_helper.py` / `deploy_config.json` / `plink.exe` / `pscp.exe` SSH 工具与凭据
## ⚠️ 验证环节避坑点(逐条遵守,不再踩)
1. **每步操作后必须验证结果**(读列表/URL/页面文本确认),不能只看 `clicked` / 函数返回 True —— phase9-13 曾因只看 clicked 漏掉绑定失败
2. **绑定权限组后必须退出重登**,admin 菜单/权限才刷新;否则用户列表/会议授权/区域列表全 not found
3. **菜单展开只展开未展开项**`is-opened` 判断),勿 `forEach(t=>t.click())`(toggle 会折叠已展开的)
4. **密码框一律 send_keys**,JS `input.value=` 不触发 Vue v-model,导致创建静默失败(点确定对话框不关闭、无报错)
5. **前台登录**:点 `.el-checkbox`(不是隐藏 native input)勾协议 + 填完验证码等 1 秒再点登录 + dismiss 协议/提示弹窗;成功标志 URL 含 `%2FHome`
6. **登录函数封装多轮重试(5-6 次)**,重新 get 页面重填重点,单次失败不退出(验证码 csba / 框架初始化 / 会话偶发问题)
7. **角色下拉 placeholder 是「搜索角色」**(非"选择角色"),用 `indexOf("角色")` 宽匹配;Element UI placeholder 文案可能反直觉,失败时 dump 实际 placeholder 排查
8. **Element UI 子菜单展开有时序**:点 `.el-submenu__title` 后子 `.el-menu-item` 需等动画(约 2s)才可见,**展开 + 点击务必分两步带间隔**(适用于运维设置下人员设置/类型标签等所有二级菜单)
9. **弹窗填值先排除 readonly/number/checkbox 框**:Element UI 弹窗常含只读下拉(placeholder"请选择")与数字框,宽泛兜底匹配会先命中只读框致 `element click intercepted`。按 placeholder 含「名称/类型/标签」精确选,不确定先 dump 弹窗内 input 列表
10. **会议授权码状态文案是「已激活」**(非"已启用"),判断启用看"已激活"或操作按钮含"停用"(可停用=已启用)
11. **部门管理「添加」是图标按钮(无文字)**,文本匹配失败;部门编辑非必需可跳过,页面能访问即可
12. **公司管理员在 系统管理→管理员设置**(菜单名就是"管理员设置",点击后即公司管理员管理页,**没有**"公司管理员设置"二级子菜单),不在"公司管理"页面
13. **新增用户时有两个"请选择"下拉**:第一个是角色(不要动),第二个是公司
14. **新增用户 2 个密码框**通过 `type=password` 按顺序定位并用 send_keys(非 JS):第一个新密码、第二个确认密码(后台规则比前台宽松,可用 `Ubains@1357`
15. **添加会议室**:先填名称,再选授权码,最后点"保 存";区域列表中"预定授权"点第一个"请选择",授权码选第一个可用项(跳过"条/页"分页)
16. **"保存"按钮文字可能是"保 存"**(中间有空格),不能 `==="保存"`,要用 `indexOf("存")>=0` 匹配
17. **会议室 checkbox 定位**:用 `closest("tr")` 找到含"测试会议室"的行再勾选,不能简单勾第一个/最后一个
18. **运维人员查询框 placeholder = "请输入名称"**(非关键字/搜索/账号/手机);填用户名回车后列表从 2→1 条精确命中即存在,返回"暂无数据"即不存在
19. **类型标签在运维设置子菜单下**(非组织管理),模块表头"名称/排序",新增记录排序默认 0
## 🎯 执行要点
- 严格按 PRD/部署指导执行,不得自行发挥或加文档外操作;遇文档未描述情况→停下报告
- 全程 Selenium 驱动 Element UI 页面交互,关键操作用 JavaScript 辅助(直接 click 可能失效)
- 所有操作保存截图和日志;全程记录时间戳
- 每个阶段完成后输出该阶段的验证结论(通过/失败 + 证据),失败如实记录不隐瞒
# -*- coding: utf-8 -*-
"""中间件验证模块包:每个中间件一个检查器"""
# -*- coding: utf-8 -*-
"""验证数据结构与中间件检查基类
定义三层验证(L1 容器状态 / L2 端口监听 / L3 功能验证)的统一结果模型。
状态分三类,是“防误判”的核心:
- PASS 正常
- FAIL 服务功能性失败(确认后可进入修复)
- ERROR 基础设施/脚本错误(容器不存在、docker 缺失、ARM 架构不兼容等)→ 永不修复
"""
from dataclasses import dataclass, field
from enum import Enum
class CheckLevel(str, Enum):
L1_STATUS = "L1-容器状态"
L2_PORT = "L2-端口监听"
L3_FUNCTION = "L3-功能验证"
class CheckStatus(str, Enum):
PASS = "PASS"
FAIL = "FAIL" # 服务功能性失败,确认真失败后可修复
ERROR = "ERROR" # 基础设施/脚本错误,永不修复
SKIP = "SKIP" # 降级或前置条件不满足,跳过
@dataclass
class CheckItem:
"""单条检查项结果"""
name: str # 检查项名称
level: CheckLevel
status: CheckStatus = CheckStatus.PASS
detail: str = "" # 结论描述
command: str = "" # 执行的命令
output: str = "" # 命令输出(用于取证)
duration: float = 0.0
attempts: list = field(default_factory=list) # 防误判重试逐次摘要
repairable: bool = True # False=凭据/认证配置类失败,restart 无法修复,不自动修复
@property
def is_failure(self):
"""确认真失败(FAIL),需进入修复流程"""
return self.status == CheckStatus.FAIL
@property
def is_error(self):
"""基础设施错误,不修复"""
return self.status == CheckStatus.ERROR
@dataclass
class CheckOutcome:
"""单个中间件的验证结果汇总(一次 verify 快照)"""
middleware: str
items: list = field(default_factory=list) # List[CheckItem]
@property
def overall_status(self):
"""整体状态:有 ERROR→ERROR;有 FAIL→FAIL;否则 PASS"""
statuses = [i.status for i in self.items]
if CheckStatus.ERROR in statuses:
return CheckStatus.ERROR
if CheckStatus.FAIL in statuses:
return CheckStatus.FAIL
return CheckStatus.PASS
@property
def confirmed_failures(self):
"""确认真失败的检查项(待修复)"""
return [i for i in self.items if i.is_failure]
@property
def has_infra_error(self):
"""是否存在基础设施错误(整体不修复)"""
return any(i.is_error for i in self.items)
class MiddlewareChecker:
"""中间件检查器基类。子类实现 verify() 执行一次三层验证并返回 CheckOutcome。"""
name = "base"
def __init__(self, cfg, ssh):
self.cfg = cfg # 该中间件配置 dict(来自 config.yaml)
self.ssh = ssh # SSHExec 实例
def verify(self):
"""执行一次完整三层验证,返回 CheckOutcome(单次快照)"""
raise NotImplementedError
# ---- 便捷工厂 ----
def _outcome(self):
return CheckOutcome(middleware=self.name)
def _item(self, name, level):
return CheckItem(name=name, level=level)
# ---------------- L1 / L2 通用检查(所有中间件复用) ----------------
def _check_l1_status(self, outcome):
"""L1 容器存活:容器不存在/docker 异常 → ERROR;非 running → FAIL"""
name = self.cfg['name']
cmd = f"docker inspect -f '{{{{.State.Status}}}}' {name}"
res = self.ssh.run(cmd, timeout=15)
item = self._item("容器运行状态", CheckLevel.L1_STATUS)
item.command = cmd
err = (res.stderr + res.stdout).lower()
if res.timed_out:
item.status, item.detail = CheckStatus.ERROR, "docker inspect 超时"
item.output = res.stderr
elif "no such object" in err or "no such container" in err:
item.status, item.detail = CheckStatus.ERROR, f"容器 {name} 不存在"
item.output = res.stderr or res.stdout
elif res.exit_code != 0:
item.status, item.detail = CheckStatus.ERROR, f"docker 命令异常(exit={res.exit_code})"
item.output = res.stderr
else:
state = res.stdout.strip()
if state == "running":
item.status, item.detail = CheckStatus.PASS, "running"
else:
item.status = CheckStatus.FAIL
item.detail = f"容器状态={state}(非 running)"
outcome.items.append(item)
return item
def _check_l2_port(self, outcome, port, host="127.0.0.1"):
"""L2 宿主端口监听:ss 查不到且 nc 也失败 → FAIL"""
port = int(port)
cmd = (f"ss -tln 2>/dev/null | grep -qE ':{port}([[:space:]]|$)' "
f"|| nc -z -w3 {host} {port} 2>/dev/null")
res = self.ssh.run(cmd, timeout=15)
item = self._item(f"端口监听 {port}", CheckLevel.L2_PORT)
item.command = cmd
if res.timed_out:
item.status, item.detail = CheckStatus.ERROR, "端口探测超时"
elif res.exit_code == 0:
item.status, item.detail = CheckStatus.PASS, f"端口 {port} 监听中"
else:
item.status, item.detail = CheckStatus.FAIL, f"端口 {port} 未监听"
item.output = (res.stdout + res.stderr).strip()
outcome.items.append(item)
return item
# -*- coding: utf-8 -*-
"""utracker 验证:FastDFS(tracker+storage 单容器,host 网络,HTTP 8888)"""
from checks.base import MiddlewareChecker, CheckLevel, CheckStatus
L3 = CheckLevel.L3_FUNCTION
class FastDfsChecker(MiddlewareChecker):
name = "utracker"
def verify(self):
outcome = self._outcome()
self._check_l1_status(outcome)
# host 网络:tracker 端口应在宿主监听
self._check_l2_port(outcome, self.cfg['tracker_port'])
self._check_l3(outcome)
return outcome
def _check_l3(self, outcome):
name = self.cfg['name']
cconf = self.cfg.get('client_conf', '/etc/fdfs/client.conf')
web = self.cfg['web_port']
# 1) 上传测试文件
item = self._item("fdfs_upload_file 上传", L3)
cmd = f"fdfs_upload_file {cconf} /etc/hosts"
res = self.ssh.docker_exec(name, cmd, timeout=25)
item.command = cmd
raw = res.stdout.strip()
path = raw.splitlines()[-1] if raw else ""
if res.timed_out:
item.status, item.detail = CheckStatus.ERROR, "上传超时"
path = ""
elif res.ok and path.startswith("group"):
item.status, item.detail = CheckStatus.PASS, f"上传成功 → {path}"
else:
item.status, item.detail = CheckStatus.FAIL, "上传未返回 group 路径"
item.output = (res.stdout + res.stderr).strip()
path = ""
outcome.items.append(item)
# 2) HTTP 下载验证
item2 = self._item("HTTP 下载验证", L3)
if path:
cmd2 = f"curl -s -o /dev/null -w '%{{http_code}}' http://127.0.0.1:{web}/{path}"
res2 = self.ssh.run(cmd2, timeout=15)
item2.command = cmd2
code = res2.stdout.strip()
if res2.timed_out:
item2.status, item2.detail = CheckStatus.ERROR, "下载超时"
elif code == '200':
item2.status, item2.detail = CheckStatus.PASS, "HTTP 200"
else:
item2.status, item2.detail = CheckStatus.FAIL, f"HTTP {code}"
item2.output = (res2.stdout + res2.stderr).strip()
else:
item2.status, item2.detail = CheckStatus.SKIP, "上传失败,跳过下载验证"
outcome.items.append(item2)
# -*- coding: utf-8 -*-
"""umysql 验证:MySQL 8.0.46(对外端口 8306)"""
from checks.base import MiddlewareChecker, CheckLevel, CheckStatus
L3 = CheckLevel.L3_FUNCTION
class MysqlChecker(MiddlewareChecker):
name = "umysql"
def verify(self):
outcome = self._outcome()
self._check_l1_status(outcome)
self._check_l2_port(outcome, self.cfg['host_port'])
self._check_l3(outcome)
return outcome
def _check_l3(self, outcome):
name = self.cfg['name']
pwd = self.cfg['root_password']
# 1) mysqladmin ping
item = self._item("mysqladmin ping", L3)
cmd = f"mysqladmin -uroot -p'{pwd}' ping"
res = self.ssh.docker_exec(name, cmd, timeout=20)
item.command = cmd
if res.timed_out:
item.status, item.detail = CheckStatus.ERROR, "ping 超时"
elif res.ok and 'alive' in res.stdout.lower():
item.status, item.detail = CheckStatus.PASS, "mysqld is alive"
else:
item.status, item.detail = CheckStatus.FAIL, "ping 未返回 alive"
item.output = (res.stdout + res.stderr).strip()
outcome.items.append(item)
# 2) SELECT 1 + 关键库存在
item2 = self._item("SELECT + 关键库存在", L3)
cmd2 = f"mysql -uroot -p'{pwd}' -N -e \"SELECT 1; SHOW DATABASES;\""
res2 = self.ssh.docker_exec(name, cmd2, timeout=20)
item2.command = cmd2
req = self.cfg.get('required_databases', [])
if res2.timed_out:
item2.status, item2.detail = CheckStatus.ERROR, "SQL 执行超时"
elif res2.ok:
dbs = res2.stdout.split()
missing = [d for d in req if d not in dbs]
if missing:
item2.status = CheckStatus.FAIL
item2.detail = f"缺失关键库 {missing}"
item2.output = res2.stdout.strip()
else:
item2.status = CheckStatus.PASS
item2.detail = f"SELECT OK,关键库 {req} 存在"
else:
item2.status, item2.detail = CheckStatus.FAIL, "SQL 执行失败"
item2.output = (res2.stdout + res2.stderr).strip()
outcome.items.append(item2)
# -*- coding: utf-8 -*-
"""unacos 验证:Nacos v2.5.2(8848;从宿主 localhost 访问以绕开 nginx 外网限制)"""
from checks.base import MiddlewareChecker, CheckLevel, CheckStatus
L3 = CheckLevel.L3_FUNCTION
class NacosChecker(MiddlewareChecker):
name = "unacos"
def verify(self):
outcome = self._outcome()
self._check_l1_status(outcome)
self._check_l2_port(outcome, self.cfg['web_port'])
self._check_l3(outcome)
return outcome
def _check_l3(self, outcome):
web = self.cfg['web_port']
user = self.cfg['username']
pwd = self.cfg['password']
# 1) Web 探活
item = self._item("Web 探活 /nacos/", L3)
cmd = f"curl -s -o /dev/null -w '%{{http_code}}' http://127.0.0.1:{web}/nacos/"
res = self.ssh.run(cmd, timeout=15)
item.command = cmd
code = res.stdout.strip()
if res.timed_out:
item.status, item.detail = CheckStatus.ERROR, "curl 超时"
elif code in ('200', '302', '301'):
item.status, item.detail = CheckStatus.PASS, f"HTTP {code}"
else:
item.status, item.detail = CheckStatus.FAIL, f"HTTP {code}"
item.output = (res.stdout + res.stderr).strip()
outcome.items.append(item)
# 2) 账号登录拿 token(验证认证可用)
item2 = self._item("账号登录 /v1/auth/login", L3)
item2.repairable = False # 登录失败多为凭据/认证配置问题,restart 无法修复
cmd2 = (f"curl -s -X POST 'http://127.0.0.1:{web}/nacos/v1/auth/login' "
f"-d 'username={user}&password={pwd}'")
res2 = self.ssh.run(cmd2, timeout=15)
item2.command = cmd2
out = res2.stdout
if res2.timed_out:
item2.status, item2.detail = CheckStatus.ERROR, "登录超时"
elif 'accessToken' in out:
item2.status, item2.detail = CheckStatus.PASS, "返回 accessToken"
else:
item2.status, item2.detail = CheckStatus.FAIL, "登录未返回 accessToken"
item2.output = out.strip()
outcome.items.append(item2)
# -*- coding: utf-8 -*-
"""unginx 验证:Nginx 1.30.2(仅 443 HTTPS)"""
from checks.base import MiddlewareChecker, CheckLevel, CheckStatus
L3 = CheckLevel.L3_FUNCTION
class NginxChecker(MiddlewareChecker):
name = "unginx"
def verify(self):
outcome = self._outcome()
self._check_l1_status(outcome)
self._check_l2_port(outcome, self.cfg['https_port'])
self._check_l3(outcome)
return outcome
def _check_l3(self, outcome):
port = self.cfg['https_port']
item = self._item("HTTPS 探活", L3)
cmd = f"curl -ks -o /dev/null -w '%{{http_code}}' https://127.0.0.1:{port}/"
res = self.ssh.run(cmd, timeout=15)
item.command = cmd
code = res.stdout.strip()
if res.timed_out:
item.status, item.detail = CheckStatus.ERROR, "curl 超时"
elif code in ('200', '301', '302'):
item.status, item.detail = CheckStatus.PASS, f"HTTPS {code}"
else:
item.status, item.detail = CheckStatus.FAIL, f"HTTPS {code}"
item.output = (res.stdout + res.stderr).strip()
outcome.items.append(item)
# -*- coding: utf-8 -*-
"""uredis 验证:Redis 8(host 网络,端口 6379,AOF 开)"""
from checks.base import MiddlewareChecker, CheckLevel, CheckStatus
L3 = CheckLevel.L3_FUNCTION
TEST_KEY = "mw_verify:test"
class RedisChecker(MiddlewareChecker):
name = "uredis"
def verify(self):
outcome = self._outcome()
self._check_l1_status(outcome)
self._check_l2_port(outcome, self.cfg['host_port'])
self._check_l3(outcome)
return outcome
def _cli(self, *args):
pwd = self.cfg['password']
return f"redis-cli -a '{pwd}' --no-auth-warning " + " ".join(args)
def _check_l3(self, outcome):
name = self.cfg['name']
# 1) PING 认证
item = self._item("PING 认证", L3)
cmd = self._cli("PING")
res = self.ssh.docker_exec(name, cmd, timeout=15)
item.command = cmd
if res.timed_out:
item.status, item.detail = CheckStatus.ERROR, "PING 超时"
elif res.ok and 'PONG' in res.stdout:
item.status, item.detail = CheckStatus.PASS, "PONG"
else:
item.status, item.detail = CheckStatus.FAIL, "未返回 PONG"
item.output = (res.stdout + res.stderr).strip()
outcome.items.append(item)
# 2) SET/GET/DEL 读写删(3 次独立调用,避免 shell 嵌套转义)
item2 = self._item("SET/GET/DEL 读写删", L3)
set_cmd = self._cli(f"SET {TEST_KEY} ok123 EX 60")
get_cmd = self._cli(f"GET {TEST_KEY}")
del_cmd = self._cli(f"DEL {TEST_KEY}")
r_set = self.ssh.docker_exec(name, set_cmd, timeout=15)
r_get = self.ssh.docker_exec(name, get_cmd, timeout=15)
r_del = self.ssh.docker_exec(name, del_cmd, timeout=15)
item2.command = f"{set_cmd} && {get_cmd} && {del_cmd}"
got = r_get.stdout.strip()
if r_set.timed_out or r_get.timed_out or r_del.timed_out:
item2.status, item2.detail = CheckStatus.ERROR, "读写删超时"
elif r_set.ok and r_get.ok and got == 'ok123' and r_del.ok:
item2.status, item2.detail = CheckStatus.PASS, "SET/GET/DEL 成功"
else:
item2.status = CheckStatus.FAIL
item2.detail = f"读写删失败(GET={got})"
item2.output = (r_set.stderr + r_get.stderr + r_del.stderr).strip()
outcome.items.append(item2)
# ===================================================================
# 中间件升级后验证 + 修复工具 配置
# 用途:每次镜像组件升级后执行 python verify_middleware.py
# 修改服务器/凭据只改本文件,代码无需改动。
# ===================================================================
server:
host: 192.168.9.76
port: 22
username: admin # root 的 SSH 登录被禁(欧拉默认),用 admin + sudo
password: 'Ubains@123'
use_sudo: true # 命令自动经 sudo 执行(docker 需要 root 权限)
sudo_password: 'Ubains@123'
connect_timeout: 15
# ---------- 六个中间件容器(u 前缀) ----------
containers:
umysql: # MySQL 8.0.46,对外端口 8306(非 3306)
name: umysql
image_expected: mysql:8.0.46
host_port: 8306
container_port: 3306
root_password: 'dNrprU&2S'
required_databases: [nacos_mysql, devops]
uredis: # Redis 8,host 网络模式
name: uredis
image_expected: redis:8
host_port: 6379
network_mode: host
password: 'dNrprU&2S'
data_path: /data/middleware/redis/data # L3 清数据路径(防删根校验用)
unacos: # Nacos v2.5.2,认证开启
name: unacos
image_expected: nacos/nacos-server:v2.5.2
web_port: 8848
grpc_port: 9848
username: nacos
password: 'dNrprU&2S'
unginx: # Nginx 1.30.2,仅 443(无 80)
name: unginx
image_expected: nginx:1.30.2
https_port: 443
uemqx: # EMQX 5.8.7,禁匿名 sha256
name: uemqx
image_expected: emqx/emqx:5.8.7
mqtt_port: 1883
dashboard_port: 18083
dashboard_user: admin
dashboard_password: 'public' # 9.76 实测为 EMQX 默认 admin/public
mqtt_user: 'mqtt@cmdb'
mqtt_password: 'mqtt@webpassw0RD'
utracker: # FastDFS(tracker+storage 单容器),host 网络
name: utracker
image_expected: ufastdfs:v2
network_mode: host
tracker_port: 22122
storage_port: 23000
web_port: 8888
client_conf: /etc/fdfs/client.conf
# ---------- 验证参数 ----------
verify:
timeout: 30 # 单条命令超时(秒)
retry_count: 3 # 防误判:连续失败确认次数
retry_intervals: [10, 15, 20] # 重试间隔(秒)
mqtt_test_topic: 'middleware_verify/test'
state_file: /data/logs/.middleware_verify_state # 冷却期状态文件(远端)
# ---------- 修复策略 ----------
repair:
enabled: true # 总开关;冒烟测试时设 false
levels:
container: true # L1:docker start / restart
account_config: true # L2:EMQX 账号重注册 + conf load
data_and_cascade: true # L3:清 Redis 数据 + Nacos 连带重启 ujava2(高危)
cooldown_seconds: 1800 # 同容器修复冷却期(30 分钟,防反复 restart)
readiness_wait: # 修复后等待就绪秒数(叠加就绪探针轮询)
umysql: 30
unacos: 40
uemqx: 35
uredis: 10
default: 15
cascade_restart: [ujava2] # nacos 修复后连带重启的容器
# 注意:L3 高危动作执行前会在控制台与报告双重 ⚠️ 告警
# -*- coding: utf-8 -*-
"""防误判核心:五重门控中的"连续失败确认 + 失败分类"
修复绝不仅因单次失败触发。Confirmer.confirm() 多次调用检查器的 verify():
- 任一次 PASS → 判为瞬态抖动,不修复;
- 遇 ERROR(基础设施/脚本错误)→ 立即停止,永不修复;
- 连续 N 次全 FAIL → 才判"确认真失败",进入修复。
"""
import time
from dataclasses import dataclass, field
from checks.base import CheckStatus
@dataclass
class ConfirmDetail:
"""防误判确认结果"""
attempts: int # 实际执行 verify 的次数
confirmed_failure: bool # 是否确认真失败(可进入修复)
transient_recovered: bool # 是否重试中恢复(瞬态抖动)
infra_error: bool # 是否基础设施错误(不修复)
history: list = field(default_factory=list) # 每次 overall_status 文本
note: str = ""
class Confirmer:
"""连续失败确认器"""
def __init__(self, retry_count=3, retry_intervals=None, sleep_fn=None):
self.retry_count = max(1, retry_count)
self.retry_intervals = retry_intervals or [10, 15, 20]
self.sleep = sleep_fn or time.sleep
def _interval(self, fail_index):
"""第 fail_index 次(0-based)失败后的等待秒数;越界取最后一个"""
if not self.retry_intervals:
return 10
return self.retry_intervals[min(fail_index, len(self.retry_intervals) - 1)]
def confirm(self, checker):
"""对 checker 反复 verify,返回 (最终 CheckOutcome, ConfirmDetail)"""
history = []
outcome = None
for attempt in range(1, self.retry_count + 1):
outcome = checker.verify()
status = outcome.overall_status
history.append(status.value)
if status == CheckStatus.PASS:
return outcome, ConfirmDetail(
attempts=attempt, confirmed_failure=False,
transient_recovered=(attempt > 1), infra_error=False,
history=history,
note="验证通过" if attempt == 1
else f"第 {attempt} 次重试恢复 → 判为瞬态抖动,不修复",
)
if status == CheckStatus.ERROR:
# 基础设施/脚本错误:立即停止,绝不修复
return outcome, ConfirmDetail(
attempts=attempt, confirmed_failure=False,
transient_recovered=False, infra_error=True,
history=history,
note="基础设施/脚本错误(非服务故障),不修复,需人工排查",
)
# status == FAIL:等待后继续重试
if attempt < self.retry_count:
self.sleep(self._interval(attempt - 1))
# 连续 N 次 FAIL → 确认真失败
return outcome, ConfirmDetail(
attempts=self.retry_count, confirmed_failure=True,
transient_recovered=False, infra_error=False,
history=history,
note=f"连续 {self.retry_count} 次验证均失败 → 确认真失败",
)
# -*- coding: utf-8 -*-
"""分级修复 + 修复前取证 + 冷却期 + 修复后复验
仅在 Confirmer 确认"真失败"且非基础设施错误时执行。分级:
L1 容器级 :docker start(Exited) / docker restart(假死)
L2 账号配置 :uemqx 业务账号重注册(幂等)
L3 数据/连带:uredis 清数据目录 + restart;unacos 连带 restart ujava2(高危)
所有动作前采集证据(docker ps + logs --tail 50),动作后等待就绪并用
Confirmer 复验(复验也走连续失败确认)。同容器冷却期内不重复修。
"""
import json
import time
from checks.base import CheckStatus
from results import RepairRecord, RepairActionRecord
class Repairer:
def __init__(self, ssh, repair_cfg, confirmer, containers_cfg, state_file):
self.ssh = ssh
self.cfg = repair_cfg or {}
self.confirmer = confirmer
self.containers_cfg = containers_cfg
self.state_file = state_file
self.levels = self.cfg.get('levels', {})
# ---------------- 门控 ----------------
def should_repair(self, confirm_detail):
"""返回 (是否修复, 原因)"""
if not self.cfg.get('enabled'):
return False, "修复总开关关闭(只读模式)"
if confirm_detail.infra_error:
return False, "基础设施/脚本错误,不修复"
if not confirm_detail.confirmed_failure:
return False, "未确认真失败(通过或瞬态恢复)"
return True, ""
def in_cooldown(self, container):
"""冷却期检查:远端状态文件记录上次修复时间"""
state = self.ssh.read_state(self.state_file)
key = f"last_repair_{container}"
if key in state:
try:
if time.time() - float(state[key]) < self.cfg.get('cooldown_seconds', 1800):
return True
except ValueError:
pass
return False
def mark_repaired(self, container):
state = self.ssh.read_state(self.state_file)
state[f"last_repair_{container}"] = str(int(time.time()))
self.ssh.write_state(self.state_file, state)
# ---------------- 取证 ----------------
def collect_evidence(self, container):
ps = self.ssh.run(
"docker ps -a --filter name=^%s$ --format "
"'table {{.Names}}\\t{{.Status}}\\t{{.Image}}'" % container, timeout=15)
logs = self.ssh.run(f"docker logs --tail 50 {container} 2>&1", timeout=15)
return f"$ docker ps:\n{ps.stdout}\n\n$ docker logs --tail 50:\n{logs.stdout}"
# ---------------- 修复主流程 ----------------
def repair(self, record, checker):
name = record.name
rr = RepairRecord()
ok, reason = self.should_repair(record.confirm)
if not ok:
rr.skipped_reason = reason
return rr
# 仅认证/凭据类失败(repairable=False)→ restart 无效,跳过自动修复
repairable = [i for i in record.outcome.items
if i.is_failure and getattr(i, 'repairable', True)]
if not repairable:
rr.skipped_reason = "仅认证/凭据类检查失败(restart 无法修复),需人工核对凭据/配置"
return rr
if self.in_cooldown(name):
rr.skipped_reason = f"冷却期内({self.cfg.get('cooldown_seconds', 1800)}s 未过),跳过"
return rr
rr.performed = True
rr.evidence_before = self.collect_evidence(name)
# L1 容器级(所有中间件)
if self.levels.get('container', True):
self._action_container(rr, name)
# L2 账号配置(uemqx)
if name == 'uemqx' and self.levels.get('account_config', True):
self._action_emqx_account(rr)
# L3 数据/连带
if self.levels.get('data_and_cascade', True):
if name == 'uredis':
self._action_redis_data(rr)
elif name == 'unacos':
self._action_nacos_cascade(rr)
# 等待就绪
waits = self.cfg.get('readiness_wait', {})
self._wait(name, waits.get(name, waits.get('default', 15)))
# 复验(同样走连续失败确认)
re_outcome, re_confirm = self.confirmer.confirm(checker)
rr.reverify_outcome = re_outcome
rr.reverify_confirm = re_confirm
if re_outcome.overall_status == CheckStatus.PASS:
rr.final_status = "PASS(复验通过)"
elif re_outcome.has_infra_error:
rr.final_status = "ERROR(复验出现基础设施错误)"
else:
rr.final_status = "FAIL(复验仍未通过)"
self.mark_repaired(name)
return rr
def _wait(self, name, seconds):
time.sleep(int(seconds))
# ---------------- L1 ----------------
def _action_container(self, rr, name):
"""running 但服务坏 → restart;非 running → start"""
res = self.ssh.run(f"docker inspect -f '{{{{.State.Running}}}}' {name}", timeout=15)
if res.ok and res.stdout.strip() == 'true':
r = self.ssh.run(f"docker restart {name}", timeout=90)
rr.actions.append(RepairActionRecord('L1', f"docker restart {name}", r.ok, r.stdout + r.stderr))
else:
r = self.ssh.run(f"docker start {name}", timeout=90)
rr.actions.append(RepairActionRecord('L1', f"docker start {name}", r.ok, r.stdout + r.stderr))
# ---------------- L2 ----------------
def _action_emqx_account(self, rr):
cfg = self.containers_cfg['uemqx']
name = cfg['name']
dport = cfg['dashboard_port']
du, dpw = cfg['dashboard_user'], cfg['dashboard_password']
mu, mp = cfg['mqtt_user'], cfg['mqtt_password']
body = json.dumps({"username": du, "password": dpw})
login_cmd = (f"curl -s -X POST http://127.0.0.1:{dport}/api/v5/login "
f"-H 'Content-Type: application/json' -d '{body}'")
lr = self.ssh.docker_exec(name, login_cmd, timeout=15)
try:
token = json.loads(lr.stdout).get('token')
except Exception:
token = None
if not token:
rr.actions.append(RepairActionRecord('L2', "EMQX admin 登录取 token", False, lr.stdout))
return
reg_body = json.dumps({"user_id": mu, "password": mp, "is_superuser": False})
reg_cmd = (f"curl -s -X POST 'http://127.0.0.1:{dport}"
f"/api/v5/authentication/password_based:built_in_database/users' "
f"-H 'Content-Type: application/json' -H 'Authorization: Bearer {token}' "
f"-d '{reg_body}'")
rr2 = self.ssh.docker_exec(name, reg_cmd, timeout=15)
text = (rr2.stdout + rr2.stderr).lower()
ok = (mu in rr2.stdout) or ('already' in text) or ('bad' not in text and rr2.exit_code == 0)
rr.actions.append(RepairActionRecord('L2', f"重注册业务账号 {mu}", ok, rr2.stdout.strip()))
# ---------------- L3 ----------------
def _action_redis_data(self, rr):
"""清空 Redis 数据目录(防删根)→ 重建。⚠高危:丢缓存数据"""
cfg = self.containers_cfg['uredis']
name = cfg['name']
path = cfg.get('data_path', '')
# 防删根:路径必须足够深且非根
if not path or path == '/' or len(path.strip('/').split('/')) < 3:
rr.actions.append(RepairActionRecord('L3', "清 Redis 数据(路径安全检查未过,跳过)", False, path))
return
self.ssh.run(f"docker stop {name}", timeout=60)
qp = self.ssh._shell_quote(path)
cmd = f'RDP={qp}; rm -rf "${{RDP:?}}/"*' # ${RDP:?} 未设/空则中止,防删根
rc = self.ssh.run(cmd, timeout=30)
rr.actions.append(RepairActionRecord('L3', f"清空 Redis 数据 {path}", rc.exit_code == 0, rc.stdout + rc.stderr))
rs = self.ssh.run(f"docker start {name}", timeout=90)
rr.actions.append(RepairActionRecord('L3', f"docker start {name}(数据重建)", rs.ok, rs.stdout + rs.stderr))
def _action_nacos_cascade(self, rr):
"""Nacos 修复后连带重启 ujava2(token 变更致 JWT 失效)。⚠高危:业务短暂中断"""
for c in self.cfg.get('cascade_restart', []):
r = self.ssh.run(f"docker restart {c}", timeout=90)
rr.actions.append(RepairActionRecord('L3', f"连带 docker restart {c}", r.ok, r.stdout + r.stderr))
# -*- coding: utf-8 -*-
"""报告:控制台彩色实时输出 + Markdown 报告文件"""
import os
from datetime import datetime
from colorama import Fore, Style, init as colorama_init
from checks.base import CheckStatus, CheckLevel
colorama_init()
_STATUS_STYLE = {
CheckStatus.PASS: (Fore.GREEN + "✓ PASS ", "通过"),
CheckStatus.FAIL: (Fore.RED + "✗ FAIL ", "失败"),
CheckStatus.ERROR: (Fore.YELLOW + "⚠ ERROR", "错误"),
CheckStatus.SKIP: (Fore.CYAN + "• SKIP ", "跳过"),
}
def _stamp():
return datetime.now().strftime("%Y%m%d_%H%M%S")
# ---------------- 最终判定 helper ----------------
def record_final_pass(record):
"""该中间件最终是否通过:修复过看复验,否则看验证结果"""
if record.repair and record.repair.performed:
rv = record.repair.reverify_outcome
return rv is not None and rv.overall_status == CheckStatus.PASS
return record.outcome.overall_status == CheckStatus.PASS
def _final_items(record):
"""取最终生效的检查项:修复过用复验结果,否则用原结果"""
if record.repair and record.repair.performed and record.repair.reverify_outcome:
return record.repair.reverify_outcome.items
return record.outcome.items
def record_service_ok(record):
"""服务级是否可用:不存在 repairable=True 的 FAIL/ERROR"""
for it in _final_items(record):
if it.status in (CheckStatus.FAIL, CheckStatus.ERROR) and getattr(it, 'repairable', True):
return False
return True
def record_has_cred_issue(record):
"""是否存在凭据/配置类异常(repairable=False 的失败)"""
return any(it.status == CheckStatus.FAIL and not getattr(it, 'repairable', True)
for it in _final_items(record))
def record_state(record):
"""三态:pass 服务正常 / warn 服务可用但有凭据异常 / fail 服务异常"""
if not record_service_ok(record):
return 'fail'
if record_has_cred_issue(record):
return 'warn'
return 'pass'
def _level_status(record, level):
items = [i for i in record.outcome.items if i.level == level]
if not items:
return "—"
if any(i.status == CheckStatus.ERROR for i in items):
return "⚠️错误"
if any(i.status == CheckStatus.FAIL for i in items):
return "❌失败"
return "✅通过"
def _md_status(status):
return {CheckStatus.PASS: "✅", CheckStatus.FAIL: "❌",
CheckStatus.ERROR: "⚠️", CheckStatus.SKIP: "➖"}.get(status, str(status))
# ---------------- 控制台输出 ----------------
def print_header(title):
print(f"\n{Fore.CYAN}{Style.BRIGHT}=== {title} ==={Style.RESET_ALL}")
def print_item(item, indent=2):
sym, _ = _STATUS_STYLE.get(item.status, ("? ", ""))
pad = " " * indent
extra = ""
if item.status != CheckStatus.PASS and not getattr(item, 'repairable', True):
extra = f" {Fore.MAGENTA}[凭据/配置类·不自动修复]{Style.RESET_ALL}"
tail = extra if extra else Style.RESET_ALL
print(f"{pad}{sym} [{item.level.value}] {item.name} — {item.detail}{tail}")
def print_confirm(detail):
"""打印防误判确认结论(仅当非首次即通过时)"""
if detail.attempts <= 1 and not detail.confirmed_failure and not detail.infra_error:
return
color = Fore.YELLOW if (detail.confirmed_failure or detail.infra_error) else Fore.GREEN
print(f" {color}[防误判] {detail.note}(历史: {' → '.join(detail.history)}){Style.RESET_ALL}")
def print_repair(record):
"""打印修复记录"""
if record is None:
return
if not record.performed:
if record.skipped_reason:
print(f" {Fore.CYAN}[修复] 跳过:{record.skipped_reason}{Style.RESET_ALL}")
return
print(f" {Fore.YELLOW}{Style.BRIGHT}[修复] 已执行 {len(record.actions)} 步动作:{Style.RESET_ALL}")
for a in record.actions:
sym = "✓" if a.success else "✗"
color = Fore.GREEN if a.success else Fore.RED
hi = " ⚠高危" if a.level == "L3" else ""
print(f" {color}{sym} ({a.level}) {a.name}{hi}{Style.RESET_ALL}")
if record.reverify_outcome is not None:
fs = record.final_status or record.reverify_outcome.overall_status.value
color = Fore.GREEN if "PASS" in fs else Fore.RED
print(f" {color}[复验] 最终状态:{fs}{Style.RESET_ALL}")
def print_final_summary(records):
print_header("最终汇总")
total = len(records)
states = [record_state(r) for r in records]
n_pass = states.count('pass')
n_warn = states.count('warn')
n_fail = states.count('fail')
repaired = sum(1 for r in records if r.repair and r.repair.performed)
color = Fore.GREEN if n_fail == 0 else Fore.RED
print(f"{color}服务正常 {n_pass}/{total},凭据异常 {n_warn},服务异常 {n_fail},"
f"触发修复 {repaired} 个{Style.RESET_ALL}")
for r in records:
st = record_state(r)
if st == 'pass':
sym, c = "✅", Fore.GREEN
elif st == 'warn':
sym, c = "⚠️", Fore.YELLOW
else:
sym, c = "❌", Fore.RED
flags = []
if r.confirm.transient_recovered:
flags.append("瞬态恢复")
if r.confirm.confirmed_failure:
flags.append("确认失败")
if r.confirm.infra_error:
flags.append("基础设施错误")
if r.repair and r.repair.performed:
flags.append(f"已修复({len(r.repair.actions)}步)")
elif r.repair and r.repair.skipped_reason:
flags.append(f"修复跳过:{r.repair.skipped_reason}")
suffix = (" [" + ", ".join(flags) + "]") if flags else ""
print(f" {c}{sym} {r.name}{suffix}{Style.RESET_ALL}")
# ---------------- Markdown 报告 ----------------
def render_report(records, meta):
lines = []
lines.append("# 中间件验证报告")
lines.append("")
lines.append(f"- 生成时间:{meta.get('time', '')}")
lines.append(f"- 服务器:{meta.get('server', '')}")
lines.append(f"- 自动修复:{'开启' if meta.get('repair') else '关闭(只读模式)'}")
if meta.get('note'):
lines.append(f"- 备注:{meta['note']}")
lines.append("")
# 汇总表
lines.append("## 汇总")
lines.append("")
lines.append("| 中间件 | L1状态 | L2端口 | L3功能 | 防误判 | 修复 | 最终 |")
lines.append("|---|---|---|---|---|---|---|")
for r in records:
l1 = _level_status(r, CheckLevel.L1_STATUS)
l2 = _level_status(r, CheckLevel.L2_PORT)
l3 = _level_status(r, CheckLevel.L3_FUNCTION)
if r.confirm.infra_error:
conf = "基础设施错误"
elif r.confirm.confirmed_failure:
conf = "确认失败"
elif r.confirm.transient_recovered:
conf = "瞬态恢复"
else:
conf = "通过"
if r.repair and r.repair.performed:
rep = f"已修复({len(r.repair.actions)}步)"
elif r.repair and r.repair.skipped_reason:
rep = f"跳过:{r.repair.skipped_reason}"
else:
rep = "—"
final = {'pass': "✅通过", 'warn': "⚠️服务可用·有凭据异常",
'fail': "❌服务异常"}[record_state(r)]
lines.append(f"| {r.name} | {l1} | {l2} | {l3} | {conf} | {rep} | {final} |")
lines.append("")
# 详情
lines.append("## 详情")
lines.append("")
for r in records:
lines.append(f"### {r.name}")
lines.append("")
for it in r.outcome.items:
lines.append(f"- **[{it.level.value}] {it.name}**:{_md_status(it.status)} — {it.detail}")
if it.status != CheckStatus.PASS:
if it.command:
lines.append(f" - 命令:`{it.command}`")
if it.output:
lines.append(" - 输出:")
lines.append(" ```")
for ln in it.output.strip().splitlines()[-15:]:
lines.append(f" {ln}")
lines.append(" ```")
lines.append(f"- **防误判**:{r.confirm.note}(历史 {' → '.join(r.confirm.history)})")
if r.repair and r.repair.performed:
lines.append("- **修复**:")
if r.repair.evidence_before:
lines.append(" - 修复前取证:")
lines.append(" ```")
for ln in r.repair.evidence_before.strip().splitlines()[-20:]:
lines.append(f" {ln}")
lines.append(" ```")
for a in r.repair.actions:
hi = " ⚠️高危" if a.level == "L3" else ""
lines.append(f" - {'✅' if a.success else '❌'} ({a.level}) {a.name}{hi}")
if r.repair.reverify_outcome:
lines.append(f" - 复验最终:{r.repair.final_status}")
elif r.repair and r.repair.skipped_reason:
lines.append(f"- **修复**:跳过({r.repair.skipped_reason})")
lines.append("")
return "\n".join(lines)
def save_report(records, meta, reports_dir):
"""生成并保存报告,返回文件路径"""
os.makedirs(reports_dir, exist_ok=True)
content = render_report(records, meta)
path = os.path.join(reports_dir, f"verify_{_stamp()}.md")
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return path
# -*- coding: utf-8 -*-
"""验证/修复结果记录结构(集中定义,避免 reporter 与 repair 循环依赖)"""
from dataclasses import dataclass, field
from checks.base import CheckOutcome
from confirmer import ConfirmDetail
@dataclass
class RepairActionRecord:
"""单步修复动作"""
level: str # L1 / L2 / L3
name: str # 动作描述
success: bool
output: str = ""
@dataclass
class RepairRecord:
"""单个中间件的修复记录"""
performed: bool = False # 是否执行了修复
skipped_reason: str = "" # 未修复原因(冷却期/ERROR/开关关)
evidence_before: str = "" # 修复前取证(容器状态+日志尾)
actions: list = field(default_factory=list) # List[RepairActionRecord]
reverify_outcome: CheckOutcome = None # 修复后复验结果
reverify_confirm: ConfirmDetail = None # 复验确认详情
final_status: str = "" # 修复后最终状态文字
@dataclass
class MiddlewareRecord:
"""单个中间件的完整记录(验证 + 确认 + 修复)"""
name: str
outcome: CheckOutcome
confirm: ConfirmDetail
repair: RepairRecord = None
# -*- coding: utf-8 -*-
"""SSH 远程执行封装(基于 paramiko)
提供在目标服务器上执行宿主命令、docker exec 容器内命令的能力,
兼容 root 直登与 admin+sudo 两种账号场景。所有中间件验证/修复命令
都经此模块发出,统一超时与异常处理。
"""
import paramiko
class ExecResult:
"""单条命令执行结果"""
def __init__(self, exit_code, stdout, stderr, timed_out=False):
self.exit_code = exit_code # 退出码;异常时为 -1
self.stdout = stdout # 标准输出(utf-8 解码)
self.stderr = stderr # 标准错误
self.timed_out = timed_out # 是否超时
@property
def ok(self):
"""命令成功(退出码 0 且未超时)"""
return self.exit_code == 0 and not self.timed_out
def __repr__(self):
return f"<ExecResult exit={self.exit_code} timed_out={self.timed_out}>"
class SSHExec:
"""SSH 执行器:封装 paramiko,提供 run / docker_exec / 状态文件读写"""
def __init__(self, server_cfg):
self.host = server_cfg['host']
self.port = server_cfg.get('port', 22)
self.username = server_cfg['username']
self.password = server_cfg['password']
self.use_sudo = server_cfg.get('use_sudo', False)
self.sudo_password = server_cfg.get('sudo_password', '')
self.connect_timeout = server_cfg.get('connect_timeout', 15)
self._client = None
# ---------------- 连接管理 ----------------
def connect(self):
"""建立 SSH 连接,失败抛异常"""
self._client = paramiko.SSHClient()
self._client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self._client.connect(
hostname=self.host, port=self.port,
username=self.username, password=self.password,
timeout=self.connect_timeout,
allow_agent=False, look_for_keys=False,
)
return self
def close(self):
if self._client:
self._client.close()
self._client = None
def __enter__(self):
return self.connect()
def __exit__(self, *exc):
self.close()
# ---------------- 命令执行 ----------------
@staticmethod
def _shell_quote(s):
"""单引号安全转义,拼接进 shell 字符串"""
return "'" + str(s).replace("'", "'\"'\"'") + "'"
def _wrap_sudo(self, cmd):
"""admin 账号时把命令包成 sudo(密码经 stdin 传入,避免进入进程列表)"""
if not self.use_sudo:
return cmd
return (f"echo {self._shell_quote(self.sudo_password)} "
f"| sudo -S -p '' bash -c {self._shell_quote(cmd)}")
def run(self, cmd, timeout=30):
"""在宿主机执行命令,返回 ExecResult。超时/异常不抛出,返回带标记的结果。"""
full = self._wrap_sudo(cmd)
try:
stdin, stdout, stderr = self._client.exec_command(full, timeout=timeout)
exit_code = stdout.channel.recv_exit_status()
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
return ExecResult(exit_code, out, err, timed_out=False)
except Exception as e:
# paramiko 超时通常抛 socket.timeout / SSHException
msg = str(e).lower()
timed_out = ('timeout' in msg) or ('timed out' in msg)
return ExecResult(-1, '', f"{type(e).__name__}: {e}", timed_out=timed_out)
def docker_exec(self, container, cmd, timeout=30, user=None):
"""在指定容器内执行命令(docker exec)。user 可指定运行用户。"""
user_opt = f" -u {user}" if user else ""
return self.run(f"docker exec{user_opt} {container} {cmd}", timeout=timeout)
# ---------------- 冷却期状态文件 ----------------
def read_state(self, path):
"""读取远端状态文件(key=value 行格式),返回 dict"""
res = self.run(f"cat {path} 2>/dev/null", timeout=10)
state = {}
for line in res.stdout.splitlines():
if '=' in line:
k, v = line.split('=', 1)
state[k.strip()] = v.strip()
return state
def write_state(self, path, state):
"""写远端状态文件(key=value 行格式)"""
content = '\n'.join(f"{k}={v}" for k, v in state.items())
cmd = (f"mkdir -p $(dirname {path}) && "
f"printf '%s\\n' {self._shell_quote(content)} > {path}")
return self.run(cmd, timeout=10)
# -*- coding: utf-8 -*-
"""中间件升级后验证 + 修复工具(入口)
用法:
python verify_middleware.py # 按 config.yaml 执行(默认开启修复)
python verify_middleware.py --no-repair # 只读模式,仅验证不修复(冒烟测试用)
python verify_middleware.py --config xx.yaml
流程:加载配置 → SSH → 逐中间件(验证→Confirmer 确认→修复→复验) → 控制台汇总 + Markdown 报告
退出码:全过 0;有任何未通过(含修复后仍失败/基础设施错误)1
"""
import os
import sys
import argparse
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
# Windows 控制台默认 GBK,强制 UTF-8 以正确显示中文(报告文件本身始终 UTF-8)
if sys.platform == 'win32':
try:
sys.stdout.reconfigure(encoding='utf-8')
sys.stderr.reconfigure(encoding='utf-8')
except Exception:
pass
import yaml
from colorama import Fore, Style
from ssh_exec import SSHExec
from confirmer import Confirmer
from repair import Repairer
from results import MiddlewareRecord, RepairRecord
import reporter
from checks.mysql import MysqlChecker
from checks.redis import RedisChecker
from checks.nacos import NacosChecker
from checks.nginx import NginxChecker
from checks.emqx import EmqxChecker
from checks.fastdfs import FastDfsChecker
# 验证顺序 = 依赖顺序:mysql → redis → emqx → nacos → fastdfs → nginx
CHECKERS = [
('umysql', MysqlChecker),
('uredis', RedisChecker),
('uemqx', EmqxChecker),
('unacos', NacosChecker),
('utracker', FastDfsChecker),
('unginx', NginxChecker),
]
def load_config(path):
with open(path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
def main():
ap = argparse.ArgumentParser(description="中间件升级后验证 + 修复工具")
ap.add_argument('--config', default=os.path.join(os.path.dirname(os.path.abspath(__file__)), 'config.yaml'))
ap.add_argument('--no-repair', action='store_true', help='只读模式:仅验证,不修复')
args = ap.parse_args()
cfg = load_config(args.config)
if args.no_repair:
cfg.setdefault('repair', {})['enabled'] = False
server = cfg['server']
vcfg = cfg.get('verify', {})
rcfg = cfg.get('repair', {})
state_file = vcfg.get('state_file', '/data/logs/.middleware_verify_state')
confirmer = Confirmer(vcfg.get('retry_count', 3), vcfg.get('retry_intervals'))
repair_on = rcfg.get('enabled', False)
reporter.print_header(
f"中间件验证 服务器 {server['host']}:{server.get('port',22)} "
f"修复={'开启' if repair_on else '关闭(只读)'}")
records = []
try:
ssh = SSHExec(server).__enter__()
except Exception as e:
print(f"{Fore.RED}SSH 连接失败:{e}{Style.RESET_ALL}")
sys.exit(2)
try:
repairer = Repairer(ssh, rcfg, confirmer, cfg['containers'], state_file)
for key, Cls in CHECKERS:
ccfg = dict(cfg['containers'][key]) # 拷贝,避免污染原配置
if key == 'uemqx':
ccfg['mqtt_test_topic'] = vcfg.get('mqtt_test_topic', 'middleware_verify/test')
checker = Cls(ccfg, ssh)
reporter.print_header(f"验证 {key}")
outcome, confirm = confirmer.confirm(checker)
for it in outcome.items:
reporter.print_item(it)
reporter.print_confirm(confirm)
record = MiddlewareRecord(name=key, outcome=outcome, confirm=confirm)
# 修复决策
if repair_on and confirm.confirmed_failure and not confirm.infra_error:
print(f" {Fore.YELLOW}确认真失败,启动修复…{Style.RESET_ALL}")
rr = repairer.repair(record, checker)
# 修复后打印复验的 items
if rr.performed and rr.reverify_outcome is not None:
print(f" {Fore.CYAN}-- 修复后复验 --{Style.RESET_ALL}")
for it in rr.reverify_outcome.items:
reporter.print_item(it, indent=4)
record.repair = rr
reporter.print_repair(rr)
else:
rr = RepairRecord()
rr.skipped_reason = ("修复关闭" if not repair_on
else "基础设施错误" if confirm.infra_error
else "未确认真失败")
record.repair = rr
records.append(record)
finally:
ssh.__exit__(None, None, None)
# 汇总 + 报告
reporter.print_final_summary(records)
meta = {
'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'server': f"{server['host']}:{server.get('port', 22)}",
'repair': repair_on,
}
reports_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'reports')
path = reporter.save_report(records, meta, reports_dir)
print(f"\n{Fore.CYAN}报告已生成:{path}{Style.RESET_ALL}")
# 退出码:服务级全部可用即 0(凭据异常算可用但报告会标注);任一服务异常 1
all_service_ok = all(reporter.record_state(r) in ('pass', 'warn') for r in records)
sys.exit(0 if all_service_ok else 1)
if __name__ == '__main__':
main()
此差异已折叠。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Generate final deployment report"""
import json, os
from datetime import datetime
now = datetime.now()
REPORTS=os.path.join(os.path.dirname(os.path.abspath(__file__)),'reports')
os.makedirs(REPORTS,exist_ok=True)
report_name = f'ARM_192.168.9.76_部署分析报告_{now.strftime("%Y%m%d")}.md'
report_path = os.path.join(REPORTS, report_name)
report = f'''# ARM Ubuntu 新统一平台部署分析报告
**服务器**: 192.168.9.76 | ARM aarch64 | openEuler 24.03 (LTS-SP3)
**部署时间**: {now.strftime("%Y-%m-%d %H:%M:%S")}
**部署包**: arm_offline_auto_unifiedPlatform.tar.gz (6.5GB)
---
## 部署流程总结
| 阶段 | 内容 | 结果 |
|------|------|------|
| 1-2 | 前置检查与MD5校验 | ✅ 成功 |
| 3-4 | 解压赋权与执行部署 | ✅ 成功 (约40分钟) |
| 5 | 服务验证 | ✅ 对外/预定/运维 3/4正常 |
| 6 | 系统授权 | ✅ 成功 (项目信息→下载激活文件→上传license→重启) |
| 7 | 生成报告 | ✅ 成功 |
| 8 | 创建公司管理员 | ✅ admin创建成功,密码已修改 |
| 9 | 权限配置 | ✅ 管理员权限组绑定成功 |
| 10 | 新增用户 | ✅ admin@test 创建成功 |
| 11 | 部门管理 | ⚠️ 页面可访问 (编辑非必需) |
| 12 | 会议授权码启用 | ✅ 批量启用成功 |
| 13 | 添加会议室 | ✅ 测试会议室创建成功 |
| 14 | 新建会议 | ✅ 前台会议创建成功 |
| 15 | 运维人员核实 | ✅ admin@test 在运维设置中存在 |
| 16 | 类型标签新增 | ✅ 测试类型新增 成功 |
---
## 详细结果
### Docker容器 (11个全部运行)
- umysql, uredis, uemqx, unacos, utracker, unginx
- ujava2, upython, upython_voice
- paperless, cardtable
### API验证
| 接口 | 状态 | 说明 |
|------|------|------|
| 对外接口 | ✅ A0076 | 正常 |
| 预定系统 | ✅ A0078 | 正常 (授权后恢复) |
| 运维集控 | ✅ | 正常 |
| 讯飞转录 | ❌ | ARM未适配py7zr (预期失败,不影响核心) |
### 业务功能验证
- 公司管理员admin创建 ✅
- 密码修改 Admin@2024ub ✅
- 权限组绑定 ✅
- 用户admin@test创建 ✅
- 会议授权码启用 ✅
- 测试会议室创建 ✅
- 前台会议创建 ✅
- 运维人员核实 ✅
- 类型标签新增 ✅
---
## 特殊说明
1. **授权问题**: 需先填写项目信息→下载激活文件→上传license,否则meeting2.0报CPU/MAC不匹配
2. **ARM Java启动慢**: meeting2.0启动约需12分钟
3. **讯飞转录**: ARM架构未适配py7zr,接口失败属预期
---
*报告生成于 {now.strftime("%Y-%m-%d %H:%M:%S")}*
'''
with open(report_path, 'w', encoding='utf-8') as f:
f.write(report)
print(f'报告已更新: {report_path}')
\ No newline at end of file
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论