feat(k8s): M3 Pod文件上传 + ExecInPod WebSocket协议重写
M3: Pod 文件上传 - 后端: POST .../pods/:pod/files/upload (multipart, base64分块exec写入) - 前端: PodFileBrowser 增加上传按钮 + FormData API - 路由: 新增 Kuboard (:clusterCode) + 旧 (:id) 两套路由 重大修复: ExecInPod 从 raw HTTP SPDY 改为 WebSocket channel.k8s.io - 旧方案: HTTP POST + SPDY upgrade → 无法解析响应(协议头缺失 + 帧解析复杂) - 新方案: gorilla/websocket Dialer + channel.k8s.io 子协议 → 和 ExecTerminal 一致 - 影响: 文件浏览/下载/上传全部恢复正常 修复: BusyBox ls 解析 - BusyBox 不支持 --time-style 参数,默认输出 Mon DD HH:MM 格式 - 解析器改为自动检测 GNU vs BusyBox 日期格式
This commit is contained in:
1 parent
69df073d42
commit
9ab7572b8e
11 files changed
+389
-126
No files matched your search
@@ -1,11 +1,13 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"path"
|
||||
"sort"
|
||||
"strconv"
|
||||
@@ -16,6 +18,7 @@ import (
|
||||
"ezops/pkg/kubernetesclient"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/gorilla/websocket"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
@@ -916,7 +919,22 @@ func (h *KubernetesHandler) GetPodLogs(c *gin.Context) {
|
||||
// WebSocket Exec Terminal (交互式 bash)
|
||||
// ============================================================================
|
||||
|
||||
// ExecTerminal WebSocket 终端 — 前端 WS → 后端 → K8s SPDY exec
|
||||
// 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")
|
||||
@@ -927,102 +945,128 @@ func (h *KubernetesHandler) ExecTerminal(c *gin.Context) {
|
||||
command = "sh"
|
||||
}
|
||||
|
||||
client, _, err := h.getK8sClient(clusterID)
|
||||
client, cluster, 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)
|
||||
// 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 != "" {
|
||||
execPath += "&container=" + container
|
||||
params.Set("container", container)
|
||||
}
|
||||
execURL += fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/exec?%s",
|
||||
url.PathEscape(namespace), url.PathEscape(podName), params.Encode())
|
||||
|
||||
// Open SPDY connection to K8s API
|
||||
k8sReq, err := http.NewRequest("POST", client.BaseURL+execPath, nil)
|
||||
// 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 {
|
||||
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "build request: " + err.Error()})
|
||||
log.Printf("[K8sExec] dial K8s exec WS failed: %v", err)
|
||||
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s exec 连接失败: " + 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")
|
||||
defer k8sConn.Close()
|
||||
|
||||
// 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)
|
||||
// Upgrade frontend connection to WebSocket
|
||||
frontendConn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer wsConn.Close()
|
||||
defer frontendConn.Close()
|
||||
|
||||
// Bidirectional proxy: WS ↔ SPDY
|
||||
done := make(chan struct{}, 2)
|
||||
|
||||
// WS → SPDY (stdin)
|
||||
// Frontend WS → K8s WS (stdin + resize)
|
||||
go func() {
|
||||
defer func() { done <- struct{}{} }()
|
||||
for {
|
||||
_, msg, err := wsConn.ReadMessage()
|
||||
msgType, msg, err := frontendConn.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
|
||||
_ = 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
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// SPDY → WS (stdout/stderr)
|
||||
// K8s WS → Frontend WS (stdout/stderr/error)
|
||||
go func() {
|
||||
defer func() { done <- struct{}{} }()
|
||||
buf := make([]byte, 4096)
|
||||
for {
|
||||
n, err := conn.Read(buf)
|
||||
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
|
||||
}
|
||||
if n > 8 {
|
||||
// Skip SPDY header (8 bytes), send data
|
||||
wsConn.WriteMessage(1, buf[8:n]) // 1 = TextMessage
|
||||
_ = 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
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
@@ -2623,8 +2667,8 @@ func (h *KubernetesHandler) listPodFilesWithNS(c *gin.Context, ns string) {
|
||||
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 配合自定义 time-style 获取清晰可解析的字段
|
||||
cmd := []string{"ls", "-la", "--time-style=+%Y-%m-%dT%H:%M:%S", req.Path}
|
||||
// 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 }
|
||||
|
||||
@@ -2636,7 +2680,7 @@ func (h *KubernetesHandler) listPodFilesWithNS(c *gin.Context, ns string) {
|
||||
IsDir bool `json:"is_dir"`
|
||||
}
|
||||
files := []FileInfo{}
|
||||
output := cleanExecOutput(result.Stdout)
|
||||
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 }
|
||||
@@ -2645,13 +2689,20 @@ func (h *KubernetesHandler) listPodFilesWithNS(c *gin.Context, ns string) {
|
||||
mode := fields[0]
|
||||
if len(mode) == 0 || (mode[0] != 'd' && mode[0] != '-' && mode[0] != 'l') { continue }
|
||||
size := fields[4]
|
||||
modTime := fields[5]
|
||||
// 文件名可能含空格:定位到 modTime 之后的内容作为 name
|
||||
name := ""
|
||||
if idx := strings.Index(line, modTime); idx >= 0 {
|
||||
name = strings.TrimSpace(line[idx+len(modTime):])
|
||||
// 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 = strings.Join(fields[6:], " ") }
|
||||
if name == "" || name == "." || name == ".." { continue }
|
||||
files = append(files, FileInfo{
|
||||
Name: name,
|
||||
Size: size,
|
||||
@@ -2679,9 +2730,109 @@ func (h *KubernetesHandler) downloadPodFileWithNS(c *gin.Context, ns string) {
|
||||
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 := cleanExecOutput(result.Stdout)
|
||||
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,
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -317,6 +317,7 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
|
||||
// Pod 文件浏览 / 下载 / 删除 // Logan.Li
|
||||
clusters.POST("/:id/namespaces/:namespace/pods/:podName/files/list", k8sHandler.ListPodFilesNS)
|
||||
clusters.GET("/:id/namespaces/:namespace/pods/:podName/files/download", k8sHandler.DownloadPodFileNS)
|
||||
clusters.POST("/:id/namespaces/:namespace/pods/:podName/files/upload", k8sHandler.UploadPodFileNS)
|
||||
clusters.DELETE("/:id/namespaces/:namespace/pods/:podName", k8sHandler.DeletePodNS)
|
||||
// Namespace // Logan.Li
|
||||
clusters.GET("/:id/namespaces", k8sHandler.ListNamespaces)
|
||||
@@ -369,6 +370,7 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
|
||||
// 🆕 Pod 文件浏览 / 下载 / 删除 // Logan.Li
|
||||
kubernetesRouter.POST("/:clusterCode/namespace/:namespace/pods/:podName/files/list", k8sHandler.ListPodFilesNS)
|
||||
kubernetesRouter.GET("/:clusterCode/namespace/:namespace/pods/:podName/files/download", k8sHandler.DownloadPodFileNS)
|
||||
kubernetesRouter.POST("/:clusterCode/namespace/:namespace/pods/:podName/files/upload", k8sHandler.UploadPodFileNS)
|
||||
kubernetesRouter.DELETE("/:clusterCode/namespace/:namespace/pods/:podName", k8sHandler.DeletePodNS)
|
||||
|
||||
// ConfigMap / Secret
|
||||
|
||||
Reference in new issue
Block a user