feat: K8S智能诊断平台 - 一键诊断+AI分析+Web仪表盘
- 后端: FastAPI + kubectl 9大检查项(Pod/Node/Event/Deploy/Service/PVC/资源/网络) - AI Agent: OpenAI兼容接口自动分析故障根因+多轮追问对话 - 前端: Vue3 + Element Plus 仪表盘, 健康评分, 问题列表, 资源监控 - 一键启动脚本 start.sh
This commit is contained in:
cnbugs
@@ -0,0 +1,523 @@
|
||||
"""K8S 集群诊断引擎 - 通过 kubectl 命令采集集群状态"""
|
||||
import asyncio
|
||||
import json
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
KUBECTL = shutil.which("kubectl") or "/home/cnbugs/bin/kubectl"
|
||||
|
||||
|
||||
async def _run_kubectl(args: list[str], timeout: int = 30) -> dict:
|
||||
"""执行 kubectl 命令，返回结构化结果"""
|
||||
cmd = [KUBECTL] + args + ["-o", "json"]
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
|
||||
if proc.returncode != 0:
|
||||
return {"error": stderr.decode().strip(), "data": None}
|
||||
return {"error": None, "data": json.loads(stdout.decode())}
|
||||
except asyncio.TimeoutError:
|
||||
return {"error": f"kubectl 命令超时 ({timeout}s)", "data": None}
|
||||
except Exception as e:
|
||||
return {"error": str(e), "data": None}
|
||||
|
||||
|
||||
async def _run_kubectl_raw(args: list[str], timeout: int = 30) -> str:
|
||||
"""执行 kubectl 命令，返回原始文本"""
|
||||
cmd = [KUBECTL] + args
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
|
||||
return stdout.decode().strip()
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def _parse_resource_quantity(q: str) -> float:
|
||||
"""解析 K8S 资源量 (如 500m, 2Gi) 为数值"""
|
||||
if not q:
|
||||
return 0
|
||||
q = str(q)
|
||||
units = {"n": 1e-9, "u": 1e-6, "m": 1e-3, "": 1, "k": 1e3, "M": 1e6, "G": 1e9, "T": 1e12,
|
||||
"Ki": 1024, "Mi": 1024**2, "Gi": 1024**3, "Ti": 1024**4}
|
||||
for suffix, mult in sorted(units.items(), key=lambda x: -len(x[0])):
|
||||
if suffix and q.endswith(suffix):
|
||||
try:
|
||||
return float(q[:-len(suffix)]) * mult
|
||||
except ValueError:
|
||||
return 0
|
||||
try:
|
||||
return float(q)
|
||||
except ValueError:
|
||||
return 0
|
||||
|
||||
|
||||
# ─── 诊断项 ───────────────────────────────────────────────
|
||||
|
||||
async def check_cluster_info() -> dict:
|
||||
"""集群基本信息"""
|
||||
ver = await _run_kubectl(["version"])
|
||||
nodes = await _run_kubectl(["get", "nodes"])
|
||||
namespaces = await _run_kubectl(["get", "namespaces"])
|
||||
return {
|
||||
"name": "cluster_info",
|
||||
"title": "集群信息",
|
||||
"version": ver,
|
||||
"nodes": nodes,
|
||||
"namespaces": namespaces,
|
||||
}
|
||||
|
||||
|
||||
async def check_pods(namespace: str = "") -> dict:
|
||||
"""Pod 状态检查"""
|
||||
args = ["get", "pods"]
|
||||
if namespace:
|
||||
args += ["-n", namespace]
|
||||
else:
|
||||
args += ["--all-namespaces"]
|
||||
result = await _run_kubectl(args)
|
||||
|
||||
issues = []
|
||||
pods_summary = {"total": 0, "running": 0, "pending": 0, "failed": 0, "unknown": 0, "crashloop": 0}
|
||||
|
||||
if result["data"] and "items" in result["data"]:
|
||||
for pod in result["data"]["items"]:
|
||||
pods_summary["total"] += 1
|
||||
phase = pod.get("status", {}).get("phase", "Unknown")
|
||||
name = pod["metadata"]["name"]
|
||||
ns = pod["metadata"].get("namespace", "default")
|
||||
|
||||
if phase == "Running":
|
||||
pods_summary["running"] += 1
|
||||
# 检查重启次数
|
||||
for cs in pod.get("status", {}).get("containerStatuses", []):
|
||||
restarts = cs.get("restartCount", 0)
|
||||
if restarts > 5:
|
||||
issues.append({
|
||||
"level": "warning",
|
||||
"resource": f"{ns}/{name}",
|
||||
"message": f"容器 {cs['name']} 重启 {restarts} 次",
|
||||
})
|
||||
state = cs.get("state", {})
|
||||
if "waiting" in state:
|
||||
reason = state["waiting"].get("reason", "")
|
||||
if reason in ("CrashLoopBackOff", "ErrImagePull", "ImagePullBackOff", "CreateContainerConfigError"):
|
||||
pods_summary["crashloop"] += 1
|
||||
issues.append({
|
||||
"level": "critical",
|
||||
"resource": f"{ns}/{name}",
|
||||
"message": f"容器 {cs['name']} 状态异常: {reason}",
|
||||
})
|
||||
elif phase == "Pending":
|
||||
pods_summary["pending"] += 1
|
||||
issues.append({"level": "warning", "resource": f"{ns}/{name}", "message": "Pod 处于 Pending 状态"})
|
||||
elif phase == "Failed":
|
||||
pods_summary["failed"] += 1
|
||||
issues.append({"level": "critical", "resource": f"{ns}/{name}", "message": "Pod 处于 Failed 状态"})
|
||||
else:
|
||||
pods_summary["unknown"] += 1
|
||||
|
||||
return {
|
||||
"name": "pods",
|
||||
"title": "Pod 状态检查",
|
||||
"summary": pods_summary,
|
||||
"issues": issues,
|
||||
"raw": result,
|
||||
}
|
||||
|
||||
|
||||
async def check_nodes() -> dict:
|
||||
"""Node 状态检查"""
|
||||
result = await _run_kubectl(["get", "nodes"])
|
||||
issues = []
|
||||
nodes_summary = {"total": 0, "ready": 0, "not_ready": 0}
|
||||
|
||||
if result["data"] and "items" in result["data"]:
|
||||
for node in result["data"]["items"]:
|
||||
nodes_summary["total"] += 1
|
||||
name = node["metadata"]["name"]
|
||||
conditions = node.get("status", {}).get("conditions", [])
|
||||
is_ready = False
|
||||
for cond in conditions:
|
||||
if cond["type"] == "Ready" and cond["status"] == "True":
|
||||
is_ready = True
|
||||
if cond["status"] == "True" and cond["type"] in ("MemoryPressure", "DiskPressure", "PIDPressure"):
|
||||
issues.append({
|
||||
"level": "critical",
|
||||
"resource": name,
|
||||
"message": f"节点存在 {cond['type']} 压力",
|
||||
})
|
||||
if is_ready:
|
||||
nodes_summary["ready"] += 1
|
||||
else:
|
||||
nodes_summary["not_ready"] += 1
|
||||
issues.append({"level": "critical", "resource": name, "message": "节点 NotReady"})
|
||||
|
||||
# 资源使用
|
||||
alloc = node.get("status", {}).get("allocatable", {})
|
||||
cap = node.get("status", {}).get("capacity", {})
|
||||
if alloc and cap:
|
||||
for res in ("cpu", "memory"):
|
||||
a = _parse_resource_quantity(alloc.get(res, "0"))
|
||||
c = _parse_resource_quantity(cap.get(res, "0"))
|
||||
if c > 0 and a / c < 0.1:
|
||||
issues.append({
|
||||
"level": "warning",
|
||||
"resource": name,
|
||||
"message": f"节点 {res} 可分配量不足 10%",
|
||||
})
|
||||
|
||||
return {
|
||||
"name": "nodes",
|
||||
"title": "Node 状态检查",
|
||||
"summary": nodes_summary,
|
||||
"issues": issues,
|
||||
"raw": result,
|
||||
}
|
||||
|
||||
|
||||
async def check_events(namespace: str = "") -> dict:
|
||||
"""异常事件检查"""
|
||||
args = ["get", "events", "--field-selector", "type!=Normal", "--sort-by=.lastTimestamp"]
|
||||
if namespace:
|
||||
args += ["-n", namespace]
|
||||
else:
|
||||
args += ["--all-namespaces"]
|
||||
result = await _run_kubectl(args)
|
||||
|
||||
issues = []
|
||||
events_list = []
|
||||
if result["data"] and "items" in result["data"]:
|
||||
for ev in result["data"]["items"][-50:]: # 最近50条
|
||||
events_list.append({
|
||||
"namespace": ev["metadata"].get("namespace", ""),
|
||||
"reason": ev.get("reason", ""),
|
||||
"message": ev.get("message", ""),
|
||||
"type": ev.get("type", ""),
|
||||
"object": f"{ev.get('involvedObject', {}).get('kind', '')}/{ev.get('involvedObject', {}).get('name', '')}",
|
||||
"lastSeen": ev.get("lastTimestamp", ""),
|
||||
"count": ev.get("count", 1),
|
||||
})
|
||||
if ev.get("type") == "Warning":
|
||||
issues.append({
|
||||
"level": "warning",
|
||||
"resource": f"{ev['metadata'].get('namespace', '')}/{ev.get('involvedObject', {}).get('name', '')}",
|
||||
"message": f"[{ev.get('reason', '')}] {ev.get('message', '')[:200]}",
|
||||
})
|
||||
|
||||
return {
|
||||
"name": "events",
|
||||
"title": "异常事件检查",
|
||||
"warning_count": len(issues),
|
||||
"events": events_list,
|
||||
"issues": issues,
|
||||
"raw": result,
|
||||
}
|
||||
|
||||
|
||||
async def check_deployments(namespace: str = "") -> dict:
|
||||
"""Deployment 状态检查"""
|
||||
args = ["get", "deployments"]
|
||||
if namespace:
|
||||
args += ["-n", namespace]
|
||||
else:
|
||||
args += ["--all-namespaces"]
|
||||
result = await _run_kubectl(args)
|
||||
|
||||
issues = []
|
||||
deploys = []
|
||||
if result["data"] and "items" in result["data"]:
|
||||
for dep in result["data"]["items"]:
|
||||
name = dep["metadata"]["name"]
|
||||
ns = dep["metadata"].get("namespace", "default")
|
||||
spec_replicas = dep.get("spec", {}).get("replicas", 0)
|
||||
status = dep.get("status", {})
|
||||
ready = status.get("readyReplicas", 0) or 0
|
||||
available = status.get("availableReplicas", 0) or 0
|
||||
updated = status.get("updatedReplicas", 0) or 0
|
||||
|
||||
deploys.append({
|
||||
"namespace": ns, "name": name,
|
||||
"desired": spec_replicas, "ready": ready,
|
||||
"available": available, "updated": updated,
|
||||
})
|
||||
if ready < spec_replicas:
|
||||
issues.append({
|
||||
"level": "critical" if ready == 0 else "warning",
|
||||
"resource": f"{ns}/{name}",
|
||||
"message": f"副本不足: 期望 {spec_replicas}, 就绪 {ready}",
|
||||
})
|
||||
if updated < spec_replicas:
|
||||
issues.append({
|
||||
"level": "warning",
|
||||
"resource": f"{ns}/{name}",
|
||||
"message": f"滚动更新未完成: 已更新 {updated}/{spec_replicas}",
|
||||
})
|
||||
|
||||
return {
|
||||
"name": "deployments",
|
||||
"title": "Deployment 状态检查",
|
||||
"deployments": deploys,
|
||||
"issues": issues,
|
||||
"raw": result,
|
||||
}
|
||||
|
||||
|
||||
async def check_services(namespace: str = "") -> dict:
|
||||
"""Service / Endpoint 检查"""
|
||||
args = ["get", "endpoints"]
|
||||
if namespace:
|
||||
args += ["-n", namespace]
|
||||
else:
|
||||
args += ["--all-namespaces"]
|
||||
result = await _run_kubectl(args)
|
||||
|
||||
issues = []
|
||||
if result["data"] and "items" in result["data"]:
|
||||
for ep in result["data"]["items"]:
|
||||
name = ep["metadata"]["name"]
|
||||
ns = ep["metadata"].get("namespace", "default")
|
||||
subsets = ep.get("subsets", [])
|
||||
has_addr = any(s.get("addresses") for s in subsets)
|
||||
if not has_addr and name != "kubernetes":
|
||||
issues.append({
|
||||
"level": "warning",
|
||||
"resource": f"{ns}/{name}",
|
||||
"message": "Service 没有可用的后端 Endpoint",
|
||||
})
|
||||
|
||||
return {
|
||||
"name": "services",
|
||||
"title": "Service/Endpoint 检查",
|
||||
"issues": issues,
|
||||
"raw": result,
|
||||
}
|
||||
|
||||
|
||||
async def check_pvc(namespace: str = "") -> dict:
|
||||
"""PVC 状态检查"""
|
||||
args = ["get", "pvc"]
|
||||
if namespace:
|
||||
args += ["-n", namespace]
|
||||
else:
|
||||
args += ["--all-namespaces"]
|
||||
result = await _run_kubectl(args)
|
||||
|
||||
issues = []
|
||||
if result["data"] and "items" in result["data"]:
|
||||
for pvc in result["data"]["items"]:
|
||||
name = pvc["metadata"]["name"]
|
||||
ns = pvc["metadata"].get("namespace", "default")
|
||||
phase = pvc.get("status", {}).get("phase", "")
|
||||
if phase != "Bound":
|
||||
issues.append({
|
||||
"level": "critical",
|
||||
"resource": f"{ns}/{name}",
|
||||
"message": f"PVC 状态异常: {phase}",
|
||||
})
|
||||
|
||||
return {
|
||||
"name": "pvc",
|
||||
"title": "PVC 存储检查",
|
||||
"issues": issues,
|
||||
"raw": result,
|
||||
}
|
||||
|
||||
|
||||
async def check_resource_usage() -> dict:
|
||||
"""节点资源使用率 (需要 metrics-server)"""
|
||||
raw = await _run_kubectl_raw(["top", "nodes", "--no-headers"])
|
||||
issues = []
|
||||
nodes_usage = []
|
||||
|
||||
if raw and "error" not in raw.lower():
|
||||
for line in raw.strip().split("\n"):
|
||||
parts = line.split()
|
||||
if len(parts) >= 5:
|
||||
name = parts[0]
|
||||
cpu_pct = parts[1].rstrip("%")
|
||||
mem_pct = parts[3].rstrip("%")
|
||||
try:
|
||||
cpu_v, mem_v = float(cpu_pct), float(mem_pct)
|
||||
nodes_usage.append({"name": name, "cpu_pct": cpu_v, "mem_pct": mem_v})
|
||||
if cpu_v > 90:
|
||||
issues.append({"level": "critical", "resource": name, "message": f"CPU 使用率 {cpu_v}%"})
|
||||
elif cpu_v > 75:
|
||||
issues.append({"level": "warning", "resource": name, "message": f"CPU 使用率 {cpu_v}%"})
|
||||
if mem_v > 90:
|
||||
issues.append({"level": "critical", "resource": name, "message": f"内存使用率 {mem_v}%"})
|
||||
elif mem_v > 75:
|
||||
issues.append({"level": "warning", "resource": name, "message": f"内存使用率 {mem_v}%"})
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
return {
|
||||
"name": "resource_usage",
|
||||
"title": "资源使用率",
|
||||
"issues": [],
|
||||
"nodes": [],
|
||||
"note": "metrics-server 不可用，跳过资源使用率检查",
|
||||
"raw": {"error": raw},
|
||||
}
|
||||
|
||||
return {
|
||||
"name": "resource_usage",
|
||||
"title": "资源使用率",
|
||||
"issues": issues,
|
||||
"nodes": nodes_usage,
|
||||
"raw": {"text": raw},
|
||||
}
|
||||
|
||||
|
||||
async def check_network_policies(namespace: str = "") -> dict:
|
||||
"""网络策略检查"""
|
||||
args = ["get", "networkpolicies"]
|
||||
if namespace:
|
||||
args += ["-n", namespace]
|
||||
else:
|
||||
args += ["--all-namespaces"]
|
||||
result = await _run_kubectl(args)
|
||||
count = len(result["data"].get("items", [])) if result["data"] else 0
|
||||
return {
|
||||
"name": "network",
|
||||
"title": "网络策略检查",
|
||||
"policy_count": count,
|
||||
"issues": [],
|
||||
"raw": result,
|
||||
}
|
||||
|
||||
|
||||
# ─── 主诊断入口 ───────────────────────────────────────────
|
||||
|
||||
ALL_CHECKS = {
|
||||
"cluster_info": check_cluster_info,
|
||||
"pods": check_pods,
|
||||
"nodes": check_nodes,
|
||||
"events": check_events,
|
||||
"deployments": check_deployments,
|
||||
"services": check_services,
|
||||
"pvc": check_pvc,
|
||||
"resource_usage": check_resource_usage,
|
||||
"network": check_network_policies,
|
||||
}
|
||||
|
||||
|
||||
async def run_full_diagnosis(namespace: str = "", checks: Optional[list[str]] = None) -> dict:
|
||||
"""执行完整诊断"""
|
||||
start = datetime.now()
|
||||
|
||||
selected = checks or list(ALL_CHECKS.keys())
|
||||
tasks = {}
|
||||
for name in selected:
|
||||
fn = ALL_CHECKS.get(name)
|
||||
if fn:
|
||||
if name in ("pods", "events", "deployments", "services", "pvc", "network"):
|
||||
tasks[name] = fn(namespace)
|
||||
else:
|
||||
tasks[name] = fn()
|
||||
|
||||
results = {}
|
||||
for name, coro in tasks.items():
|
||||
try:
|
||||
results[name] = await coro
|
||||
except Exception as e:
|
||||
results[name] = {"name": name, "title": name, "error": str(e), "issues": []}
|
||||
|
||||
# 汇总
|
||||
all_issues = []
|
||||
for r in results.values():
|
||||
for issue in r.get("issues", []):
|
||||
issue["check"] = r.get("title", r.get("name", ""))
|
||||
all_issues.append(issue)
|
||||
|
||||
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()
|
||||
|
||||
# 健康评分 (100 分制)
|
||||
score = 100
|
||||
score -= len(critical) * 15
|
||||
score -= len(warning) * 5
|
||||
score = max(0, score)
|
||||
|
||||
if score >= 90:
|
||||
status = "healthy"
|
||||
elif score >= 60:
|
||||
status = "warning"
|
||||
else:
|
||||
status = "critical"
|
||||
|
||||
return {
|
||||
"timestamp": start.isoformat(),
|
||||
"elapsed_seconds": round(elapsed, 2),
|
||||
"namespace": namespace or "all",
|
||||
"score": score,
|
||||
"status": status,
|
||||
"critical_count": len(critical),
|
||||
"warning_count": len(warning),
|
||||
"issues": all_issues,
|
||||
"checks": results,
|
||||
}
|
||||
|
||||
|
||||
def build_diagnosis_report(diagnosis: dict) -> str:
|
||||
"""将诊断结果构建为 AI 可读的文本报告"""
|
||||
lines = []
|
||||
lines.append(f"# K8S 集群诊断报告")
|
||||
lines.append(f"时间: {diagnosis['timestamp']}")
|
||||
lines.append(f"命名空间: {diagnosis['namespace']}")
|
||||
lines.append(f"健康评分: {diagnosis['score']}/100 ({diagnosis['status']})")
|
||||
lines.append(f"严重问题: {diagnosis['critical_count']} 个, 警告: {diagnosis['warning_count']} 个")
|
||||
lines.append("")
|
||||
|
||||
if diagnosis["issues"]:
|
||||
lines.append("## 发现的问题")
|
||||
for i, issue in enumerate(diagnosis["issues"], 1):
|
||||
icon = "🔴" if issue["level"] == "critical" else "🟡"
|
||||
lines.append(f"{i}. {icon} [{issue['check']}] {issue['resource']}: {issue['message']}")
|
||||
lines.append("")
|
||||
|
||||
# Pod 摘要
|
||||
pods = diagnosis["checks"].get("pods", {})
|
||||
if pods.get("summary"):
|
||||
s = pods["summary"]
|
||||
lines.append(f"## Pod 状态: 总计 {s['total']}, Running {s['running']}, Pending {s['pending']}, Failed {s['failed']}, CrashLoop {s.get('crashloop', 0)}")
|
||||
|
||||
# Node 摘要
|
||||
nodes = diagnosis["checks"].get("nodes", {})
|
||||
if nodes.get("summary"):
|
||||
s = nodes["summary"]
|
||||
lines.append(f"## Node 状态: 总计 {s['total']}, Ready {s['ready']}, NotReady {s['not_ready']}")
|
||||
|
||||
# Deployment 摘要
|
||||
deps = diagnosis["checks"].get("deployments", {})
|
||||
if deps.get("deployments"):
|
||||
lines.append("## Deployment 状态:")
|
||||
for d in deps["deployments"]:
|
||||
lines.append(f" - {d['namespace']}/{d['name']}: {d['ready']}/{d['desired']} ready")
|
||||
|
||||
# 异常事件
|
||||
events = diagnosis["checks"].get("events", {})
|
||||
if events.get("events"):
|
||||
lines.append(f"\n## 最近异常事件 (共 {len(events['events'])} 条):")
|
||||
for ev in events["events"][-20:]:
|
||||
lines.append(f" - [{ev['type']}] {ev['object']}: {ev['reason']} - {ev['message'][:150]}")
|
||||
|
||||
# 资源使用
|
||||
usage = diagnosis["checks"].get("resource_usage", {})
|
||||
if usage.get("nodes"):
|
||||
lines.append("\n## 节点资源使用:")
|
||||
for n in usage["nodes"]:
|
||||
lines.append(f" - {n['name']}: CPU {n['cpu_pct']}%, 内存 {n['mem_pct']}%")
|
||||
|
||||
return "\n".join(lines)
|
||||
Reference in New Issue
Block a user