feat(k8s): 实时日志流 + WebSocket exec 交互终端 + 前端流式日志查看器
后端: - GetPodLogsStream: 新增 K8s follow=true 流式日志读取 - GetPodLogs handler: 支持 stream=true → chunked transfer 实时推流 - ExecTerminal: WebSocket → K8s SPDY exec 双向代理 (交互式 bash) - 路由: GET /ws/k8s-exec/:id/namespaces/:ns/pods/:pod (WebSocket) 前端 DeploymentDetail: - 新增「实时流」按钮: fetch ReadableStream + TextDecoder 逐块渲染 - 日志内容自动滚动追加 CD/CI 镜像更新: - POST /update-image 已可用 (前端表单+后端 API 均完成)
This commit is contained in:
attacker committed
1 parent
1a913549fc
commit
77045622fc
4 files changed
+201
-5
No files matched your search
@@ -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
|
||||
// ============================================================================
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user