cnbugs
475 lines
20 KiB
Python
475 lines
20 KiB
Python
"""K8S 智能诊断平台 - FastAPI 主应用"""
|
|
import os
|
|
import json
|
|
import asyncio
|
|
import httpx
|
|
from datetime import datetime
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
|
|
from backend.config import PORT, AI_API_BASE, AI_MODEL, HISTORY_DB_PATH
|
|
from backend.diagnosis import (
|
|
run_full_diagnosis, build_diagnosis_report, ALL_CHECKS,
|
|
check_connectivity,
|
|
check_cluster_info, check_pods, check_nodes, check_events,
|
|
check_deployments, check_services, check_pvc, check_resource_usage,
|
|
check_network_policies,
|
|
)
|
|
from backend.ai_agent import (
|
|
analyze_with_ai, chat_with_ai,
|
|
analyze_with_ai_stream, chat_with_ai_stream,
|
|
run_diagnostic_agent, resume_diagnostic_agent, take_pending_approval,
|
|
)
|
|
from backend.agent_tools import execute_kubectl
|
|
from backend.history_store import HistoryStore
|
|
|
|
app = FastAPI(title="K8S 智能诊断平台", version="1.0.0")
|
|
history_store = HistoryStore(HISTORY_DB_PATH)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# 存储最近诊断结果 (内存)
|
|
_last_diagnosis: Optional[dict] = None
|
|
_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 = [
|
|
("cluster_info", "集群信息", check_cluster_info),
|
|
("nodes", "Node 状态检查", check_nodes),
|
|
("pods", "Pod 状态检查", check_pods),
|
|
("deployments", "Deployment 状态检查", check_deployments),
|
|
("services", "Service/Endpoint 检查", check_services),
|
|
("events", "异常事件检查", check_events),
|
|
("pvc", "PVC 存储检查", check_pvc),
|
|
("resource_usage", "资源使用率", check_resource_usage),
|
|
("network", "网络策略检查", check_network_policies),
|
|
]
|
|
|
|
|
|
class DiagnoseRequest(BaseModel):
|
|
namespace: str = ""
|
|
checks: Optional[list[str]] = None
|
|
|
|
|
|
class AnalyzeRequest(BaseModel):
|
|
question: str = ""
|
|
|
|
|
|
class ChatRequest(BaseModel):
|
|
messages: list[dict]
|
|
|
|
|
|
class AgentApprovalRequest(BaseModel):
|
|
approval_id: str
|
|
approved: bool
|
|
|
|
|
|
@app.get("/api/health")
|
|
async def health():
|
|
return {"status": "ok", "time": datetime.now().isoformat()}
|
|
|
|
|
|
@app.get("/api/checks")
|
|
async def list_checks():
|
|
"""列出所有可用诊断项"""
|
|
return {
|
|
"checks": [
|
|
{"name": name, "title": title}
|
|
for name, title, _ in CHECK_ORDER
|
|
]
|
|
}
|
|
|
|
|
|
@app.get("/api/diagnose/stream")
|
|
async def diagnose_stream(namespace: str = ""):
|
|
"""SSE 流式诊断 - 实时推送每一步进度"""
|
|
global _last_diagnosis, _last_report
|
|
|
|
async def event_generator():
|
|
start = datetime.now()
|
|
|
|
# ★ 第一步: 连通性检查
|
|
yield f"data: {json.dumps({'type': 'progress', 'step': 0, 'total': len(CHECK_ORDER) + 1, 'name': 'connectivity', 'title': '集群连通性检查', 'status': 'running'}, ensure_ascii=False)}\n\n"
|
|
|
|
conn = await check_connectivity()
|
|
if not conn["connected"]:
|
|
yield f"data: {json.dumps({'type': 'progress', 'step': 0, 'total': len(CHECK_ORDER) + 1, 'name': 'connectivity', 'title': '集群连通性检查', 'status': 'done', 'issues': 1}, ensure_ascii=False)}\n\n"
|
|
|
|
diagnosis = {
|
|
"timestamp": start.isoformat(),
|
|
"elapsed_seconds": round((datetime.now() - start).total_seconds(), 2),
|
|
"namespace": namespace or "all",
|
|
"score": 0,
|
|
"status": "critical",
|
|
"connected": False,
|
|
"connectivity_error": conn["error"],
|
|
"connectivity_suggestion": conn.get("suggestion", ""),
|
|
"critical_count": 1,
|
|
"warning_count": 0,
|
|
"issues": [{
|
|
"level": "critical",
|
|
"check": "集群连通性",
|
|
"resource": "API Server",
|
|
"message": f"无法连接到集群: {conn['error'][:300]}",
|
|
}],
|
|
"checks": {},
|
|
}
|
|
report = build_diagnosis_report(diagnosis)
|
|
_save_current_diagnosis(diagnosis, report)
|
|
yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis, 'history_id': _last_diagnosis_id}, ensure_ascii=False)}\n\n"
|
|
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"
|
|
|
|
# ★ 第二步: 执行各项检查
|
|
results = {}
|
|
all_issues = []
|
|
total = len(CHECK_ORDER) + 1
|
|
|
|
for idx, (name, title, fn) in enumerate(CHECK_ORDER, 1):
|
|
yield f"data: {json.dumps({'type': 'progress', 'step': idx, 'total': total, 'name': name, 'title': title, 'status': 'running'}, ensure_ascii=False)}\n\n"
|
|
|
|
try:
|
|
if name in ("pods", "events", "deployments", "services", "pvc", "network"):
|
|
result = await fn(namespace)
|
|
else:
|
|
result = await fn()
|
|
results[name] = result
|
|
issue_count = len(result.get("issues", []))
|
|
for issue in result.get("issues", []):
|
|
issue["check"] = title
|
|
all_issues.append(issue)
|
|
except Exception as e:
|
|
results[name] = {
|
|
"name": name, "title": title, "error": str(e),
|
|
"issues": [{"level": "critical", "resource": name, "message": f"检查执行异常: {str(e)}"}],
|
|
}
|
|
issue_count = 1
|
|
all_issues.append({"level": "critical", "check": title, "resource": name, "message": f"检查执行异常: {str(e)}"})
|
|
|
|
yield f"data: {json.dumps({'type': 'progress', 'step': idx, 'total': total, 'name': name, 'title': title, 'status': 'done', 'issues': issue_count}, ensure_ascii=False)}\n\n"
|
|
|
|
# 汇总
|
|
critical = [i for i in all_issues if i["level"] == "critical"]
|
|
warning = [i for i in all_issues if i["level"] == "warning"]
|
|
elapsed = (datetime.now() - start).total_seconds()
|
|
score = max(0, 100 - len(critical) * 15 - len(warning) * 5)
|
|
status = "healthy" if score >= 90 else ("warning" if score >= 60 else "critical")
|
|
|
|
diagnosis = {
|
|
"timestamp": start.isoformat(),
|
|
"elapsed_seconds": round(elapsed, 2),
|
|
"namespace": namespace or "all",
|
|
"score": score,
|
|
"status": status,
|
|
"connected": True,
|
|
"critical_count": len(critical),
|
|
"warning_count": len(warning),
|
|
"issues": all_issues,
|
|
"checks": results,
|
|
}
|
|
|
|
report = build_diagnosis_report(diagnosis)
|
|
_save_current_diagnosis(diagnosis, report)
|
|
|
|
yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis, 'history_id': _last_diagnosis_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.post("/api/diagnose")
|
|
async def diagnose(req: DiagnoseRequest):
|
|
"""执行一键诊断 (非流式，兼容旧接口)"""
|
|
global _last_diagnosis, _last_report
|
|
diagnosis = await run_full_diagnosis(namespace=req.namespace, checks=req.checks)
|
|
report = build_diagnosis_report(diagnosis)
|
|
_save_current_diagnosis(diagnosis, report)
|
|
return {**diagnosis, "history_id": _last_diagnosis_id}
|
|
|
|
|
|
@app.get("/api/diagnosis/latest")
|
|
async def get_latest():
|
|
"""获取最近一次诊断结果，服务重启后从 SQLite 恢复。"""
|
|
global _last_diagnosis, _last_report, _last_diagnosis_id
|
|
if _last_diagnosis is None:
|
|
latest = history_store.latest()
|
|
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")
|
|
async def analyze(req: AnalyzeRequest):
|
|
"""AI 分析诊断结果 (非流式，兼容旧接口)"""
|
|
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(), "analysis", "standard", req.question)
|
|
result = await analyze_with_ai(_last_report, question=req.question)
|
|
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")
|
|
async def analyze_stream(question: str = ""):
|
|
"""SSE 流式 AI 分析 - 逐字推送分析内容并持久化"""
|
|
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(), "analysis", "stream", question)
|
|
|
|
async def event_generator():
|
|
process_steps = [
|
|
("report", "读取诊断报告", "已加载本次集群诊断上下文"),
|
|
("risk", "识别关键风险", "正在梳理严重问题、警告和潜在影响"),
|
|
("solution", "生成修复方案", "正在为高优先级问题生成可执行操作"),
|
|
("summary", "整理分析结论", "正在组织根因、影响和预防建议"),
|
|
]
|
|
|
|
yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL, 'run_id': run_id}, ensure_ascii=False)}\n\n"
|
|
content = ""
|
|
try:
|
|
for step, title, detail in process_steps:
|
|
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)
|
|
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):
|
|
content += chunk
|
|
yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, 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:
|
|
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"
|
|
except httpx.ConnectError:
|
|
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:
|
|
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,
|
|
execution_mode=execution_mode,
|
|
):
|
|
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_kubectl(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.post("/api/chat")
|
|
async def chat(req: ChatRequest):
|
|
"""AI 多轮对话 (非流式，兼容旧接口)"""
|
|
result = await chat_with_ai(req.messages, context=_last_report)
|
|
return result
|
|
|
|
|
|
@app.post("/api/chat/stream")
|
|
async def chat_stream(req: ChatRequest):
|
|
"""SSE 流式 AI 对话 - 逐字推送回复"""
|
|
async def event_generator():
|
|
yield f"data: {json.dumps({'type': 'start'}, ensure_ascii=False)}\n\n"
|
|
try:
|
|
async for chunk in chat_with_ai_stream(req.messages, context=_last_report):
|
|
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"
|
|
except Exception as e:
|
|
yield f"data: {json.dumps({'type': 'error', 'message': f'AI 对话失败: {str(e)}'}, 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/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")
|
|
async def get_report():
|
|
"""获取诊断报告文本"""
|
|
if not _last_report:
|
|
raise HTTPException(status_code=404, detail="暂无诊断报告")
|
|
return {"report": _last_report}
|
|
|
|
|
|
# 静态文件服务 (前端)
|
|
static_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "frontend", "dist")
|
|
if os.path.isdir(static_dir):
|
|
app.mount("/", StaticFiles(directory=static_dir, html=True), name="static")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=PORT)
|