package controller import ( "encoding/json" "fmt" "io" "log" "net/http" "net/url" "strconv" "strings" "time" "ezops/model" k8sclient "ezops/pkg/kubernetesclient" "github.com/gin-gonic/gin" "github.com/gorilla/websocket" ) 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, model.Response{Code: 500, Message: err.Error()}) return } pods, err := client.ListPods() if err != nil { c.JSON(http.StatusInternalServerError, model.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, model.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} model.Response{data=[]model.KubernetesDeployment} // @Router /kubernetes-clusters/{id}/deployments [get] 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, model.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, model.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, model.Response{Code: 500, Message: "获取日志失败: " + err.Error()}) return } c.JSON(http.StatusOK, model.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) { namespace := c.Param("namespace") podName := c.Param("podName") container := c.Query("container") command := c.Query("command") if command == "" { command = "bash" } client, cluster, err := h.getK8sClientByCode(c) if err != nil { c.JSON(http.StatusInternalServerError, model.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, model.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, model.Response{Code: 400, Message: "参数错误"}) return } if len(req.Command) == 0 { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "command 不能为空"}) return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: err.Error()}) return } result, err := client.ExecInPod(namespace, podName, req.Container, req.Command) if err != nil { c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: "执行失败: " + err.Error()}) return } c.JSON(http.StatusOK, model.Response{Code: 200, Data: result}) } // ============================================================================ // Scale (副本数加减) // ============================================================================ // ScaleDeployment 缩放 Deployment 副本数 func (h *KubernetesHandler) ListPodsNS(c *gin.Context) { h.listPodsWithNS(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) 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, model.Response{Code: 500, Message: err.Error()}); return } pods, err := client.ListPods() if err != nil { c.JSON(500, model.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"` Conditions []k8sclient.PodCondition `json:"conditions,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, Conditions: p.Status.Conditions, } 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, model.Response{Code: 200, Data: result}) } 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, model.Response{Code: 500, Message: err.Error()}); return } if stream { body, err := client.GetPodLogsStreamWithOptions(ns, podName, opts) if err != nil { c.JSON(500, model.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, model.Response{Code: 500, Message: "获取日志失败: " + err.Error()}); return } c.JSON(200, model.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, model.Response{Code: 400, Message: "参数错误"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, model.Response{Code: 500, Message: err.Error()}); return } result, err := client.ExecInPod(ns, podName, req.Container, req.Command) if err != nil { c.JSON(500, model.Response{Code: 500, Message: "执行失败: " + err.Error()}); return } c.JSON(200, model.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) ListPodsAllNS(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, model.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, model.Response{Code: 500, Message: err.Error()}); return } pods, err := client.ListPods() if err != nil { c.JSON(500, model.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, model.Response{Code: 200, Data: result}) } // ListHPAsAllNS lists all HPAs across all namespaces using clusterCode. 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 } // 清理常见后缀,大小写不敏感,支持 GiB / G / GB / bytes 等 // Logan.Li val = strings.TrimSpace(val) val = strings.ToUpper(val) val = strings.TrimSuffix(val, "IB") val = strings.TrimSuffix(val, "B") var num float64 var unit string fmt.Sscanf(val, "%f%s", &num, &unit) unit = strings.TrimSpace(unit) switch unit { case "KI": return int64(num / 1024) case "MI": return int64(num) case "GI": return int64(num * 1024) case "TI": return int64(num * 1024 * 1024) default: // 纯数字通常是 bytes if unit == "" && num > 0 { return int64(num / 1024 / 1024) } 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) GetPodMetrics(c *gin.Context) { clusterID := h.resolveClusterCode(c.Param("clusterCode")) if clusterID == 0 { c.JSON(500, model.Response{Code: 500, Message: "cluster not found"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, model.Response{Code: 500, Message: err.Error()}); return } ns := c.Query("namespace") metrics, err := client.ListPodMetrics(ns) if err != nil { c.JSON(500, model.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, model.Response{Code: 200, Data: result}) } // ============================================================================ // Events (kubectl get events) // ============================================================================ // ListEvents returns cluster events (optionally filtered by namespace and type) 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, model.Response{Code: 400, Message: "缺少 podName"}); return } client, _, err := h.getK8sClient(clusterID) if err != nil { c.JSON(500, model.Response{Code: 500, Message: err.Error()}); return } if err := client.DeletePod(ns, podName); err != nil { c.JSON(500, model.Response{Code: 500, Message: "删除 Pod 失败: " + err.Error()}); return } c.JSON(200, model.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 内指定路径的文件