attacker
e4f1018ca1
- 新增 pkg/wshub/manager.go 统一 WebSocket 连接管理器 (WSManager) - 6 个 WebSocket handler 全部迁移: WebShell/kubectl/PodExec/Audit/Metrics/Pipeline - 心跳机制: ping(30s) + pong handler + read deadline(60s) 自动检测死连接 - 连接注册表: 元数据(类型/用户/远程地址/连接时间/最后活跃) - 每用户连接限制 MaxConnsPerUser=15 防资源耗尽 - 监控端点: GET /api/v1/ws/stats + GET /api/v1/ws/connections - 优雅关闭: CloseAll() 发送 CloseGoingAway 帧并关闭所有连接 - 消除 3 个重复 websocket.Upgrader 变量统一为 WSManager 内部 upgrader - 6/6 A3 测试通过, 17/17 回归测试通过
673 lines
22 KiB
Go
673 lines
22 KiB
Go
package controller
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"net/http"
|
||
"net/url"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"ezops/middleware"
|
||
"ezops/model"
|
||
k8sclient "ezops/pkg/kubernetesclient"
|
||
"ezops/pkg/wshub"
|
||
|
||
"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 (tracked by WSManager)
|
||
userID := middleware.GetUserID(c)
|
||
username := middleware.GetUsername(c)
|
||
frontendConn, info, err := h.wsMgr.Upgrade(c, wshub.TypePodExec, userID, username)
|
||
if err != nil {
|
||
return
|
||
}
|
||
defer h.wsMgr.Unregister(info)
|
||
h.wsMgr.StartHeartbeat(frontendConn, info)
|
||
|
||
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 内指定路径的文件
|