cnbugs
d5d7e45ded
- 后端: FastAPI + kubectl 9大检查项(Pod/Node/Event/Deploy/Service/PVC/资源/网络) - AI Agent: OpenAI兼容接口自动分析故障根因+多轮追问对话 - 前端: Vue3 + Element Plus 仪表盘, 健康评分, 问题列表, 资源监控 - 一键启动脚本 start.sh
524 lines
19 KiB
Python
524 lines
19 KiB
Python
"""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)
|