diff --git a/backend/internal/api/handlers/k8s.go b/backend/internal/api/handlers/k8s.go index d2ff797..b3fb2f6 100644 --- a/backend/internal/api/handlers/k8s.go +++ b/backend/internal/api/handlers/k8s.go @@ -3,9 +3,11 @@ package handlers import ( "encoding/json" "fmt" + "io" "log" "net/http" "strconv" + "strings" "time" "ezops/internal/models" @@ -797,13 +799,14 @@ func (h *K8sHandler) SyncCluster(c *gin.Context) { // Pod Logs // ============================================================================ -// GetPodLogs 获取 Pod 日志 +// GetPodLogs 获取 Pod 日志（支持 stream=true 实时流） func (h *K8sHandler) 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 { @@ -811,6 +814,35 @@ func (h *K8sHandler) GetPodLogs(c *gin.Context) { 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()}) @@ -820,6 +852,119 @@ func (h *K8sHandler) GetPodLogs(c *gin.Context) { 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 SPDY exec +func (h *K8sHandler) 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 = "sh" + } + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + // Build exec URL with SPDY upgrade + execPath := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/exec?command=%s&stdin=true&stdout=true&stderr=true&tty=true", + namespace, podName, command) + if container != "" { + execPath += "&container=" + container + } + + // Open SPDY connection to K8s API + k8sReq, err := http.NewRequest("POST", client.BaseURL+execPath, nil) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "build request: " + err.Error()}) + return + } + k8sReq.Header.Set("Connection", "Upgrade") + k8sReq.Header.Set("Upgrade", "SPDY/3.1") + k8sReq.Header.Set("X-Stream-Protocol-Version", "v4.channel.k8s.io") + k8sReq.Header.Set("X-Stream-Protocol-Version", "v3.channel.k8s.io") + + // Use raw TCP dial for SPDY upgrade + dialer := client.HTTPClient.Transport.(*http.Transport).DialContext + conn, err := dialer(c.Request.Context(), "tcp", strings.TrimPrefix(strings.TrimPrefix(client.BaseURL, "https://"), "http://")) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "dial K8s: " + err.Error()}) + return + } + defer conn.Close() + + if err := k8sReq.Write(conn); err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "write request: " + err.Error()}) + return + } + + // Read the 101 Switching Protocols response + buf := make([]byte, 4096) + n, err := conn.Read(buf) + if err != nil || n == 0 { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "read upgrade response: " + fmt.Sprint(err)}) + return + } + respStr := string(buf[:n]) + if !strings.Contains(respStr, "101") { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "upgrade failed: " + respStr[:min(len(respStr), 200)]}) + return + } + + // Upgrade Gin to WebSocket + wsConn, err := upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + return + } + defer wsConn.Close() + + // Bidirectional proxy: WS ↔ SPDY + done := make(chan struct{}, 2) + + // WS → SPDY (stdin) + go func() { + defer func() { done <- struct{}{} }() + for { + _, msg, err := wsConn.ReadMessage() + if err != nil { + return + } + // SPDY stdin frame: channel 0 + frame := make([]byte, len(msg)+8) + frame[0] = 0 // channel 0 (stdin) + copy(frame[8:], msg) + if _, err := conn.Write(frame); err != nil { + return + } + } + }() + + // SPDY → WS (stdout/stderr) + go func() { + defer func() { done <- struct{}{} }() + buf := make([]byte, 4096) + for { + n, err := conn.Read(buf) + if err != nil { + return + } + if n > 8 { + // Skip SPDY header (8 bytes), send data + wsConn.WriteMessage(1, buf[8:n]) // 1 = TextMessage + } + } + }() + + <-done +} + // ============================================================================ // Exec / File Browser // ============================================================================ diff --git a/backend/internal/api/routes/routes.go b/backend/internal/api/routes/routes.go index b3b72f4..41142ec 100644 --- a/backend/internal/api/routes/routes.go +++ b/backend/internal/api/routes/routes.go @@ -64,6 +64,10 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c metricsWSHandler := handlers.NewMetricsWSHandler(metricsHub) r.GET("/ws/metrics", middleware.JWTAuth(&cfg.JWT), metricsWSHandler.ServeWS) + // WebSocket K8s exec 交互终端 // Logan.Li + k8sHandler := handlers.NewK8sHandler(db) + r.GET("/ws/k8s-exec/:id/namespaces/:namespace/pods/:podName", middleware.JWTAuth(&cfg.JWT), k8sHandler.ExecTerminal) + // 公开路由 authHandler := handlers.NewUserHandler(db, cfg) api.POST("/auth/login", authHandler.Login) @@ -231,7 +235,6 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c } // K8s集群路由 - k8sHandler := handlers.NewK8sHandler(db) clusters := authorized.Group("/k8s-clusters") { clusters.GET("", k8sHandler.ListClusters) diff --git a/backend/pkg/k8sclient/client.go b/backend/pkg/k8sclient/client.go index 5344b1f..5175e81 100644 --- a/backend/pkg/k8sclient/client.go +++ b/backend/pkg/k8sclient/client.go @@ -467,6 +467,25 @@ func (c *Client) GetPodLogs(namespace, podName, container string, tailLines int) return string(body), nil } +// GetPodLogsStream returns the raw streaming response body for follow=true +func (c *Client) GetPodLogsStream(namespace, podName, container string, tailLines int) (io.ReadCloser, error) { + path := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/log?follow=true&tailLines=%d", namespace, podName, tailLines) + if container != "" { + path += "&container=" + container + } + url := c.BaseURL + path + resp, err := c.HTTPClient.Get(url) + if err != nil { + return nil, fmt.Errorf("GET logs stream: %w", err) + } + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + return nil, fmt.Errorf("K8s API %d: %s", resp.StatusCode, string(body[:min(len(body), 300)])) + } + return resp.Body, nil +} + // --- Scale --- // ScaleDeployment patches the replicas count for a deployment diff --git a/frontend/src/components/k8s/DeploymentDetail.tsx b/frontend/src/components/k8s/DeploymentDetail.tsx index 130ca4a..139d39e 100644 --- a/frontend/src/components/k8s/DeploymentDetail.tsx +++ b/frontend/src/components/k8s/DeploymentDetail.tsx @@ -388,9 +388,38 @@ export function DeploymentDetail({ clusterId, namespace, deploymentName, onBack

点击 Pod 列表中的「日志」按钮查看

) : ( - 
-                  {logs || '无日志内容'}
-
+
+
+ + {logPod} +
+ 
+                    {logs || '无日志内容'}
+
+
)}