package handlers import ( "encoding/base64" "encoding/json" "fmt" "io" "log" "net/http" "net/url" "path" "sort" "strconv" "strings" "time" "ezops/internal/models" "ezops/pkg/kubernetesclient" "github.com/gin-gonic/gin" "github.com/gorilla/websocket" "gorm.io/gorm" ) // KubernetesHandler K8s处理器 type KubernetesHandler struct { db *gorm.DB } // NewKubernetesHandler 创建K8s处理器 func NewKubernetesHandler(db *gorm.DB) *KubernetesHandler { return &KubernetesHandler{db: db} } // ListClusters 获取集群列表 // @Summary 获取集群列表 // @Description 获取所有K8s集群列表 // @Tags K8s集群 // @Produce json // @Security Bearer // @Param page query int false "页码" // @Param page_size query int false "每页数量" // @Success 200 {object} models.Response{data=models.ListResponse} // @Router /kubernetes-clusters [get] func (h *KubernetesHandler) ListClusters(c *gin.Context) { var query models.PageQuery c.ShouldBindQuery(&query) var total int64 h.db.Model(&models.KubernetesCluster{}).Count(&total) var clusters []models.KubernetesCluster h.db.Offset(query.GetOffset()).Limit(query.GetLimit()).Find(&clusters) c.JSON(http.StatusOK, models.Response{ Code: 200, Message: "success", Data: models.ListResponse{ List: clusters, Total: total, }, }) } // GetCluster 获取集群详情 // @Summary 获取集群详情 // @Description 获取指定K8s集群的详细信息 // @Tags K8s集群 // @Produce json // @Security Bearer // @Param id path int true "集群ID" // @Success 200 {object} models.Response{data=models.KubernetesCluster} // @Router /kubernetes-clusters/{id} [get] func (h *KubernetesHandler) GetCluster(c *gin.Context) { id, _ := strconv.Atoi(c.Param("id")) var cluster models.KubernetesCluster if err := h.db.First(&cluster, id).Error; err != nil { c.JSON(http.StatusNotFound, models.Response{ Code: 404, Message: "集群不存在", }) return } c.JSON(http.StatusOK, models.Response{ Code: 200, Message: "success", Data: cluster, }) } // CreateCluster 创建集群 // @Summary 创建集群 // @Description 创建新K8s集群 // @Tags K8s集群 // @Accept json // @Produce json // @Security Bearer // @Param request body models.CreateKubernetesClusterRequest true "创建请求" // @Success 200 {object} models.Response{data=models.KubernetesCluster} // @Router /kubernetes-clusters [post] func (h *KubernetesHandler) CreateCluster(c *gin.Context) { var req models.CreateKubernetesClusterRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{ Code: 400, Message: "请求参数错误: " + err.Error(), }) return } cluster := models.KubernetesCluster{ Name: req.Name, Code: req.Code, Version: req.Version, ConfigType: req.ConfigType, APIServer: req.APIServer, APIEndpoint: req.APIServer, // auto-sync Token: req.Token, KubeConfig: req.KubeConfig, Description: req.Description, Status: "healthy", } if err := h.db.Create(&cluster).Error; err != nil { c.JSON(http.StatusInternalServerError, models.Response{ Code: 500, Message: "创建失败", }) return } c.JSON(http.StatusOK, models.Response{ Code: 200, Message: "创建成功", Data: cluster, }) } // UpdateCluster 更新集群 // @Summary 更新集群 // @Description 更新指定K8s集群信息 // @Tags K8s集群 // @Accept json // @Produce json // @Security Bearer // @Param id path int true "集群ID" // @Param request body models.CreateKubernetesClusterRequest true "更新请求" // @Success 200 {object} models.Response{data=models.KubernetesCluster} // @Router /kubernetes-clusters/{id} [put] func (h *KubernetesHandler) UpdateCluster(c *gin.Context) { id, _ := strconv.Atoi(c.Param("id")) // 用 map 接收,避免 binding:"required" 校验 // Logan.Li var req map[string]interface{} if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{ Code: 400, Message: "请求参数错误: " + err.Error(), }) return } var cluster models.KubernetesCluster if err := h.db.First(&cluster, id).Error; err != nil { c.JSON(http.StatusNotFound, models.Response{ Code: 404, Message: "集群不存在", }) return } // 逐字段更新:空值不覆盖原值(防止凭据被清空) // Logan.Li updates := map[string]interface{}{} if v, ok := req["name"]; ok && v != nil && v != "" { updates["name"] = v } if v, ok := req["version"]; ok { updates["version"] = v } if v, ok := req["description"]; ok { updates["description"] = v } if v, ok := req["config_type"]; ok { updates["config_type"] = v } if v, ok := req["api_server"]; ok && v != nil && v != "" { updates["api_server"] = v updates["api_endpoint"] = v } // Token / KubeConfig:空值不覆盖(前端编辑时可能拿不到原值) if v, ok := req["token"]; ok && v != nil && v != "" { updates["token"] = v } if v, ok := req["kube_config"]; ok && v != nil && v != "" { updates["kube_config"] = v } if len(updates) == 0 { c.JSON(http.StatusOK, models.Response{ Code: 200, Message: "无更新内容", Data: cluster, }) return } if err := h.db.Model(&cluster).Updates(updates).Error; err != nil { c.JSON(http.StatusInternalServerError, models.Response{ Code: 500, Message: "更新失败", }) return } h.db.First(&cluster, id) c.JSON(http.StatusOK, models.Response{ Code: 200, Message: "更新成功", Data: cluster, }) } // GetClusterDetail 返回集群完整信息(含 token/kube_config,供编辑回填) // Logan.Li func (h *KubernetesHandler) GetClusterDetail(c *gin.Context) { id, _ := strconv.Atoi(c.Param("id")) var cluster models.KubernetesCluster if err := h.db.First(&cluster, id).Error; err != nil { c.JSON(http.StatusNotFound, models.Response{ Code: 404, Message: "集群不存在", }) return } // 返回时手动脱敏:token/kube_config 只返回是否存在标记,不返回原文 c.JSON(http.StatusOK, models.Response{ Code: 200, Message: "success", Data: map[string]interface{}{ "id": cluster.ID, "name": cluster.Name, "code": cluster.Code, "version": cluster.Version, "status": cluster.Status, "config_type": cluster.ConfigType, "api_server": cluster.APIServer, "api_endpoint": cluster.APIEndpoint, "description": cluster.Description, "has_token": cluster.Token != "", "has_kube_config": cluster.KubeConfig != "", "node_count": cluster.NodeCount, "pod_count": cluster.PodCount, "labels": cluster.Labels, "created_at": cluster.CreatedAt, "updated_at": cluster.UpdatedAt, }, }) } // DeleteCluster 删除集群 // @Summary 删除集群 // @Description 删除指定K8s集群 // @Tags K8s集群 // @Produce json // @Security Bearer // @Param id path int true "集群ID" // @Success 200 {object} models.Response // @Router /kubernetes-clusters/{id} [delete] func (h *KubernetesHandler) DeleteCluster(c *gin.Context) { id, _ := strconv.Atoi(c.Param("id")) if err := h.db.Delete(&models.KubernetesCluster{}, id).Error; err != nil { c.JSON(http.StatusInternalServerError, models.Response{ Code: 500, Message: "删除失败", }) return } c.JSON(http.StatusOK, models.Response{ Code: 200, Message: "删除成功", }) } // ListNodes 获取节点列表 // @Summary 获取节点列表 // @Description 获取指定集群的节点列表 // @Tags K8s集群 // @Produce json // @Security Bearer // @Param id path int true "集群ID" // @Success 200 {object} models.Response{data=[]models.KubernetesNode} // @Router /kubernetes-clusters/{id}/nodes [get] func (h *KubernetesHandler) ListNodes(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) 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 } // Attempt to fetch metrics (best-effort, metrics-server may not be installed) metrics := map[string]k8sclient.NodeMetricsItem{} if m, err := client.GetNodeMetrics(); err == nil { for _, item := range m.Items { metrics[item.Metadata.Name] = item } } 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"` // Metrics (optional, omitted if empty) CPUUsageM string `json:"cpu_usage_m,omitempty"` // e.g. "122m" MemUsageMi string `json:"mem_usage_mi,omitempty"` // e.g. "2105Mi" } 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 } } s := 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"]), } if m, ok := metrics[n.Metadata.Name]; ok { // K8s metrics API returns CPU in nanocores ("80410173n"), memory in Ki ("2154292Ki") s.CPUUsageM = formatCpuFromNano(m.Usage["cpu"]) s.MemUsageMi = formatMemFromKi(m.Usage["memory"]) } result = append(result, s) } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result}) } // ListPods 获取Pod列表 // @Summary 获取Pod列表 // @Description 获取指定集群的Pod列表 // @Tags K8s集群 // @Produce json // @Security Bearer // @Param id path int true "集群ID" // @Param namespace query string false "命名空间" // @Success 200 {object} models.Response{data=[]models.KubernetesPod} // @Router /kubernetes-clusters/{id}/pods [get] func (h *KubernetesHandler) ListPods(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Query("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 } // M2: inject pod metrics (best-effort) // Logan.Li podMetrics := buildPodMetricsMap(client, namespace) 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"` CPUUsageM string `json:"cpu_usage_m,omitempty"` // M2: e.g. "50m" MemUsageMi string `json:"mem_usage_mi,omitempty"` // M2: e.g. "128Mi" } 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 } ps := 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, } if m, ok := podMetrics[p.Metadata.Namespace+"/"+p.Metadata.Name]; ok { ps.CPUUsageM = m.CPU ps.MemUsageMi = m.Mem } result = append(result, ps) } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result}) } // ListDeployments 获取Deployment列表 // @Summary 获取Deployment列表 // @Description 获取指定集群的Deployment列表 // @Tags K8s集群 // @Produce json // @Security Bearer // @Param id path int true "集群ID" // @Param namespace query string false "命名空间" // @Success 200 {object} models.Response{data=[]models.KubernetesDeployment} // @Router /kubernetes-clusters/{id}/deployments [get] func (h *KubernetesHandler) ListDeployments(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Query("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 } 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列表 // @Summary 获取Service列表 // @Description 获取指定集群的Service列表 // @Tags K8s集群 // @Produce json // @Security Bearer // @Param id path int true "集群ID" // @Param namespace query string false "命名空间" // @Success 200 {object} models.Response{data=[]models.KubernetesService} // @Router /kubernetes-clusters/{id}/services [get] func (h *KubernetesHandler) ListServices(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Query("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 } 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列表 // @Summary 获取Ingress列表 // @Description 获取指定集群的Ingress列表 // @Tags K8s集群 // @Produce json // @Security Bearer // @Param id path int true "集群ID" // @Param namespace query string false "命名空间" // @Success 200 {object} models.Response{data=[]models.KubernetesIngress} // @Router /kubernetes-clusters/{id}/ingresses [get] func (h *KubernetesHandler) ListIngresses(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Query("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 } 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 func (h *KubernetesHandler) GetTopology(c *gin.Context) { clusterID := c.Param("id") var layout models.TopologyLayout result := h.db.Where("cluster_id = ?", clusterID).First(&layout) if result.Error != nil { c.JSON(http.StatusOK, models.Response{Code: 200, Data: &models.TopologyLayout{}}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Data: layout}) } // SaveTopology 保存集群拓扑布局 // Logan.Li func (h *KubernetesHandler) SaveTopology(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) var req struct { LayoutData string `json:"layout_data"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"}) return } var layout models.TopologyLayout h.db.Where("cluster_id = ?", clusterID).First(&layout) layout.ClusterID = uint(clusterID) layout.LayoutData = req.LayoutData if layout.ID == 0 { h.db.Create(&layout) } else { h.db.Save(&layout) } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "保存成功"}) } // SyncCluster 同步集群资源(连接真实K8s API拉取数据) func (h *KubernetesHandler) SyncCluster(c *gin.Context) { id, _ := strconv.Atoi(c.Param("id")) var cluster models.KubernetesCluster 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{} // Fetch node metrics (best-effort — metrics-server may not be installed) // Logan.Li nodeMetrics := map[string]k8sclient.NodeMetricsItem{} if m, err := client.GetNodeMetrics(); err == nil { for _, item := range m.Items { nodeMetrics[item.Metadata.Name] = item } } else { log.Printf("[K8s] sync: GetNodeMetrics failed (metrics-server not installed?): %v", err) } // Fetch pod metrics (best-effort) // Logan.Li podMetrics := map[string]k8sclient.PodMetricsItem{} // key: "namespace/podname" if pm, err := client.ListPodMetrics(""); err == nil { for _, item := range pm.Items { podMetrics[item.Metadata.Namespace+"/"+item.Metadata.Name] = item } } // 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 kubernetes_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() // Compute CPU/memory usage percentage from metrics-server data // Logan.Li cpuUsagePct, memUsagePct := 0.0, 0.0 if m, ok := nodeMetrics[n.Metadata.Name]; ok { cpuCores := parseIntResource(n.Status.Capacity["cpu"]) memMB := parseMemResource(n.Status.Capacity["memory"]) if cpuCores > 0 { cpuMillicores := parseCpuNanoToMilli(m.Usage["cpu"]) cpuUsagePct = float64(cpuMillicores) / float64(cpuCores*1000) * 100 } if memMB > 0 { memMi := parseMemKiToMi(m.Usage["memory"]) memUsagePct = float64(memMi) / float64(memMB) * 100 } } err2 := h.db.Exec(`INSERT INTO kubernetes_nodes (cluster_id, name, ip, role, status, os, kubelet_version, cpu, memory, max_pods, cpu_usage, memory_usage, 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"]), cpuUsagePct, memUsagePct, 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.KubernetesPod{}) 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.KubernetesPod{ 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.KubernetesDeployment{}) 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.KubernetesDeployment{ 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.KubernetesService{}) 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.KubernetesService{ 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.KubernetesIngress{}) 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.KubernetesIngress{ 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, }) } // ============================================================================ // Pod Logs // ============================================================================ // GetPodLogs 获取 Pod 日志(支持 stream=true 实时流) func (h *KubernetesHandler) GetPodLogs(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") podName := c.Param("podName") container := c.Query("container") tailLines, _ := strconv.Atoi(c.DefaultQuery("tail", "200")) stream := c.Query("stream") == "true" client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } if stream { // Real-time streaming via chunked transfer body, err := client.GetPodLogsStream(namespace, podName, container, tailLines) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "获取日志流失败: " + err.Error()}) return } defer body.Close() c.Header("Content-Type", "text/plain; charset=utf-8") c.Header("Cache-Control", "no-cache") c.Header("Connection", "keep-alive") c.Header("X-Content-Type-Options", "nosniff") c.Stream(func(w io.Writer) bool { buf := make([]byte, 4096) n, err := body.Read(buf) if n > 0 { w.Write(buf[:n]) return true } if err != nil { return false } return true }) return } logs, err := client.GetPodLogs(namespace, podName, container, tailLines) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "获取日志失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Data: gin.H{"logs": logs, "pod": podName, "namespace": namespace}}) } // ============================================================================ // WebSocket Exec Terminal (交互式 bash) // ============================================================================ // ExecTerminal WebSocket 终端 — 前端 WS → 后端 → K8s WebSocket channel 协议 // // 直接通过 K8s API 的 exec 端点(WebSocket channel.k8s.io v4 协议)建立 // 交互式终端,无需在容器中安装 kubectl。 // // 数据流: // // Frontend WS (JSON) ←→ Backend ←→ K8s API WS (channel frames) // // K8s channel 协议: 每条消息 [1字节 channel ID] + [payload] // // 0 = stdin (client→server) // 1 = stdout (server→client) // 2 = stderr (server→client) // 3 = error (server→client, JSON) // 4 = resize (client→server, JSON {"Height","Width"}) func (h *KubernetesHandler) ExecTerminal(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") podName := c.Param("podName") container := c.Query("container") command := c.Query("command") if command == "" { command = "bash" } client, cluster, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } // Build exec WebSocket URL — https→wss / http→ws execURL := client.BaseURL if strings.HasPrefix(execURL, "https://") { execURL = "wss://" + strings.TrimPrefix(execURL, "https://") } else { execURL = "ws://" + strings.TrimPrefix(execURL, "http://") } params := url.Values{} params.Set("command", command) params.Set("stdin", "true") params.Set("stdout", "true") params.Set("stderr", "true") params.Set("tty", "true") if container != "" { params.Set("container", container) } execURL += fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/exec?%s", url.PathEscape(namespace), url.PathEscape(podName), params.Encode()) // Dial K8s API server via WebSocket with channel protocol subprotocol k8sDialer := &websocket.Dialer{ TLSClientConfig: client.TLSConfig, Subprotocols: []string{"v4.channel.k8s.io", "v3.channel.k8s.io", "v5.channel.k8s.io"}, HandshakeTimeout: 10 * time.Second, } k8sHeader := http.Header{} if client.BearerToken != "" { k8sHeader.Set("Authorization", "Bearer "+client.BearerToken) } k8sConn, _, err := k8sDialer.Dial(execURL, k8sHeader) if err != nil { log.Printf("[K8sExec] dial K8s exec WS failed: %v", err) c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s exec 连接失败: " + err.Error()}) return } defer k8sConn.Close() // Upgrade frontend connection to WebSocket frontendConn, err := upgrader.Upgrade(c.Writer, c.Request, nil) if err != nil { return } defer frontendConn.Close() done := make(chan struct{}, 2) // Frontend WS → K8s WS (stdin + resize) go func() { defer func() { done <- struct{}{} }() for { msgType, msg, err := frontendConn.ReadMessage() if err != nil { return } _ = msgType // Parse frontend JSON protocol: {status, data} or {status, cols, rows} var wsMsg WSMessage if err := json.Unmarshal(msg, &wsMsg); err != nil { continue } if wsMsg.Status == 0 && wsMsg.Data != "" { // stdin: prepend channel 0 byte frame := make([]byte, 1+len(wsMsg.Data)) frame[0] = 0 // channel 0 = stdin copy(frame[1:], wsMsg.Data) if err := k8sConn.WriteMessage(websocket.BinaryMessage, frame); err != nil { return } } else if wsMsg.Status == 1 && wsMsg.Cols > 0 { // resize: channel 4 + JSON resizeJSON, _ := json.Marshal(struct { Height uint16 `json:"Height"` Width uint16 `json:"Width"` }{Height: uint16(wsMsg.Rows), Width: uint16(wsMsg.Cols)}) frame := make([]byte, 1+len(resizeJSON)) frame[0] = 4 // channel 4 = resize copy(frame[1:], resizeJSON) if err := k8sConn.WriteMessage(websocket.BinaryMessage, frame); err != nil { return } } } }() // K8s WS → Frontend WS (stdout/stderr/error) go func() { defer func() { done <- struct{}{} }() for { msgType, msg, err := k8sConn.ReadMessage() if err != nil { // K8s connection closed — notify frontend frontendConn.WriteJSON(WSMessage{Status: 2, Data: "\r\n\x1b[33m[K8s 连接已关闭]\x1b[0m\r\n"}) return } _ = msgType if len(msg) == 0 { continue } channel := msg[0] payload := msg[1:] switch channel { case 1, 2: // stdout / stderr → forward to frontend as terminal output frontendConn.WriteJSON(WSMessage{Status: 0, Data: string(payload)}) case 3: // error → forward as error message errMsg := string(payload) frontendConn.WriteJSON(WSMessage{Status: 2, Data: "\r\n\x1b[31m" + errMsg + "\x1b[0m\r\n"}) } } }() <-done _ = cluster } // ============================================================================ // Exec / File Browser // ============================================================================ // ExecInPod 在 Pod 中执行命令 (文件浏览 / 单次命令) func (h *KubernetesHandler) ExecInPod(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") podName := c.Param("podName") var req struct { Command []string `json:"command"` Container string `json:"container"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"}) return } if len(req.Command) == 0 { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "command 不能为空"}) return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } result, err := client.ExecInPod(namespace, podName, req.Container, req.Command) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "执行失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Data: result}) } // ============================================================================ // Scale (副本数加减) // ============================================================================ // ScaleDeployment 缩放 Deployment 副本数 func (h *KubernetesHandler) ScaleDeployment(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") name := c.Param("name") var req struct { Replicas int `json:"replicas"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"}) return } if req.Replicas < 0 { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "副本数不能为负数"}) return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } if err := client.ScaleDeployment(namespace, name, req.Replicas); err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "扩缩容失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Message: fmt.Sprintf("已设置为 %d 副本", req.Replicas)}) } // RestartDeployment 重启 Deployment (滚动重启) func (h *KubernetesHandler) RestartDeployment(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") name := c.Param("name") client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } if err := client.RestartDeployment(namespace, name); err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "重启失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "重启指令已下发"}) } // PauseDeployment 暂停 Deployment 滚动更新(spec.paused=true) func (h *KubernetesHandler) PauseDeployment(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") name := c.Param("name") client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } if err := client.PauseDeployment(namespace, name); err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "暂停更新失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "已暂停滚动更新"}) } // ResumeDeployment 恢复 Deployment 滚动更新(spec.paused=false) func (h *KubernetesHandler) ResumeDeployment(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") name := c.Param("name") client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } if err := client.ResumeDeployment(namespace, name); err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "恢复更新失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "已恢复滚动更新"}) } // ============================================================================ // HPA 管理 // ============================================================================ // ListHPAs 获取 HPA 列表 func (h *KubernetesHandler) ListHPAs(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } hpas, err := client.ListHPAs() 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, Data: hpas.Items}) } // CreateHPA 创建 HPA func (h *KubernetesHandler) CreateHPA(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) var req k8sclient.CreateHPARequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()}) return } if req.Name == "" || req.Namespace == "" || req.TargetName == "" { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "name/namespace/targetName 必填"}) return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } hpa, err := client.CreateHPA(req) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "创建 HPA 失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "HPA 创建成功", Data: hpa}) } // DeleteHPA 删除 HPA func (h *KubernetesHandler) DeleteHPA(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") name := c.Param("name") client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } if err := client.DeleteHPA(namespace, name); err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "删除 HPA 失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "HPA 已删除"}) } // ============================================================================ // CronJob 管理 // ============================================================================ // ListCronJobs 获取 CronJob 列表 func (h *KubernetesHandler) ListCronJobs(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } cronjobs, err := client.ListCronJobs() 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, Data: cronjobs.Items}) } // CreateCronJob 创建 CronJob func (h *KubernetesHandler) CreateCronJob(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) var req k8sclient.CreateCronJobRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()}) return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } cj, err := client.CreateCronJob(req) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "创建 CronJob 失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "CronJob 创建成功", Data: cj}) } // SuspendCronJob 暂停/恢复 CronJob func (h *KubernetesHandler) SuspendCronJob(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") name := c.Param("name") var req struct { Suspend bool `json:"suspend"` } c.ShouldBindJSON(&req) client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } if err := client.SuspendCronJob(namespace, name, req.Suspend); err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "操作失败: " + err.Error()}) return } action := "恢复" if req.Suspend { action = "暂停" } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "CronJob 已" + action}) } // DeleteCronJob 删除 CronJob func (h *KubernetesHandler) DeleteCronJob(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") name := c.Param("name") client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } if err := client.DeleteCronJob(namespace, name); err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "删除 CronJob 失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "CronJob 已删除"}) } // TriggerCronJob 手动触发 CronJob func (h *KubernetesHandler) TriggerCronJob(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") name := c.Param("name") client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } if err := client.TriggerCronJob(namespace, name); err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "触发失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Message: "手动触发成功"}) } // ============================================================================ // Deployment CI 集成 (Webhook 更新镜像) // ============================================================================ // UpdateDeploymentImage 更新 Deployment 容器镜像 (CI webhook) func (h *KubernetesHandler) UpdateDeploymentImage(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) namespace := c.Param("namespace") name := c.Param("name") var req struct { Image string `json:"image"` ContainerName string `json:"container_name"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"}) return } if req.Image == "" { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "image 不能为空"}) return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } if err := client.UpdateDeploymentImage(namespace, name, req.ContainerName, req.Image); err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "更新镜像失败: " + err.Error()}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Message: fmt.Sprintf("镜像已更新为 %s,正在滚动更新...", req.Image)}) } // ============================================================================ // Namespace // ============================================================================ // ListNamespaces 获取命名空间列表 func (h *KubernetesHandler) ListNamespaces(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) // Support clusterCode param from new routes if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } namespaces, err := client.ListNamespaces() 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, Data: namespaces}) } // ============================================================================ // ConfigMap & Secret (Kuboard-style namespace resources) // ============================================================================ // ListConfigMaps 获取命名空间下的 ConfigMap 列表 func (h *KubernetesHandler) ListConfigMaps(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } namespace := c.Param("namespace") client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } cms, err := client.ListConfigMaps(namespace) 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, Data: cms.Items}) } // ListSecrets 获取命名空间下的 Secret 列表(仅名称+类型,不含值) func (h *KubernetesHandler) ListSecrets(c *gin.Context) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } namespace := c.Param("namespace") client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) return } secrets, err := client.ListSecrets(namespace) 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, Data: secrets.Items}) } // ============================================================================ // Helpers // ============================================================================ // resolveClusterCode resolves a cluster code to its numeric ID func (h *KubernetesHandler) resolveClusterCode(code string) int { if code == "" { return 0 } var cluster models.KubernetesCluster if err := h.db.Where("code = ?", code).First(&cluster).Error; err != nil { return 0 } return int(cluster.ID) } // getK8sClientByCode resolves cluster by code, falls back to ID func (h *KubernetesHandler) getK8sClientByCode(c *gin.Context) (*k8sclient.Client, *models.KubernetesCluster, error) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } if clusterID == 0 { return nil, nil, fmt.Errorf("cluster not found") } return h.getK8sClient(clusterID) } // ============================================================================ // NS wrappers — read namespace from path (Kuboard-style routes) // ============================================================================ func (h *KubernetesHandler) ListDeploymentsNS(c *gin.Context) { h.listDeploymentsWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) ListPodsNS(c *gin.Context) { h.listPodsWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) ListServicesNS(c *gin.Context) { h.listServicesWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) ListIngressesNS(c *gin.Context) { h.listIngressesWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) ListHPAsNS(c *gin.Context) { h.listHPAsWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) CreateHPANS(c *gin.Context) { h.createHPAWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) DeleteHPANS(c *gin.Context) { h.deleteHPAWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) ListCronJobsNS(c *gin.Context) { h.listCronJobsWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) CreateCronJobNS(c *gin.Context) { h.createCronJobWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) SuspendCronJobNS(c *gin.Context) { h.suspendCronJobWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) TriggerCronJobNS(c *gin.Context) { h.triggerCronJobWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) DeleteCronJobNS(c *gin.Context) { h.deleteCronJobWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) ScaleDeploymentNS(c *gin.Context) { h.scaleDeploymentWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) RestartDeploymentNS(c *gin.Context) { h.restartDeploymentWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) UpdateDeploymentImageNS(c *gin.Context) { h.updateDeploymentImageWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) PauseDeploymentNS(c *gin.Context) { h.pauseDeploymentWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) ResumeDeploymentNS(c *gin.Context) { h.resumeDeploymentWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) GetPodLogsNS(c *gin.Context) { h.getPodLogsWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) ExecInPodNS(c *gin.Context) { h.execInPodWithNS(c, c.Param("namespace")) } // Internal helpers that resolve clusterCode and inject namespace func (h *KubernetesHandler) listDeploymentsWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } deps, err := client.ListDeployments() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } 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"` Paused bool `json:"paused"` Selector map[string]string `json:"selector,omitempty"` // matchLabels — 前端 Service Tab 关联用 } result := []DepSummary{} for _, d := range deps.Items { if d.Metadata.Namespace != ns { 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, Paused: d.Spec.Paused, Selector: func() map[string]string { if d.Spec.Selector != nil { return d.Spec.Selector.MatchLabels }; return nil }(), }) } c.JSON(200, models.Response{Code: 200, Data: result}) } func (h *KubernetesHandler) listPodsWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } pods, err := client.ListPods() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } // M2: inject pod metrics (best-effort) // Logan.Li podMetrics := buildPodMetricsMap(client, ns) 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"` CPUUsageM string `json:"cpu_usage_m,omitempty"` MemUsageMi string `json:"mem_usage_mi,omitempty"` } result := []PodSummary{} for _, p := range pods.Items { if p.Metadata.Namespace != ns { 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 } ps := 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, } if m, ok := podMetrics[p.Metadata.Namespace+"/"+p.Metadata.Name]; ok { ps.CPUUsageM = m.CPU ps.MemUsageMi = m.Mem } result = append(result, ps) } c.JSON(200, models.Response{Code: 200, Data: result}) } func (h *KubernetesHandler) listServicesWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } svcs, err := client.ListServices() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } type SvcPortDetail struct { Name string `json:"name,omitempty"` Port int `json:"port"` TargetPort interface{} `json:"targetPort"` NodePort int `json:"nodePort,omitempty"` Protocol string `json:"protocol,omitempty"` } type SvcSummary struct { Namespace string `json:"namespace"` Name string `json:"name"` Type string `json:"type"` ClusterIP string `json:"cluster_ip"` Ports int `json:"ports"` PortList []SvcPortDetail `json:"portList,omitempty"` Selector map[string]string `json:"selector,omitempty"` } result := []SvcSummary{} for _, s := range svcs.Items { if s.Metadata.Namespace != ns { continue } ports := []SvcPortDetail{} for _, p := range s.Spec.Ports { ports = append(ports, SvcPortDetail{ Name: p.Name, Port: p.Port, TargetPort: p.TargetPort, NodePort: p.NodePort, Protocol: p.Protocol, }) } 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), PortList: ports, Selector: s.Spec.Selector, }) } c.JSON(200, models.Response{Code: 200, Data: result}) } func (h *KubernetesHandler) listIngressesWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } ings, err := client.ListIngresses() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } type PathDetail struct { Path string `json:"path"` PathType string `json:"pathType,omitempty"` BackendSvc string `json:"backendService,omitempty"` BackendPort string `json:"backendPort,omitempty"` } type RuleDetail struct { Host string `json:"host"` Paths []PathDetail `json:"paths"` } type IngSummary struct { Namespace string `json:"namespace"` Name string `json:"name"` Hosts []string `json:"hosts"` TLS bool `json:"tls"` IngressClass string `json:"ingressClass,omitempty"` Rules []RuleDetail `json:"rules,omitempty"` } result := []IngSummary{} for _, ing := range ings.Items { if ing.Metadata.Namespace != ns { continue } hosts := []string{} rules := []RuleDetail{} for _, rule := range ing.Spec.Rules { if rule.Host != "" { hosts = append(hosts, rule.Host) } rd := RuleDetail{Host: rule.Host, Paths: []PathDetail{}} if rule.HTTP != nil { for _, p := range rule.HTTP.Paths { pd := PathDetail{Path: p.Path, PathType: p.PathType, BackendSvc: p.Backend.Service.Name} if p.Backend.Service.Port.Number > 0 { pd.BackendPort = strconv.Itoa(p.Backend.Service.Port.Number) } else if p.Backend.Service.Port.Name != "" { pd.BackendPort = p.Backend.Service.Port.Name } rd.Paths = append(rd.Paths, pd) } } rules = append(rules, rd) } // ingressClassName from annotations ingClass := "" if ing.Metadata.Annotations != nil { ingClass = ing.Metadata.Annotations["kubernetes.io/ingress.class"] } result = append(result, IngSummary{ Namespace: ing.Metadata.Namespace, Name: ing.Metadata.Name, Hosts: hosts, TLS: len(ing.Spec.TLS) > 0, IngressClass: ingClass, Rules: rules, }) } c.JSON(200, models.Response{Code: 200, Data: result}) } func (h *KubernetesHandler) listHPAsWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } hpas, err := client.ListHPAs() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } type HPASummary struct { Namespace string `json:"namespace"` Name string `json:"name"` TargetKind string `json:"target_kind"` TargetName string `json:"target_name"` MinReplicas int `json:"min_replicas"` MaxReplicas int `json:"max_replicas"` CPUPercent int `json:"cpu_percent"` MemPercent int `json:"mem_percent"` CurrentReplicas int `json:"current_replicas"` } result := []HPASummary{} for _, h := range hpas.Items { if h.Metadata.Namespace != ns { continue } cpu := 0; mem := 0 for _, m := range h.Spec.Metrics { if m.Resource != nil { if m.Resource.Name == "cpu" { cpu = m.Resource.Target.AverageUtilization } if m.Resource.Name == "memory" { mem = m.Resource.Target.AverageUtilization } } } result = append(result, HPASummary{ Namespace: h.Metadata.Namespace, Name: h.Metadata.Name, TargetKind: h.Spec.ScaleTargetRef.Kind, TargetName: h.Spec.ScaleTargetRef.Name, MinReplicas: h.Spec.MinReplicas, MaxReplicas: h.Spec.MaxReplicas, CPUPercent: cpu, MemPercent: mem, CurrentReplicas: h.Status.CurrentReplicas, }) } c.JSON(200, models.Response{Code: 200, Data: result}) } func (h *KubernetesHandler) createHPAWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } var req k8sclient.CreateHPARequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误"}); return } if req.Namespace == "" { req.Namespace = ns } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } hpa, err := client.CreateHPA(req) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "创建失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Data: hpa}) } func (h *KubernetesHandler) deleteHPAWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.DeleteHPA(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "删除失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "已删除"}) } func (h *KubernetesHandler) listCronJobsWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } cjs, err := client.ListCronJobs() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } filtered := []k8sclient.CronJobItem{} for _, cj := range cjs.Items { if cj.Metadata.Namespace == ns { filtered = append(filtered, cj) } } c.JSON(200, models.Response{Code: 200, Data: filtered}) } func (h *KubernetesHandler) createCronJobWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } var req k8sclient.CreateCronJobRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误"}); return } if req.Namespace == "" { req.Namespace = ns } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } cj, err := client.CreateCronJob(req) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "创建失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Data: cj}) } func (h *KubernetesHandler) suspendCronJobWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } var req struct { Suspend bool `json:"suspend"` }; c.ShouldBindJSON(&req) client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.SuspendCronJob(ns, c.Param("name"), req.Suspend); err != nil { c.JSON(500, models.Response{Code: 500, Message: "操作失败: " + err.Error()}); return } action := "恢复"; if req.Suspend { action = "暂停" } c.JSON(200, models.Response{Code: 200, Message: "CronJob 已" + action}) } func (h *KubernetesHandler) triggerCronJobWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.TriggerCronJob(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "触发失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "手动触发成功"}) } func (h *KubernetesHandler) deleteCronJobWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.DeleteCronJob(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "删除失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "已删除"}) } func (h *KubernetesHandler) scaleDeploymentWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } var req struct { Replicas int `json:"replicas"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.ScaleDeployment(ns, c.Param("name"), req.Replicas); err != nil { c.JSON(500, models.Response{Code: 500, Message: "扩缩容失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: fmt.Sprintf("已设置为 %d 副本", req.Replicas)}) } func (h *KubernetesHandler) restartDeploymentWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.RestartDeployment(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "重启失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "重启指令已下发"}) } func (h *KubernetesHandler) updateDeploymentImageWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } var req struct { Image string `json:"image"`; ContainerName string `json:"container_name"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.UpdateDeploymentImage(ns, c.Param("name"), req.ContainerName, req.Image); err != nil { c.JSON(500, models.Response{Code: 500, Message: "更新镜像失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: fmt.Sprintf("镜像已更新为 %s", req.Image)}) } func (h *KubernetesHandler) pauseDeploymentWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.PauseDeployment(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "暂停更新失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "已暂停滚动更新"}) } func (h *KubernetesHandler) resumeDeploymentWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.ResumeDeployment(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "恢复更新失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "已恢复滚动更新"}) } func (h *KubernetesHandler) getPodLogsWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } podName := c.Param("podName") stream := c.Query("stream") == "true" // 日志增强参数:container / sinceTime / previous / tailLines(兼容 tail) // Logan.Li tailLines := 0 if v := c.Query("tailLines"); v != "" { tailLines, _ = strconv.Atoi(v) } if tailLines == 0 { tailLines, _ = strconv.Atoi(c.DefaultQuery("tail", "200")) } opts := k8sclient.LogOptions{ Container: c.Query("container"), TailLines: tailLines, SinceTime: c.Query("sinceTime"), Previous: c.Query("previous") == "true", } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if stream { body, err := client.GetPodLogsStreamWithOptions(ns, podName, opts) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "获取日志流失败: " + err.Error()}); return } defer body.Close() c.Header("Content-Type", "text/plain; charset=utf-8") c.Header("Cache-Control", "no-cache") c.Stream(func(w io.Writer) bool { buf := make([]byte, 4096); n, err := body.Read(buf) if n > 0 { w.Write(buf[:n]); return true } return err == nil }) return } logs, err := client.GetPodLogsWithOptions(ns, podName, opts) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "获取日志失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "success", Data: gin.H{"logs": logs, "pod": podName, "namespace": ns}}) } func (h *KubernetesHandler) execInPodWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } podName := c.Param("podName") var req struct { Command []string `json:"command"`; Container string `json:"container"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } result, err := client.ExecInPod(ns, podName, req.Container, req.Command) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "执行失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Data: result}) } // ============================================================================ // All-namespace handlers (clusterCode-based, no namespace filter) // ============================================================================ // ListDeploymentsAllNS lists all deployments across all namespaces using clusterCode. func (h *KubernetesHandler) ListDeploymentsAllNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } deps, err := client.ListDeployments() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } 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 { 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(200, models.Response{Code: 200, Data: result}) } // ListPodsAllNS lists all pods across all namespaces using clusterCode. func (h *KubernetesHandler) ListPodsAllNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } pods, err := client.ListPods() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } // M2: inject pod metrics (best-effort) // Logan.Li podMetrics := buildPodMetricsMap(client, "") 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"` CPUUsageM string `json:"cpu_usage_m,omitempty"` MemUsageMi string `json:"mem_usage_mi,omitempty"` } result := []PodSummary{} 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 } ps := 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, } if m, ok := podMetrics[p.Metadata.Namespace+"/"+p.Metadata.Name]; ok { ps.CPUUsageM = m.CPU ps.MemUsageMi = m.Mem } result = append(result, ps) } c.JSON(200, models.Response{Code: 200, Data: result}) } // ListHPAsAllNS lists all HPAs across all namespaces using clusterCode. func (h *KubernetesHandler) ListHPAsAllNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } hpas, err := client.ListHPAs() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } type HPASummary struct { Namespace string `json:"namespace"` Name string `json:"name"` TargetKind string `json:"target_kind"` TargetName string `json:"target_name"` MinReplicas int `json:"min_replicas"` MaxReplicas int `json:"max_replicas"` CPUPercent int `json:"cpu_percent"` MemPercent int `json:"mem_percent"` CurrentReplicas int `json:"current_replicas"` } result := []HPASummary{} for _, h := range hpas.Items { cpu := 0; mem := 0 for _, m := range h.Spec.Metrics { if m.Resource != nil { if m.Resource.Name == "cpu" { cpu = m.Resource.Target.AverageUtilization } if m.Resource.Name == "memory" { mem = m.Resource.Target.AverageUtilization } } } result = append(result, HPASummary{ Namespace: h.Metadata.Namespace, Name: h.Metadata.Name, TargetKind: h.Spec.ScaleTargetRef.Kind, TargetName: h.Spec.ScaleTargetRef.Name, MinReplicas: h.Spec.MinReplicas, MaxReplicas: h.Spec.MaxReplicas, CPUPercent: cpu, MemPercent: mem, CurrentReplicas: h.Status.CurrentReplicas, }) } c.JSON(200, models.Response{Code: 200, Data: result}) } // ============================================================================ // Helpers // ============================================================================ // buildPodMetricsMap fetches pod metrics and returns a map keyed by "namespace/podname" → aggregated CPU/Memory strings. // Best-effort: returns empty map if metrics-server is unavailable. // Logan.Li M2 func buildPodMetricsMap(client *k8sclient.Client, namespace string) map[string]struct{ CPU, Mem string } { m := map[string]struct{ CPU, Mem string }{} if pm, err := client.ListPodMetrics(namespace); err == nil { for _, item := range pm.Items { totalCPU := 0 totalMemMi := 0 for _, c := range item.Containers { totalCPU += parseCpuNanoToMilli(c.Usage["cpu"]) totalMemMi += parseMemKiToMi(c.Usage["memory"]) } key := item.Metadata.Namespace + "/" + item.Metadata.Name m[key] = struct{ CPU, Mem string }{ CPU: fmt.Sprintf("%dm", totalCPU), Mem: fmt.Sprintf("%dMi", totalMemMi), } } } else { log.Printf("[K8s] ListPodMetrics failed (metrics-server not installed?): %v", err) } return m } 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) } } // formatCpuFromNano converts K8s metrics CPU (nanocores "80410173n") to millicores "80m" func formatCpuFromNano(raw string) string { if raw == "" { return "" } val := raw val = strings.TrimSuffix(val, "n") var nano int64 fmt.Sscanf(val, "%d", &nano) milli := nano / 1000000 if milli < 1 { milli = 1 } return fmt.Sprintf("%dm", milli) } // parseCpuNanoToMilli converts K8s metrics CPU (nanocores "100354992n") to millicores (int) // Logan.Li func parseCpuNanoToMilli(raw string) int { if raw == "" { return 0 } val := strings.TrimSuffix(raw, "n") var nano int64 fmt.Sscanf(val, "%d", &nano) milli := nano / 1000000 if milli < 1 { milli = 1 } return int(milli) } // formatMemFromKi converts K8s metrics memory (Ki "2154292Ki") to Mi "2105Mi" func formatMemFromKi(raw string) string { if raw == "" { return "" } val := strings.TrimSuffix(raw, "Ki") var ki int64 fmt.Sscanf(val, "%d", &ki) mi := ki / 1024 if mi < 1 { mi = 1 } return fmt.Sprintf("%dMi", mi) } // parseMemKiToMi converts K8s metrics memory (Ki "2154292Ki") to Mi (int) // Logan.Li func parseMemKiToMi(raw string) int { if raw == "" { return 0 } val := strings.TrimSuffix(raw, "Ki") var ki int64 fmt.Sscanf(val, "%d", &ki) return int(ki / 1024) } // GetResourceYAMLNS returns the YAML of a K8s resource func (h *KubernetesHandler) GetResourceYAMLNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } ns := c.Param("namespace") resource := c.Param("resource") name := c.Param("name") // Build API path based on resource type var apiPath string switch resource { case "deployments", "deployment": apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s", ns, name) case "statefulsets", "statefulset": apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/statefulsets/%s", ns, name) case "daemonsets", "daemonset": apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/daemonsets/%s", ns, name) case "replicasets", "replicaset": apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/replicasets/%s", ns, name) case "pods", "pod": apiPath = fmt.Sprintf("/api/v1/namespaces/%s/pods/%s", ns, name) case "services", "service": apiPath = fmt.Sprintf("/api/v1/namespaces/%s/services/%s", ns, name) case "ingresses", "ingress": apiPath = fmt.Sprintf("/apis/networking.k8s.io/v1/namespaces/%s/ingresses/%s", ns, name) case "configmaps", "configmap": apiPath = fmt.Sprintf("/api/v1/namespaces/%s/configmaps/%s", ns, name) case "secrets", "secret": apiPath = fmt.Sprintf("/api/v1/namespaces/%s/secrets/%s", ns, name) case "pvcs", "persistentvolumeclaim": apiPath = fmt.Sprintf("/api/v1/namespaces/%s/persistentvolumeclaims/%s", ns, name) case "hpas", "hpa", "horizontalpodautoscaler": apiPath = fmt.Sprintf("/apis/autoscaling/v2/namespaces/%s/horizontalpodautoscalers/%s", ns, name) case "cronjobs", "cronjob": apiPath = fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", ns, name) default: c.JSON(400, models.Response{Code: 400, Message: "unsupported resource type: " + resource}); return } yamlBytes, err := client.GetResourceYAML(apiPath) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "获取YAML失败: " + err.Error()}); return } c.Header("Content-Type", "application/yaml") c.String(200, string(yamlBytes)) } // UpdateResourceYAMLNS applies YAML to update a K8s resource func (h *KubernetesHandler) UpdateResourceYAMLNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } ns := c.Param("namespace") resource := c.Param("resource") name := c.Param("name") yamlData, err := io.ReadAll(c.Request.Body) if err != nil { c.JSON(400, models.Response{Code: 400, Message: "读取body失败"}); return } var apiPath string switch resource { case "deployments", "deployment": apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s", ns, name) case "statefulsets", "statefulset": apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/statefulsets/%s", ns, name) case "daemonsets", "daemonset": apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/daemonsets/%s", ns, name) case "replicasets", "replicaset": apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/replicasets/%s", ns, name) case "pods", "pod": apiPath = fmt.Sprintf("/api/v1/namespaces/%s/pods/%s", ns, name) case "services", "service": apiPath = fmt.Sprintf("/api/v1/namespaces/%s/services/%s", ns, name) case "ingresses", "ingress": apiPath = fmt.Sprintf("/apis/networking.k8s.io/v1/namespaces/%s/ingresses/%s", ns, name) case "configmaps", "configmap": apiPath = fmt.Sprintf("/api/v1/namespaces/%s/configmaps/%s", ns, name) case "secrets", "secret": apiPath = fmt.Sprintf("/api/v1/namespaces/%s/secrets/%s", ns, name) case "pvcs", "persistentvolumeclaim": apiPath = fmt.Sprintf("/api/v1/namespaces/%s/persistentvolumeclaims/%s", ns, name) case "hpas", "hpa", "horizontalpodautoscaler": apiPath = fmt.Sprintf("/apis/autoscaling/v2/namespaces/%s/horizontalpodautoscalers/%s", ns, name) case "cronjobs", "cronjob": apiPath = fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", ns, name) default: c.JSON(400, models.Response{Code: 400, Message: "unsupported resource type: " + resource}); return } if err := client.UpdateResourceYAML(apiPath, yamlData); err != nil { c.JSON(500, models.Response{Code: 500, Message: "更新失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "YAML已应用"}) } // resourceAPIPath builds the K8s API path for common resource types (shared by Get/Update/Delete) func resourceAPIPath(ns, resource, name string) (string, error) { switch resource { case "deployments", "deployment": return fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s", ns, name), nil case "statefulsets", "statefulset": return fmt.Sprintf("/apis/apps/v1/namespaces/%s/statefulsets/%s", ns, name), nil case "daemonsets", "daemonset": return fmt.Sprintf("/apis/apps/v1/namespaces/%s/daemonsets/%s", ns, name), nil case "replicasets", "replicaset": return fmt.Sprintf("/apis/apps/v1/namespaces/%s/replicasets/%s", ns, name), nil case "pods", "pod": return fmt.Sprintf("/api/v1/namespaces/%s/pods/%s", ns, name), nil case "services", "service": return fmt.Sprintf("/api/v1/namespaces/%s/services/%s", ns, name), nil case "ingresses", "ingress": return fmt.Sprintf("/apis/networking.k8s.io/v1/namespaces/%s/ingresses/%s", ns, name), nil case "configmaps", "configmap": return fmt.Sprintf("/api/v1/namespaces/%s/configmaps/%s", ns, name), nil case "secrets", "secret": return fmt.Sprintf("/api/v1/namespaces/%s/secrets/%s", ns, name), nil case "pvcs", "persistentvolumeclaim": return fmt.Sprintf("/api/v1/namespaces/%s/persistentvolumeclaims/%s", ns, name), nil case "hpas", "hpa", "horizontalpodautoscaler": return fmt.Sprintf("/apis/autoscaling/v2/namespaces/%s/horizontalpodautoscalers/%s", ns, name), nil case "cronjobs", "cronjob": return fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", ns, name), nil default: return "", fmt.Errorf("unsupported resource type: %s", resource) } } // DeleteResourceYAMLNS deletes a K8s resource (Service/Ingress/ConfigMap etc.) func (h *KubernetesHandler) DeleteResourceYAMLNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } ns := c.Param("namespace") resource := c.Param("resource") name := c.Param("name") apiPath, err := resourceAPIPath(ns, resource, name) if err != nil { c.JSON(400, models.Response{Code: 400, Message: err.Error()}); return } if err := client.DeleteResource(apiPath); err != nil { c.JSON(500, models.Response{Code: 500, Message: "删除失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "资源已删除"}) } // getK8sClient builds a K8s API client from cluster's stored kubeconfig func (h *KubernetesHandler) getK8sClient(clusterID int) (*k8sclient.Client, *models.KubernetesCluster, error) { var cluster models.KubernetesCluster 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 } // ============================================================================ // PVC (存储) handlers // ============================================================================ // ListPVCsNS lists PersistentVolumeClaims in a namespace (code-based) func (h *KubernetesHandler) ListPVCsNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } ns := c.Param("namespace") pvcs, err := client.ListPVCs(ns) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } type PVCSummary struct { Namespace string `json:"namespace"` Name string `json:"name"` Phase string `json:"phase"` Size string `json:"size"` StorageClass string `json:"storage_class"` } result := []PVCSummary{} for _, p := range pvcs.Items { size := "" if p.Status.Capacity != nil { if s, ok := p.Status.Capacity["storage"]; ok { size = s } } result = append(result, PVCSummary{ Namespace: p.Metadata.Namespace, Name: p.Metadata.Name, Phase: p.Status.Phase, Size: size, StorageClass: p.Spec.StorageClassName, }) } c.JSON(200, models.Response{Code: 200, Data: result}) } // DeletePVCNS deletes a PVC (code-based) func (h *KubernetesHandler) DeletePVCNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.DeletePVC(c.Param("namespace"), c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "删除失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "已删除"}) } // ============================================================================ // StatefulSet / DaemonSet handlers (code-based) // ============================================================================ // ListStatefulSetsNS lists StatefulSets in a namespace func (h *KubernetesHandler) ListStatefulSetsNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } sts, err := client.ListStatefulSets() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } ns := c.Param("namespace") type StsSummary struct { Namespace string `json:"namespace"` Name string `json:"name"` Replicas int `json:"replicas"` Ready int `json:"ready"` Image string `json:"image"` } result := []StsSummary{} for _, s := range sts.Items { if s.Metadata.Namespace != ns { continue } image := ""; if len(s.Spec.Template.Spec.Containers) > 0 { image = s.Spec.Template.Spec.Containers[0].Image } result = append(result, StsSummary{ Namespace: s.Metadata.Namespace, Name: s.Metadata.Name, Replicas: s.Spec.Replicas, Ready: s.Status.ReadyReplicas, Image: image, }) } c.JSON(200, models.Response{Code: 200, Data: result}) } // ListDaemonSetsNS lists DaemonSets in a namespace func (h *KubernetesHandler) ListDaemonSetsNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } ds, err := client.ListDaemonSets() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } ns := c.Param("namespace") type DsSummary struct { Namespace string `json:"namespace"` Name string `json:"name"` Desired int `json:"desired"` Current int `json:"current"` Ready int `json:"ready"` Image string `json:"image"` } result := []DsSummary{} for _, d := range ds.Items { if d.Metadata.Namespace != ns { continue } image := ""; if len(d.Spec.Template.Spec.Containers) > 0 { image = d.Spec.Template.Spec.Containers[0].Image } result = append(result, DsSummary{ Namespace: d.Metadata.Namespace, Name: d.Metadata.Name, Desired: d.Status.DesiredNumberScheduled, Current: d.Status.CurrentNumberScheduled, Ready: d.Status.NumberReady, Image: image, }) } c.JSON(200, models.Response{Code: 200, Data: result}) } // ============================================================================ // Pod Metrics (kubectl top pods) — per-container CPU/Memory usage via metrics-server // ============================================================================ // GetPodMetrics returns pod-level CPU/Memory metrics func (h *KubernetesHandler) GetPodMetrics(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } ns := c.Query("namespace") metrics, err := client.ListPodMetrics(ns) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "metrics API error: " + err.Error()}); return } type ContainerMetric struct { Name string `json:"name"` CPU string `json:"cpu"` Memory string `json:"memory"` } type PodMetric struct { Namespace string `json:"namespace"` Name string `json:"name"` Containers []ContainerMetric `json:"containers"` } result := []PodMetric{} for _, pm := range metrics.Items { containers := []ContainerMetric{} for _, c := range pm.Containers { containers = append(containers, ContainerMetric{c.Name, c.Usage["cpu"], c.Usage["memory"]}) } result = append(result, PodMetric{pm.Metadata.Namespace, pm.Metadata.Name, containers}) } c.JSON(200, models.Response{Code: 200, Data: result}) } // ============================================================================ // Events (kubectl get events) // ============================================================================ // ListEvents returns cluster events (optionally filtered by namespace and type) func (h *KubernetesHandler) ListEvents(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } ns := c.Query("namespace") eventType := c.Query("type") // Normal or Warning events, err := client.ListEvents(ns) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } type EventSummary struct { Namespace string `json:"namespace"` Name string `json:"name"` Type string `json:"type"` Reason string `json:"reason"` Message string `json:"message"` Object string `json:"object"` Count int `json:"count"` FirstTime string `json:"first_time"` LastTime string `json:"last_time"` } result := []EventSummary{} for _, e := range events.Items { if eventType != "" && e.Type != eventType { continue } object := fmt.Sprintf("%s/%s", e.InvolvedObject.Kind, e.InvolvedObject.Name) if e.InvolvedObject.Namespace != "" { object = fmt.Sprintf("%s/%s/%s", e.InvolvedObject.Namespace, e.InvolvedObject.Kind, e.InvolvedObject.Name) } firstTime := ""; if !e.FirstTimestamp.IsZero() { firstTime = e.FirstTimestamp.Format(time.RFC3339) } lastTime := ""; if !e.LastTimestamp.IsZero() { lastTime = e.LastTimestamp.Format(time.RFC3339) } result = append(result, EventSummary{ Namespace: e.Metadata.Namespace, Name: e.Metadata.Name, Type: e.Type, Reason: e.Reason, Message: e.Message, Object: object, Count: int(e.Count), FirstTime: firstTime, LastTime: lastTime, }) } c.JSON(200, models.Response{Code: 200, Data: result}) } // ============================================================================ // Capacity Analysis — CPU/Memory request/limit per Deployment // ============================================================================ // GetCapacityAnalysis returns resource analysis for all deployments func (h *KubernetesHandler) GetCapacityAnalysis(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } ns := c.Query("namespace") deps, err := client.ListDeployments() if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } type ContainerResource struct { Name string `json:"name"` CPUReq string `json:"cpu_request"` CPULim string `json:"cpu_limit"` MemReq string `json:"mem_request"` MemLim string `json:"mem_limit"` } type CapacityRow struct { Namespace string `json:"namespace"` Name string `json:"name"` Replicas int `json:"replicas"` Containers []ContainerResource `json:"containers"` } result := []CapacityRow{} for _, d := range deps.Items { if ns != "" && d.Metadata.Namespace != ns { continue } containers := []ContainerResource{} for _, c := range d.Spec.Template.Spec.Containers { cpuReq := c.Resources.Requests["cpu"] cpuLim := c.Resources.Limits["cpu"] memReq := c.Resources.Requests["memory"] memLim := c.Resources.Limits["memory"] containers = append(containers, ContainerResource{c.Name, cpuReq, cpuLim, memReq, memLim}) } result = append(result, CapacityRow{d.Metadata.Namespace, d.Metadata.Name, d.Spec.Replicas, containers}) } c.JSON(200, models.Response{Code: 200, Data: result}) } // ============================================================================ // YAML Apply (kubectl apply -f) // ============================================================================ // ApplyYAML applies a YAML manifest to create or update a resource func (h *KubernetesHandler) ApplyYAML(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } var req struct { YAML string `json:"yaml"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误: " + err.Error()}); return } if req.YAML == "" { c.JSON(400, models.Response{Code: 400, Message: "YAML 内容不能为空"}); return } result, err := client.ApplyYAML(req.YAML) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "Apply 失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: "资源已应用", Data: result}) } // ============================================================================ // ReplicaSet 列表 / Deployment 回滚 / 历史版本 // Logan.Li // ============================================================================ // ListReplicaSetsNS 列出 namespace 下的 ReplicaSet func (h *KubernetesHandler) ListReplicaSetsNS(c *gin.Context) { h.listReplicaSetsWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) listReplicaSetsWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } items, err := client.ListReplicaSetsRaw(ns) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } type RSSummary 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"` CreatedAt string `json:"created_at"` Revision string `json:"revision"` OwnerDeployment string `json:"owner_deployment"` } result := []RSSummary{} for _, rs := range items { image := "" if len(rs.Spec.Template.Spec.Containers) > 0 { image = rs.Spec.Template.Spec.Containers[0].Image } revision := rs.Metadata.Annotations["deployment.kubernetes.io/revision"] owner := "" for _, o := range rs.Metadata.OwnerReferences { if o.Kind == "Deployment" { owner = o.Name; break } } result = append(result, RSSummary{ Namespace: rs.Metadata.Namespace, Name: rs.Metadata.Name, Replicas: rs.Spec.Replicas, AvailableReplicas: rs.Status.AvailableReplicas, ReadyReplicas: rs.Status.ReadyReplicas, Image: image, CreatedAt: rs.Metadata.CreationTimestamp, Revision: revision, OwnerDeployment: owner, }) } c.JSON(200, models.Response{Code: 200, Message: "success", Data: result}) } // ListDeploymentRevisionsNS Deployment 发布历史(按 revision 倒序) func (h *KubernetesHandler) ListDeploymentRevisionsNS(c *gin.Context) { h.listDeploymentRevisionsWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) listDeploymentRevisionsWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } deploymentName := c.Param("name") if deploymentName == "" { c.JSON(400, models.Response{Code: 400, Message: "缺少 deployment name"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } items, err := client.ListReplicaSetsRaw(ns) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return } type Revision struct { Revision string `json:"revision"` ReplicaSet string `json:"replica_set"` Replicas int `json:"replicas"` Image string `json:"image"` CreatedAt string `json:"created_at"` Current bool `json:"current"` } result := []Revision{} for _, rs := range items { isOwner := false for _, o := range rs.Metadata.OwnerReferences { if o.Kind == "Deployment" && o.Name == deploymentName { isOwner = true; break } } if !isOwner { continue } image := "" if len(rs.Spec.Template.Spec.Containers) > 0 { image = rs.Spec.Template.Spec.Containers[0].Image } result = append(result, Revision{ Revision: rs.Metadata.Annotations["deployment.kubernetes.io/revision"], ReplicaSet: rs.Metadata.Name, Replicas: rs.Spec.Replicas, Image: image, CreatedAt: rs.Metadata.CreationTimestamp, Current: rs.Spec.Replicas > 0, }) } // 按 revision 数字倒序(最新在前) sort.Slice(result, func(i, j int) bool { ri, _ := strconv.Atoi(result[i].Revision) rj, _ := strconv.Atoi(result[j].Revision) return ri > rj }) c.JSON(200, models.Response{Code: 200, Message: "success", Data: result}) } // RollbackDeploymentNS 回滚 Deployment 到指定 revision 或指定 ReplicaSet func (h *KubernetesHandler) RollbackDeploymentNS(c *gin.Context) { h.rollbackDeploymentWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) rollbackDeploymentWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } deploymentName := c.Param("name") if deploymentName == "" { c.JSON(400, models.Response{Code: 400, Message: "缺少 deployment name"}); return } var req struct { Revision int `json:"revision"` ReplicaSet string `json:"replicaSet"` } _ = c.ShouldBindJSON(&req) // 允许空 body(但下面会校验) client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } targetRS := req.ReplicaSet // 通过 revision 定位 ReplicaSet if targetRS == "" && req.Revision > 0 { items, err := client.ListReplicaSetsRaw(ns) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "获取 ReplicaSet 列表失败: " + err.Error()}); return } for _, rs := range items { isOwner := false for _, o := range rs.Metadata.OwnerReferences { if o.Kind == "Deployment" && o.Name == deploymentName { isOwner = true; break } } if !isOwner { continue } if rs.Metadata.Annotations["deployment.kubernetes.io/revision"] == strconv.Itoa(req.Revision) { targetRS = rs.Metadata.Name break } } if targetRS == "" { c.JSON(404, models.Response{Code: 404, Message: fmt.Sprintf("revision %d 的 ReplicaSet 未找到", req.Revision)}) return } } if targetRS == "" { c.JSON(400, models.Response{Code: 400, Message: "需要 revision 或 replicaSet 参数"}); return } // 取 ReplicaSet 的 template,patch 到 Deployment.spec.template 实现回滚 rsJSON, err := client.GetReplicaSet(ns, targetRS) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "获取 ReplicaSet 失败: " + err.Error()}); return } specMap, ok := rsJSON["spec"].(map[string]interface{}) if !ok { c.JSON(500, models.Response{Code: 500, Message: "ReplicaSet 缺少 spec"}); return } template, ok := specMap["template"].(map[string]interface{}) if !ok { c.JSON(500, models.Response{Code: 500, Message: "ReplicaSet 缺少 template"}); return } if err := client.PatchDeploymentSpecTemplate(ns, deploymentName, template); err != nil { c.JSON(500, models.Response{Code: 500, Message: "回滚失败: " + err.Error()}); return } c.JSON(200, models.Response{ Code: 200, Message: fmt.Sprintf("Deployment %s 已回滚到 %s", deploymentName, targetRS), Data: gin.H{"replicaSet": targetRS, "revision": req.Revision}, }) } // ============================================================================ // Pod 删除 // Logan.Li // ============================================================================ // DeletePodNS 删除指定 Pod func (h *KubernetesHandler) DeletePodNS(c *gin.Context) { h.deletePodWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) deletePodWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } podName := c.Param("podName") if podName == "" { c.JSON(400, models.Response{Code: 400, Message: "缺少 podName"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } if err := client.DeletePod(ns, podName); err != nil { c.JSON(500, models.Response{Code: 500, Message: "删除 Pod 失败: " + err.Error()}); return } c.JSON(200, models.Response{Code: 200, Message: fmt.Sprintf("Pod %s 已删除", podName)}) } // ============================================================================ // Pod 文件列表 / 下载(复用 exec 能力) // Logan.Li // ============================================================================ // cleanExecOutput 清理 exec 输出中残留的 SPDY 帧头等非打印字节(best-effort) func cleanExecOutput(raw string) string { var b strings.Builder for _, r := range raw { if r == '\n' || r == '\r' || r == ' ' || (r >= 0x20 && r < 0x7F) { b.WriteRune(r) } } return b.String() } // ListPodFilesNS 列出 Pod 内指定路径的文件 func (h *KubernetesHandler) ListPodFilesNS(c *gin.Context) { h.listPodFilesWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) listPodFilesWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } podName := c.Param("podName") var req struct { Path string `json:"path"` Container string `json:"container"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误: " + err.Error()}); return } if req.Path == "" { req.Path = "/" } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } // ls -la — BusyBox ignores --time-style, use default "Mon DD HH:MM" format cmd := []string{"ls", "-la", req.Path} result, err := client.ExecInPod(ns, podName, req.Container, cmd) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "执行 ls 失败: " + err.Error()}); return } type FileInfo struct { Name string `json:"name"` Size string `json:"size"` Mode string `json:"mode"` ModTime string `json:"mod_time"` IsDir bool `json:"is_dir"` } files := []FileInfo{} output := result.Stdout // WebSocket output is already clean text for _, line := range strings.Split(output, "\n") { line = strings.TrimSpace(line) if line == "" || strings.HasPrefix(line, "total ") { continue } fields := strings.Fields(line) if len(fields) < 7 { continue } mode := fields[0] if len(mode) == 0 || (mode[0] != 'd' && mode[0] != '-' && mode[0] != 'l') { continue } size := fields[4] // BusyBox ls date format: "Jun 23 02:17" (3 fields: month day time) // GNU ls --time-style: "2026-06-23T02:17:00" (1 field) // Detect format: if fields[5] contains ":" it's time-only (BusyBox style) var name, modTime string if strings.Contains(fields[5], ":") || strings.Contains(fields[5], "T") { // GNU style: single timestamp field modTime = fields[5] name = strings.Join(fields[6:], " ") } else { // BusyBox style: "Mon DD HH:MM" = 3 fields starting at index 5 modTime = strings.Join(fields[5:8], " ") name = strings.Join(fields[8:], " ") } if name == "" || name == "." || name == ".." { continue } files = append(files, FileInfo{ Name: name, Size: size, Mode: mode, ModTime: modTime, IsDir: mode[0] == 'd', }) } c.JSON(200, models.Response{Code: 200, Message: "success", Data: gin.H{"path": req.Path, "files": files}}) } // DownloadPodFileNS 下载 Pod 内指定路径的文件(通过 exec cat) func (h *KubernetesHandler) DownloadPodFileNS(c *gin.Context) { h.downloadPodFileWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) downloadPodFileWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } podName := c.Param("podName") filePath := c.Query("path") container := c.Query("container") if filePath == "" { c.JSON(400, models.Response{Code: 400, Message: "path 参数不能为空"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return } cmd := []string{"cat", filePath} result, err := client.ExecInPod(ns, podName, container, cmd) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "读取文件失败: " + err.Error()}); return } content := result.Stdout fileName := path.Base(filePath) c.Header("Content-Disposition", fmt.Sprintf("attachment; filename=\"%s\"", fileName)) c.Header("Content-Type", "application/octet-stream") c.String(200, content) } // UploadPodFileNS 上传文件到 Pod 容器内指定目录(通过 exec base64 分块写入) func (h *KubernetesHandler) UploadPodFileNS(c *gin.Context) { h.uploadPodFileWithNS(c, c.Param("namespace")) } func (h *KubernetesHandler) uploadPodFileWithNS(c *gin.Context, ns string) { clusterID, _ := strconv.Atoi(c.Param("id")) if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) } podName := c.Param("podName") container := c.PostForm("container") targetDir := c.PostForm("path") if targetDir == "" { targetDir = "/" } // 读取上传的文件 file, header, err := c.Request.FormFile("file") if err != nil { c.JSON(400, models.Response{Code: 400, Message: "文件上传失败: " + err.Error()}) return } defer file.Close() // 限制文件大小 10MB const maxUploadSize = 10 << 20 // 10MB if header.Size > maxUploadSize { c.JSON(400, models.Response{Code: 400, Message: fmt.Sprintf("文件过大: %d bytes, 上限 %d bytes", header.Size, maxUploadSize)}) return } content, err := io.ReadAll(file) if err != nil { c.JSON(500, models.Response{Code: 500, Message: "读取文件失败: " + err.Error()}) return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}) return } // 目标完整路径 targetFile := strings.TrimRight(targetDir, "/") + "/" + header.Filename // 确保目标目录存在 _, _ = client.ExecInPod(ns, podName, container, []string{"mkdir", "-p", targetDir}) // base64 分块写入:每块 3000 字节 base64 (~2250 bytes 二进制),保证 URL 长度安全 encoded := base64.StdEncoding.EncodeToString(content) const chunkSize = 3000 first := true for i := 0; i < len(encoded); i += chunkSize { end := i + chunkSize if end > len(encoded) { end = len(encoded) } chunk := encoded[i:end] var cmd []string if first { // 首块:创建/覆盖文件 cmd = []string{"sh", "-c", fmt.Sprintf("echo -n '%s' | base64 -d > '%s'", chunk, targetFile)} first = false } else { // 后续块:追加 cmd = []string{"sh", "-c", fmt.Sprintf("echo -n '%s' | base64 -d >> '%s'", chunk, targetFile)} } result, err := client.ExecInPod(ns, podName, container, cmd) if err != nil { stderr := "" if result != nil { stderr = result.Stderr } c.JSON(500, models.Response{Code: 500, Message: fmt.Sprintf("写入文件失败 (chunk %d): %s\n%s", i/chunkSize, err.Error(), stderr)}) return } } // 验证文件大小 verifyResult, verifyErr := client.ExecInPod(ns, podName, container, []string{"stat", "-c", "%s", targetFile}) writtenSize := "" if verifyErr == nil && verifyResult != nil { writtenSize = strings.TrimSpace(verifyResult.Stdout) } log.Printf("[UploadPodFile] %s/%s → %s (%d bytes, %d chunks)", ns, podName, targetFile, len(content), (len(encoded)+chunkSize-1)/chunkSize) c.JSON(200, models.Response{ Code: 200, Message: "上传成功", Data: gin.H{ "path": targetFile, "name": header.Filename, "size": len(content), "verify": writtenSize, }, }) }