rlwj-lab/aiops
0
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: K8s 集群实时代理 — 直接调用 K8s API 不存储 MySQL

Browse Source 
- 新增 pkg/k8sclient 轻量级 K8s REST 客户端（解析 kubeconfig TLS 证书）
- ListNodes/Pods/Deployments/Services/Ingresses 改为实时代理 K8s API
- 集群同步端点 POST /:id/sync（拉取后仍存 MySQL 用于离线缓存储备）
- 前端 k8sApi 新增 syncCluster 方法
- 修复 SNMP MaxRepetitions uint8→uint32 类型错误
- 修复 ServicePort.TargetPort 兼容 int/string 双类型
- 创建 k8s_nodes/pods/deployments/services/ingresses 五张表
This commit is contained in:
attacker committed 2026-06-19 06:32:18 +00:00
1 parent a1d4e2f18a
commit b25ad8705f
6 files changed
+821 -53
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/internal/api/handlers/k8s.go
+440 -51
View File
@@ -1,10 +1,15 @@
package handlers
import (
"encoding/json"
"fmt"
"log"
"net/http"
"strconv"
"time"
"ezops/internal/models"
"ezops/pkg/k8sclient"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
@@ -224,14 +229,65 @@ func (h *K8sHandler) DeleteCluster(c *gin.Context) {
func (h *K8sHandler) ListNodes(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
var nodes []models.K8sNode
h.db.Where("cluster_id = ?", clusterID).Find(&nodes)
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
nodes, err := client.ListNodes()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: nodes,
})
// Convert to simplified response
type NodeSummary struct {
Name string `json:"name"`
IP string `json:"ip"`
Role string `json:"role"`
Status string `json:"status"`
OS string `json:"os"`
Kubelet string `json:"kubelet_version"`
CPU int `json:"cpu"`
Memory int64 `json:"memory_mb"`
MaxPods int `json:"max_pods"`
}
result := []NodeSummary{}
for _, n := range nodes.Items {
role := "worker"
for k := range n.Metadata.Labels {
if k == "node-role.kubernetes.io/master" || k == "node-role.kubernetes.io/control-plane" {
role = "master"
break
}
}
nodeStatus := "NotReady"
for _, cond := range n.Status.Conditions {
if cond.Type == "Ready" && cond.Status == "True" {
nodeStatus = "Ready"
break
}
}
ip := ""
for _, addr := range n.Status.Addresses {
if addr.Type == "InternalIP" {
ip = addr.Address
break
}
}
result = append(result, NodeSummary{
Name: n.Metadata.Name,
IP: ip,
Role: role,
Status: nodeStatus,
OS: n.Status.NodeInfo.OSImage,
Kubelet: n.Status.NodeInfo.KubeletVersion,
CPU: parseIntResource(n.Status.Capacity["cpu"]),
Memory: parseMemResource(n.Status.Capacity["memory"]),
MaxPods: parseIntResource(n.Status.Capacity["pods"]),
})
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// ListPods 获取Pod列表
@@ -248,19 +304,50 @@ func (h *K8sHandler) ListPods(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Query("namespace")
db := h.db.Where("cluster_id = ?", clusterID)
if namespace != "" {
db = db.Where("namespace = ?", namespace)
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
pods, err := client.ListPods()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
var pods []models.K8sPod
db.Find(&pods)
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: pods,
})
type PodSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Status string `json:"status"`
NodeName string `json:"node_name"`
IP string `json:"ip"`
RestartCount int `json:"restart_count"`
Image string `json:"image"`
}
result := []PodSummary{}
for _, p := range pods.Items {
if namespace != "" && p.Metadata.Namespace != namespace {
continue
}
image := ""
if len(p.Spec.Containers) > 0 {
image = p.Spec.Containers[0].Image
}
restarts := 0
if len(p.Status.ContainerStatuses) > 0 {
restarts = p.Status.ContainerStatuses[0].RestartCount
}
result = append(result, PodSummary{
Namespace: p.Metadata.Namespace,
Name: p.Metadata.Name,
Status: p.Status.Phase,
NodeName: p.Spec.NodeName,
IP: p.Status.PodIP,
RestartCount: restarts,
Image: image,
})
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// ListDeployments 获取Deployment列表
@@ -277,19 +364,44 @@ func (h *K8sHandler) ListDeployments(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Query("namespace")
db := h.db.Where("cluster_id = ?", clusterID)
if namespace != "" {
db = db.Where("namespace = ?", namespace)
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
deps, err := client.ListDeployments()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
var deployments []models.K8sDeployment
db.Find(&deployments)
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: deployments,
})
type DepSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Replicas int `json:"replicas"`
AvailableReplicas int `json:"available_replicas"`
ReadyReplicas int `json:"ready_replicas"`
Image string `json:"image"`
}
result := []DepSummary{}
for _, d := range deps.Items {
if namespace != "" && d.Metadata.Namespace != namespace {
continue
}
image := ""
if len(d.Spec.Template.Spec.Containers) > 0 {
image = d.Spec.Template.Spec.Containers[0].Image
}
result = append(result, DepSummary{
Namespace: d.Metadata.Namespace,
Name: d.Metadata.Name,
Replicas: d.Spec.Replicas,
AvailableReplicas: d.Status.AvailableReplicas,
ReadyReplicas: d.Status.ReadyReplicas,
Image: image,
})
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// ListServices 获取Service列表
@@ -306,19 +418,38 @@ func (h *K8sHandler) ListServices(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Query("namespace")
db := h.db.Where("cluster_id = ?", clusterID)
if namespace != "" {
db = db.Where("namespace = ?", namespace)
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
svcs, err := client.ListServices()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
var services []models.K8sService
db.Find(&services)
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: services,
})
type SvcSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Type string `json:"type"`
ClusterIP string `json:"cluster_ip"`
Ports int `json:"port_count"`
}
result := []SvcSummary{}
for _, s := range svcs.Items {
if namespace != "" && s.Metadata.Namespace != namespace {
continue
}
result = append(result, SvcSummary{
Namespace: s.Metadata.Namespace,
Name: s.Metadata.Name,
Type: s.Spec.Type,
ClusterIP: s.Spec.ClusterIP,
Ports: len(s.Spec.Ports),
})
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// ListIngresses 获取Ingress列表
@@ -335,19 +466,42 @@ func (h *K8sHandler) ListIngresses(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Query("namespace")
db := h.db.Where("cluster_id = ?", clusterID)
if namespace != "" {
db = db.Where("namespace = ?", namespace)
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
ings, err := client.ListIngresses()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
var ingresses []models.K8sIngress
db.Find(&ingresses)
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: ingresses,
})
type IngSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Hosts []string `json:"hosts"`
TLS bool `json:"tls"`
}
result := []IngSummary{}
for _, ing := range ings.Items {
if namespace != "" && ing.Metadata.Namespace != namespace {
continue
}
hosts := []string{}
for _, rule := range ing.Spec.Rules {
if rule.Host != "" {
hosts = append(hosts, rule.Host)
}
}
result = append(result, IngSummary{
Namespace: ing.Metadata.Namespace,
Name: ing.Metadata.Name,
Hosts: hosts,
TLS: len(ing.Spec.TLS) > 0,
})
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// GetTopology 获取集群拓扑布局 // Logan.Li
@@ -385,3 +539,238 @@ func (h *K8sHandler) SaveTopology(c *gin.Context) {
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "保存成功"})
}
// SyncCluster 同步集群资源（连接真实K8s API拉取数据）
func (h *K8sHandler) SyncCluster(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
var cluster models.K8sCluster
if err := h.db.First(&cluster, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "集群不存在"})
return
}
if cluster.KubeConfig == "" && cluster.Token == "" {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "集群未配置凭证"})
return
}
client, err := k8sclient.NewClient(cluster.KubeConfig)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "连接K8s失败: " + err.Error()})
return
}
result := k8sclient.SyncResult{}
// Sync nodes
if nodes, err := client.ListNodes(); err == nil {
// Use raw SQL for reliability (avoid GORM soft-delete / auto-timestamp issues)
h.db.Exec("DELETE FROM k8s_nodes WHERE cluster_id = ?", cluster.ID)
for _, n := range nodes.Items {
role := "worker"
if _, ok := n.Metadata.Labels["node-role.kubernetes.io/master"]; ok {
role = "master"
} else if _, ok := n.Metadata.Labels["node-role.kubernetes.io/control-plane"]; ok {
role = "master"
}
nodeStatus := "NotReady"
for _, cond := range n.Status.Conditions {
if cond.Type == "Ready" && cond.Status == "True" {
nodeStatus = "Ready"
break
}
}
ip := ""
for _, addr := range n.Status.Addresses {
if addr.Type == "InternalIP" {
ip = addr.Address
break
}
}
labelsJSON, _ := json.Marshal(n.Metadata.Labels)
now := time.Now()
err2 := h.db.Exec(`INSERT INTO k8s_nodes (cluster_id, name, ip, role, status, os, kubelet_version, cpu, memory, max_pods, labels, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
cluster.ID, n.Metadata.Name, ip, role, nodeStatus, n.Status.NodeInfo.OSImage, n.Status.NodeInfo.KubeletVersion,
parseIntResource(n.Status.Capacity["cpu"]), parseMemResource(n.Status.Capacity["memory"]), parseIntResource(n.Status.Capacity["pods"]),
string(labelsJSON), now, now).Error
if err2 != nil {
log.Printf("[K8s] insert node %s error: %v", n.Metadata.Name, err2)
} else {
result.Nodes++
}
}
} else {
log.Printf("[K8s] sync nodes error: %v", err)
}
// Sync pods
if pods, err := client.ListPods(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&models.K8sPod{})
for _, p := range pods.Items {
image := ""
if len(p.Spec.Containers) > 0 {
image = p.Spec.Containers[0].Image
}
restarts := 0
if len(p.Status.ContainerStatuses) > 0 {
restarts = p.Status.ContainerStatuses[0].RestartCount
}
labelsJSON, _ := json.Marshal(p.Metadata.Labels)
h.db.Create(&models.K8sPod{
ClusterID: cluster.ID,
Namespace: p.Metadata.Namespace,
Name: p.Metadata.Name,
Status: p.Status.Phase,
NodeName: p.Spec.NodeName,
IP: p.Status.PodIP,
Image: image,
RestartCount: restarts,
Labels: string(labelsJSON),
})
result.Pods++
}
} else {
log.Printf("[K8s] sync pods error: %v", err)
}
// Sync deployments
if deps, err := client.ListDeployments(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&models.K8sDeployment{})
for _, d := range deps.Items {
image := ""
if len(d.Spec.Template.Spec.Containers) > 0 {
image = d.Spec.Template.Spec.Containers[0].Image
}
labelsJSON, _ := json.Marshal(d.Metadata.Labels)
h.db.Create(&models.K8sDeployment{
ClusterID: cluster.ID,
Namespace: d.Metadata.Namespace,
Name: d.Metadata.Name,
Replicas: d.Spec.Replicas,
AvailableReplicas: d.Status.AvailableReplicas,
ReadyReplicas: d.Status.ReadyReplicas,
Image: image,
Labels: string(labelsJSON),
})
result.Deployments++
}
} else {
log.Printf("[K8s] sync deps error: %v", err)
}
// Sync services
if svcs, err := client.ListServices(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&models.K8sService{})
for _, s := range svcs.Items {
portsJSON, _ := json.Marshal(s.Spec.Ports)
selectorJSON, _ := json.Marshal(s.Spec.Selector)
labelsJSON, _ := json.Marshal(s.Metadata.Labels)
h.db.Create(&models.K8sService{
ClusterID: cluster.ID,
Namespace: s.Metadata.Namespace,
Name: s.Metadata.Name,
Type: s.Spec.Type,
ClusterIP: s.Spec.ClusterIP,
Ports: string(portsJSON),
Selector: string(selectorJSON),
Labels: string(labelsJSON),
})
result.Services++
}
}
// Sync ingresses
if ings, err := client.ListIngresses(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&models.K8sIngress{})
for _, ing := range ings.Items {
hosts := []string{}
for _, rule := range ing.Spec.Rules {
hosts = append(hosts, rule.Host)
}
hostsJSON, _ := json.Marshal(hosts)
tls := len(ing.Spec.TLS) > 0
tlsHosts := []string{}
if tls {
for _, t := range ing.Spec.TLS {
tlsHosts = append(tlsHosts, t.Hosts...)
}
}
tlsHostsJSON, _ := json.Marshal(tlsHosts)
labelsJSON, _ := json.Marshal(ing.Metadata.Labels)
h.db.Create(&models.K8sIngress{
ClusterID: cluster.ID,
Namespace: ing.Metadata.Namespace,
Name: ing.Metadata.Name,
Hosts: string(hostsJSON),
TLS: tls,
TLSHosts: string(tlsHostsJSON),
Labels: string(labelsJSON),
})
result.Ingresses++
}
}
// Update cluster stats
now := time.Now()
h.db.Model(&cluster).Updates(map[string]interface{}{
"node_count": result.Nodes,
"pod_count": result.Pods,
"deployment_count": result.Deployments,
"service_count": result.Services,
"last_sync_at": &now,
})
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: fmt.Sprintf("同步完成: %d nodes, %d pods, %d deployments, %d services, %d ingresses",
result.Nodes, result.Pods, result.Deployments, result.Services, result.Ingresses),
Data: result,
})
}
func parseIntResource(val string) int {
if val == "" {
return 0
}
var i int
fmt.Sscanf(val, "%d", &i)
return i
}
func parseMemResource(val string) int64 {
if val == "" {
return 0
}
var num float64
var unit string
fmt.Sscanf(val, "%f%s", &num, &unit)
switch unit {
case "Ki":
return int64(num / 1024)
case "Mi":
return int64(num)
case "Gi":
return int64(num * 1024)
default:
return int64(num / 1024 / 1024)
}
}
// getK8sClient builds a K8s API client from cluster's stored kubeconfig
func (h *K8sHandler) getK8sClient(clusterID int) (*k8sclient.Client, *models.K8sCluster, error) {
var cluster models.K8sCluster
if err := h.db.First(&cluster, clusterID).Error; err != nil {
return nil, nil, fmt.Errorf("cluster not found")
}
if cluster.KubeConfig == "" {
return nil, nil, fmt.Errorf("no kubeconfig")
}
client, err := k8sclient.NewClient(cluster.KubeConfig)
if err != nil {
return nil, nil, err
}
return client, &cluster, nil
}
backend/internal/api/routes/routes.go
+1
View File
@@ -239,6 +239,7 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
clusters.POST("", k8sHandler.CreateCluster)
clusters.PUT("/:id", k8sHandler.UpdateCluster)
clusters.DELETE("/:id", k8sHandler.DeleteCluster)
clusters.POST("/:id/sync", k8sHandler.SyncCluster) // 同步集群资源 // Logan.Li
clusters.GET("/:id/nodes", k8sHandler.ListNodes)
clusters.GET("/:id/pods", k8sHandler.ListPods)
clusters.GET("/:id/deployments", k8sHandler.ListDeployments)