20 Commits

Author SHA1 Message Date
cnbugs 14cebeaee2 更新 README.md 2026-07-25 22:19:46 +08:00
cnbugs 523671154a 上传文件至「故障报告」 2026-07-25 17:26:41 +08:00
Your Name e26a3e559f feat: 支持 shell 变量定义(VAR=value command)、export、多行命令
完整 shell 能力支持:
- VAR=value command 语法(自动跳过赋值检测实际 binary)
- export VAR=value 支持(export 加入只读白名单)
- 多行命令(\n 分割后逐段检查)
- && || ; 等连接符的武器级命令检测
- _split_into_commands 重写:使用 while 索引遍历,正确处理 && / || / |
- _command_has_dangerous_cmds 跳过 VAR=value 和 export 前缀
- classify_agent_command 检测 VAR=value kubectl 语法路由到 kubectl 分类器
- execute_command 支持变量赋值后的 binary 定位
- 新增 shell 内置白名单:echo printf cd pwd export source . sleep timeout
  nohup unset shift return local readonly trap type command set shopt eval
  exec alias unalias declare typeset read mapfile
- 白名单只读命令新增:timeout、nohup、sleep 等
2026-07-25 16:25:41 +08:00
Your Name a964a23b8f feat: 支持多行命令
多行命令通过 create_subprocess_shell 执行,逐行检查白名单:
- 每行独立检查 binary 是否在白名单中
- 取所有行的最高权限模式(如有任何一行是 write,整体需审批)
- 去除 _parse_command 中的多行拒绝逻辑
- _has_shell_operators 检测到 \n 也返回 True
2026-07-25 15:27:43 +08:00
Your Name defb09ef97 chore: ctr 全部放行(None),支持 k8s.io 等任意子命令命名空间 2026-07-25 15:16:04 +08:00
Your Name 841eb26ea0 feat: 新增 sleep/timeout/nohup 命令支持
- sleep: 诊断脚本中常用等待
- timeout: 限制命令执行时间
- nohup: 后台运行保护
2026-07-25 14:55:24 +08:00
Your Name 91ab429e5b feat: cat 完全放行,不再限制路径
- 移除 _SAFE_CAT_FILES / _SAFE_CAT_DIRS / _is_safe_cat_path 及相关检查
- cat 在 SHELL_READ_ONLY_COMMANDS 为 None(任意参数均可)
- cat /etc/shadow 等所有路径现在都能正常通过
2026-07-25 14:54:19 +08:00
Your Name 00d95c4c5e feat: 全面放开 shell 命令连接符(; && || > >> < 等)
现在 AI Agent 支持在单条命令中使用:
- 管道 (|): crictl ps -a | grep nginx | head -5
- 重定向 (> / >>): journalctl -u kubelet > /tmp/log
- 分号 (;): systemctl status kubelet; df -h
- 逻辑连接符 (&& / ||): df -h && free -m

安全策略:
- FORBIDDEN_TOKENS 清空,所有 shell 操作符均放行
- 新增 _split_into_commands() 按操作符分割命令段
- 新增 _command_has_dangerous_cmds() 扫描所有段中的武器级命令
- 永远拒绝: sh/bash/dash/zsh/dd/format/.../parted 等
- 检查第一条命令的白名单,cat 不安全路径检查
- 只读白名单新增 echo printf cd pwd export test [
2026-07-25 14:48:32 +08:00
Your Name 3c3b7abda5 feat: rm/cp/mv/chmod/chown 等文件操作命令支持(需人工审批)
- SHELL_WRITE_COMMANDS 新增 rm cp mv chmod chown mkdir touch ln
  所有文件操作命令任意参数都需人工审批
- _PIPE_DANGEROUS_TARGETS 移除 rm mv chmod chown,它们现在在
  写白名单中审批执行,不再直接拒绝
- 管道分支增加写表子命令不匹配时的回退处理
2026-07-25 14:42:30 +08:00
Your Name 8852e6d0ca fix: 允许 AI Agent 使用管道(|)和重定向(>)进行精准排查
之前禁止所有 shell 操作符,导致 crictl ps -a|tail / df -h | grep vda
这类常用诊断管道无法使用。

改动:
1. FORBIDDEN_TOKENS 移除了 | > >> $,保留 ; || && < << `
2. 新增 _has_shell_operators() 检测命令是否需要 shell 执行
3. 新增 _pipe_to_dangerous_target() 阻止管道后执行 sh/bash/rm 等危险命令
4. classify_shell_command 对含管道命令: 只检查管道前 binary 白名单,
   子命令不做限制(管道后工具多样,精确分类无意义)
5. execute_command 对 needs_shell=True 的命令用 create_subprocess_shell 执行
6. AI Prompt 第1条安全规则改为允许管道和重定向
7. 分号仍然禁止;cat 不安全路径检查对管道场景同样生效
2026-07-25 14:40:12 +08:00
Your Name b19975c369 feat: 移除 etcdutl 引用,增加 ctr/nerdctl 示例,实现 Agent 断点续传
- AI Agent Prompt 移除不存在的 etcdutl 命令引用
- Prompt 增加 containerd(ctr/nerdctl) 和 Linux 基础调优命令示例
- processing_runs 表新增 context_json 字段,保存 Agent 执行上下文
- 每次命令执行后自动保存上下文到 DB
2026-07-25 14:34:05 +08:00
Your Name a5399bd69d chore: 移除不存在的 etcdutl 命令及其相关逻辑和测试 2026-07-25 14:23:04 +08:00
Your Name 4b54e49a0d feat: 扩充 shell 命令白名单(含 containerd/ctr/nerdctl),解除 AI Agent 轮数上限
- 新增 ~50 个 Linux 基础命令(du/top/iostat/sar/curl/tcpdump 等)
- 新增 containerd 系列: ctr (containers/images/tasks/snapshots),
  nerdctl (ps/logs/inspect/stats 等)
- ip 命令补充 maddr/monitor/tunnel/tuntap 只读操作
- _run_agent_loop 改为 while True 循环,max_steps=0 表示不限轮数
- main.py 传入 max_steps=0,AI Agent 持续诊断直到问题解决
- 更新 Agent system prompt 第7条:明确无轮数上限
2026-07-25 14:17:56 +08:00
cnbugs 1f37cd3268 fix: etcdutl snapshot 未知子命令应拒绝而非默认当写
修复验证中发现的安全漏洞:
- etcdutl snapshot <bogus> 原会默认返回 write (需审批),
  现改为显式拒绝 (ValueError)
- 新增 _ETCDUTL_SNAPSHOT_WRITE = {restore, make} 显式枚举写子命令
- 同时补充 etcdctl defrag 为写操作 (需审批)

验证: 65 项 ad-hoc 检查全部通过, 56 个单元测试通过
2026-07-25 13:52:19 +08:00
cnbugs 4f37c65f18 feat: Agent 支持 etcdctl/etcdutl 诊断命令
etcd 是 K8S 控制平面的关键组件,缺少 etcd 诊断会导致很多故障无法
继续排查。本次将 etcdctl 和 etcdutl 纳入命令白名单:

只读 (自动执行):
- etcdctl endpoint status/health/hashkv - 端点健康
- etcdctl member list - 成员列表
- etcdctl alarm list - 告警
- etcdctl auth/user/role status|list - 认证与权限查看
- etcdctl check perf/datascale - 性能检查
- etcdctl get/watch - 读取 key
- etcdctl version
- etcdutl snapshot status/hash - 快照检查
- etcdutl version

写操作 (人工审批):
- etcdctl put/del/compact/txn - 数据修改
- etcdctl snapshot save - 保存快照
- etcdctl member add/remove/update - 成员管理
- etcdctl alarm disarm - 解除告警
- etcdctl auth enable/disable - 认证开关
- etcdctl user/role add/delete/grant/revoke - 权限管理
- etcdutl snapshot restore - 恢复快照

实现细节:
- etcdctl/etcdutl 命令支持全局 flag (--endpoints/--cacert 等在子命令前)
- 两段式命令按 (verb, subverb) 组合精确分类,避免 member list (只读)
  和 member remove (写) 混淆
- 新增 12 个 etcd 命令分类测试,全部 56 个测试通过
2026-07-25 13:50:34 +08:00
cnbugs 32bea0d4ea feat: AI 诊断 Agent 支持执行系统诊断 shell 命令
扩展 Agent 命令工具,除 kubectl 外新增节点宿主机系统诊断命令支持:

- agent_tools.py: 新增 shell 命令白名单分类器 (classify_shell_command)
  - 只读白名单: systemctl status/is-active、journalctl、crictl ps/logs、
    docker ps/logs、df、free、ss、ip addr、ps、mount、lsmod、dmesg、
    ping、nslookup、cat (限 /proc /sys /etc/kubernetes 等安全路径) 等 30+ 命令
  - 写白名单: systemctl restart/stop/start/reload 等需人工审批
  - 统一分发器 classify_agent_command 按命令前缀路由
  - 统一执行入口 execute_command 走 subprocess_exec (无 shell 注入风险)
- ai_agent.py: 更新 AGENT_SYSTEM_PROMPT 告知 AI 两类可用工具及安全规则
  - 三个调用点 (初始探测/分类/执行) 切换到统一接口
- main.py: 审批执行处改用 execute_command
- tests: 新增 19 个用例覆盖 shell 命令分类、混合执行、白名单拒绝
  - 全部 44 个测试通过
- 前端/README: 文案与文档补充 shell 命令支持说明

安全策略不变: 禁止 shell 操作符/管道/重定向/多命令拼接,
非白名单命令 (rm/vi/apt 等) 直接拒绝。
2026-07-25 13:38:48 +08:00
cnbugs f2254646c1 fix: 实现 Agent 修复验证闭环 2026-07-25 13:10:21 +08:00
cnbugs 47abfef238 fix: 允许 Agent 持续自主诊断 2026-07-25 12:50:38 +08:00
cnbugs fdac1eaffc feat: 持久化诊断历史并修复 Agent 命令执行 2026-07-25 12:40:53 +08:00
cnbugs 4f3a7833a0 feat: 增加 AI Agent 自主诊断执行模式 2026-07-25 12:21:54 +08:00
14 changed files with 3308 additions and 27 deletions
+6
View File
@@ -15,6 +15,12 @@ venv/
# Environment # Environment
.env .env
# Runtime data
data/
*.db
*.db-shm
*.db-wal
# IDE # IDE
.vscode/ .vscode/
.idea/ .idea/
+24 -1
View File
@@ -2,7 +2,7 @@
一键诊断 Kubernetes 集群健康状态,接入 AI Agent 自动分析故障根因并给出修复建议。 一键诊断 Kubernetes 集群健康状态,接入 AI Agent 自动分析故障根因并给出修复建议。
![界面预览](/home/cnbugs/.hermes/cache/screenshots/browser_screenshot_b258ca152c1e489fa186583946ecbd9e.png) ![image-20260725221146269.png](https://wx-img.cnbugs.com/image-20260725221146269.png)
## 功能特性 ## 功能特性
@@ -13,6 +13,21 @@
- **健康评分**:100 分制量化集群健康度(严重问题 -15 分,警告 -5 分) - **健康评分**:100 分制量化集群健康度(严重问题 -15 分,警告 -5 分)
- **AI 智能分析**:接入 OpenAI 兼容接口,自动分析诊断报告 - **AI 智能分析**:接入 OpenAI 兼容接口,自动分析诊断报告
- 根因定位、影响评估、修复命令、优先级排序、预防建议 - 根因定位、影响评估、修复命令、优先级排序、预防建议
- **AI 自主诊断(双执行模式)**:启动前可选择命令执行方式
- **手动执行命令**:Agent 只生成命令,用户复制到自己的终端执行,服务端不会运行
- **AI 执行命令**:自动执行只读诊断命令,覆盖 kubectl 与节点宿主机两个维度
- kubectl: `get``describe``logs``top` 等只读命令
- 系统诊断: `systemctl status``journalctl``crictl ps``df``free``ss``ip addr` 等只读命令
- 实时展示思考步骤、执行命令和真实输出
- 即使选择 AI 执行,修改命令人工确认:
- kubectl: `apply``delete``patch``scale``rollout``exec`
- 系统: `systemctl restart/stop/start`
- 命令白名单严格限制,禁止 shell、管道、重定向和多命令拼接
- **历史记录持久化**:使用 SQLite 保存全部诊断与处理轨迹
- 保存完整诊断结果、文本报告、AI 分析结果
- 保存 Agent 思考、命令、真实输出、审批和执行结果
- 支持按状态、命名空间查询并查看完整详情
- 服务重启后仍可读取历史和恢复最近一次诊断
- **AI 追问对话**:基于诊断上下文多轮追问,深入排查问题 - **AI 追问对话**:基于诊断上下文多轮追问,深入排查问题
- **Web 仪表盘**Vue3 + Element Plus,实时展示诊断结果 - **Web 仪表盘**Vue3 + Element Plus,实时展示诊断结果
@@ -44,6 +59,9 @@ PORT=8900
# Kubeconfig 路径 (留空使用默认 ~/.kube/config) # Kubeconfig 路径 (留空使用默认 ~/.kube/config)
KUBECONFIG_PATH= KUBECONFIG_PATH=
# 历史记录 SQLite 文件(K8S 部署时请挂载持久卷到该路径)
HISTORY_DB_PATH=/app/data/history.db
``` ```
## API 接口 ## API 接口
@@ -55,6 +73,10 @@ KUBECONFIG_PATH=
| POST | `/api/diagnose` | 执行诊断 `{"namespace":"", "checks":[]}` | | POST | `/api/diagnose` | 执行诊断 `{"namespace":"", "checks":[]}` |
| GET | `/api/diagnosis/latest` | 获取最近诊断结果 | | GET | `/api/diagnosis/latest` | 获取最近诊断结果 |
| POST | `/api/analyze` | AI 分析诊断结果 `{"question":""}` | | POST | `/api/analyze` | AI 分析诊断结果 `{"question":""}` |
| GET | `/api/agent/diagnose/stream` | 自主诊断 SSE 流,`execution_mode=manual|ai` |
| POST | `/api/agent/approve` | 批准或拒绝待执行命令 `{"approval_id":"...","approved":true}` |
| GET | `/api/history/diagnoses` | 查询全部诊断历史,支持 `status``namespace` 过滤 |
| GET | `/api/history/diagnoses/{id}` | 查询诊断详情及全部 AI/Agent 处理轨迹 |
| POST | `/api/chat` | AI 多轮对话 `{"messages":[...]}` | | POST | `/api/chat` | AI 多轮对话 `{"messages":[...]}` |
| GET | `/api/report` | 获取诊断报告文本 | | GET | `/api/report` | 获取诊断报告文本 |
@@ -72,6 +94,7 @@ k8smanager-cli/
│ ├── main.py # FastAPI 主应用 │ ├── main.py # FastAPI 主应用
│ ├── config.py # 配置管理 │ ├── config.py # 配置管理
│ ├── diagnosis.py # K8S 诊断引擎 (9 大检查项) │ ├── diagnosis.py # K8S 诊断引擎 (9 大检查项)
│ ├── agent_tools.py # Agent 命令工具与安全策略 (kubectl + 系统诊断白名单)
│ └── ai_agent.py # AI Agent 分析模块 │ └── ai_agent.py # AI Agent 分析模块
├── frontend/ ├── frontend/
│ ├── src/App.vue # 仪表盘界面 │ ├── src/App.vue # 仪表盘界面
+576
View File
@@ -0,0 +1,576 @@
"""AI Agent 可用的命令工具和安全策略。
支持两类命令:
1. kubectl 命令 - 通过 classify_kubectl_command 分类
2. 系统诊断 shell 命令 - 通过 classify_shell_command 分类 (白名单)
classify_agent_command 统一分发,execute_command 自动选择执行方式。
所有命令均通过 subprocess_exec 执行 (无 shell),禁止管道、重定向和命令组合。
"""
import asyncio
import shlex
from dataclasses import dataclass
from backend.diagnosis import KUBECTL
# ── kubectl 命令分类 ──────────────────────────────────────
READ_ONLY_VERBS = {
"api-resources", "api-versions", "auth", "cluster-info", "config", "describe",
"diff", "explain", "get", "logs", "top", "version", "wait",
}
WRITE_OR_INTERACTIVE_VERBS = {
"annotate", "apply", "attach", "autoscale", "cordon", "cp", "create",
"debug", "delete", "drain", "edit", "exec", "expose", "label", "patch",
"port-forward", "replace", "rollout", "run", "scale", "set", "taint",
"uncordon",
}
GLOBAL_FLAGS_WITH_VALUE = {
"--as", "--as-group", "--as-uid", "--cache-dir", "--certificate-authority",
"--client-certificate", "--client-key", "--cluster", "--context", "--kubeconfig",
"--namespace", "-n", "--profile", "--request-timeout", "--server", "-s",
"--tls-server-name", "--token", "--user",
}
FORBIDDEN_TOKENS = set() # 全部 shell 操作符均允许;安全靠白名单+危险命令检查
# 永远拒绝的武器级危险命令(无论是否在白名单中)
_ALWAYS_DENY_CMDS = {"sh", "bash", "dash", "zsh",
"dd", "format", "mkfs", "fdisk", "parted"}
@dataclass(frozen=True)
class CommandDecision:
mode: str # "read" (只读,自动执行) 或 "write" (修改,需审批)
requires_approval: bool
verb: str # kubectl 子命令 或 shell 命令名
args: list[str]
needs_shell: bool = False # True 时需用 shell=True 执行
def _has_shell_operators(command: str) -> bool:
"""检测命令是否包含 shell 操作符(管道、重定向、分号、逻辑连接符、多行等)。"""
stripped = command.strip()
if not stripped:
return False
if "\n" in stripped or "\r" in stripped:
return True # 多行命令需要 shell=True
in_single = in_double = False
for i, ch in enumerate(stripped):
if ch == "'" and not in_double:
in_single = not in_single
elif ch == '"' and not in_single:
in_double = not in_double
if in_single or in_double:
continue
if ch in ('|', '>', ';', '&', '<'):
# 单纯 & 不算(后台任务标记)
if ch == '&':
if i + 1 < len(stripped) and stripped[i + 1] in ('&',):
return True # &&
continue
return True
return False
def _split_into_commands(command: str) -> list[str]:
"""按 shell 连接符(; && || | \n 等)分割命令块。
返回每个可执行段,忽略引号内的操作符。重定向符号 > < 不分割。
"""
segments = []
buf = []
in_single = in_double = False
i = 0
while i < len(command):
ch = command[i]
if ch == "'" and not in_double:
in_single = not in_single
elif ch == '"' and not in_single:
in_double = not in_double
if not in_single and not in_double:
if ch in (';', '\n'):
if buf:
segments.append(''.join(buf).strip())
buf = []
i += 1
continue
if ch == '|':
if i + 1 < len(command) and command[i + 1] == '|':
# || — 分割命令
if buf:
segments.append(''.join(buf).strip())
buf = []
i += 2
continue
# 单 | — 分割命令
if buf:
segments.append(''.join(buf).strip())
buf = []
i += 1
continue
if ch == '&':
if i + 1 < len(command) and command[i + 1] == '&':
# && — 分割命令
if buf:
segments.append(''.join(buf).strip())
buf = []
i += 2
continue
# 单独 & — 保留
buf.append(ch)
i += 1
if buf:
segments.append(''.join(buf).strip())
return [s for s in segments if s]
def _command_has_dangerous_cmds(command: str) -> bool:
"""检查命令中任何可执行段是否包含武器级危险命令。"""
for segment in _split_into_commands(command):
try:
seg_args = shlex.split(segment)
except ValueError:
continue
if not seg_args:
continue
# 跳过 VAR=value 赋值
idx = 0
while idx < len(seg_args) and "=" in seg_args[idx] and not seg_args[idx].startswith("-"):
idx += 1
if idx < len(seg_args) and seg_args[idx] in _ALWAYS_DENY_CMDS:
return True
# 也检查 export VAR=value 后的命令
if idx < len(seg_args) and seg_args[idx] == "export":
idx2 = idx + 1
while idx2 < len(seg_args) and "=" in seg_args[idx2]:
idx2 += 1
if idx2 < len(seg_args) and seg_args[idx2] in _ALWAYS_DENY_CMDS:
return True
return False
def _parse_command(command: str) -> list[str]:
"""解析命令,检查合法性。
允许所有 shell 操作符(| > ; && || < 等)以及多行命令,安全靠白名单检查。
武器级危险命令 sh/bash/dd/mkfs 等在 classify 阶段处理。
"""
if not isinstance(command, str) or not command.strip():
raise ValueError("命令不能为空")
if _command_has_dangerous_cmds(command):
raise ValueError("命令包含不允许的危险命令(sh/bash/dd/mkfs 等),请手动执行")
try:
return shlex.split(command)
except ValueError as exc:
raise ValueError(f"命令格式错误: {exc}") from exc
def parse_kubectl_command(command: str) -> list[str]:
"""仅解析单条 kubectl 命令,不允许 shell、重定向或命令组合。"""
parts = _parse_command(command)
if not parts or parts[0] != "kubectl":
raise ValueError("只允许执行 kubectl 命令")
if len(parts) < 2:
raise ValueError("kubectl 命令缺少操作类型")
return parts[1:]
def _find_verb(args: list[str]) -> str:
index = 0
while index < len(args):
arg = args[index]
if arg == "--":
break
if arg in GLOBAL_FLAGS_WITH_VALUE:
index += 2
continue
if any(arg.startswith(flag + "=") for flag in GLOBAL_FLAGS_WITH_VALUE if flag.startswith("--")):
index += 1
continue
if arg.startswith("-"):
index += 1
continue
return arg.lower()
raise ValueError("无法识别 kubectl 操作类型")
READ_ONLY_CONFIG_SUBCOMMANDS = {"current-context", "get-clusters", "get-contexts", "get-users", "view"}
def classify_kubectl_command(command: str) -> CommandDecision:
args = parse_kubectl_command(command)
verb = _find_verb(args)
if verb == "config":
try:
verb_index = args.index(next(arg for arg in args if arg.lower() == "config"))
subcommand = args[verb_index + 1].lower()
except (StopIteration, ValueError, IndexError):
return CommandDecision("write", True, verb, args)
if subcommand in READ_ONLY_CONFIG_SUBCOMMANDS:
return CommandDecision("read", False, verb, args)
return CommandDecision("write", True, verb, args)
if verb in READ_ONLY_VERBS:
return CommandDecision("read", False, verb, args)
if verb in WRITE_OR_INTERACTIVE_VERBS:
return CommandDecision("write", True, verb, args)
return CommandDecision("write", True, verb, args)
# ── 系统诊断 shell 命令分类 (白名单) ─────────────────────
# 只读诊断命令白名单:value 为 None 表示该命令本身只读,任意参数均可;
# 为集合时,第一个非 flag 参数必须属于该集合。
SHELL_READ_ONLY_COMMANDS: dict[str, set[str] | None] = {
"systemctl": {
"status", "is-active", "is-enabled", "is-failed", "list-units",
"list-sockets", "list-jobs", "show", "cat", "list-unit-files",
"list-dependencies",
},
"journalctl": None, # 只读,任意参数 (-u, -n, --no-pager, --since 等)
"crictl": {"ps", "inspect", "logs", "info", "version", "images", "stats"},
"ctr": None,
"nerdctl": {"ps", "logs", "inspect", "info", "version", "images", "stats", "top", "events", "system", "namespace", "network", "volume"},
"docker": {"ps", "logs", "inspect", "stats", "version", "info", "images", "top"},
"df": None, "du": None, "free": None, "uptime": None, "uname": None, "hostname": None,
"ip": {"addr", "address", "route", "link", "neighbor", "neigh", "rule", "maddr", "monitor", "tunnel", "tuntap"},
"ss": None, "netstat": None, "lsblk": None, "ls": None,
"cat": None, "head": None, "tail": None, "wc": None, "grep": None, "egrep": None, "fgrep": None,
"ps": None, "pstree": None, "mount": None, "lsmod": None, "lscpu": None, "lsof": None,
"dmesg": None, "ping": None, "nslookup": None, "dig": None, "getent": None,
"timedatectl": None, "date": None, "stat": None, "file": None, "blkid": None,
"top": None, "htop": None, "iostat": None, "vmstat": None, "mpstat": None,
"pidstat": None, "sar": None, "nproc": None, "env": None, "which": None,
"readlink": None, "realpath": None, "find": None, "sort": None, "uniq": None,
"awk": None, "sed": None, "cut": None, "tr": None, "tee": None,
"basename": None, "dirname": None, "xargs": None, "bc": None, "expr": None,
"curl": None, "wget": None, "traceroute": None, "tracepath": None, "mtr": None,
"nc": None, "nmap": None, "telnet": None, "tcpdump": None, "ethtool": None,
"lspci": None, "lsusb": None, "dmidecode": None, "sysctl": None, "udevadm": None,
"echo": None, "printf": None, "cd": None, "pwd": None, "export": None,
"test": None, "[": None, "sleep": None, "source": None, ".": None,
"timeout": None, "nohup": None, "unset": None, "shift": None, "return": None,
"local": None, "readonly": None, "trap": None, "type": None, "command": None,
"set": None, "shopt": None, "eval": None, "exec": None, "alias": None,
"unalias": None, "declare": None, "typeset": None, "read": None, "mapfile": None,
}
# etcdctl 由 classify_shell_command 中的专用块处理 (两段式命令)
# 从通用白名单中移除,避免被单段式逻辑误判
# etcdctl 单段式只读 verb
_ETCDCTL_READ_VERBS = {"get", "watch", "version", "lock", "lease", "move-leader", "leader"}
# etcdctl 单段式写 verb (需人工审批)
_ETCDCTL_WRITE_VERBS = {"put", "del", "delete", "compact", "txn", "snapshot", "defrag"}
# etcdctl 两段式 (verb, subverb) 只读组合
_ETCDCTL_READ_SUBVERBS = {
"endpoint": {"status", "health", "hashkv"},
"member": {"list", "list-urls"},
"alarm": {"list"},
"auth": {"status"},
"user": {"list", "get"},
"role": {"list", "get"},
"check": {"perf", "datascale"},
}
# etcdctl 两段式 (verb, subverb) 写组合 (需人工审批)
_ETCDCTL_WRITE_SUBVERBS = {
"member": {"add", "remove", "update"},
"alarm": {"disarm"},
"auth": {"enable", "disable"},
"user": {"add", "delete", "grant-role", "revoke-role", "passwd"},
"role": {"add", "delete", "grant-permission", "revoke-permission"},
}
# 系统修改命令白名单:需要人工审批
SHELL_WRITE_COMMANDS: dict[str, set[str] | None] = {
"systemctl": {"restart", "start", "stop", "reload", "enable", "disable", "daemon-reload"},
# 文件操作命令:任意参数都需审批
"rm": None,
"cp": None,
"mv": None,
"chmod": None,
"chown": None,
"mkdir": None,
"touch": None,
"ln": None,
}
# ip 命令的写操作动词 (出现即拒绝自动执行)
_IP_WRITE_ACTIONS = {"set", "add", "del", "delete", "change", "replace", "append", "prepend"}
def _first_non_flag(args: list[str]) -> str | None:
for arg in args:
if not arg.startswith("-"):
return arg.lower()
return None
def parse_shell_command(command: str) -> list[str]:
"""解析单条 shell 诊断命令,禁止 shell 操作符和命令组合。"""
parts = _parse_command(command)
if not parts:
raise ValueError("命令不能为空")
return parts
def classify_shell_command(command: str) -> CommandDecision:
"""对白名单内的系统诊断命令进行分类。
systemctl 同时出现在只读和写白名单中:restart/stop 等走写表(需审批),
status/is-active 等走只读表(自动执行)。因此先检查写表,再检查只读表。
含管道 (|) 或重定向 (>) 的命令:只检查管道前的 binary 是否在白名单内,
并设置 needs_shell=True 以便用 shell 方式执行。
"""
needs_shell = _has_shell_operators(command)
if needs_shell:
# 含 shell 操作符或为多行命令:先检查武器级危险命令
if _command_has_dangerous_cmds(command):
raise ValueError("命令包含不允许的危险命令(sh/bash/dd/mkfs 等),请手动执行")
# 多行或含操作符的命令:按行分割后,对每行逐段检查,取最高权限模式
lines = command.split("\n") if "\n" in command else [command]
overall_mode = "read"
overall_approval = False
first_binary = None
first_segment = ""
for line in lines:
line = line.strip()
if not line or line.startswith("#"):
continue
segments = _split_into_commands(line)
for segment in segments:
# 检测 kubectl 命令
if segment.startswith("kubectl ") or segment == "kubectl":
try:
kdec = classify_kubectl_command(segment)
except (ValueError, IndexError) as exc:
raise ValueError(str(exc))
if first_binary is None:
first_binary = "kubectl"
first_segment = segment
if kdec.mode == "write":
overall_mode = "write"
overall_approval = True
continue
try:
seg_parts = shlex.split(segment)
except ValueError:
continue
if not seg_parts:
continue
# 跳过 VAR=value 赋值前缀(含 export
idx = 0
while idx < len(seg_parts):
if "=" in seg_parts[idx] and not seg_parts[idx].startswith("-"):
idx += 1
elif seg_parts[idx] == "export":
idx += 1
else:
break
if idx >= len(seg_parts):
continue # 纯赋值段,跳过
binary = seg_parts[idx]
check_parts = seg_parts[idx:]
if first_binary is None:
first_binary = binary
if binary not in SHELL_READ_ONLY_COMMANDS and binary not in SHELL_WRITE_COMMANDS:
raise ValueError(f"不允许执行该命令: {binary}")
if binary in SHELL_WRITE_COMMANDS:
allowed = SHELL_WRITE_COMMANDS[binary]
if allowed is None:
overall_mode = "write"
overall_approval = True
else:
verb = _first_non_flag(check_parts[1:])
if verb is not None and verb in allowed:
overall_mode = "write"
overall_approval = True
elif binary in SHELL_READ_ONLY_COMMANDS:
ro_allowed = SHELL_READ_ONLY_COMMANDS[binary]
if ro_allowed is None or (verb is not None and verb in ro_allowed):
pass # 该段是只读,不改变 overall
else:
raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
else:
raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
if first_binary is None:
raise ValueError("命令不能为空")
return CommandDecision(overall_mode, overall_approval, first_binary, [command], needs_shell=True)
parts = parse_shell_command(command)
binary = parts[0]
# 支持 VAR=value command 语法
start_idx = 0
while start_idx < len(parts) and "=" in parts[start_idx] and not parts[start_idx].startswith("-"):
start_idx += 1
if start_idx > 0:
if start_idx < len(parts):
binary = parts[start_idx]
effective_parts = parts[start_idx:]
else:
# 纯赋值行如 LOG_DIR=/tmp/log — 放行
return CommandDecision("read", False, parts[0], parts[1:])
verb = _first_non_flag(effective_parts[1:])
else:
effective_parts = parts
verb = _first_non_flag(parts[1:])
if binary in SHELL_WRITE_COMMANDS:
allowed = SHELL_WRITE_COMMANDS[binary]
if allowed is None or (verb is not None and verb in allowed):
return CommandDecision("write", True, binary, effective_parts[1:])
# etcdctl: 单段式或两段式命令,verb/subverb 均跳过全局 flag (--endpoints 等)
# member/alarm/auth 等同时有只读和写子命令,需按 (verb, subverb) 组合精确判断
if binary == "etcdctl":
non_flags = [a for a in effective_parts[1:] if not a.startswith("-")]
ev = non_flags[0].lower() if non_flags else None
esub = non_flags[1].lower() if len(non_flags) > 1 else None
# 单段式 verb
if ev in _ETCDCTL_READ_VERBS:
return CommandDecision("read", False, ev, effective_parts[1:])
if ev in _ETCDCTL_WRITE_VERBS:
return CommandDecision("write", True, ev, effective_parts[1:])
# 两段式: 先查写子命令 (精确匹配 subverb),再查只读子命令
if ev in _ETCDCTL_WRITE_SUBVERBS and esub in _ETCDCTL_WRITE_SUBVERBS[ev]:
return CommandDecision("write", True, ev, effective_parts[1:])
if ev in _ETCDCTL_READ_SUBVERBS:
if esub is None or esub in _ETCDCTL_READ_SUBVERBS[ev]:
return CommandDecision("read", False, ev, effective_parts[1:])
raise ValueError(f"etcdctl {ev} 不支持子命令: {esub}")
raise ValueError(f"etcdctl 不支持该子命令: {ev}")
if binary in SHELL_READ_ONLY_COMMANDS:
allowed = SHELL_READ_ONLY_COMMANDS[binary]
if allowed is not None and (verb is None or verb not in allowed):
raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
if binary == "ip" and any(arg in _IP_WRITE_ACTIONS for arg in effective_parts[1:]):
raise ValueError("ip 命令包含写操作,请使用只读形式如 'ip addr show'")
return CommandDecision("read", False, binary, effective_parts[1:])
raise ValueError(f"不允许执行该命令: {binary}")
def classify_agent_command(command: str) -> CommandDecision:
"""统一分类 Agent 命令:kubectl 或白名单系统诊断命令。"""
stripped = (command or "").strip()
# 检测 VAR=value kubectl 语法(有变量赋值时)
rest = stripped
while True:
try:
first = rest.split(None, 1)[0]
except IndexError:
break
if "=" in first and not first.startswith("-"):
idx = rest.index("=")
# 必须是合法变量赋值 VAR=value
parts_split = rest.split(None, 1)
if len(parts_split) < 2:
break
rest = parts_split[1]
elif first == "export":
parts_split = rest.split(None, 1)
if len(parts_split) < 2:
break
rest = parts_split[1]
else:
break
if rest.startswith("kubectl ") or rest == "kubectl":
return classify_kubectl_command(rest)
if stripped.startswith("kubectl ") or stripped == "kubectl":
return classify_kubectl_command(stripped)
return classify_shell_command(stripped)
# ── 命令执行 ──────────────────────────────────────────────
async def execute_command(command: str, timeout: int = 30) -> dict:
"""执行 Agent 命令 (kubectl 或白名单系统诊断命令)。
含管道/重定向的命令通过 shell=True 执行,其他用 subprocess_exec 直接执行。
"""
decision = classify_agent_command(command)
parts = _parse_command(command)
# 支持 VAR=value command 语法,跳过赋值找实际 binary
cmd_idx = 0
while cmd_idx < len(parts) and "=" in parts[cmd_idx] and not parts[cmd_idx].startswith("-"):
cmd_idx += 1
if cmd_idx > 0 and cmd_idx < len(parts):
parts = parts[cmd_idx:]
if parts[0] == "kubectl":
executable = KUBECTL
args = decision.args
else:
executable = parts[0]
args = parts[1:]
if decision.needs_shell:
proc = await asyncio.create_subprocess_shell(
command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
else:
proc = await asyncio.create_subprocess_exec(
executable,
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
return {
"command": command,
"mode": decision.mode,
"returncode": proc.returncode,
"stdout": stdout.decode(errors="replace")[:20000].strip(),
"stderr": stderr.decode(errors="replace")[:10000].strip(),
}
except asyncio.TimeoutError:
proc.kill()
await proc.communicate()
return {
"command": command,
"mode": decision.mode,
"returncode": -1,
"stdout": "",
"stderr": f"命令执行超时 ({timeout}s)",
}
async def execute_kubectl(command: str, timeout: int = 30) -> dict:
"""执行单条 kubectl 命令 (兼容旧接口)。"""
decision = classify_kubectl_command(command)
proc = await asyncio.create_subprocess_exec(
KUBECTL,
*decision.args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
return {
"command": command,
"mode": decision.mode,
"returncode": proc.returncode,
"stdout": stdout.decode(errors="replace")[:20000].strip(),
"stderr": stderr.decode(errors="replace")[:10000].strip(),
}
except asyncio.TimeoutError:
proc.kill()
await proc.communicate()
return {
"command": command,
"mode": decision.mode,
"returncode": -1,
"stdout": "",
"stderr": f"命令执行超时 ({timeout}s)",
}
+391
View File
@@ -1,7 +1,12 @@
"""AI Agent 分析模块 - 调用 OpenAI 兼容接口分析诊断结果 (支持流式输出)""" """AI Agent 分析模块 - 调用 OpenAI 兼容接口分析诊断结果 (支持流式输出)"""
import json import json
import re
import uuid
from collections import Counter
from typing import Optional
import httpx import httpx
from backend.config import AI_API_BASE, AI_API_KEY, AI_MODEL from backend.config import AI_API_BASE, AI_API_KEY, AI_MODEL
from backend.agent_tools import classify_agent_command, execute_command
SYSTEM_PROMPT = """你是一位资深的 Kubernetes 运维专家和 SRE 工程师。你的任务是分析 K8S 集群诊断报告,找出根本原因并给出修复建议。 SYSTEM_PROMPT = """你是一位资深的 Kubernetes 运维专家和 SRE 工程师。你的任务是分析 K8S 集群诊断报告,找出根本原因并给出修复建议。
@@ -20,6 +25,392 @@ _HEADERS = {
} }
AGENT_SYSTEM_PROMPT = """你是 Kubernetes 自主诊断 Agent。你的首要职责是主动调用命令获取现场证据,而不是只给用户一段脚本或操作建议。
可用工具:
1. kubectl 命令 - 查询集群内资源状态。例如: kubectl get pods -A, kubectl describe node <name>, kubectl logs <pod> -n <ns>
2. 系统诊断 shell 命令 - 查询节点宿主机状态。例如: systemctl status kubelet, journalctl -u kubelet -n 100 --no-pager, crictl ps, df -h, free -m, ss -tlnp, ip addr, top, du -sh, curl, tcpdump
3. containerd 命令 - 查询容器运行时状态。例如: ctr containers list, ctr images list, ctr tasks ls, nerdctl ps -a, nerdctl logs <container>, nerdctl stats
4. etcd 诊断命令 - 检查 etcd 集群健康与数据。例如: etcdctl endpoint status, etcdctl endpoint health, etcdctl member list, etcdctl alarm list, etcdctl get /kubernetes --prefix
5. Linux 基础调优命令 - 例如: iostat -xz 1 3, vmstat 1 3, mpstat -P ALL 1 3, pidstat -d 1 3, lsof -i :<port>, ethtool <iface>, tcpdump -i any -c 10 -nn
行为要求:
1. 如果证据不足,必须返回 command 动作并逐条调查;至少检查与故障资源直接相关的状态、describe、events 或 logs。
2. 禁止在 final 中仅提供"请执行以下命令"的脚本。只有在已经依据真实命令输出形成结论时才能返回 final。
3. 每次只提出一条命令,服务端会执行并把真实输出返回给你,然后你继续下一轮判断。
安全规则:
1. 可以使用管道 (|)、重定向 (>/>>)、逻辑连接符 (&&/||) 或分号 (;) 组合多条命令进行精确排查。不得使用反引号 (`` ` ``) 执行命令替换。
2. kubectl 的 get、describe、logs、top 等只读命令可以自动执行。
3. 系统诊断只读命令 (systemctl status、journalctl、crictl ps、df、free、ss、ip addr 等) 可以自动执行。
4. apply、delete、patch、scale、rollout、exec 等修改或交互命令必须暂停并请求人工批准。
5. systemctl restart/stop/start 等系统修改命令必须暂停并请求人工批准。
6. 不要猜测命令输出;必须依据实际工具结果继续判断。
7. 只要问题没有完全解决(verified=true 且 remaining_issues=0),就要持续执行命令诊断,没有轮数上限。
每轮必须只返回一个 JSON 对象,不要使用 Markdown 代码块:
- 继续调查:{"action":"command","command":"kubectl ...","reason":"为什么执行"}
- 完成分析:{"action":"final","analysis":"中文 Markdown 分析结论","verified":true,"remaining_issues":0}
只有完成修复后的全量复检,并确认 remaining_issues 为 0 时才能返回 final。只定位根因、执行命令成功或部分资源恢复都不能结束。
"""
_PENDING_APPROVALS: dict[str, dict] = {}
def _parse_agent_action(content: str) -> dict:
text = content.strip()
candidates = [text]
fenced = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL | re.IGNORECASE)
if fenced:
candidates.insert(0, fenced.group(1))
first_brace = text.find("{")
last_brace = text.rfind("}")
if first_brace >= 0 and last_brace > first_brace:
candidates.append(text[first_brace:last_brace + 1])
action = None
for candidate in candidates:
try:
action = json.loads(candidate)
break
except json.JSONDecodeError:
continue
if action is None:
raise ValueError("AI 未返回合法 JSON 动作")
if not isinstance(action, dict) or action.get("action") not in {"command", "final"}:
raise ValueError("AI 返回了不支持的动作")
if action["action"] == "command" and not action.get("command"):
raise ValueError("AI 命令动作缺少 command")
if action["action"] == "final" and not action.get("analysis"):
raise ValueError("AI 最终动作缺少 analysis")
if action["action"] == "final":
if action.get("verified") is not True or action.get("remaining_issues") != 0:
raise ValueError("AI 最终动作必须确认 verified=true 且 remaining_issues=0")
return action
async def _complete_chat(messages: list[dict]) -> str:
timeout = httpx.Timeout(90, connect=10)
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(
f"{AI_API_BASE}/chat/completions",
headers=_HEADERS,
json={
"model": AI_MODEL,
"messages": messages,
"temperature": 0.1,
"max_tokens": 4096,
"stream": False,
},
)
response.raise_for_status()
payload = response.json()
return payload["choices"][0]["message"]["content"]
async def run_diagnostic_agent(
report: str,
question: str = "",
max_steps: int = 30,
execution_mode: str = "ai",
run_id: str = "",
history_store=None,
):
"""自主诊断循环:AI 模式执行只读命令,手动模式等待用户执行。"""
if execution_mode not in {"ai", "manual"}:
yield {"type": "error", "message": "不支持的命令执行模式,仅支持 ai 或 manual"}
return
user_content = f"当前 K8S 诊断报告:\n\n{report}"
if question:
user_content += f"\n\n用户关注点:{question}"
messages = [
{"role": "system", "content": AGENT_SYSTEM_PROMPT},
{"role": "user", "content": user_content},
]
command_counts = Counter()
if execution_mode == "ai":
initial_command = "kubectl get nodes -o wide"
command_counts[initial_command] += 1
yield {
"type": "command", "step": 0, "command": initial_command,
"reason": "建立集群节点状态基线,确保 AI 执行模式实际采集现场证据",
"mode": "read",
}
initial_result = await execute_command(initial_command)
yield {"type": "result", **initial_result}
messages.append({
"role": "user",
"content": "AI 执行模式已自动采集第一份现场证据:\n" + json.dumps(initial_result, ensure_ascii=False),
})
async for event in _run_agent_loop(
messages=messages,
execution_mode=execution_mode,
max_steps=max_steps,
start_step=1,
command_counts=command_counts,
report=report,
question=question,
run_id=run_id,
history_store=history_store,
):
yield event
async def _run_agent_loop(
messages: list[dict],
execution_mode: str,
max_steps: int,
start_step: int,
command_counts: Counter,
report: str,
question: str,
run_id: str = "",
history_store=None,
):
invalid_response_count = 0
step = start_step
while True:
if max_steps > 0 and step > max_steps:
break
yield {"type": "thinking", "step": step, "message": f"AI 正在规划第 {step} 轮诊断"}
try:
content = await _complete_chat(messages)
action = _parse_agent_action(content)
except ValueError as exc:
if "最终动作必须确认" in str(exc):
invalid_response_count = 0
messages.extend([
{"role": "assistant", "content": content},
{
"role": "user",
"content": (
"不能结束诊断:尚未通过修复后的全量复检确认所有问题清零。"
"请继续返回 command,检查节点、Pod、Deployment、事件以及原故障资源;"
"只有 verified=true 且 remaining_issues=0 时才可返回 final。"
),
},
])
yield {"type": "verification_required", "message": "尚未确认所有问题清零,Agent 继续执行诊断验证"}
continue
invalid_response_count += 1
if invalid_response_count >= 3:
yield {
"type": "error",
"message": "AI 连续 3 次未返回合法 JSON 动作,已停止本次诊断以避免无效循环",
}
return
messages.extend([
{"role": "assistant", "content": content},
{
"role": "user",
"content": (
f"上一次响应无法解析:{exc}。请不要输出解释、脚本或多条命令,"
"严格只返回一个 JSON 对象:"
'{"action":"command","command":"kubectl ...","reason":"..."} '
"或在全量复检问题清零后返回 final。"
),
},
])
yield {
"type": "response_rejected",
"message": f"AI 响应格式无效,正在自动纠正并重试({invalid_response_count}/3",
"reason": str(exc),
}
continue
except Exception as exc:
yield {"type": "error", "message": str(exc)}
return
if action["action"] == "final":
invalid_response_count = 0
yield {
"type": "final", "content": action["analysis"], "model": AI_MODEL,
"verified": True, "remaining_issues": 0,
}
return
command = action["command"]
invalid_response_count = 0
reason = action.get("reason", "AI 需要更多集群证据")
command_counts[command] += 1
if command_counts[command] > 2:
messages.extend([
{"role": "assistant", "content": content},
{
"role": "user",
"content": (
f"命令 {command!r} 已重复执行两次,不能再次执行。"
"请基于已有输出选择不同的只读诊断命令,或在证据充分时返回 final。"
),
},
])
yield {
"type": "command_rejected", "command": command,
"reason": "相同命令已执行两次,请更换诊断方向",
}
continue
try:
decision = classify_agent_command(command)
except ValueError as exc:
messages.extend([
{"role": "assistant", "content": content},
{"role": "user", "content": f"命令被安全策略拒绝:{exc}。请改用单条合法 kubectl 或系统诊断命令。"},
])
yield {"type": "command_rejected", "command": command, "reason": str(exc)}
continue
yield {
"type": "command", "step": step, "command": command,
"reason": reason, "mode": decision.mode,
}
if execution_mode == "manual":
yield {
"type": "manual_command", "command_id": uuid.uuid4().hex,
"command": command, "reason": reason, "mode": decision.mode,
"message": "请在终端手动执行命令,并根据结果继续排查",
}
# 手动模式也保存上下文,支持断点续传
self_save_context(run_id, history_store, messages, command_counts, step, step + 1,
max_steps, report, question, execution_mode, content)
return
if decision.requires_approval:
approval_id = uuid.uuid4().hex
_PENDING_APPROVALS[approval_id] = {
"command": command,
"reason": reason,
"created_by": AI_MODEL,
"run_id": run_id,
"report": report,
"question": question,
"execution_mode": execution_mode,
"history_store": history_store,
"messages": messages + [{"role": "assistant", "content": content}],
"next_step": step + 1,
"max_steps": max_steps,
"command_counts": dict(command_counts),
}
# 保存上下文到 DB,避免浏览器关闭/刷新丢失进度
self_save_context(run_id, history_store, messages, command_counts, step, step + 1,
max_steps, report, question, execution_mode, content)
yield {
"type": "approval_required", "approval_id": approval_id,
"command": command, "reason": reason,
}
return
result = await execute_command(command)
yield {"type": "result", **result}
messages.extend([
{"role": "assistant", "content": content},
{"role": "user", "content": "命令实际执行结果:\n" + json.dumps(result, ensure_ascii=False)},
])
# 每次执行后保存上下文,支持断点续传
self_save_context(run_id, history_store, messages, command_counts, step, step + 1,
max_steps, report, question, execution_mode)
step += 1
if max_steps > 0:
yield {
"type": "error",
"message": (
f"自主诊断已达到安全上限({max_steps} 轮),但 Agent 仍未形成有效结论。"
"系统已停止继续执行以避免无限循环,请检查是否重复执行相同命令或 AI 服务响应异常。"
),
}
# max_steps=0 表示不限轮数,不输出错误 — Agent 会持续执行直到完成或遇到致命错误
async def resume_diagnostic_agent(pending: dict, result: dict):
"""审批命令执行后,把真实结果交回原 Agent 上下文并继续诊断。"""
messages = list(pending.get("messages") or [])
messages.append({
"role": "user",
"content": (
"用户已批准命令,以下是实际执行结果。请继续自主诊断并验证问题是否解决:\n"
+ json.dumps(result, ensure_ascii=False)
),
})
run_id = pending.get("run_id", "")
history_store = pending.get("history_store")
async for event in _run_agent_loop(
messages=messages,
execution_mode=pending.get("execution_mode", "ai"),
max_steps=int(pending.get("max_steps", 30)),
start_step=int(pending.get("next_step", 1)),
command_counts=Counter(pending.get("command_counts") or {}),
report=pending.get("report", ""),
question=pending.get("question", ""),
run_id=run_id,
history_store=history_store,
):
yield event
def take_pending_approval(approval_id: str) -> dict | None:
return _PENDING_APPROVALS.pop(approval_id, None)
def self_save_context(
run_id: str, history_store,
messages: list, command_counts: Counter,
step: int, next_step: int,
max_steps: int, report: str, question: str,
execution_mode: str,
last_response_content: str = "",
):
"""保存 Agent 执行上下文到 DB,支持断点续传。"""
if not run_id or not history_store:
return
context = {
"messages": messages,
"command_counts": dict(command_counts),
"step": step,
"next_step": next_step,
"max_steps": max_steps,
"report": report,
"question": question,
"execution_mode": execution_mode,
"last_response": last_response_content,
}
history_store.update_context(run_id, context)
async def resume_from_context(
ctx: dict,
run_id: str = "",
history_store=None,
):
"""从 DB 保存的上下文恢复 Agent 诊断,不重跑历史命令,直接进入下一轮。"""
messages = list(ctx.get("messages") or [])
command_counts = Counter(ctx.get("command_counts") or {})
next_step = int(ctx.get("next_step", 1))
max_steps = int(ctx.get("max_steps", 0))
report = ctx.get("report", "")
question = ctx.get("question", "")
execution_mode = ctx.get("execution_mode", "ai")
async for event in _run_agent_loop(
messages=messages,
execution_mode=execution_mode,
max_steps=max_steps,
start_step=next_step,
command_counts=command_counts,
report=report,
question=question,
run_id=run_id,
history_store=history_store,
):
yield event
async def _stream_chat(messages: list[dict]): async def _stream_chat(messages: list[dict]):
"""通用流式请求 - yield 文本 chunk""" """通用流式请求 - yield 文本 chunk"""
# connect=10s 快速失败; read=90s 防止 AI 服务无响应时前端无限卡死 # connect=10s 快速失败; read=90s 防止 AI 服务无响应时前端无限卡死
+1
View File
@@ -9,3 +9,4 @@ AI_API_KEY = os.getenv("AI_API_KEY", "hermes")
AI_MODEL = os.getenv("AI_MODEL", "qwen3.8-max-preview") AI_MODEL = os.getenv("AI_MODEL", "qwen3.8-max-preview")
PORT = int(os.getenv("PORT", "8900")) PORT = int(os.getenv("PORT", "8900"))
KUBECONFIG_PATH = os.getenv("KUBECONFIG_PATH", "") or None KUBECONFIG_PATH = os.getenv("KUBECONFIG_PATH", "") or None
HISTORY_DB_PATH = os.getenv("HISTORY_DB_PATH", os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "history.db"))
+211
View File
@@ -0,0 +1,211 @@
"""SQLite 持久化诊断历史、AI 分析和 Agent 处理记录。"""
import json
import os
import sqlite3
import uuid
from datetime import datetime
from typing import Optional
def _now() -> str:
return datetime.now().isoformat(timespec="seconds")
class HistoryStore:
def __init__(self, path: str):
self.path = path
parent = os.path.dirname(os.path.abspath(path))
os.makedirs(parent, exist_ok=True)
self._init_db()
def _connect(self):
connection = sqlite3.connect(self.path, timeout=30)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA journal_mode=WAL")
connection.execute("PRAGMA foreign_keys=ON")
return connection
def _init_db(self):
with self._connect() as db:
db.executescript("""
CREATE TABLE IF NOT EXISTS diagnoses (
id TEXT PRIMARY KEY,
created_at TEXT NOT NULL,
namespace TEXT NOT NULL,
score INTEGER NOT NULL,
status TEXT NOT NULL,
critical_count INTEGER NOT NULL DEFAULT 0,
warning_count INTEGER NOT NULL DEFAULT 0,
elapsed_seconds REAL NOT NULL DEFAULT 0,
diagnosis_json TEXT NOT NULL,
report TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_diagnoses_created_at ON diagnoses(created_at DESC);
CREATE INDEX IF NOT EXISTS idx_diagnoses_status ON diagnoses(status);
CREATE TABLE IF NOT EXISTS processing_runs (
id TEXT PRIMARY KEY,
diagnosis_id TEXT NOT NULL,
kind TEXT NOT NULL,
mode TEXT NOT NULL DEFAULT '',
question TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL,
summary TEXT NOT NULL DEFAULT '',
started_at TEXT NOT NULL,
finished_at TEXT,
context_json TEXT NOT NULL DEFAULT '',
FOREIGN KEY(diagnosis_id) REFERENCES diagnoses(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_runs_diagnosis ON processing_runs(diagnosis_id, started_at);
CREATE TABLE IF NOT EXISTS processing_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
run_id TEXT NOT NULL,
sequence INTEGER NOT NULL,
type TEXT NOT NULL,
created_at TEXT NOT NULL,
payload_json TEXT NOT NULL,
FOREIGN KEY(run_id) REFERENCES processing_runs(id) ON DELETE CASCADE,
UNIQUE(run_id, sequence)
);
""")
# 兼容旧表:确保 context_json 列存在
self._ensure_column(db, "processing_runs", "context_json", "TEXT NOT NULL DEFAULT ''")
def _ensure_column(self, db, table: str, column: str, col_def: str):
try:
db.execute(f"ALTER TABLE {table} ADD COLUMN {column} {col_def}")
except sqlite3.OperationalError as e:
if "duplicate column" not in str(e).lower():
raise
def save_diagnosis(self, diagnosis: dict, report: str) -> str:
record_id = uuid.uuid4().hex
created_at = diagnosis.get("timestamp") or _now()
with self._connect() as db:
db.execute(
"""INSERT INTO diagnoses
(id, created_at, namespace, score, status, critical_count, warning_count,
elapsed_seconds, diagnosis_json, report)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
record_id, created_at, diagnosis.get("namespace", "all"),
int(diagnosis.get("score", 0)), diagnosis.get("status", "unknown"),
int(diagnosis.get("critical_count", 0)), int(diagnosis.get("warning_count", 0)),
float(diagnosis.get("elapsed_seconds", 0)),
json.dumps(diagnosis, ensure_ascii=False), report,
),
)
return record_id
def list_diagnoses(self, status: str = "", namespace: str = "", limit: int = 100, offset: int = 0) -> list[dict]:
conditions = []
params: list = []
if status:
conditions.append("status = ?")
params.append(status)
if namespace:
conditions.append("namespace = ?")
params.append(namespace)
where = " WHERE " + " AND ".join(conditions) if conditions else ""
params.extend([max(1, min(limit, 500)), max(0, offset)])
with self._connect() as db:
rows = db.execute(
f"""SELECT id, created_at, namespace, score, status, critical_count,
warning_count, elapsed_seconds,
(SELECT COUNT(*) FROM processing_runs r WHERE r.diagnosis_id = diagnoses.id) AS run_count
FROM diagnoses{where}
ORDER BY created_at DESC, rowid DESC LIMIT ? OFFSET ?""",
params,
).fetchall()
return [dict(row) for row in rows]
def get_diagnosis(self, record_id: str) -> Optional[dict]:
with self._connect() as db:
row = db.execute("SELECT * FROM diagnoses WHERE id = ?", (record_id,)).fetchone()
if row is None:
return None
run_rows = db.execute(
"SELECT * FROM processing_runs WHERE diagnosis_id = ? ORDER BY started_at, rowid",
(record_id,),
).fetchall()
runs = []
for run_row in run_rows:
run = dict(run_row)
event_rows = db.execute(
"SELECT type, created_at, payload_json FROM processing_events WHERE run_id = ? ORDER BY sequence",
(run["id"],),
).fetchall()
run["events"] = [
{"type": event["type"], "created_at": event["created_at"], **json.loads(event["payload_json"])}
for event in event_rows
]
run["context_json"] = run.get("context_json") or ""
if run["context_json"]:
run["context"] = json.loads(run["context_json"])
else:
run["context"] = None
runs.append(run)
return {
"id": row["id"], "created_at": row["created_at"],
"diagnosis": json.loads(row["diagnosis_json"]),
"report": row["report"], "runs": runs,
}
def latest(self) -> Optional[dict]:
records = self.list_diagnoses(limit=1)
return self.get_diagnosis(records[0]["id"]) if records else None
def create_run(self, diagnosis_id: str, kind: str, mode: str = "", question: str = "") -> str:
run_id = uuid.uuid4().hex
with self._connect() as db:
db.execute(
"""INSERT INTO processing_runs
(id, diagnosis_id, kind, mode, question, status, started_at)
VALUES (?, ?, ?, ?, ?, 'running', ?)""",
(run_id, diagnosis_id, kind, mode, question, _now()),
)
return run_id
def update_context(self, run_id: str, context: dict):
"""保存/更新 Agent 执行上下文到 processing_runs,支持断点续传。"""
with self._connect() as db:
db.execute(
"UPDATE processing_runs SET context_json = ? WHERE id = ?",
(json.dumps(context, ensure_ascii=False), run_id),
)
def get_context(self, run_id: str) -> Optional[dict]:
"""读取 Agent 执行上下文,如果不存在或为空则返回 None。"""
with self._connect() as db:
row = db.execute(
"SELECT context_json, diagnosis_id, report FROM processing_runs pr JOIN diagnoses d ON pr.diagnosis_id = d.id WHERE pr.id = ?",
(run_id,),
).fetchone()
if row is None:
return None
ctx_str = row["context_json"] or ""
if not ctx_str:
return None
ctx = json.loads(ctx_str)
ctx["report"] = row["report"]
return ctx
def append_event(self, run_id: str, event_type: str, payload: dict):
with self._connect() as db:
sequence = db.execute(
"SELECT COALESCE(MAX(sequence), 0) + 1 FROM processing_events WHERE run_id = ?",
(run_id,),
).fetchone()[0]
db.execute(
"""INSERT INTO processing_events (run_id, sequence, type, created_at, payload_json)
VALUES (?, ?, ?, ?, ?)""",
(run_id, sequence, event_type, _now(), json.dumps(payload, ensure_ascii=False)),
)
def finish_run(self, run_id: str, status: str, summary: str = ""):
with self._connect() as db:
db.execute(
"UPDATE processing_runs SET status = ?, summary = ?, finished_at = ? WHERE id = ?",
(status, summary, _now(), run_id),
)
+255 -25
View File
@@ -11,7 +11,7 @@ from fastapi.responses import StreamingResponse
from pydantic import BaseModel from pydantic import BaseModel
from typing import Optional from typing import Optional
from backend.config import PORT, AI_API_BASE, AI_MODEL from backend.config import PORT, AI_API_BASE, AI_MODEL, HISTORY_DB_PATH
from backend.diagnosis import ( from backend.diagnosis import (
run_full_diagnosis, build_diagnosis_report, ALL_CHECKS, run_full_diagnosis, build_diagnosis_report, ALL_CHECKS,
check_connectivity, check_connectivity,
@@ -22,9 +22,14 @@ from backend.diagnosis import (
from backend.ai_agent import ( from backend.ai_agent import (
analyze_with_ai, chat_with_ai, analyze_with_ai, chat_with_ai,
analyze_with_ai_stream, chat_with_ai_stream, analyze_with_ai_stream, chat_with_ai_stream,
run_diagnostic_agent, resume_diagnostic_agent, resume_from_context,
take_pending_approval,
) )
from backend.agent_tools import execute_command
from backend.history_store import HistoryStore
app = FastAPI(title="K8S 智能诊断平台", version="1.0.0") app = FastAPI(title="K8S 智能诊断平台", version="1.0.0")
history_store = HistoryStore(HISTORY_DB_PATH)
app.add_middleware( app.add_middleware(
CORSMiddleware, CORSMiddleware,
@@ -36,6 +41,25 @@ app.add_middleware(
# 存储最近诊断结果 (内存) # 存储最近诊断结果 (内存)
_last_diagnosis: Optional[dict] = None _last_diagnosis: Optional[dict] = None
_last_report: str = "" _last_report: str = ""
_last_diagnosis_id: Optional[str] = None
def _save_current_diagnosis(diagnosis: dict, report: str) -> str:
global _last_diagnosis, _last_report, _last_diagnosis_id
_last_diagnosis = diagnosis
_last_report = report
_last_diagnosis_id = history_store.save_diagnosis(diagnosis, report)
return _last_diagnosis_id
def _ensure_current_history() -> str:
global _last_diagnosis_id
if _last_diagnosis_id:
return _last_diagnosis_id
if _last_diagnosis is None:
raise RuntimeError("暂无诊断记录")
_last_diagnosis_id = history_store.save_diagnosis(_last_diagnosis, _last_report)
return _last_diagnosis_id
# 诊断项执行顺序 (带中文标题) # 诊断项执行顺序 (带中文标题)
CHECK_ORDER = [ CHECK_ORDER = [
@@ -64,6 +88,11 @@ class ChatRequest(BaseModel):
messages: list[dict] messages: list[dict]
class AgentApprovalRequest(BaseModel):
approval_id: str
approved: bool
@app.get("/api/health") @app.get("/api/health")
async def health(): async def health():
return {"status": "ok", "time": datetime.now().isoformat()} return {"status": "ok", "time": datetime.now().isoformat()}
@@ -114,9 +143,9 @@ async def diagnose_stream(namespace: str = ""):
}], }],
"checks": {}, "checks": {},
} }
_last_diagnosis = diagnosis report = build_diagnosis_report(diagnosis)
_last_report = build_diagnosis_report(diagnosis) _save_current_diagnosis(diagnosis, report)
yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis, 'history_id': _last_diagnosis_id}, ensure_ascii=False)}\n\n"
return return
yield f"data: {json.dumps({'type': 'progress', 'step': 0, 'total': len(CHECK_ORDER) + 1, 'name': 'connectivity', 'title': '集群连通性检查', 'status': 'done', 'issues': 0}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'progress', 'step': 0, 'total': len(CHECK_ORDER) + 1, 'name': 'connectivity', 'title': '集群连通性检查', 'status': 'done', 'issues': 0}, ensure_ascii=False)}\n\n"
@@ -169,10 +198,10 @@ async def diagnose_stream(namespace: str = ""):
"checks": results, "checks": results,
} }
_last_diagnosis = diagnosis report = build_diagnosis_report(diagnosis)
_last_report = build_diagnosis_report(diagnosis) _save_current_diagnosis(diagnosis, report)
yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis, 'history_id': _last_diagnosis_id}, ensure_ascii=False)}\n\n"
return StreamingResponse( return StreamingResponse(
event_generator(), event_generator(),
@@ -185,17 +214,24 @@ async def diagnose_stream(namespace: str = ""):
async def diagnose(req: DiagnoseRequest): async def diagnose(req: DiagnoseRequest):
"""执行一键诊断 (非流式,兼容旧接口)""" """执行一键诊断 (非流式,兼容旧接口)"""
global _last_diagnosis, _last_report global _last_diagnosis, _last_report
_last_diagnosis = await run_full_diagnosis(namespace=req.namespace, checks=req.checks) diagnosis = await run_full_diagnosis(namespace=req.namespace, checks=req.checks)
_last_report = build_diagnosis_report(_last_diagnosis) report = build_diagnosis_report(diagnosis)
return _last_diagnosis _save_current_diagnosis(diagnosis, report)
return {**diagnosis, "history_id": _last_diagnosis_id}
@app.get("/api/diagnosis/latest") @app.get("/api/diagnosis/latest")
async def get_latest(): async def get_latest():
"""获取最近一次诊断结果""" """获取最近一次诊断结果,服务重启后从 SQLite 恢复。"""
global _last_diagnosis, _last_report, _last_diagnosis_id
if _last_diagnosis is None: if _last_diagnosis is None:
raise HTTPException(status_code=404, detail="暂无诊断记录,请先执行诊断") latest = history_store.latest()
return _last_diagnosis if latest is None:
raise HTTPException(status_code=404, detail="暂无诊断记录,请先执行诊断")
_last_diagnosis = latest["diagnosis"]
_last_report = latest["report"]
_last_diagnosis_id = latest["id"]
return {**_last_diagnosis, "history_id": _last_diagnosis_id}
@app.post("/api/analyze") @app.post("/api/analyze")
@@ -203,19 +239,25 @@ async def analyze(req: AnalyzeRequest):
"""AI 分析诊断结果 (非流式,兼容旧接口)""" """AI 分析诊断结果 (非流式,兼容旧接口)"""
global _last_diagnosis, _last_report global _last_diagnosis, _last_report
if _last_diagnosis is None: if _last_diagnosis is None:
_last_diagnosis = await run_full_diagnosis() diagnosis = await run_full_diagnosis()
_last_report = build_diagnosis_report(_last_diagnosis) report = build_diagnosis_report(diagnosis)
_save_current_diagnosis(diagnosis, report)
run_id = history_store.create_run(_ensure_current_history(), "analysis", "standard", req.question)
result = await analyze_with_ai(_last_report, question=req.question) result = await analyze_with_ai(_last_report, question=req.question)
return result history_store.append_event(run_id, "analysis", result)
history_store.finish_run(run_id, "completed" if result.get("success") else "failed", result.get("analysis", ""))
return {**result, "run_id": run_id}
@app.get("/api/analyze/stream") @app.get("/api/analyze/stream")
async def analyze_stream(question: str = ""): async def analyze_stream(question: str = ""):
"""SSE 流式 AI 分析 - 逐字推送分析内容""" """SSE 流式 AI 分析 - 逐字推送分析内容并持久化"""
global _last_diagnosis, _last_report global _last_diagnosis, _last_report
if _last_diagnosis is None: if _last_diagnosis is None:
_last_diagnosis = await run_full_diagnosis() diagnosis = await run_full_diagnosis()
_last_report = build_diagnosis_report(_last_diagnosis) report = build_diagnosis_report(diagnosis)
_save_current_diagnosis(diagnosis, report)
run_id = history_store.create_run(_ensure_current_history(), "analysis", "stream", question)
async def event_generator(): async def event_generator():
process_steps = [ process_steps = [
@@ -225,23 +267,190 @@ async def analyze_stream(question: str = ""):
("summary", "整理分析结论", "正在组织根因、影响和预防建议"), ("summary", "整理分析结论", "正在组织根因、影响和预防建议"),
] ]
yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL, 'run_id': run_id}, ensure_ascii=False)}\n\n"
content = ""
try: try:
for step, title, detail in process_steps: for step, title, detail in process_steps:
yield f"data: {json.dumps({'type': 'process', 'step': step, 'title': title, 'detail': detail, 'status': 'running'}, ensure_ascii=False)}\n\n" process_event = {'step': step, 'title': title, 'detail': detail, 'status': 'running'}
history_store.append_event(run_id, "process", process_event)
yield f"data: {json.dumps({'type': 'process', **process_event}, ensure_ascii=False)}\n\n"
await asyncio.sleep(0) await asyncio.sleep(0)
yield f"data: {json.dumps({'type': 'process', 'step': step, 'title': title, 'detail': detail, 'status': 'done'}, ensure_ascii=False)}\n\n" process_event = {'step': step, 'title': title, 'detail': detail, 'status': 'done'}
history_store.append_event(run_id, "process", process_event)
yield f"data: {json.dumps({'type': 'process', **process_event}, ensure_ascii=False)}\n\n"
async for chunk in analyze_with_ai_stream(_last_report, question=question): async for chunk in analyze_with_ai_stream(_last_report, question=question):
content += chunk
yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n"
yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n" history_store.append_event(run_id, "final", {"content": content, "model": AI_MODEL})
history_store.finish_run(run_id, "completed", content)
yield f"data: {json.dumps({'type': 'done', 'run_id': run_id}, ensure_ascii=False)}\n\n"
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
err = f"AI 接口返回错误: {e.response.status_code} - {str(e)[:300]}" err = f"AI 接口返回错误: {e.response.status_code} - {str(e)[:300]}"
history_store.append_event(run_id, "error", {"message": err})
history_store.finish_run(run_id, "failed", err)
yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n"
except httpx.ConnectError: except httpx.ConnectError:
yield f"data: {json.dumps({'type': 'error', 'message': f'无法连接 AI 服务 ({AI_API_BASE}),请检查 .env 中的 AI_API_BASE 配置'}, ensure_ascii=False)}\n\n" err = f"无法连接 AI 服务 ({AI_API_BASE}),请检查 .env 中的 AI_API_BASE 配置"
history_store.append_event(run_id, "error", {"message": err})
history_store.finish_run(run_id, "failed", err)
yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n"
except Exception as e: except Exception as e:
yield f"data: {json.dumps({'type': 'error', 'message': f'AI 分析失败: {str(e)}'}, ensure_ascii=False)}\n\n" err = f"AI 分析失败: {str(e)}"
history_store.append_event(run_id, "error", {"message": err})
history_store.finish_run(run_id, "failed", err)
yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
@app.get("/api/agent/diagnose/stream")
async def agent_diagnose_stream(question: str = "", execution_mode: str = "ai"):
"""自主诊断 Agent:支持手动执行和安全 AI 执行模式。"""
if execution_mode not in {"ai", "manual"}:
raise HTTPException(status_code=400, detail="execution_mode 仅支持 ai 或 manual")
global _last_diagnosis, _last_report
if _last_diagnosis is None:
diagnosis = await run_full_diagnosis()
report = build_diagnosis_report(diagnosis)
_save_current_diagnosis(diagnosis, report)
run_id = history_store.create_run(_ensure_current_history(), "agent", execution_mode, question)
async def event_generator():
yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL, 'mode': execution_mode, 'run_id': run_id}, ensure_ascii=False)}\n\n"
final_status = "completed"
summary = ""
try:
async for event in run_diagnostic_agent(
_last_report,
question=question,
max_steps=0,
execution_mode=execution_mode,
run_id=run_id,
history_store=history_store,
):
history_store.append_event(run_id, event.get("type", "event"), {k: v for k, v in event.items() if k != "type"})
if event.get("type") == "final":
summary = event.get("content", "")
elif event.get("type") == "error":
final_status = "failed"
summary = event.get("message", "")
elif event.get("type") in {"manual_command", "approval_required"}:
final_status = "waiting"
if event.get("type") == "approval_required":
from backend.ai_agent import _PENDING_APPROVALS
pending = _PENDING_APPROVALS.get(event.get("approval_id"))
if pending is not None:
pending["run_id"] = run_id
yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
history_store.finish_run(run_id, final_status, summary)
yield f"data: {json.dumps({'type': 'done', 'run_id': run_id}, ensure_ascii=False)}\n\n"
except Exception as exc:
message = f"自主诊断失败: {exc}"
history_store.append_event(run_id, "error", {"message": message})
history_store.finish_run(run_id, "failed", message)
yield f"data: {json.dumps({'type': 'error', 'message': message}, ensure_ascii=False)}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
@app.post("/api/agent/approve")
async def execute_approved_agent_command(req: AgentApprovalRequest):
"""人工批准后执行一次待审批的修改命令。"""
pending = take_pending_approval(req.approval_id)
if pending is None:
raise HTTPException(status_code=404, detail="审批请求不存在或已经处理")
if not req.approved:
result = {"approved": False, "command": pending["command"], "message": "用户已拒绝执行"}
if pending.get("run_id"):
history_store.append_event(pending["run_id"], "approval_rejected", result)
history_store.finish_run(pending["run_id"], "rejected", result["message"])
return result
result = await execute_command(pending["command"])
approved_event = {"approved": True, "reason": pending["reason"], **result}
async def event_generator():
yield f"data: {json.dumps({'type': 'approved_result', **approved_event}, ensure_ascii=False)}\n\n"
run_id = pending.get("run_id", "")
if run_id:
history_store.append_event(run_id, "approved_result", approved_event)
final_status = "completed"
summary = result.get("stdout") or result.get("stderr", "")
async for event in resume_diagnostic_agent(pending, result):
if run_id:
history_store.append_event(run_id, event.get("type", "event"), {k: v for k, v in event.items() if k != "type"})
if event.get("type") == "final":
summary = event.get("content", "")
elif event.get("type") == "error":
final_status = "failed"
summary = event.get("message", "")
elif event.get("type") == "approval_required":
final_status = "waiting"
from backend.ai_agent import _PENDING_APPROVALS
next_pending = _PENDING_APPROVALS.get(event.get("approval_id"))
if next_pending is not None:
next_pending["run_id"] = run_id
yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
if run_id:
history_store.finish_run(run_id, final_status, summary)
yield f"data: {json.dumps({'type': 'done', 'run_id': run_id}, ensure_ascii=False)}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
@app.get("/api/agent/resume-stream/{run_id}")
async def agent_resume_stream(run_id: str):
"""从 DB 上下文恢复 Agent 诊断,不重跑历史命令,直接继续诊断。"""
ctx = history_store.get_context(run_id)
if ctx is None:
raise HTTPException(status_code=404, detail="该运行记录不存在或不支持断点续传")
ctx.pop("report", None) # get_context 已经注入了 report
async def event_generator():
yield f"data: {json.dumps({'type': 'start', 'mode': ctx.get('execution_mode', 'ai'), 'run_id': run_id, 'resumed': True}, ensure_ascii=False)}\n\n"
final_status = "completed"
summary = ""
try:
async for event in resume_from_context(
ctx=ctx,
run_id=run_id,
history_store=history_store,
):
history_store.append_event(run_id, event.get("type", "event"), {k: v for k, v in event.items() if k != "type"})
if event.get("type") == "final":
summary = event.get("content", "")
elif event.get("type") == "error":
final_status = "failed"
summary = event.get("message", "")
elif event.get("type") in {"manual_command", "approval_required"}:
final_status = "waiting"
if event.get("type") == "approval_required":
from backend.ai_agent import _PENDING_APPROVALS
pending = _PENDING_APPROVALS.get(event.get("approval_id"))
if pending is not None:
pending["run_id"] = run_id
yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
history_store.finish_run(run_id, final_status, summary)
yield f"data: {json.dumps({'type': 'done', 'run_id': run_id}, ensure_ascii=False)}\n\n"
except Exception as exc:
message = f"断点续传诊断失败: {exc}"
history_store.append_event(run_id, "error", {"message": message})
history_store.finish_run(run_id, "failed", message)
yield f"data: {json.dumps({'type': 'error', 'message': message}, ensure_ascii=False)}\n\n"
return StreamingResponse( return StreamingResponse(
event_generator(), event_generator(),
@@ -276,6 +485,27 @@ async def chat_stream(req: ChatRequest):
) )
@app.get("/api/history/diagnoses")
async def list_diagnosis_history(
status: str = "",
namespace: str = "",
limit: int = 100,
offset: int = 0,
):
"""查询全部历史诊断记录,可按状态和命名空间过滤。"""
items = history_store.list_diagnoses(status=status, namespace=namespace, limit=limit, offset=offset)
return {"total": len(items), "items": items, "limit": limit, "offset": offset}
@app.get("/api/history/diagnoses/{record_id}")
async def get_diagnosis_history(record_id: str):
"""查询单次诊断及其全部 AI 分析、Agent 命令和处理记录。"""
record = history_store.get_diagnosis(record_id)
if record is None:
raise HTTPException(status_code=404, detail="历史诊断记录不存在")
return record
@app.get("/api/report") @app.get("/api/report")
async def get_report(): async def get_report():
"""获取诊断报告文本""" """获取诊断报告文本"""
+328 -1
View File
@@ -14,9 +14,17 @@
<el-button type="primary" :loading="diagnosing" @click="runDiagnosis" :icon="Refresh" :disabled="diagnosing"> <el-button type="primary" :loading="diagnosing" @click="runDiagnosis" :icon="Refresh" :disabled="diagnosing">
{{ diagnosing ? '诊断中...' : '一键诊断' }} {{ diagnosing ? '诊断中...' : '一键诊断' }}
</el-button> </el-button>
<el-button type="success" :loading="analyzing" @click="runAnalysis" :icon="MagicStick" :disabled="!diagnosis || diagnosing"> <el-button type="success" :loading="analyzing" @click="runAnalysis" :icon="MagicStick" :disabled="!diagnosis || diagnosing || agentRunning">
{{ analyzing ? 'AI 分析中...' : 'AI 分析' }} {{ analyzing ? 'AI 分析中...' : 'AI 分析' }}
</el-button> </el-button>
<el-radio-group v-model="agentExecutionMode" :disabled="agentRunning" class="agent-mode-switch">
<el-radio-button value="manual">手动执行命令</el-radio-button>
<el-radio-button value="ai">AI 执行命令</el-radio-button>
</el-radio-group>
<el-button @click="openHistory" :disabled="diagnosing || analyzing || agentRunning">历史记录</el-button>
<el-button type="warning" :loading="agentRunning" @click="runAgentDiagnosis" :disabled="!diagnosis || diagnosing || analyzing">
{{ agentRunning ? 'Agent 诊断中...' : '自主诊断' }}
</el-button>
</div> </div>
</el-header> </el-header>
@@ -257,6 +265,62 @@
<div class="ai-content" v-html="renderedAnalysis"></div> <div class="ai-content" v-html="renderedAnalysis"></div>
</el-card> </el-card>
<!-- 自主诊断 Agent -->
<el-card v-if="diagnosis && !diagnosing && (agentRunning || agentEvents.length)" class="section-card agent-card" shadow="never">
<template #header>
<div class="card-header">
<span>🧠 AI Agent 自主诊断</span>
<div class="ai-header-meta">
<el-tag :type="activeAgentMode === 'ai' ? 'success' : 'info'" size="small">
{{ activeAgentMode === 'ai' ? 'AI 执行模式' : '手动执行模式' }}
</el-tag>
<el-tag v-if="activeAgentMode === 'ai'" type="warning" size="small">修改命令人工确认</el-tag>
</div>
</div>
</template>
<el-alert
:title="activeAgentMode === 'ai'
? 'AI 执行模式:Agent 自动执行只读 kubectl 和系统诊断命令 (systemctl status、journalctl、crictl ps、df 等),修改或交互命令仍需人工批准。'
: '手动执行模式:Agent 只生成安全的诊断命令,不会在服务器上执行,请复制命令到终端手动运行。'"
type="info"
:closable="false"
show-icon
class="agent-alert"
/>
<div class="agent-timeline">
<div v-for="(event, index) in agentEvents" :key="index" class="agent-event" :class="'event-' + event.type">
<div class="agent-event-title">
<el-icon v-if="event.type === 'thinking'" class="is-loading" color="#409eff"><Loading /></el-icon>
<el-icon v-else-if="event.type === 'result' || event.type === 'final'" color="#67c23a"><CircleCheckFilled /></el-icon>
<el-icon v-else-if="event.type === 'approval_required'" color="#e6a23c"><WarningFilled /></el-icon>
<el-icon v-else color="#909399"><Monitor /></el-icon>
<strong>{{ agentEventTitle(event) }}</strong>
<el-tag v-if="event.mode" :type="event.mode === 'read' ? 'success' : 'warning'" size="small">
{{ event.mode === 'read' ? '只读' : '需审批' }}
</el-tag>
</div>
<p v-if="event.message" class="agent-event-message">{{ event.message }}</p>
<p v-if="event.reason" class="agent-event-message">目的{{ event.reason }}</p>
<pre v-if="event.command" class="agent-command">{{ event.command }}</pre>
<pre v-if="event.stdout || event.stderr" class="agent-output">{{ event.stdout || event.stderr }}</pre>
<div v-if="event.type === 'final'" class="ai-content" v-html="renderMd(event.content)"></div>
<div v-if="event.type === 'manual_command'" class="manual-actions">
<el-button type="primary" plain @click="copyAgentCommand(event.command)">复制命令</el-button>
<span>请在你的终端中执行Agent 不会自动运行该命令</span>
</div>
<div v-if="event.type === 'approval_required' && !event.resolved" class="approval-actions">
<el-button type="danger" @click="resolveApproval(event, true)" :loading="event.executing">确认执行</el-button>
<el-button @click="resolveApproval(event, false)" :disabled="event.executing">拒绝</el-button>
</div>
<el-alert v-if="event.approvalResult" :title="event.approvalResult" :type="event.approvalSuccess ? 'success' : 'info'" :closable="false" />
</div>
<div v-if="agentRunning" class="agent-running">
<el-icon class="is-loading" color="#409eff"><Loading /></el-icon>
<span>Agent 正在分析命令结果并规划下一步...</span>
</div>
</div>
</el-card>
<!-- AI 对话 (连接成功或失败都显示) --> <!-- AI 对话 (连接成功或失败都显示) -->
<el-card v-if="diagnosis && !diagnosing" class="section-card" shadow="never"> <el-card v-if="diagnosis && !diagnosing" class="section-card" shadow="never">
<template #header><span>💬 追问 AI 助手</span></template> <template #header><span>💬 追问 AI 助手</span></template>
@@ -274,6 +338,61 @@
</div> </div>
</el-card> </el-card>
</el-main> </el-main>
<el-drawer v-model="historyVisible" title="历史诊断与处理记录" size="75%" @open="loadHistory">
<div class="history-toolbar">
<el-select v-model="historyStatus" placeholder="全部状态" clearable style="width: 150px" @change="loadHistory">
<el-option label="健康" value="healthy" />
<el-option label="警告" value="warning" />
<el-option label="严重" value="critical" />
</el-select>
<el-input v-model="historyNamespace" placeholder="命名空间" clearable style="width: 180px" @keyup.enter="loadHistory" />
<el-button type="primary" @click="loadHistory" :loading="historyLoading">查询</el-button>
</div>
<el-table :data="historyItems" stripe highlight-current-row @row-click="loadHistoryDetail" v-loading="historyLoading">
<el-table-column prop="created_at" label="诊断时间" min-width="180" />
<el-table-column prop="namespace" label="命名空间" min-width="110" />
<el-table-column prop="score" label="评分" width="80" />
<el-table-column label="状态" width="90">
<template #default="{ row }"><el-tag :type="historyStatusType(row.status)">{{ historyStatusText(row.status) }}</el-tag></template>
</el-table-column>
<el-table-column prop="critical_count" label="严重" width="75" />
<el-table-column prop="warning_count" label="警告" width="75" />
<el-table-column prop="run_count" label="处理记录" width="100" />
<el-table-column label="操作" width="100">
<template #default="{ row }"><el-button link type="primary" @click.stop="loadHistoryDetail(row)">查看详情</el-button></template>
</el-table-column>
</el-table>
<div v-if="historyDetail" class="history-detail">
<el-divider content-position="left">诊断详情</el-divider>
<el-descriptions :column="4" border>
<el-descriptions-item label="时间">{{ historyDetail.created_at }}</el-descriptions-item>
<el-descriptions-item label="命名空间">{{ historyDetail.diagnosis.namespace }}</el-descriptions-item>
<el-descriptions-item label="健康评分">{{ historyDetail.diagnosis.score }}</el-descriptions-item>
<el-descriptions-item label="状态">{{ historyStatusText(historyDetail.diagnosis.status) }}</el-descriptions-item>
</el-descriptions>
<el-collapse class="history-collapse">
<el-collapse-item title="完整诊断报告" name="report">
<pre class="history-report">{{ historyDetail.report }}</pre>
</el-collapse-item>
<el-collapse-item v-for="run in historyDetail.runs" :key="run.id" :name="run.id">
<template #title>
<span>{{ runKindText(run.kind) }} · {{ run.started_at }} · {{ runStatusText(run.status) }}</span>
</template>
<el-descriptions :column="3" size="small" border>
<el-descriptions-item label="模式">{{ run.mode || '-' }}</el-descriptions-item>
<el-descriptions-item label="问题">{{ run.question || '-' }}</el-descriptions-item>
<el-descriptions-item label="结束时间">{{ run.finished_at || '-' }}</el-descriptions-item>
</el-descriptions>
<div v-for="(event, index) in run.events" :key="index" class="history-event">
<div class="history-event-head"><el-tag size="small">{{ event.type }}</el-tag><span>{{ event.created_at }}</span></div>
<pre>{{ formatHistoryEvent(event) }}</pre>
</div>
</el-collapse-item>
</el-collapse>
</div>
</el-drawer>
</div> </div>
</template> </template>
@@ -288,10 +407,20 @@ const API = '/api'
const namespace = ref('') const namespace = ref('')
const diagnosing = ref(false) const diagnosing = ref(false)
const analyzing = ref(false) const analyzing = ref(false)
const agentRunning = ref(false)
const agentExecutionMode = ref('manual')
const activeAgentMode = ref('manual')
const agentEvents = ref([])
const diagnosis = ref(null) const diagnosis = ref(null)
const aiAnalysis = ref('') const aiAnalysis = ref('')
const aiModel = ref('') const aiModel = ref('')
const analysisProcess = ref([]) const analysisProcess = ref([])
const historyVisible = ref(false)
const historyLoading = ref(false)
const historyItems = ref([])
const historyDetail = ref(null)
const historyStatus = ref('')
const historyNamespace = ref('')
const chatMessages = ref([]) const chatMessages = ref([])
const chatInput = ref('') const chatInput = ref('')
const chatLoading = ref(false) const chatLoading = ref(false)
@@ -462,6 +591,176 @@ async function runAnalysis() {
} }
} }
function openHistory() {
historyVisible.value = true
}
async function loadHistory() {
historyLoading.value = true
historyDetail.value = null
try {
const params = new URLSearchParams({ limit: '200' })
if (historyStatus.value) params.set('status', historyStatus.value)
if (historyNamespace.value) params.set('namespace', historyNamespace.value)
const resp = await fetch(`${API}/history/diagnoses?${params.toString()}`)
if (!resp.ok) throw new Error(`HTTP ${resp.status}`)
const data = await resp.json()
historyItems.value = data.items || []
} catch (e) {
ElMessage.error('加载历史记录失败: ' + e.message)
} finally {
historyLoading.value = false
}
}
async function loadHistoryDetail(row) {
historyLoading.value = true
try {
const resp = await fetch(`${API}/history/diagnoses/${row.id}`)
if (!resp.ok) throw new Error(`HTTP ${resp.status}`)
historyDetail.value = await resp.json()
} catch (e) {
ElMessage.error('加载历史详情失败: ' + e.message)
} finally {
historyLoading.value = false
}
}
function historyStatusType(status) {
return status === 'healthy' ? 'success' : status === 'warning' ? 'warning' : status === 'critical' ? 'danger' : 'info'
}
function historyStatusText(status) {
return { healthy: '健康', warning: '警告', critical: '严重' }[status] || status
}
function runKindText(kind) {
return { analysis: 'AI 分析', agent: 'Agent 自主诊断', chat: 'AI 对话' }[kind] || kind
}
function runStatusText(status) {
return { running: '运行中', completed: '已完成', waiting: '等待处理', failed: '失败', rejected: '已拒绝' }[status] || status
}
function formatHistoryEvent(event) {
const { type, created_at, ...payload } = event
return JSON.stringify(payload, null, 2)
}
async function runAgentDiagnosis() {
activeAgentMode.value = agentExecutionMode.value
agentRunning.value = true
agentEvents.value = []
try {
const params = new URLSearchParams()
params.set('execution_mode', activeAgentMode.value)
if (namespace.value) params.set('question', `重点检查命名空间 ${namespace.value}`)
const resp = await fetch(`${API}/agent/diagnose/stream?${params.toString()}`)
if (!resp.ok || !resp.body) throw new Error(`HTTP ${resp.status}`)
const reader = resp.body.getReader()
const decoder = new TextDecoder()
let buffer = ''
while (true) {
const { done, value } = await reader.read()
if (done) break
buffer += decoder.decode(value, { stream: true })
const lines = buffer.split('\n')
buffer = lines.pop() || ''
for (const line of lines) {
if (!line.startsWith('data: ')) continue
try {
const event = JSON.parse(line.slice(6))
if (!['start', 'done'].includes(event.type)) agentEvents.value.push(event)
if (event.type === 'error') ElMessage.error(event.message)
} catch (e) {
// ignore partial events
}
}
}
} catch (e) {
ElMessage.error('自主诊断失败: ' + e.message)
} finally {
agentRunning.value = false
}
}
function agentEventTitle(event) {
const titles = {
thinking: 'Agent 思考', command: '准备执行命令', result: '命令执行结果',
manual_command: '请手动执行命令',
approval_required: '发现修改命令,等待审批', command_rejected: '命令已被安全策略拒绝',
verification_required: '需要继续复检', response_rejected: 'AI 响应格式已自动纠正',
final: '自主诊断结论', error: 'Agent 执行异常',
}
return titles[event.type] || event.type
}
async function copyAgentCommand(command) {
try {
await navigator.clipboard.writeText(command)
ElMessage.success('命令已复制,请在终端中手动执行')
} catch (e) {
ElMessage.error('复制失败,请手动选择命令文本')
}
}
async function resolveApproval(event, approved) {
event.executing = true
try {
const resp = await fetch(`${API}/agent/approve`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ approval_id: event.approval_id, approved }),
})
if (!resp.ok) {
const error = await resp.json()
throw new Error(error.detail || `HTTP ${resp.status}`)
}
if (!approved) {
const result = await resp.json()
event.resolved = true
event.approvalSuccess = false
event.approvalResult = '已拒绝执行该命令'
ElMessage.info(event.approvalResult)
return
}
if (!resp.body) throw new Error('审批接口未返回流式响应')
event.resolved = true
agentRunning.value = true
const reader = resp.body.getReader()
const decoder = new TextDecoder()
let buffer = ''
while (true) {
const { done, value } = await reader.read()
if (done) break
buffer += decoder.decode(value, { stream: true })
const lines = buffer.split('\n')
buffer = lines.pop() || ''
for (const line of lines) {
if (!line.startsWith('data: ')) continue
const nextEvent = JSON.parse(line.slice(6))
if (nextEvent.type === 'approved_result') {
event.approvalSuccess = nextEvent.returncode === 0
event.approvalResult = nextEvent.returncode === 0 ? '命令执行成功,Agent 继续诊断中' : `命令执行失败:${nextEvent.stderr || nextEvent.returncode}`
agentEvents.value.push({ type: 'result', ...nextEvent })
} else if (!['start', 'done'].includes(nextEvent.type)) {
agentEvents.value.push(nextEvent)
if (nextEvent.type === 'error') ElMessage.error(nextEvent.message)
}
}
}
ElMessage.success('审批命令已执行,Agent 已继续诊断')
} catch (e) {
ElMessage.error('审批处理失败: ' + e.message)
} finally {
event.executing = false
agentRunning.value = false
}
}
async function sendChat() { async function sendChat() {
const q = chatInput.value.trim() const q = chatInput.value.trim()
if (!q || chatLoading.value) return if (!q || chatLoading.value) return
@@ -611,6 +910,34 @@ body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'PingFang SC'
.ai-content table { border-collapse: collapse; width: 100%; margin: 12px 0; } .ai-content table { border-collapse: collapse; width: 100%; margin: 12px 0; }
.ai-content th, .ai-content td { border: 1px solid #ebeef5; padding: 8px 12px; text-align: left; } .ai-content th, .ai-content td { border: 1px solid #ebeef5; padding: 8px 12px; text-align: left; }
.agent-card { border: 1px solid #f3d19e; }
.agent-mode-switch { flex-shrink: 0; }
.agent-alert { margin-bottom: 16px; }
.agent-timeline { display: flex; flex-direction: column; gap: 12px; }
.agent-event { padding: 14px; border: 1px solid #ebeef5; border-radius: 8px; background: #fafafa; }
.agent-event.event-thinking { border-color: #a0cfff; background: #ecf5ff; }
.agent-event.event-approval_required { border-color: #eebe77; background: #fdf6ec; }
.agent-event.event-final { border-color: #b3e19d; background: #f0f9eb; }
.agent-event-title { display: flex; align-items: center; gap: 8px; color: #303133; }
.agent-event-message { margin-top: 8px; color: #606266; font-size: 13px; }
.agent-command, .agent-output {
margin-top: 10px; padding: 12px; border-radius: 6px; overflow-x: auto;
white-space: pre-wrap; word-break: break-all; font-size: 13px;
}
.agent-command { color: #d4d4d4; background: #1e1e1e; }
.agent-output { max-height: 320px; color: #303133; background: #f4f4f5; }
.approval-actions { display: flex; gap: 8px; margin-top: 12px; }
.manual-actions { display: flex; align-items: center; gap: 12px; margin-top: 12px; color: #606266; font-size: 13px; }
.agent-running { display: flex; align-items: center; gap: 8px; color: #409eff; padding: 8px; }
.history-toolbar { display: flex; gap: 10px; margin-bottom: 16px; }
.history-detail { margin-top: 20px; }
.history-collapse { margin-top: 16px; }
.history-report { max-height: 480px; padding: 14px; overflow: auto; border-radius: 6px; background: #f4f4f5; white-space: pre-wrap; }
.history-event { margin-top: 10px; padding: 12px; border: 1px solid #ebeef5; border-radius: 6px; }
.history-event-head { display: flex; align-items: center; gap: 10px; color: #909399; font-size: 12px; }
.history-event pre { max-height: 300px; margin-top: 8px; padding: 10px; overflow: auto; background: #f4f4f5; white-space: pre-wrap; }
.chat-box { max-height: 400px; overflow-y: auto; padding: 12px 0; display: flex; flex-direction: column; gap: 12px; } .chat-box { max-height: 400px; overflow-y: auto; padding: 12px 0; display: flex; flex-direction: column; gap: 12px; }
.chat-msg { display: flex; } .chat-msg { display: flex; }
.chat-msg.user { justify-content: flex-end; } .chat-msg.user { justify-content: flex-end; }
+97
View File
@@ -0,0 +1,97 @@
import asyncio
import json
import unittest
from unittest.mock import patch
import backend.main as main
def parse_sse(event):
return json.loads(event[6:].strip())
class AgentApiTest(unittest.TestCase):
def test_agent_stream_forwards_agent_events(self):
async def fake_agent(report, question="", max_steps=6, execution_mode="ai"):
self.assertEqual(execution_mode, "manual")
yield {"type": "thinking", "step": 1, "message": "规划"}
yield {"type": "final", "content": "完成"}
async def run():
main._last_diagnosis = {"score": 60}
main._last_report = "report"
with patch.object(main, "run_diagnostic_agent", fake_agent):
response = await main.agent_diagnose_stream(question="检查 Pod", execution_mode="manual")
return [parse_sse(event) async for event in response.body_iterator]
events = asyncio.run(run())
self.assertEqual(events[0]["type"], "start")
self.assertEqual(events[-1]["type"], "done")
self.assertEqual(events[-2]["type"], "final")
def test_config_view_is_read_only_and_does_not_require_approval(self):
from backend.agent_tools import classify_kubectl_command
decision = classify_kubectl_command("kubectl config view")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_execute_approved_command_continues_agent_with_command_result(self):
approval = {
"command": "kubectl rollout restart deployment/api -n prod",
"reason": "重启工作负载",
"run_id": "",
"report": "diagnosis report",
"question": "修复 API",
"execution_mode": "ai",
"messages": [{"role": "system", "content": "context"}],
"next_step": 2,
"max_steps": 30,
"command_counts": {"kubectl rollout restart deployment/api -n prod": 1},
}
async def fake_execute(command):
return {"command": command, "mode": "write", "returncode": 0, "stdout": "restarted", "stderr": ""}
async def fake_resume(pending, result):
self.assertEqual(result["stdout"], "restarted")
yield {"type": "thinking", "step": 2, "message": "继续验证"}
yield {"type": "command", "command": "kubectl get pods -n prod", "mode": "read", "reason": "验证恢复"}
yield {"type": "result", "command": "kubectl get pods -n prod", "mode": "read", "returncode": 0, "stdout": "Running", "stderr": ""}
yield {"type": "final", "content": "故障已恢复", "model": "test"}
async def run():
with patch.object(main, "take_pending_approval", return_value=approval), patch.object(main, "execute_command", fake_execute), patch.object(main, "resume_diagnostic_agent", fake_resume):
response = await main.execute_approved_agent_command(main.AgentApprovalRequest(approval_id="abc", approved=True))
return [parse_sse(event) async for event in response.body_iterator]
events = asyncio.run(run())
self.assertEqual(events[0]["type"], "approved_result")
self.assertEqual(events[1]["type"], "thinking")
self.assertEqual(events[-2]["type"], "final")
self.assertEqual(events[-1]["type"], "done")
def test_execute_approved_command_once(self):
approval = {"command": "kubectl scale deployment/api --replicas=2", "reason": "恢复副本"}
async def fake_execute(command):
return {"command": command, "mode": "write", "returncode": 0, "stdout": "scaled", "stderr": ""}
async def fake_resume(pending, result):
yield {"type": "final", "content": "验证完成", "model": "test"}
async def run():
with patch.object(main, "take_pending_approval", return_value=approval), patch.object(main, "execute_command", fake_execute), patch.object(main, "resume_diagnostic_agent", fake_resume):
response = await main.execute_approved_agent_command(main.AgentApprovalRequest(approval_id="abc", approved=True))
return [parse_sse(event) async for event in response.body_iterator]
events = asyncio.run(run())
self.assertEqual(events[0]["type"], "approved_result")
self.assertEqual(events[0]["returncode"], 0)
self.assertEqual(events[0]["stdout"], "scaled")
self.assertEqual(events[-1]["type"], "done")
if __name__ == "__main__":
unittest.main()
+214
View File
@@ -0,0 +1,214 @@
import unittest
from backend.agent_tools import (
classify_agent_command,
classify_kubectl_command,
classify_shell_command,
parse_kubectl_command,
parse_shell_command,
)
class KubectlCommandPolicyTest(unittest.TestCase):
def test_allows_read_only_kubectl_command_for_autonomous_execution(self):
decision = classify_kubectl_command("kubectl get pods -n production -o wide")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_requires_approval_for_mutating_kubectl_command(self):
decision = classify_kubectl_command("kubectl rollout restart deployment/api -n production")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
def test_rejects_shell_operators_and_non_kubectl_commands(self):
# 武器级危险命令(sh/bash)仍被拒绝
with self.assertRaises(ValueError):
parse_kubectl_command("kubectl get pods | sh")
# 非 kubectl 命令仍被拒绝
for command in (
"bash -c kubectl get pods",
"sudo kubectl get pods",
):
with self.subTest(command=command):
with self.assertRaises(ValueError):
parse_kubectl_command(command)
def test_rejects_exec_even_when_command_looks_read_only(self):
decision = classify_kubectl_command("kubectl exec api-0 -- cat /etc/hosts")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
class ShellCommandPolicyTest(unittest.TestCase):
def test_allows_read_only_systemctl_status_for_autonomous_execution(self):
decision = classify_shell_command("systemctl status kubelet")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_allows_journalctl_with_flags(self):
decision = classify_shell_command("journalctl -u kubelet -n 100 --no-pager")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_allows_crictl_ps(self):
decision = classify_shell_command("crictl ps -a")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_allows_df_and_free(self):
for command in ("df -h", "free -m", "uptime"):
with self.subTest(command=command):
decision = classify_shell_command(command)
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_allows_ip_addr(self):
decision = classify_shell_command("ip addr show")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_requires_approval_for_systemctl_restart(self):
decision = classify_shell_command("systemctl restart kubelet")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
def test_rejects_unknown_binary(self):
# rm 现在在写白名单中(需审批),不会直接拒绝
decision = classify_shell_command("rm -rf /tmp/cache")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
# 完全不在任何白名单中的命令才拒绝
with self.assertRaises(ValueError):
classify_shell_command("hacktool -x")
def test_rejects_shell_operators_in_shell_command(self):
# 所有 shell 操作符(; | > && ||)现在都允许通过分类器
# 分号 ;
decision = classify_shell_command("systemctl status kubelet; df -h")
self.assertEqual(decision.mode, "read")
self.assertTrue(decision.needs_shell)
# 管道 |
decision = classify_shell_command("df -h | grep vda")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
self.assertTrue(decision.needs_shell)
# 重定向 >
decision = classify_shell_command("journalctl -u kubelet > /tmp/log")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
self.assertTrue(decision.needs_shell)
# 武器级 sh/bash 仍拒绝
with self.assertRaises(ValueError):
classify_shell_command("crictl ps -a | bash")
with self.assertRaises(ValueError):
classify_shell_command("crictl ps -a && sh -c 'echo x'")
def test_cat_only_allows_safe_paths(self):
# cat 完全放行,任意路径均可
decision = classify_shell_command("cat /etc/os-release")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_cat_rejects_unsafe_path(self):
# cat 不再限制路径
decision = classify_shell_command("cat /etc/shadow")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
class EtcdCommandPolicyTest(unittest.TestCase):
def test_etcdctl_endpoint_status_is_read_only(self):
decision = classify_shell_command("etcdctl endpoint status")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_etcdctl_with_global_flags_still_classifies(self):
decision = classify_shell_command(
"etcdctl --endpoints=https://127.0.0.1:2379 --cacert=/etc/ssl/ca.pem endpoint health"
)
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_etcdctl_member_list_is_read_only(self):
decision = classify_shell_command("etcdctl member list")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_etcdctl_get_and_version_are_read_only(self):
for command in ("etcdctl get /kubernetes", "etcdctl version", "etcdctl alarm list"):
with self.subTest(command=command):
decision = classify_shell_command(command)
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_etcdctl_alarm_list_is_read_only(self):
decision = classify_shell_command("etcdctl alarm list")
self.assertEqual(decision.mode, "read")
self.assertFalse(decision.requires_approval)
def test_etcdctl_put_requires_approval(self):
decision = classify_shell_command("etcdctl put key value")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
def test_etcdctl_member_remove_requires_approval(self):
decision = classify_shell_command("etcdctl member remove abc123")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
def test_etcdctl_snapshot_save_requires_approval(self):
decision = classify_shell_command("etcdctl snapshot save /tmp/etcd.db")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
def test_etcdctl_compact_requires_approval(self):
decision = classify_shell_command("etcdctl compact 12345")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
def test_etcdctl_auth_enable_requires_approval(self):
decision = classify_shell_command("etcdctl auth enable")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
def test_etcdctl_rejects_unknown_subcommand(self):
with self.assertRaises(ValueError):
classify_shell_command("etcdctl bogus command")
class AgentCommandDispatchTest(unittest.TestCase):
def test_dispatches_kubectl_command_to_kubectl_classifier(self):
decision = classify_agent_command("kubectl get pods -A")
self.assertEqual(decision.mode, "read")
self.assertEqual(decision.verb, "get")
def test_dispatches_shell_command_to_shell_classifier(self):
decision = classify_agent_command("systemctl status kubelet")
self.assertEqual(decision.mode, "read")
self.assertEqual(decision.verb, "systemctl")
def test_dispatches_write_shell_command_to_shell_classifier(self):
decision = classify_agent_command("systemctl restart kubelet")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
def test_dispatches_write_kubectl_command(self):
decision = classify_agent_command("kubectl delete pod api-0 -n prod")
self.assertEqual(decision.mode, "write")
self.assertTrue(decision.requires_approval)
if __name__ == "__main__":
unittest.main()
+328
View File
@@ -0,0 +1,328 @@
import asyncio
import unittest
from unittest.mock import patch
from backend.ai_agent import run_diagnostic_agent
async def collect_events(stream):
return [event async for event in stream]
class DiagnosticAgentTest(unittest.TestCase):
def test_executes_read_only_tool_and_returns_final_answer(self):
responses = iter([
'{"action":"command","command":"kubectl get pods -A","reason":"查看异常 Pod"}',
'{"action":"final","analysis":"发现一个异常 Pod","verified":true,"remaining_issues":0}',
])
async def fake_completion(messages):
return next(responses)
async def fake_execute(command):
return {"command": command, "mode": "read", "returncode": 0, "stdout": "pod-a Running", "stderr": ""}
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
return await collect_events(run_diagnostic_agent("report", max_steps=3))
events = asyncio.run(run())
self.assertEqual(
[event["type"] for event in events],
["command", "result", "thinking", "command", "result", "thinking", "final"],
)
self.assertEqual(events[3]["mode"], "read")
self.assertEqual(events[-1]["content"], "发现一个异常 Pod")
def test_pauses_mutating_command_for_human_approval(self):
async def fake_completion(messages):
return '{"action":"command","command":"kubectl rollout restart deployment/api -n prod","reason":"重启故障工作负载"}'
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion):
return await collect_events(run_diagnostic_agent("report", max_steps=3))
events = asyncio.run(run())
self.assertEqual(events[-1]["type"], "approval_required")
self.assertEqual(events[-1]["command"], "kubectl rollout restart deployment/api -n prod")
self.assertTrue(events[-1]["approval_id"])
def test_manual_mode_never_executes_command_and_waits_for_result(self):
async def fake_completion(messages):
return '{"action":"command","command":"kubectl get pods -A","reason":"查看 Pod"}'
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command") as execute:
events = await collect_events(run_diagnostic_agent("report", max_steps=3, execution_mode="manual"))
return events, execute
events, execute = asyncio.run(run())
execute.assert_not_called()
self.assertEqual(events[-1]["type"], "manual_command")
self.assertEqual(events[-1]["command"], "kubectl get pods -A")
self.assertTrue(events[-1]["command_id"])
def test_ai_mode_keeps_write_command_behind_approval(self):
async def fake_completion(messages):
return '{"action":"command","command":"kubectl delete pod api-0 -n prod","reason":"删除故障 Pod"}'
executed = []
async def run():
async def fake_execute(command):
executed.append(command)
return {"command": command, "mode": "read", "returncode": 0, "stdout": "nodes ready", "stderr": ""}
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
return await collect_events(run_diagnostic_agent("report", max_steps=3, execution_mode="ai"))
events = asyncio.run(run())
self.assertEqual(executed, ["kubectl get nodes -o wide"])
self.assertEqual(events[-1]["type"], "approval_required")
def test_rejects_unknown_execution_mode(self):
async def run():
return await collect_events(run_diagnostic_agent("report", execution_mode="invalid"))
events = asyncio.run(run())
self.assertEqual(events[-1]["type"], "error")
self.assertIn("执行模式", events[-1]["message"])
def test_ai_mode_forces_initial_read_only_probe_before_model_planning(self):
responses = iter([
'{"action":"final","analysis":"根据采集结果完成分析","verified":true,"remaining_issues":0}',
])
async def fake_completion(messages):
return next(responses)
executed = []
async def fake_execute(command):
executed.append(command)
return {"command": command, "mode": "read", "returncode": 0, "stdout": "node-a Ready", "stderr": ""}
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
return await collect_events(run_diagnostic_agent("report", max_steps=3, execution_mode="ai"))
events = asyncio.run(run())
self.assertEqual(executed, ["kubectl get nodes -o wide"])
self.assertEqual([event["type"] for event in events[:2]], ["command", "result"])
self.assertEqual(events[-1]["type"], "final")
def test_parse_agent_action_accepts_markdown_json_fence(self):
from backend.ai_agent import _parse_agent_action
action = _parse_agent_action(
'我将先检查 Pod。\n```json\n{"action":"command","command":"kubectl get pods -A","reason":"查看 Pod"}\n```'
)
self.assertEqual(action["action"], "command")
self.assertEqual(action["command"], "kubectl get pods -A")
def test_ai_mode_continues_beyond_six_commands_until_model_finishes(self):
command_count = 8
responses = iter([
*[
(
'{"action":"command","command":"kubectl get pods -A '
f'--field-selector metadata.name=pod-{index}","reason":"继续收集证据"}}'
)
for index in range(command_count)
],
'{"action":"final","analysis":"已完成根因分析","verified":true,"remaining_issues":0}',
])
executed = []
async def fake_completion(messages):
return next(responses)
async def fake_execute(command):
executed.append(command)
return {"command": command, "mode": "read", "returncode": 0, "stdout": "evidence", "stderr": ""}
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=20))
events = asyncio.run(run())
self.assertEqual(len(executed), command_count + 1)
self.assertEqual(events[-1]["type"], "final")
self.assertEqual(events[-1]["content"], "已完成根因分析")
self.assertNotIn("最大轮次", events[-1]["content"])
def test_safety_limit_reports_agent_error_instead_of_sending_user_to_manual_diagnosis(self):
async def fake_completion(messages):
return '{"action":"command","command":"kubectl get pods -A","reason":"继续调查"}'
async def fake_execute(command):
return {"command": command, "mode": "read", "returncode": 0, "stdout": "same result", "stderr": ""}
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=2))
events = asyncio.run(run())
self.assertEqual(events[-1]["type"], "error")
self.assertIn("安全上限", events[-1]["message"])
self.assertNotIn("人工排查", events[-1]["message"])
def test_agent_rejects_final_until_verification_reports_no_remaining_issues(self):
responses = iter([
'{"action":"final","analysis":"已经修复"}',
'{"action":"command","command":"kubectl get pods -A","reason":"执行全量复检"}',
'{"action":"final","analysis":"所有工作负载恢复正常","verified":true,"remaining_issues":0}',
])
executed = []
async def fake_completion(messages):
return next(responses)
async def fake_execute(command):
executed.append(command)
stdout = "node-a Ready" if command == "kubectl get nodes -o wide" else "all pods Running"
return {"command": command, "mode": "read", "returncode": 0, "stdout": stdout, "stderr": ""}
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=5))
events = asyncio.run(run())
self.assertEqual(executed, ["kubectl get nodes -o wide", "kubectl get pods -A"])
self.assertEqual(events[-1]["type"], "final")
self.assertTrue(events[-1]["verified"])
self.assertEqual(events[-1]["remaining_issues"], 0)
def test_parse_final_requires_explicit_verification_fields(self):
from backend.ai_agent import _parse_agent_action
with self.assertRaises(ValueError):
_parse_agent_action('{"action":"final","analysis":"看起来正常"}')
action = _parse_agent_action(
'{"action":"final","analysis":"全部正常","verified":true,"remaining_issues":0}'
)
self.assertTrue(action["verified"])
self.assertEqual(action["remaining_issues"], 0)
def test_invalid_json_response_is_retried_instead_of_stopping_agent(self):
responses = iter([
"我需要继续检查,但没有按格式输出",
'{"action":"command","command":"kubectl get pods -A","reason":"继续检查"}',
'{"action":"final","analysis":"检查完成","verified":true,"remaining_issues":0}',
])
executed = []
async def fake_completion(messages):
return next(responses)
async def fake_execute(command):
executed.append(command)
return {"command": command, "mode": "read", "returncode": 0, "stdout": "healthy", "stderr": ""}
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
return await collect_events(run_diagnostic_agent("report", max_steps=5))
events = asyncio.run(run())
self.assertTrue(any(event["type"] == "response_rejected" for event in events))
self.assertEqual(events[-1]["type"], "final")
self.assertEqual(executed, ["kubectl get nodes -o wide", "kubectl get pods -A"])
def test_stops_after_three_consecutive_invalid_json_responses(self):
async def fake_completion(messages):
return "still invalid"
async def fake_execute(command):
return {"command": command, "mode": "read", "returncode": 0, "stdout": "node Ready", "stderr": ""}
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
return await collect_events(run_diagnostic_agent("report", max_steps=10))
events = asyncio.run(run())
self.assertEqual(events[-1]["type"], "error")
self.assertIn("连续 3 次", events[-1]["message"])
def test_stops_when_model_returns_invalid_action(self):
async def fake_completion(messages):
return "not-json"
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion):
return await collect_events(run_diagnostic_agent("report", max_steps=3))
events = asyncio.run(run())
self.assertEqual(events[-1]["type"], "error")
self.assertIn("JSON", events[-1]["message"])
def test_agent_can_mix_kubectl_and_shell_commands(self):
"""Agent 应能在同一轮诊断中交替使用 kubectl 和系统诊断 shell 命令。"""
responses = iter([
'{"action":"command","command":"kubectl get nodes -o wide","reason":"查看节点状态"}',
'{"action":"command","command":"systemctl status kubelet","reason":"检查 kubelet 服务"}',
'{"action":"command","command":"journalctl -u kubelet -n 50 --no-pager","reason":"查看 kubelet 日志"}',
'{"action":"final","analysis":"kubelet 服务正常,节点就绪","verified":true,"remaining_issues":0}',
])
executed = []
async def fake_completion(messages):
return next(responses)
async def fake_execute(command):
executed.append(command)
if command.startswith("kubectl"):
stdout = "node-a Ready"
elif command.startswith("systemctl"):
stdout = "active (running)"
else:
stdout = "no errors"
return {"command": command, "mode": "read", "returncode": 0, "stdout": stdout, "stderr": ""}
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=10))
events = asyncio.run(run())
# 初始探测 + 3 条 Agent 命令 = 4 次执行
self.assertEqual(
executed,
["kubectl get nodes -o wide", "kubectl get nodes -o wide", "systemctl status kubelet", "journalctl -u kubelet -n 50 --no-pager"],
)
# 所有命令都是只读模式 (自动执行)
command_events = [e for e in events if e["type"] == "command"]
self.assertTrue(all(e["mode"] == "read" for e in command_events))
self.assertEqual(events[-1]["type"], "final")
def test_agent_rejects_non_whitelisted_shell_command(self):
"""Agent 应拒绝不在白名单内的 shell 命令 (如 rm)。"""
responses = iter([
'{"action":"command","command":"rm -rf /tmp/test","reason":"清理测试文件"}',
'{"action":"final","analysis":"已完成","verified":true,"remaining_issues":0}',
])
executed = []
async def fake_completion(messages):
return next(responses)
async def fake_execute(command):
executed.append(command)
return {"command": command, "mode": "read", "returncode": 0, "stdout": "", "stderr": ""}
async def run():
with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=5))
events = asyncio.run(run())
# rm 命令应被拒绝
rejected = [e for e in events if e["type"] == "command_rejected"]
self.assertTrue(rejected)
self.assertIn("不允许执行该命令", rejected[0]["reason"])
# rm 不应被执行
self.assertNotIn("rm -rf /tmp/test", executed)
self.assertEqual(events[-1]["type"], "final")
if __name__ == "__main__":
unittest.main()
+43
View File
@@ -0,0 +1,43 @@
import asyncio
import os
import tempfile
import unittest
from unittest.mock import patch
import backend.main as main
from backend.history_store import HistoryStore
class HistoryApiTest(unittest.TestCase):
def setUp(self):
fd, self.path = tempfile.mkstemp(suffix=".db")
os.close(fd)
self.store = HistoryStore(self.path)
self.original_store = main.history_store
main.history_store = self.store
def tearDown(self):
main.history_store = self.original_store
os.remove(self.path)
def test_lists_and_gets_full_diagnosis_history(self):
record_id = self.store.save_diagnosis(
{"timestamp": "2026-07-25T12:00:00", "namespace": "prod", "score": 50, "status": "critical"},
"report",
)
records = asyncio.run(main.list_diagnosis_history())
detail = asyncio.run(main.get_diagnosis_history(record_id))
self.assertEqual(records["total"], 1)
self.assertEqual(records["items"][0]["id"], record_id)
self.assertEqual(detail["report"], "report")
def test_get_history_returns_404_for_unknown_id(self):
with self.assertRaises(main.HTTPException) as context:
asyncio.run(main.get_diagnosis_history("missing"))
self.assertEqual(context.exception.status_code, 404)
if __name__ == "__main__":
unittest.main()
+69
View File
@@ -0,0 +1,69 @@
import os
import tempfile
import unittest
from backend.history_store import HistoryStore
class HistoryStoreTest(unittest.TestCase):
def setUp(self):
fd, self.path = tempfile.mkstemp(suffix=".db")
os.close(fd)
self.store = HistoryStore(self.path)
def tearDown(self):
os.remove(self.path)
def test_persists_diagnosis_and_can_query_after_reopen(self):
diagnosis = {
"timestamp": "2026-07-25T12:00:00",
"namespace": "prod",
"score": 65,
"status": "warning",
"critical_count": 1,
"warning_count": 4,
"issues": [{"level": "critical", "message": "pod failed"}],
"checks": {},
}
record_id = self.store.save_diagnosis(diagnosis, "report text")
reopened = HistoryStore(self.path)
records = reopened.list_diagnoses()
detail = reopened.get_diagnosis(record_id)
self.assertEqual(len(records), 1)
self.assertEqual(records[0]["namespace"], "prod")
self.assertEqual(detail["diagnosis"]["score"], 65)
self.assertEqual(detail["report"], "report text")
def test_records_all_processing_events_in_order(self):
diagnosis_id = self.store.save_diagnosis(
{"timestamp": "2026-07-25T12:00:00", "namespace": "all", "score": 80, "status": "warning"},
"report",
)
run_id = self.store.create_run(diagnosis_id, "agent", "ai", "check pods")
self.store.append_event(run_id, "command", {"command": "kubectl get pods -A"})
self.store.append_event(run_id, "result", {"returncode": 0, "stdout": "ok"})
self.store.finish_run(run_id, "completed", "done")
detail = self.store.get_diagnosis(diagnosis_id)
self.assertEqual(len(detail["runs"]), 1)
self.assertEqual(detail["runs"][0]["status"], "completed")
self.assertEqual([event["type"] for event in detail["runs"][0]["events"]], ["command", "result"])
def test_lists_all_diagnoses_newest_first_and_filters_status(self):
first = {"timestamp": "2026-07-25T11:00:00", "namespace": "dev", "score": 100, "status": "healthy"}
second = {"timestamp": "2026-07-25T12:00:00", "namespace": "prod", "score": 20, "status": "critical"}
self.store.save_diagnosis(first, "first")
self.store.save_diagnosis(second, "second")
all_records = self.store.list_diagnoses()
critical = self.store.list_diagnoses(status="critical")
self.assertEqual([item["namespace"] for item in all_records], ["prod", "dev"])
self.assertEqual([item["namespace"] for item in critical], ["prod"])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,765 @@
# Kubernetes 集群故障分析报告
**报告时间**: 2026-07-25T07:36:59
**集群健康评分**: 0/100 (Critical)
**问题严重程度**: 4个严重问题, 30个警告问题
------
## 一、故障概述
本次集群故障表现为严重的控制平面组件异常,导致集群处于不可用状态。主要症状包括:
- **etcd** (集群核心数据库) 完全不可用,两个节点均处于CrashLoopBackOff状态
- **kube-apiserver** (API服务器) 三个节点全部异常,无法响应请求
- **kube-scheduler** 和 **kube-controller-manager** 也出现不稳定状态
- 整个集群基本功能瘫痪,无法处理任何工作负载部署请求
------
## 二、根本原因分析 (Root Cause Analysis)
### 🔴 **问题1: etcd集群完全崩溃**
**根本原因**: etcd集群的持久化存储或网络配置出现问题
- 两个etcd节点 (ansible-client-01/02) 均出现503状态码错误
- 高频次重启 (38次/30次) 表明容器启动后立即失败
- 启动探针失败返回503,表明etcd服务无法正常初始化
**可能的技术原因**:
1. **存储问题**: etcd数据目录损坏或磁盘空间不足
2. **网络分区**: etcd节点之间无法建立集群通信
3. **证书过期**: etcd的TLS证书可能已过期
4. **资源不足**: etcd容器内存或CPU资源不足
### 🔴 **问题2: kube-apiserver无法启动**
**根本原因**: kube-apiserver依赖的etcd集群不可用
- 所有三个apiserver节点 (ansible-client-01/02, ubuntu2204) 都失败
- 错误信息显示"connection reset by peer"和"connection refused"
- 启动探针返回500错误,表明服务无法初始化
**依赖链分析**:
```
etcd不可用 → apiserver无法连接etcd → apiserver启动失败 → 其他组件依赖apiserver失败
```
### 🟡 **问题3: kube-scheduler/kube-controller-manager不稳定**
**根本原因**: 由于apiserver不可用,这些组件无法正常工作
- 依赖apiserver获取集群状态和调度决策
- 在等待apiserver恢复过程中进入BackOff状态
### 🟡 **问题4: 节点存储异常**
**根本原因**: ubuntu2204节点的Docker容器存储驱动配置错误
- "invalid capacity 0 on image filesystem"错误
- 容器运行时无法正确计算存储空间
------
## 三、影响评估
### 🔴 **业务影响 (Critical)**
1. **集群完全不可用**: 无法部署、更新或管理任何工作负载
2. **现有应用可能受影响**: 依赖Kubernetes API的所有服务将失败
3. **数据丢失风险**: etcd是集群状态存储,如果损坏会导致元数据丢失
4. **恢复时间延长**: 控制平面完全崩溃需要完整恢复流程
### 🟡 **运维影响 (High)**
1. **无法进行任何kubectl操作**: 所有API请求失败
2. **自动修复失效**: HPA、自愈等机制无法工作
3. **监控告警风暴**: 持续产生大量错误事件
------
## 四、优先级排序与修复方案
### **优先级1: 恢复etcd集群 (最紧急)**
**问题**: etcd集群是Kubernetes的基础,必须首先恢复
**修复步骤**:
#### 步骤1: 检查etcd节点状态
```bash
# SSH到ansible-client-01节点
ssh root@<ansible-client-01-ip>
# 检查etcd容器日志
docker logs etcd-ansible-client-01 --tail 100
# 检查etcd端点状态
ETCDCTL_API=3 etcdctl endpoint health \
--endpoints=https://192.168.40.11:2379,https://192.168.40.12:2379 \
--cacert=/etc/kubernetes/pki/etcd/ca.crt \
--cert=/etc/kubernetes/pki/etcd/server.crt \
--key=/etc/kubernetes/pki/etcd/server.key
```
#### 步骤2: 检查存储和网络
```bash
# 检查磁盘空间
df -h /var/lib/etcd
# 检查网络连接
telnet 192.168.40.11 2379
telnet 192.168.40.12 2380 # etcd peer端口
# 检查证书是否过期
openssl x509 -in /etc/kubernetes/pki/etcd/server.crt -text -noout | grep "Not After"
```
#### 步骤3: 修复方案
**方案A: 如果存储损坏**
```bash
# 备份现有数据目录
mv /var/lib/etcd /var/lib/etcd.backup.$(date +%s)
# 创建新的etcd数据目录
mkdir -p /var/lib/etcd
chown -R etcd:etcd /var/lib/etcd
# 重启etcd服务
systemctl restart etcd
```
**方案B: 如果证书过期**
```bash
# 重新生成etcd证书
kubeadm init phase certs etcd-server
kubeadm init phase certs etcd-peer
kubeadm init phase certs etcd-healthcheck-client
# 更新kube-apiserver的etcd客户端证书
kubeadm init phase certs apiserver-etcd-client
# 重启etcd和kube-apiserver
systemctl restart etcd
systemctl restart kubelet
```
**方案C: 完全重建etcd集群 (最后手段)**
```bash
# 在其中一个节点重建etcd集群
# 1. 停止所有etcd实例
systemctl stop etcd
# 2. 备份数据
mv /var/lib/etcd /var/lib/etcd.bak
# 3. 重新初始化集群
ETCDCTL_API=3 etcdctl member remove <member-id>
ETCDCTL_API=3 etcdctl member add ansible-client-01 --peer-urls=https://192.168.40.11:2380
# 4. 启动新的单节点etcd
etcd --name ansible-client-01 \
--initial-advertise-peer-urls https://192.168.40.11:2380 \
--listen-peer-urls https://192.168.40.11:2380 \
--advertise-client-urls https://192.168.40.11:2379 \
--listen-client-urls https://192.168.40.11:2379,https://127.0.0.1:2379 \
--initial-cluster-token etcd-cluster-01 \
--initial-cluster ansible-client-01=https://192.168.40.11:2380 \
--initial-cluster-state new
```
### **优先级2: 恢复kube-apiserver**
**前提**: etcd集群已恢复
**修复步骤**:
#### 步骤1: 检查apiserver配置
```bash
# 检查kube-apiserver Pod规格
kubectl get pod kube-apiserver-ansible-client-01 -n kube-system -o yaml
# 检查启动命令参数
kubectl get pod kube-apiserver-ansible-client-01 -n kube-system -o jsonpath='{.spec.containers[0].command}'
```
#### 步骤2: 检查资源限制
```bash
# 查看apiserver的资源使用
kubectl top pod kube-apiserver-ansible-client-01 -n kube-system
# 检查节点资源
kubectl describe node ansible-client-01
```
#### 步骤3: 修复方案
**增加启动探针超时时间** (临时方案):
```yaml
# 编辑kube-apiserver Pod规范
kubectl edit pod kube-apiserver-ansible-client-01 -n kube-system
# 添加或修改以下字段:
spec:
containers:
- name: kube-apiserver
startupProbe:
failureThreshold: 60 # 增加失败阈值
periodSeconds: 5 # 增加检查周期
timeoutSeconds: 15 # 增加超时时间
```
**检查并修复资源限制**:
```yaml
# 确保有足够的资源
resources:
requests:
cpu: "500m"
memory: "2Gi"
limits:
cpu: "2000m"
memory: "4Gi"
```
### **优先级3: 恢复kube-controller-manager和kube-scheduler**
**修复步骤**:
```bash
# 检查这些组件的依赖
kubectl get event -n kube-system --field-selector reason=BackOff
# 重启这些组件 (如果etcd和apiserver已恢复)
systemctl restart kubelet
```
### **优先级4: 修复节点存储问题**
**问题**: ubuntu2204节点的Docker存储驱动配置错误
**修复步骤**:
```bash
# SSH到ubuntu2204节点
ssh root@<ubuntu2204-ip>
# 检查Docker存储驱动
docker info | grep "Storage Driver"
# 检查存储空间
df -h /var/lib/docker
# 修复方案1: 重启Docker服务
systemctl restart docker
# 修复方案2: 如果存储驱动配置错误
# 编辑Docker配置文件
vi /etc/docker/daemon.json
# 确保包含正确的storage-driver配置,例如:
{
"storage-driver": "overlay2"
}
# 重启Docker
systemctl restart docker
systemctl restart kubelet
```
------
## 五、验证修复
### 步骤1: 验证etcd健康状态
```bash
# 检查etcd集群状态
ETCDCTL_API=3 etcdctl endpoint health \
--endpoints=https://192.168.40.11:2379,https://192.168.40.12:2379 \
--cacert=/etc/kubernetes/pki/etcd/ca.crt \
--cert=/etc/kubernetes/pki/etcd/server.crt \
--key=/etc/kubernetes/pki/etcd/server.key
# 检查etcd集群成员列表
ETCDCTL_API=3 etcdctl member list \
--endpoints=https://192.168.40.11:2379 \
--cacert=/etc/kubernetes/pki/etcd/ca.crt \
--cert=/etc/kubernetes/pki/etcd/server.crt \
--key=/etc/kubernetes/pki/etcd/server.key
```
### 步骤2: 验证kube-apiserver状态
```bash
# 检查apiserver Pod状态
kubectl get pods -n kube-system | grep kube-apiserver
# 检查apiserver健康状态
kubectl get --raw='/readyz?verbose'
# 检查集群状态
kubectl cluster-info
kubectl get nodes
```
### 步骤3: 运行完整集群诊断
```bash
# 使用kubectl检查组件状态
kubectl get componentstatuses
# 检查所有系统Pod
kubectl get pods -n kube-system
# 检查集群事件
kubectl get events -n kube-system --sort-by='.lastTimestamp'
```
------
## 六、预防措施
### 1. **监控告警增强**
```yaml
# 添加以下Prometheus告警规则
groups:
- name: kubernetes-system
rules:
- alert: EtcdMembersDown
expr: etcd_server_has_leader{job="etcd"} == 0
for: 2m
labels:
severity: critical
annotations:
summary: "Etcd member has no leader"
- alert: EtcdHighCommitDuration
expr: etcd_disk_wal_fsync_duration_seconds{job="etcd"} > 1
for: 10m
labels:
severity: warning
```
### 2. **定期维护任务**
- **每周**: 检查etcd数据库大小,执行碎片整理
```bash
etcdctl defrag --endpoints=<endpoint>
```
- **每月**: 检查证书过期时间
```bash
kubeadm certs check-expiration
```
- **每季度**: 演练etcd备份恢复流程
### 3. **资源规划和配置优化**
- 为etcd分配专用存储和高性能磁盘
- 设置合适的资源限制和请求
- 启用etcd自动备份和远程存储
### 4. **网络策略优化**
- 确保etcd节点间网络延迟<10ms
- 配置专用网络段给控制平面组件
- 启用网络策略限制不必要的流量
### 5. **升级和补丁管理**
```bash
# 定期升级Kubernetes版本
kubeadm upgrade plan
kubeadm upgrade apply
# 及时应用安全补丁
apt update && apt upgrade
```
### 6. **文档和演练**
- 维护详细的集群架构文档
- 定期进行灾难恢复演练
- 建立标准化的操作手册
------
## 七、附录
### 附录A: 关键命令参考
```bash
# etcd健康检查
ETCDCTL_API=3 etcdctl endpoint health --endpoints=<endpoint> --cacert=<ca> --cert=<cert> --key=<key>
# etcd备份
ETCDCTL_API=3 etcdctl snapshot save /backup/etcd-snapshot.db
# 集群证书检查
kubeadm certs check-expiration
# 节点资源检查
kubectl describe node <node-name>
```
### 附录B: 相关文件位置
- etcd数据目录: `/var/lib/etcd/`
- Kubernetes证书: `/etc/kubernetes/pki/`
- etcd证书: `/etc/kubernetes/pki/etcd/`
- kubelet配置: `/var/lib/kubelet/config.yaml`
### 附录C: 重要端口列表
| 组件 | 端口 | 协议 | 用途 |
|------|
继续
### 附录C: 重要端口列表
| 组件 | 端口 | 协议 | 用途 |
| ----------------------- | ----- | ----- | ---------------------- |
| kube-apiserver | 6443 | HTTPS | API服务器安全端口 |
| etcd | 2379 | HTTPS | etcd客户端通信端口 |
| etcd | 2380 | HTTPS | etcd集群节点间通信端口 |
| kubelet | 10250 | HTTPS | kubelet API端口 |
| kube-scheduler | 10259 | HTTPS | 调度器端口 |
| kube-controller-manager | 10257 | HTTPS | 控制器管理器端口 |
------
### 附录D: 本次故障事件时间线(还原)
根据诊断报告中的重启次数和事件记录,推断出以下故障演进时间线:
```
时间线还原 (推测)
阶段1: etcd 首先出现异常
├── ansible-client-01 的 etcd 容器开始 CrashLoop
├── ansible-client-02 的 etcd 容器开始 CrashLoop
├── 启动探针返回 HTTP 503
└── etcd 集群丧失 quorum(多数派共识)
阶段2: kube-apiserver 因 etcd 不可用而连锁失败
├── ansible-client-01 apiserver: connection reset by peer
├── ansible-client-02 apiserver: connection refused
├── ubuntu2204 apiserver: HTTP 500 错误
└── 所有 apiserver 进入 CrashLoopBackOff
阶段3: 控制平面其他组件受波及
├── kube-controller-manager 重启 6 次后进入 BackOff
├── kube-scheduler 重启 7 次后进入 BackOff
└── TLS handshake timeout(因 apiserver 不可用)
阶段4: 业务层面影响显现
├── ubuntu2204 节点报告 InvalidDiskCapacity
├── 所有 kubectl 操作失败
├── 集群健康评分降至 0/100
└── 集群完全不可用
```
------
### 附录E: etcd 数据备份与恢复 SOP(标准操作流程)
#### E.1 etcd 定期备份脚本
```bash
#!/bin/bash
# etcd-backup.sh - 定期备份 etcd 数据
# 建议通过 cron 每小时执行一次
BACKUP_DIR="/backup/etcd"
DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_FILE="${BACKUP_DIR}/etcd-snapshot-${DATE}.db"
ETCD_ENDPOINT="https://192.168.40.11:2379"
ETCD_CACERT="/etc/kubernetes/pki/etcd/ca.crt"
ETCD_CERT="/etc/kubernetes/pki/etcd/server.crt"
ETCD_KEY="/etc/kubernetes/pki/etcd/server.key"
# 创建备份目录
mkdir -p ${BACKUP_DIR}
# 执行备份
ETCDCTL_API=3 etcdctl snapshot save ${BACKUP_FILE} \
--endpoints=${ETCD_ENDPOINT} \
--cacert=${ETCD_CACERT} \
--cert=${ETCD_CERT} \
--key=${ETCD_KEY}
# 验证备份
ETCDCTL_API=3 etcdctl snapshot status ${BACKUP_FILE} --write-table
# 保留最近 30 个备份
ls -t ${BACKUP_DIR}/etcd-snapshot-*.db | tail -n +31 | xargs rm -f
echo "[${DATE}] etcd backup completed: ${BACKUP_FILE}"
```
#### E.2 etcd 灾难恢复流程
```bash
#!/bin/bash
# etcd-restore.sh - 从备份恢复 etcd
SNAPSHOT_FILE=$1 # 备份文件路径
ETCD_DATA_DIR="/var/lib/etcd"
ETCD_NAME="ansible-client-01"
ETCD_INITIAL_CLUSTER="ansible-client-01=https://192.168.40.11:2380,ansible-client-02=https://192.168.40.12:2380"
if [ -z "${SNAPSHOT_FILE}" ]; then
echo "Usage: $0 <snapshot-file>"
exit 1
fi
echo "=== Step 1: 停止所有 etcd 实例 ==="
systemctl stop etcd
sleep 5
echo "=== Step 2: 备份当前数据目录 ==="
if [ -d "${ETCD_DATA_DIR}" ]; then
mv ${ETCD_DATA_DIR} ${ETCD_DATA_DIR}.bak.$(date +%s)
fi
echo "=== Step 3: 从快照恢复数据 ==="
ETCDCTL_API=3 etcdctl snapshot restore ${SNAPSHOT_FILE} \
--name ${ETCD_NAME} \
--initial-cluster ${ETCD_INITIAL_CLUSTER} \
--initial-advertise-peer-urls https://192.168.40.11:2380 \
--data-dir ${ETCD_DATA_DIR}
echo "=== Step 4: 设置权限 ==="
chown -R etcd:etcd ${ETCD_DATA_DIR}
echo "=== Step 5: 启动 etcd ==="
systemctl start etcd
echo "=== Step 6: 验证恢复结果 ==="
sleep 10
ETCDCTL_API=3 etcdctl endpoint health \
--endpoints=https://192.168.40.11:2379 \
--cacert=/etc/kubernetes/pki/etcd/ca.crt \
--cert=/etc/kubernetes/pki/etcd/server.crt \
--key=/etc/kubernetes/pki/etcd/server.key
ETCDCTL_API=3 etcdctl member list \
--endpoints=https://192.168.40.11:2379 \
--cacert=/etc/kubernetes/pki/etcd/ca.crt \
--cert=/etc/kubernetes/pki/etcd/server.crt \
--key=/etc/kubernetes/pki/etcd/server.key \
--write-table
echo "=== 恢复完成 ==="
```
#### E.3 etcd 快照恢复后恢复集群其余节点
```bash
#!/bin/bash
# 在恢复第一个 etcd 节点成功后,将其他节点加入集群
# Step 1: 在已恢复的节点上,移除旧成员
ETCDCTL_API=3 etcdctl member remove <old-member-id> \
--endpoints=https://192.168.40.11:2379 \
--cacert=/etc/kubernetes/pki/etcd/ca.crt \
--cert=/etc/kubernetes/pki/etcd/server.crt \
--key=/etc/kubernetes/pki/etcd/server.key
# Step 2: 添加新成员
ETCDCTL_API=3 etcdctl member add ansible-client-02 \
--peer-urls=https://192.168.40.12:2380 \
--endpoints=https://192.168.40.11:2379 \
--cacert=/etc/kubernetes/pki/etcd/ca.crt \
--cert=/etc/kubernetes/pki/etcd/server.crt \
--key=/etc/kubernetes/pki/etcd/server.key
# Step 3: SSH到 ansible-client-02,清理数据目录并启动
# ssh root@192.168.40.12
# rm -rf /var/lib/etcd/*
# systemctl restart etcd
# Step 4: 验证集群成员
ETCDCTL_API=3 etcdctl member list \
--endpoints=https://192.168.40.11:2379,https://192.168.40.12:2379 \
--cacert=/etc/kubernetes/pki/etcd/ca.crt \
--cert=/etc/kubernetes/pki/etcd/server.crt \
--key=/etc/kubernetes/pki/etcd/server.key \
--write-table
```
------
### 附录F: 推荐的 CronJob 定时备份配置
```yaml
apiVersion: batch/v1
kind: CronJob
metadata:
name: etcd-backup
namespace: kube-system
spec:
schedule: "0 */1 * * *" # 每小时执行一次
concurrencyPolicy: Forbid
successfulJobsHistoryLimit: 3
failedJobsHistoryLimit: 3
jobTemplate:
spec:
template:
spec:
nodeSelector:
kubernetes.io/hostname: ansible-client-01
hostNetwork: true
tolerations:
- key: node-role.kubernetes.io/master
effect: NoSchedule
containers:
- name: etcd-backup
image: bitnami/etcd:3.5.0
command:
- /bin/sh
- -c
- |
DATE=$(date +%Y%m%d_%H%M%S)
ETCDCTL_API=3 etcdctl snapshot save /backup/etcd-snapshot-${DATE}.db \
--endpoints=https://127.0.0.1:2379 \
--cacert=/etc/kubernetes/pki/etcd/ca.crt \
--cert=/etc/kubernetes/pki/etcd/server.crt \
--key=/etc/kubernetes/pki/etcd/server.key
echo "Backup completed: etcd-snapshot-${DATE}.db"
env:
- name: ETCDCTL_API
value: "3"
volumeMounts:
- name: etcd-certs
mountPath: /etc/kubernetes/pki/etcd
readOnly: true
- name: backup-dir
mountPath: /backup
restartPolicy: OnFailure
volumes:
- name: etcd-certs
hostPath:
path: /etc/kubernetes/pki/etcd
type: DirectoryOrCreate
- name: backup-dir
hostPath:
path: /backup/etcd
type: DirectoryOrCreate
```
------
### 附录G: 集群健康巡检 Checklist
| 序号 | 检查项 | 命令 | 正常状态 | 检查频率 |
| ---- | -------------------- | ------------------------------------------------------------ | ---------------- | -------- |
| 1 | etcd 集群健康 | `etcdctl endpoint health` | 所有节点 healthy | 每5分钟 |
| 2 | etcd 数据库大小 | `etcdctl endpoint status --write-table` | < 2GB | 每小时 |
| 3 | etcd 磁盘 fsync 延迟 | Prometheus: `etcd_disk_wal_fsync_duration_seconds` | < 10ms | 实时 |
| 4 | apiserver 可用性 | `kubectl get --raw='/readyz?verbose'` | 全部 ok | 每分钟 |
| 5 | 证书过期时间 | `kubeadm certs check-expiration` | > 30天 | 每周 |
| 6 | 节点资源使用率 | `kubectl top nodes` | CPU<80%, Mem<80% | 每小时 |
| 7 | 系统 Pod 状态 | `kubectl get pods -n kube-system` | 全部 Running | 每5分钟 |
| 8 | PVC 使用率 | `kubectl get pvc -A` | < 80% | 每小时 |
| 9 | 控制平面组件重启次数 | `kubectl get pods -n kube-system -o wide` | 重启次数=0 | 每小时 |
| 10 | 集群事件异常 | `kubectl get events -n kube-system --field-selector type=Warning` | 无新增 Warning | 每小时 |
------
### 附录H: 紧急联系人和应急流程
```
┌─────────────────────────────────────────────────────────┐
│ 集群故障应急响应流程 │
├─────────────────────────────────────────────────────────┤
│ │
│ Level 1 (P0): 集群完全不可用 │
│ ├── 控制平面全部异常 │
│ ├── etcd 集群丧失 quorum │
│ ├── 立即响应,全员介入 │
│ └── 预期恢复时间: < 1 小时 │
│ │
│ Level 2 (P1): 控制平面部分异常 │
│ ├── 单个 apiserver 节点故障 │
│ ├── 部分组件 CrashLoop │
│ ├── 30分钟内响应 │
│ └── 预期恢复时间: < 2 小时 │
│ │
│ Level 3 (P2): Worker 节点异常 │
│ ├── 单个节点 NotReady │
│ ├── 工作负载自动迁移 │
│ ├── 1小时内响应 │
│ └── 预期恢复时间: < 4 小时 │
│ │
│ Level 4 (P3): 非关键告警 │
│ ├── 证书即将过期(>30天) │
│ ├── 资源使用率偏高 │
│ ├── 24小时内响应 │
│ └── 计划性维护窗口处理 │
│ │
└─────────────────────────────────────────────────────────┘
```
------
## 八、总结与行动项
### 📋 立即执行(今日内)
| 序号 | 行动项 | 负责人 | 优先级 | 预期完成时间 |
| ---- | ------------------------------ | ------------ | ------ | ------------------ |
| 1 | 排查 etcd 崩溃根因并恢复集群 | SRE团队 | P0 | 立即 |
| 2 | 恢复 kube-apiserver 至正常状态 | SRE团队 | P0 | etcd恢复后30分钟内 |
| 3 | 验证集群功能恢复正常 | SRE团队 | P0 | apiserver恢复后 |
| 4 | 检查是否有数据丢失 | DBA/平台团队 | P0 | 集群恢复后 |
### 📋 短期执行(本周内)
| 序号 | 行动项 | 负责人 | 优先级 | 预期完成时间 |
| ---- | ---------------------------- | -------- | ------ | ------------ |
| 5 | 部署 etcd 自动备份 CronJob | SRE团队 | P1 | 3天内 |
| 6 | 配置 etcd 监控告警规则 | 监控团队 | P1 | 3天内 |
| 7 | 修复 ubuntu2204 节点存储问题 | 运维团队 | P1 | 2天内 |
| 8 | 更新证书管理流程 | SRE团队 | P2 | 一周内 |
### 📋 中长期执行(本月内)
| 序号 | 行动项 | 负责人 | 优先级 | 预期完成时间 |
| ---- | ---------------------------- | -------- | ------ | ------------ |
| 9 | 建立完整的灾难恢复预案 | SRE团队 | P1 | 两周内 |
| 10 | 实施 etcd 自动碎片整理 | 平台团队 | P2 | 两周内 |
| 11 | 升级 Kubernetes 至最新稳定版 | 平台团队 | P2 | 一个月内 |
| 12 | 建立定期巡检机制 | SRE团队 | P2 | 两周内 |
| 13 | 进行灾难恢复演练 | 全团队 | P2 | 一个月内 |
------
### 📝 故障复盘要点
**本次故障的核心教训:**
1. **etcd 是集群的心脏**:etcd 故障会导致整个集群瘫痪,必须将其作为最高优先级保护对象
2. **缺少自动备份机制**:如果之前有 etcd 定期备份,恢复时间可以大幅缩短
3. **监控告警不足**:etcd 的健康指标应该有完善的监控和告警,在问题初期就能发现
4. **证书管理需要自动化**:证书过期是 Kubernetes 集群最常见的故障原因之一
5. **需要建立 SOP 文档**:标准操作流程可以在故障时减少决策时间,提高恢复效率
> **注:** 本报告基于诊断数据推断。实际修复时请根据现场日志和具体情况调整方案。建议在修复前先对当前状态做完整快照/备份,以防操作失误导致进一步损坏。