rlwj-lab/aiops
0
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a8b4af440989bcac7f42ba156bd4570b84a5a6ca
aiops/backend/internal/api/handlers/kubernetes.go
T
attacker 2e66908e20 feat(k8s): M5 Deployment 暂停更新 + CI/CD 集成 
- 后端: PauseDeployment/ResumeDeployment 客户端方法 (PATCH spec.paused)
- 后端: ID路由 + Kuboard code路由 pause/resume handlers
- 后端: Deployment 列表 summary 增加 paused 字段
- 前端: 操作菜单增加「暂停更新/恢复更新」动态切换 + 暂停 Badge
- 前端: CI/CD 发布按钮跳转流水线页面 (namespace+deployment 上下文)
- 前端: api.ts 增加 pauseDeployment/resumeDeployment 方法
- 验证: 5项端到端测试全部通过 (Python urllib)
2026-06-23 03:06:13 +08:00

2903 lines
105 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package handlers
import (
"encoding/base64"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"net/url"
"path"
"sort"
"strconv"
"strings"
"time"
"ezops/internal/models"
"ezops/pkg/kubernetesclient"
"github.com/gin-gonic/gin"
"github.com/gorilla/websocket"
"gorm.io/gorm"
)
// KubernetesHandler K8s处理器
type KubernetesHandler struct {
db *gorm.DB
}
// NewKubernetesHandler 创建K8s处理器
func NewKubernetesHandler(db *gorm.DB) *KubernetesHandler {
return &KubernetesHandler{db: db}
}
// ListClusters 获取集群列表
// @Summary 获取集群列表
// @Description 获取所有K8s集群列表
// @Tags K8s集群
// @Produce json
// @Security Bearer
// @Param page query int false "页码"
// @Param page_size query int false "每页数量"
// @Success 200 {object} models.Response{data=models.ListResponse}
// @Router /kubernetes-clusters [get]
func (h *KubernetesHandler) ListClusters(c *gin.Context) {
var query models.PageQuery
c.ShouldBindQuery(&query)
var total int64
h.db.Model(&models.KubernetesCluster{}).Count(&total)
var clusters []models.KubernetesCluster
h.db.Offset(query.GetOffset()).Limit(query.GetLimit()).Find(&clusters)
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: models.ListResponse{
List: clusters,
Total: total,
},
})
}
// GetCluster 获取集群详情
// @Summary 获取集群详情
// @Description 获取指定K8s集群的详细信息
// @Tags K8s集群
// @Produce json
// @Security Bearer
// @Param id path int true "集群ID"
// @Success 200 {object} models.Response{data=models.KubernetesCluster}
// @Router /kubernetes-clusters/{id} [get]
func (h *KubernetesHandler) GetCluster(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
var cluster models.KubernetesCluster
if err := h.db.First(&cluster, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{
Code: 404,
Message: "集群不存在",
})
return
}
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: cluster,
})
}
// CreateCluster 创建集群
// @Summary 创建集群
// @Description 创建新K8s集群
// @Tags K8s集群
// @Accept json
// @Produce json
// @Security Bearer
// @Param request body models.CreateKubernetesClusterRequest true "创建请求"
// @Success 200 {object} models.Response{data=models.KubernetesCluster}
// @Router /kubernetes-clusters [post]
func (h *KubernetesHandler) CreateCluster(c *gin.Context) {
var req models.CreateKubernetesClusterRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{
Code: 400,
Message: "请求参数错误: " + err.Error(),
})
return
}
cluster := models.KubernetesCluster{
Name: req.Name,
Code: req.Code,
Version: req.Version,
ConfigType: req.ConfigType,
APIServer: req.APIServer,
APIEndpoint: req.APIServer, // auto-sync
Token: req.Token,
KubeConfig: req.KubeConfig,
Description: req.Description,
Status: "healthy",
}
if err := h.db.Create(&cluster).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{
Code: 500,
Message: "创建失败",
})
return
}
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "创建成功",
Data: cluster,
})
}
// UpdateCluster 更新集群
// @Summary 更新集群
// @Description 更新指定K8s集群信息
// @Tags K8s集群
// @Accept json
// @Produce json
// @Security Bearer
// @Param id path int true "集群ID"
// @Param request body models.CreateKubernetesClusterRequest true "更新请求"
// @Success 200 {object} models.Response{data=models.KubernetesCluster}
// @Router /kubernetes-clusters/{id} [put]
func (h *KubernetesHandler) UpdateCluster(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
// 用 map 接收，避免 binding:"required" 校验 // Logan.Li
var req map[string]interface{}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{
Code: 400,
Message: "请求参数错误: " + err.Error(),
})
return
}
var cluster models.KubernetesCluster
if err := h.db.First(&cluster, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{
Code: 404,
Message: "集群不存在",
})
return
}
// 逐字段更新：空值不覆盖原值（防止凭据被清空） // Logan.Li
updates := map[string]interface{}{}
if v, ok := req["name"]; ok && v != nil && v != "" {
updates["name"] = v
}
if v, ok := req["version"]; ok {
updates["version"] = v
}
if v, ok := req["description"]; ok {
updates["description"] = v
}
if v, ok := req["config_type"]; ok {
updates["config_type"] = v
}
if v, ok := req["api_server"]; ok && v != nil && v != "" {
updates["api_server"] = v
updates["api_endpoint"] = v
}
// Token / KubeConfig：空值不覆盖（前端编辑时可能拿不到原值）
if v, ok := req["token"]; ok && v != nil && v != "" {
updates["token"] = v
}
if v, ok := req["kube_config"]; ok && v != nil && v != "" {
updates["kube_config"] = v
}
if len(updates) == 0 {
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "无更新内容",
Data: cluster,
})
return
}
if err := h.db.Model(&cluster).Updates(updates).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{
Code: 500,
Message: "更新失败",
})
return
}
h.db.First(&cluster, id)
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "更新成功",
Data: cluster,
})
}
// GetClusterDetail 返回集群完整信息（含 token/kube_config，供编辑回填） // Logan.Li
func (h *KubernetesHandler) GetClusterDetail(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
var cluster models.KubernetesCluster
if err := h.db.First(&cluster, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{
Code: 404,
Message: "集群不存在",
})
return
}
// 返回时手动脱敏：token/kube_config 只返回是否存在标记，不返回原文
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: map[string]interface{}{
"id": cluster.ID,
"name": cluster.Name,
"code": cluster.Code,
"version": cluster.Version,
"status": cluster.Status,
"config_type": cluster.ConfigType,
"api_server": cluster.APIServer,
"api_endpoint": cluster.APIEndpoint,
"description": cluster.Description,
"has_token": cluster.Token != "",
"has_kube_config": cluster.KubeConfig != "",
"node_count": cluster.NodeCount,
"pod_count": cluster.PodCount,
"labels": cluster.Labels,
"created_at": cluster.CreatedAt,
"updated_at": cluster.UpdatedAt,
},
})
}
// DeleteCluster 删除集群
// @Summary 删除集群
// @Description 删除指定K8s集群
// @Tags K8s集群
// @Produce json
// @Security Bearer
// @Param id path int true "集群ID"
// @Success 200 {object} models.Response
// @Router /kubernetes-clusters/{id} [delete]
func (h *KubernetesHandler) DeleteCluster(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
if err := h.db.Delete(&models.KubernetesCluster{}, id).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{
Code: 500,
Message: "删除失败",
})
return
}
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "删除成功",
})
}
// ListNodes 获取节点列表
// @Summary 获取节点列表
// @Description 获取指定集群的节点列表
// @Tags K8s集群
// @Produce json
// @Security Bearer
// @Param id path int true "集群ID"
// @Success 200 {object} models.Response{data=[]models.KubernetesNode}
// @Router /kubernetes-clusters/{id}/nodes [get]
func (h *KubernetesHandler) ListNodes(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
nodes, err := client.ListNodes()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
// Attempt to fetch metrics (best-effort, metrics-server may not be installed)
metrics := map[string]k8sclient.NodeMetricsItem{}
if m, err := client.GetNodeMetrics(); err == nil {
for _, item := range m.Items {
metrics[item.Metadata.Name] = item
}
}
type NodeSummary struct {
Name string `json:"name"`
IP string `json:"ip"`
Role string `json:"role"`
Status string `json:"status"`
OS string `json:"os"`
Kubelet string `json:"kubelet_version"`
CPU int `json:"cpu"`
Memory int64 `json:"memory_mb"`
MaxPods int `json:"max_pods"`
// Metrics (optional, omitted if empty)
CPUUsageM string `json:"cpu_usage_m,omitempty"` // e.g. "122m"
MemUsageMi string `json:"mem_usage_mi,omitempty"` // e.g. "2105Mi"
}
result := []NodeSummary{}
for _, n := range nodes.Items {
role := "worker"
for k := range n.Metadata.Labels {
if k == "node-role.kubernetes.io/master" || k == "node-role.kubernetes.io/control-plane" {
role = "master"
break
}
}
nodeStatus := "NotReady"
for _, cond := range n.Status.Conditions {
if cond.Type == "Ready" && cond.Status == "True" {
nodeStatus = "Ready"
break
}
}
ip := ""
for _, addr := range n.Status.Addresses {
if addr.Type == "InternalIP" {
ip = addr.Address
break
}
}
s := NodeSummary{
Name: n.Metadata.Name,
IP: ip,
Role: role,
Status: nodeStatus,
OS: n.Status.NodeInfo.OSImage,
Kubelet: n.Status.NodeInfo.KubeletVersion,
CPU: parseIntResource(n.Status.Capacity["cpu"]),
Memory: parseMemResource(n.Status.Capacity["memory"]),
MaxPods: parseIntResource(n.Status.Capacity["pods"]),
}
if m, ok := metrics[n.Metadata.Name]; ok {
// K8s metrics API returns CPU in nanocores ("80410173n"), memory in Ki ("2154292Ki")
s.CPUUsageM = formatCpuFromNano(m.Usage["cpu"])
s.MemUsageMi = formatMemFromKi(m.Usage["memory"])
}
result = append(result, s)
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// ListPods 获取Pod列表
// @Summary 获取Pod列表
// @Description 获取指定集群的Pod列表
// @Tags K8s集群
// @Produce json
// @Security Bearer
// @Param id path int true "集群ID"
// @Param namespace query string false "命名空间"
// @Success 200 {object} models.Response{data=[]models.KubernetesPod}
// @Router /kubernetes-clusters/{id}/pods [get]
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, models.Response{Code: 500, Message: err.Error()})
return
}
pods, err := client.ListPods()
if err != nil {
c.JSON(http.StatusInternalServerError, models.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, models.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} models.Response{data=[]models.KubernetesDeployment}
// @Router /kubernetes-clusters/{id}/deployments [get]
func (h *KubernetesHandler) ListDeployments(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, models.Response{Code: 500, Message: err.Error()})
return
}
deps, err := client.ListDeployments()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
type DepSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Replicas int `json:"replicas"`
AvailableReplicas int `json:"available_replicas"`
ReadyReplicas int `json:"ready_replicas"`
Image string `json:"image"`
}
result := []DepSummary{}
for _, d := range deps.Items {
if namespace != "" && d.Metadata.Namespace != namespace {
continue
}
image := ""
if len(d.Spec.Template.Spec.Containers) > 0 {
image = d.Spec.Template.Spec.Containers[0].Image
}
result = append(result, DepSummary{
Namespace: d.Metadata.Namespace,
Name: d.Metadata.Name,
Replicas: d.Spec.Replicas,
AvailableReplicas: d.Status.AvailableReplicas,
ReadyReplicas: d.Status.ReadyReplicas,
Image: image,
})
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// ListServices 获取Service列表
// @Summary 获取Service列表
// @Description 获取指定集群的Service列表
// @Tags K8s集群
// @Produce json
// @Security Bearer
// @Param id path int true "集群ID"
// @Param namespace query string false "命名空间"
// @Success 200 {object} models.Response{data=[]models.KubernetesService}
// @Router /kubernetes-clusters/{id}/services [get]
func (h *KubernetesHandler) ListServices(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, models.Response{Code: 500, Message: err.Error()})
return
}
svcs, err := client.ListServices()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
type SvcSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Type string `json:"type"`
ClusterIP string `json:"cluster_ip"`
Ports int `json:"port_count"`
}
result := []SvcSummary{}
for _, s := range svcs.Items {
if namespace != "" && s.Metadata.Namespace != namespace {
continue
}
result = append(result, SvcSummary{
Namespace: s.Metadata.Namespace,
Name: s.Metadata.Name,
Type: s.Spec.Type,
ClusterIP: s.Spec.ClusterIP,
Ports: len(s.Spec.Ports),
})
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// ListIngresses 获取Ingress列表
// @Summary 获取Ingress列表
// @Description 获取指定集群的Ingress列表
// @Tags K8s集群
// @Produce json
// @Security Bearer
// @Param id path int true "集群ID"
// @Param namespace query string false "命名空间"
// @Success 200 {object} models.Response{data=[]models.KubernetesIngress}
// @Router /kubernetes-clusters/{id}/ingresses [get]
func (h *KubernetesHandler) ListIngresses(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, models.Response{Code: 500, Message: err.Error()})
return
}
ings, err := client.ListIngresses()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
type IngSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Hosts []string `json:"hosts"`
TLS bool `json:"tls"`
}
result := []IngSummary{}
for _, ing := range ings.Items {
if namespace != "" && ing.Metadata.Namespace != namespace {
continue
}
hosts := []string{}
for _, rule := range ing.Spec.Rules {
if rule.Host != "" {
hosts = append(hosts, rule.Host)
}
}
result = append(result, IngSummary{
Namespace: ing.Metadata.Namespace,
Name: ing.Metadata.Name,
Hosts: hosts,
TLS: len(ing.Spec.TLS) > 0,
})
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
// GetTopology 获取集群拓扑布局 // Logan.Li
func (h *KubernetesHandler) GetTopology(c *gin.Context) {
clusterID := c.Param("id")
var layout models.TopologyLayout
result := h.db.Where("cluster_id = ?", clusterID).First(&layout)
if result.Error != nil {
c.JSON(http.StatusOK, models.Response{Code: 200, Data: &models.TopologyLayout{}})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: layout})
}
// SaveTopology 保存集群拓扑布局 // Logan.Li
func (h *KubernetesHandler) SaveTopology(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
var req struct {
LayoutData string `json:"layout_data"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"})
return
}
var layout models.TopologyLayout
h.db.Where("cluster_id = ?", clusterID).First(&layout)
layout.ClusterID = uint(clusterID)
layout.LayoutData = req.LayoutData
if layout.ID == 0 {
h.db.Create(&layout)
} else {
h.db.Save(&layout)
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "保存成功"})
}
// SyncCluster 同步集群资源（连接真实K8s API拉取数据）
func (h *KubernetesHandler) SyncCluster(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
var cluster models.KubernetesCluster
if err := h.db.First(&cluster, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "集群不存在"})
return
}
if cluster.KubeConfig == "" && cluster.Token == "" {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "集群未配置凭证"})
return
}
client, err := k8sclient.NewClient(cluster.KubeConfig)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "连接K8s失败: " + err.Error()})
return
}
result := k8sclient.SyncResult{}
// Fetch node metrics (best-effort — metrics-server may not be installed) // Logan.Li
nodeMetrics := map[string]k8sclient.NodeMetricsItem{}
if m, err := client.GetNodeMetrics(); err == nil {
for _, item := range m.Items {
nodeMetrics[item.Metadata.Name] = item
}
} else {
log.Printf("[K8s] sync: GetNodeMetrics failed (metrics-server not installed?): %v", err)
}
// Fetch pod metrics (best-effort) // Logan.Li
podMetrics := map[string]k8sclient.PodMetricsItem{} // key: "namespace/podname"
if pm, err := client.ListPodMetrics(""); err == nil {
for _, item := range pm.Items {
podMetrics[item.Metadata.Namespace+"/"+item.Metadata.Name] = item
}
}
// Sync nodes
if nodes, err := client.ListNodes(); err == nil {
// Use raw SQL for reliability (avoid GORM soft-delete / auto-timestamp issues)
h.db.Exec("DELETE FROM kubernetes_nodes WHERE cluster_id = ?", cluster.ID)
for _, n := range nodes.Items {
role := "worker"
if _, ok := n.Metadata.Labels["node-role.kubernetes.io/master"]; ok {
role = "master"
} else if _, ok := n.Metadata.Labels["node-role.kubernetes.io/control-plane"]; ok {
role = "master"
}
nodeStatus := "NotReady"
for _, cond := range n.Status.Conditions {
if cond.Type == "Ready" && cond.Status == "True" {
nodeStatus = "Ready"
break
}
}
ip := ""
for _, addr := range n.Status.Addresses {
if addr.Type == "InternalIP" {
ip = addr.Address
break
}
}
labelsJSON, _ := json.Marshal(n.Metadata.Labels)
now := time.Now()
// Compute CPU/memory usage percentage from metrics-server data // Logan.Li
cpuUsagePct, memUsagePct := 0.0, 0.0
if m, ok := nodeMetrics[n.Metadata.Name]; ok {
cpuCores := parseIntResource(n.Status.Capacity["cpu"])
memMB := parseMemResource(n.Status.Capacity["memory"])
if cpuCores > 0 {
cpuMillicores := parseCpuNanoToMilli(m.Usage["cpu"])
cpuUsagePct = float64(cpuMillicores) / float64(cpuCores*1000) * 100
}
if memMB > 0 {
memMi := parseMemKiToMi(m.Usage["memory"])
memUsagePct = float64(memMi) / float64(memMB) * 100
}
}
err2 := h.db.Exec(`INSERT INTO kubernetes_nodes (cluster_id, name, ip, role, status, os, kubelet_version, cpu, memory, max_pods, cpu_usage, memory_usage, labels, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
cluster.ID, n.Metadata.Name, ip, role, nodeStatus, n.Status.NodeInfo.OSImage, n.Status.NodeInfo.KubeletVersion,
parseIntResource(n.Status.Capacity["cpu"]), parseMemResource(n.Status.Capacity["memory"]), parseIntResource(n.Status.Capacity["pods"]),
cpuUsagePct, memUsagePct, string(labelsJSON), now, now).Error
if err2 != nil {
log.Printf("[K8s] insert node %s error: %v", n.Metadata.Name, err2)
} else {
result.Nodes++
}
}
} else {
log.Printf("[K8s] sync nodes error: %v", err)
}
// Sync pods
if pods, err := client.ListPods(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&models.KubernetesPod{})
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
}
labelsJSON, _ := json.Marshal(p.Metadata.Labels)
h.db.Create(&models.KubernetesPod{
ClusterID: cluster.ID,
Namespace: p.Metadata.Namespace,
Name: p.Metadata.Name,
Status: p.Status.Phase,
NodeName: p.Spec.NodeName,
IP: p.Status.PodIP,
Image: image,
RestartCount: restarts,
Labels: string(labelsJSON),
})
result.Pods++
}
} else {
log.Printf("[K8s] sync pods error: %v", err)
}
// Sync deployments
if deps, err := client.ListDeployments(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&models.KubernetesDeployment{})
for _, d := range deps.Items {
image := ""
if len(d.Spec.Template.Spec.Containers) > 0 {
image = d.Spec.Template.Spec.Containers[0].Image
}
labelsJSON, _ := json.Marshal(d.Metadata.Labels)
h.db.Create(&models.KubernetesDeployment{
ClusterID: cluster.ID,
Namespace: d.Metadata.Namespace,
Name: d.Metadata.Name,
Replicas: d.Spec.Replicas,
AvailableReplicas: d.Status.AvailableReplicas,
ReadyReplicas: d.Status.ReadyReplicas,
Image: image,
Labels: string(labelsJSON),
})
result.Deployments++
}
} else {
log.Printf("[K8s] sync deps error: %v", err)
}
// Sync services
if svcs, err := client.ListServices(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&models.KubernetesService{})
for _, s := range svcs.Items {
portsJSON, _ := json.Marshal(s.Spec.Ports)
selectorJSON, _ := json.Marshal(s.Spec.Selector)
labelsJSON, _ := json.Marshal(s.Metadata.Labels)
h.db.Create(&models.KubernetesService{
ClusterID: cluster.ID,
Namespace: s.Metadata.Namespace,
Name: s.Metadata.Name,
Type: s.Spec.Type,
ClusterIP: s.Spec.ClusterIP,
Ports: string(portsJSON),
Selector: string(selectorJSON),
Labels: string(labelsJSON),
})
result.Services++
}
}
// Sync ingresses
if ings, err := client.ListIngresses(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&models.KubernetesIngress{})
for _, ing := range ings.Items {
hosts := []string{}
for _, rule := range ing.Spec.Rules {
hosts = append(hosts, rule.Host)
}
hostsJSON, _ := json.Marshal(hosts)
tls := len(ing.Spec.TLS) > 0
tlsHosts := []string{}
if tls {
for _, t := range ing.Spec.TLS {
tlsHosts = append(tlsHosts, t.Hosts...)
}
}
tlsHostsJSON, _ := json.Marshal(tlsHosts)
labelsJSON, _ := json.Marshal(ing.Metadata.Labels)
h.db.Create(&models.KubernetesIngress{
ClusterID: cluster.ID,
Namespace: ing.Metadata.Namespace,
Name: ing.Metadata.Name,
Hosts: string(hostsJSON),
TLS: tls,
TLSHosts: string(tlsHostsJSON),
Labels: string(labelsJSON),
})
result.Ingresses++
}
}
// Update cluster stats
now := time.Now()
h.db.Model(&cluster).Updates(map[string]interface{}{
"node_count": result.Nodes,
"pod_count": result.Pods,
"deployment_count": result.Deployments,
"service_count": result.Services,
"last_sync_at": &now,
})
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: fmt.Sprintf("同步完成: %d nodes, %d pods, %d deployments, %d services, %d ingresses",
result.Nodes, result.Pods, result.Deployments, result.Services, result.Ingresses),
Data: result,
})
}
// ============================================================================
// Pod Logs
// ============================================================================
// GetPodLogs 获取 Pod 日志（支持 stream=true 实时流）
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, models.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, 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()})
return
}
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 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")
podName := c.Param("podName")
container := c.Query("container")
command := c.Query("command")
if command == "" {
command = "sh"
}
client, cluster, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.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, models.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, models.Response{Code: 400, Message: "参数错误"})
return
}
if len(req.Command) == 0 {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "command 不能为空"})
return
}
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
result, err := client.ExecInPod(namespace, podName, req.Container, req.Command)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "执行失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: result})
}
// ============================================================================
// Scale (副本数加减)
// ============================================================================
// ScaleDeployment 缩放 Deployment 副本数
func (h *KubernetesHandler) ScaleDeployment(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Param("namespace")
name := c.Param("name")
var req struct {
Replicas int `json:"replicas"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"})
return
}
if req.Replicas < 0 {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "副本数不能为负数"})
return
}
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
if err := client.ScaleDeployment(namespace, name, req.Replicas); err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "扩缩容失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: fmt.Sprintf("已设置为 %d 副本", req.Replicas)})
}
// RestartDeployment 重启 Deployment (滚动重启)
func (h *KubernetesHandler) RestartDeployment(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Param("namespace")
name := c.Param("name")
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
if err := client.RestartDeployment(namespace, name); err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "重启失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "重启指令已下发"})
}
// PauseDeployment 暂停 Deployment 滚动更新（spec.paused=true）
func (h *KubernetesHandler) PauseDeployment(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Param("namespace")
name := c.Param("name")
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
if err := client.PauseDeployment(namespace, name); err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "暂停更新失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "已暂停滚动更新"})
}
// ResumeDeployment 恢复 Deployment 滚动更新（spec.paused=false）
func (h *KubernetesHandler) ResumeDeployment(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Param("namespace")
name := c.Param("name")
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
if err := client.ResumeDeployment(namespace, name); err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "恢复更新失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "已恢复滚动更新"})
}
// ============================================================================
// HPA 管理
// ============================================================================
// ListHPAs 获取 HPA 列表
func (h *KubernetesHandler) ListHPAs(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
hpas, err := client.ListHPAs()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: hpas.Items})
}
// CreateHPA 创建 HPA
func (h *KubernetesHandler) CreateHPA(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
var req k8sclient.CreateHPARequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
if req.Name == "" || req.Namespace == "" || req.TargetName == "" {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "name/namespace/targetName 必填"})
return
}
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
hpa, err := client.CreateHPA(req)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "创建 HPA 失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "HPA 创建成功", Data: hpa})
}
// DeleteHPA 删除 HPA
func (h *KubernetesHandler) DeleteHPA(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Param("namespace")
name := c.Param("name")
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
if err := client.DeleteHPA(namespace, name); err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "删除 HPA 失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "HPA 已删除"})
}
// ============================================================================
// CronJob 管理
// ============================================================================
// ListCronJobs 获取 CronJob 列表
func (h *KubernetesHandler) ListCronJobs(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
cronjobs, err := client.ListCronJobs()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: cronjobs.Items})
}
// CreateCronJob 创建 CronJob
func (h *KubernetesHandler) CreateCronJob(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
var req k8sclient.CreateCronJobRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
cj, err := client.CreateCronJob(req)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "创建 CronJob 失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "CronJob 创建成功", Data: cj})
}
// SuspendCronJob 暂停/恢复 CronJob
func (h *KubernetesHandler) SuspendCronJob(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Param("namespace")
name := c.Param("name")
var req struct {
Suspend bool `json:"suspend"`
}
c.ShouldBindJSON(&req)
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
if err := client.SuspendCronJob(namespace, name, req.Suspend); err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "操作失败: " + err.Error()})
return
}
action := "恢复"
if req.Suspend {
action = "暂停"
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "CronJob 已" + action})
}
// DeleteCronJob 删除 CronJob
func (h *KubernetesHandler) DeleteCronJob(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Param("namespace")
name := c.Param("name")
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
if err := client.DeleteCronJob(namespace, name); err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "删除 CronJob 失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "CronJob 已删除"})
}
// TriggerCronJob 手动触发 CronJob
func (h *KubernetesHandler) TriggerCronJob(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Param("namespace")
name := c.Param("name")
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
if err := client.TriggerCronJob(namespace, name); err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "触发失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "手动触发成功"})
}
// ============================================================================
// Deployment CI 集成 (Webhook 更新镜像)
// ============================================================================
// UpdateDeploymentImage 更新 Deployment 容器镜像 (CI webhook)
func (h *KubernetesHandler) UpdateDeploymentImage(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
namespace := c.Param("namespace")
name := c.Param("name")
var req struct {
Image string `json:"image"`
ContainerName string `json:"container_name"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"})
return
}
if req.Image == "" {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "image 不能为空"})
return
}
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
if err := client.UpdateDeploymentImage(namespace, name, req.ContainerName, req.Image); err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "更新镜像失败: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: fmt.Sprintf("镜像已更新为 %s，正在滚动更新...", req.Image)})
}
// ============================================================================
// Namespace
// ============================================================================
// ListNamespaces 获取命名空间列表
func (h *KubernetesHandler) ListNamespaces(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
// Support clusterCode param from new routes
if clusterID == 0 {
clusterID = h.resolveClusterCode(c.Param("clusterCode"))
}
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
namespaces, err := client.ListNamespaces()
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: namespaces})
}
// ============================================================================
// ConfigMap & Secret (Kuboard-style namespace resources)
// ============================================================================
// ListConfigMaps 获取命名空间下的 ConfigMap 列表
func (h *KubernetesHandler) ListConfigMaps(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 {
clusterID = h.resolveClusterCode(c.Param("clusterCode"))
}
namespace := c.Param("namespace")
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
cms, err := client.ListConfigMaps(namespace)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: cms.Items})
}
// ListSecrets 获取命名空间下的 Secret 列表（仅名称+类型，不含值）
func (h *KubernetesHandler) ListSecrets(c *gin.Context) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 {
clusterID = h.resolveClusterCode(c.Param("clusterCode"))
}
namespace := c.Param("namespace")
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()})
return
}
secrets, err := client.ListSecrets(namespace)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: secrets.Items})
}
// ============================================================================
// Helpers
// ============================================================================
// resolveClusterCode resolves a cluster code to its numeric ID
func (h *KubernetesHandler) resolveClusterCode(code string) int {
if code == "" {
return 0
}
var cluster models.KubernetesCluster
if err := h.db.Where("code = ?", code).First(&cluster).Error; err != nil {
return 0
}
return int(cluster.ID)
}
// getK8sClientByCode resolves cluster by code, falls back to ID
func (h *KubernetesHandler) getK8sClientByCode(c *gin.Context) (*k8sclient.Client, *models.KubernetesCluster, error) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 {
clusterID = h.resolveClusterCode(c.Param("clusterCode"))
}
if clusterID == 0 {
return nil, nil, fmt.Errorf("cluster not found")
}
return h.getK8sClient(clusterID)
}
// ============================================================================
// NS wrappers — read namespace from path (Kuboard-style routes)
// ============================================================================
func (h *KubernetesHandler) ListDeploymentsNS(c *gin.Context) {
h.listDeploymentsWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) ListPodsNS(c *gin.Context) {
h.listPodsWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) ListServicesNS(c *gin.Context) {
h.listServicesWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) ListIngressesNS(c *gin.Context) {
h.listIngressesWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) ListHPAsNS(c *gin.Context) {
h.listHPAsWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) CreateHPANS(c *gin.Context) {
h.createHPAWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) DeleteHPANS(c *gin.Context) {
h.deleteHPAWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) ListCronJobsNS(c *gin.Context) {
h.listCronJobsWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) CreateCronJobNS(c *gin.Context) {
h.createCronJobWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) SuspendCronJobNS(c *gin.Context) {
h.suspendCronJobWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) TriggerCronJobNS(c *gin.Context) {
h.triggerCronJobWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) DeleteCronJobNS(c *gin.Context) {
h.deleteCronJobWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) ScaleDeploymentNS(c *gin.Context) {
h.scaleDeploymentWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) RestartDeploymentNS(c *gin.Context) {
h.restartDeploymentWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) UpdateDeploymentImageNS(c *gin.Context) {
h.updateDeploymentImageWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) PauseDeploymentNS(c *gin.Context) {
h.pauseDeploymentWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) ResumeDeploymentNS(c *gin.Context) {
h.resumeDeploymentWithNS(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) listDeploymentsWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
deps, err := client.ListDeployments()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type DepSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Replicas int `json:"replicas"`
AvailableReplicas int `json:"available_replicas"`
ReadyReplicas int `json:"ready_replicas"`
Image string `json:"image"`
Paused bool `json:"paused"`
}
result := []DepSummary{}
for _, d := range deps.Items {
if d.Metadata.Namespace != ns { continue }
image := ""; if len(d.Spec.Template.Spec.Containers) > 0 { image = d.Spec.Template.Spec.Containers[0].Image }
result = append(result, DepSummary{
Namespace: d.Metadata.Namespace,
Name: d.Metadata.Name,
Replicas: d.Spec.Replicas,
AvailableReplicas: d.Status.AvailableReplicas,
ReadyReplicas: d.Status.ReadyReplicas,
Image: image,
Paused: d.Spec.Paused,
})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
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, models.Response{Code: 500, Message: err.Error()}); return }
pods, err := client.ListPods()
if err != nil { c.JSON(500, models.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"`
}
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,
}
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, models.Response{Code: 200, Data: result})
}
func (h *KubernetesHandler) listServicesWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
svcs, err := client.ListServices()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type SvcSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Type string `json:"type"`
ClusterIP string `json:"cluster_ip"`
Ports int `json:"ports"`
}
result := []SvcSummary{}
for _, s := range svcs.Items {
if s.Metadata.Namespace != ns { continue }
result = append(result, SvcSummary{
Namespace: s.Metadata.Namespace,
Name: s.Metadata.Name,
Type: s.Spec.Type,
ClusterIP: s.Spec.ClusterIP,
Ports: len(s.Spec.Ports),
})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
func (h *KubernetesHandler) listIngressesWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
ings, err := client.ListIngresses()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type IngSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Hosts []string `json:"hosts"`
TLS bool `json:"tls"`
}
result := []IngSummary{}
for _, ing := range ings.Items {
if ing.Metadata.Namespace != ns { continue }
hosts := []string{}
for _, rule := range ing.Spec.Rules { if rule.Host != "" { hosts = append(hosts, rule.Host) } }
result = append(result, IngSummary{
Namespace: ing.Metadata.Namespace,
Name: ing.Metadata.Name,
Hosts: hosts,
TLS: len(ing.Spec.TLS) > 0,
})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
func (h *KubernetesHandler) listHPAsWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
hpas, err := client.ListHPAs()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type HPASummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
TargetKind string `json:"target_kind"`
TargetName string `json:"target_name"`
MinReplicas int `json:"min_replicas"`
MaxReplicas int `json:"max_replicas"`
CPUPercent int `json:"cpu_percent"`
MemPercent int `json:"mem_percent"`
CurrentReplicas int `json:"current_replicas"`
}
result := []HPASummary{}
for _, h := range hpas.Items {
if h.Metadata.Namespace != ns { continue }
cpu := 0; mem := 0
for _, m := range h.Spec.Metrics {
if m.Resource != nil {
if m.Resource.Name == "cpu" { cpu = m.Resource.Target.AverageUtilization }
if m.Resource.Name == "memory" { mem = m.Resource.Target.AverageUtilization }
}
}
result = append(result, HPASummary{
Namespace: h.Metadata.Namespace,
Name: h.Metadata.Name,
TargetKind: h.Spec.ScaleTargetRef.Kind,
TargetName: h.Spec.ScaleTargetRef.Name,
MinReplicas: h.Spec.MinReplicas,
MaxReplicas: h.Spec.MaxReplicas,
CPUPercent: cpu,
MemPercent: mem,
CurrentReplicas: h.Status.CurrentReplicas,
})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
func (h *KubernetesHandler) createHPAWithNS(c *gin.Context, ns string) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) }
var req k8sclient.CreateHPARequest
if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误"}); return }
if req.Namespace == "" { req.Namespace = ns }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
hpa, err := client.CreateHPA(req)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "创建失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Data: hpa})
}
func (h *KubernetesHandler) deleteHPAWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.DeleteHPA(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "删除失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Message: "已删除"})
}
func (h *KubernetesHandler) listCronJobsWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
cjs, err := client.ListCronJobs()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
filtered := []k8sclient.CronJobItem{}
for _, cj := range cjs.Items { if cj.Metadata.Namespace == ns { filtered = append(filtered, cj) } }
c.JSON(200, models.Response{Code: 200, Data: filtered})
}
func (h *KubernetesHandler) createCronJobWithNS(c *gin.Context, ns string) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) }
var req k8sclient.CreateCronJobRequest
if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误"}); return }
if req.Namespace == "" { req.Namespace = ns }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
cj, err := client.CreateCronJob(req)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "创建失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Data: cj})
}
func (h *KubernetesHandler) suspendCronJobWithNS(c *gin.Context, ns string) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) }
var req struct { Suspend bool `json:"suspend"` }; c.ShouldBindJSON(&req)
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.SuspendCronJob(ns, c.Param("name"), req.Suspend); err != nil { c.JSON(500, models.Response{Code: 500, Message: "操作失败: " + err.Error()}); return }
action := "恢复"; if req.Suspend { action = "暂停" }
c.JSON(200, models.Response{Code: 200, Message: "CronJob 已" + action})
}
func (h *KubernetesHandler) triggerCronJobWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.TriggerCronJob(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "触发失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Message: "手动触发成功"})
}
func (h *KubernetesHandler) deleteCronJobWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.DeleteCronJob(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "删除失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Message: "已删除"})
}
func (h *KubernetesHandler) scaleDeploymentWithNS(c *gin.Context, ns string) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) }
var req struct { Replicas int `json:"replicas"` }
if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.ScaleDeployment(ns, c.Param("name"), req.Replicas); err != nil { c.JSON(500, models.Response{Code: 500, Message: "扩缩容失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Message: fmt.Sprintf("已设置为 %d 副本", req.Replicas)})
}
func (h *KubernetesHandler) restartDeploymentWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.RestartDeployment(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "重启失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Message: "重启指令已下发"})
}
func (h *KubernetesHandler) updateDeploymentImageWithNS(c *gin.Context, ns string) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) }
var req struct { Image string `json:"image"`; ContainerName string `json:"container_name"` }
if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.UpdateDeploymentImage(ns, c.Param("name"), req.ContainerName, req.Image); err != nil { c.JSON(500, models.Response{Code: 500, Message: "更新镜像失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Message: fmt.Sprintf("镜像已更新为 %s", req.Image)})
}
func (h *KubernetesHandler) pauseDeploymentWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.PauseDeployment(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "暂停更新失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Message: "已暂停滚动更新"})
}
func (h *KubernetesHandler) resumeDeploymentWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.ResumeDeployment(ns, c.Param("name")); err != nil { c.JSON(500, models.Response{Code: 500, Message: "恢复更新失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Message: "已恢复滚动更新"})
}
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, models.Response{Code: 500, Message: err.Error()}); return }
if stream {
body, err := client.GetPodLogsStreamWithOptions(ns, podName, opts)
if err != nil { c.JSON(500, 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.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, models.Response{Code: 500, Message: "获取日志失败: " + err.Error()}); return }
c.JSON(200, models.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, models.Response{Code: 400, Message: "参数错误"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
result, err := client.ExecInPod(ns, podName, req.Container, req.Command)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "执行失败: " + err.Error()}); return }
c.JSON(200, models.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) ListDeploymentsAllNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
deps, err := client.ListDeployments()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type DepSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Replicas int `json:"replicas"`
AvailableReplicas int `json:"available_replicas"`
ReadyReplicas int `json:"ready_replicas"`
Image string `json:"image"`
}
result := []DepSummary{}
for _, d := range deps.Items {
image := ""; if len(d.Spec.Template.Spec.Containers) > 0 { image = d.Spec.Template.Spec.Containers[0].Image }
result = append(result, DepSummary{
Namespace: d.Metadata.Namespace,
Name: d.Metadata.Name,
Replicas: d.Spec.Replicas,
AvailableReplicas: d.Status.AvailableReplicas,
ReadyReplicas: d.Status.ReadyReplicas,
Image: image,
})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
// ListPodsAllNS lists all pods across all namespaces using clusterCode.
func (h *KubernetesHandler) ListPodsAllNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
pods, err := client.ListPods()
if err != nil { c.JSON(500, models.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, models.Response{Code: 200, Data: result})
}
// ListHPAsAllNS lists all HPAs across all namespaces using clusterCode.
func (h *KubernetesHandler) ListHPAsAllNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
hpas, err := client.ListHPAs()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type HPASummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
TargetKind string `json:"target_kind"`
TargetName string `json:"target_name"`
MinReplicas int `json:"min_replicas"`
MaxReplicas int `json:"max_replicas"`
CPUPercent int `json:"cpu_percent"`
MemPercent int `json:"mem_percent"`
CurrentReplicas int `json:"current_replicas"`
}
result := []HPASummary{}
for _, h := range hpas.Items {
cpu := 0; mem := 0
for _, m := range h.Spec.Metrics {
if m.Resource != nil {
if m.Resource.Name == "cpu" { cpu = m.Resource.Target.AverageUtilization }
if m.Resource.Name == "memory" { mem = m.Resource.Target.AverageUtilization }
}
}
result = append(result, HPASummary{
Namespace: h.Metadata.Namespace,
Name: h.Metadata.Name,
TargetKind: h.Spec.ScaleTargetRef.Kind,
TargetName: h.Spec.ScaleTargetRef.Name,
MinReplicas: h.Spec.MinReplicas,
MaxReplicas: h.Spec.MaxReplicas,
CPUPercent: cpu,
MemPercent: mem,
CurrentReplicas: h.Status.CurrentReplicas,
})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
// ============================================================================
// Helpers
// ============================================================================
// buildPodMetricsMap fetches pod metrics and returns a map keyed by "namespace/podname" → aggregated CPU/Memory strings.
// Best-effort: returns empty map if metrics-server is unavailable. // Logan.Li M2
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
}
var num float64
var unit string
fmt.Sscanf(val, "%f%s", &num, &unit)
switch unit {
case "Ki":
return int64(num / 1024)
case "Mi":
return int64(num)
case "Gi":
return int64(num * 1024)
default:
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) GetResourceYAMLNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
ns := c.Param("namespace")
resource := c.Param("resource")
name := c.Param("name")
// Build API path based on resource type
var apiPath string
switch resource {
case "deployments", "deployment":
apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s", ns, name)
case "statefulsets", "statefulset":
apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/statefulsets/%s", ns, name)
case "daemonsets", "daemonset":
apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/daemonsets/%s", ns, name)
case "replicasets", "replicaset":
apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/replicasets/%s", ns, name)
case "pods", "pod":
apiPath = fmt.Sprintf("/api/v1/namespaces/%s/pods/%s", ns, name)
case "services", "service":
apiPath = fmt.Sprintf("/api/v1/namespaces/%s/services/%s", ns, name)
case "ingresses", "ingress":
apiPath = fmt.Sprintf("/apis/networking.k8s.io/v1/namespaces/%s/ingresses/%s", ns, name)
case "configmaps", "configmap":
apiPath = fmt.Sprintf("/api/v1/namespaces/%s/configmaps/%s", ns, name)
case "secrets", "secret":
apiPath = fmt.Sprintf("/api/v1/namespaces/%s/secrets/%s", ns, name)
case "pvcs", "persistentvolumeclaim":
apiPath = fmt.Sprintf("/api/v1/namespaces/%s/persistentvolumeclaims/%s", ns, name)
case "hpas", "hpa", "horizontalpodautoscaler":
apiPath = fmt.Sprintf("/apis/autoscaling/v2/namespaces/%s/horizontalpodautoscalers/%s", ns, name)
case "cronjobs", "cronjob":
apiPath = fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", ns, name)
default:
c.JSON(400, models.Response{Code: 400, Message: "unsupported resource type: " + resource}); return
}
yamlBytes, err := client.GetResourceYAML(apiPath)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "获取YAML失败: " + err.Error()}); return }
c.Header("Content-Type", "application/yaml")
c.String(200, string(yamlBytes))
}
// UpdateResourceYAMLNS applies YAML to update a K8s resource
func (h *KubernetesHandler) UpdateResourceYAMLNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
ns := c.Param("namespace")
resource := c.Param("resource")
name := c.Param("name")
yamlData, err := io.ReadAll(c.Request.Body)
if err != nil { c.JSON(400, models.Response{Code: 400, Message: "读取body失败"}); return }
var apiPath string
switch resource {
case "deployments", "deployment":
apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s", ns, name)
case "statefulsets", "statefulset":
apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/statefulsets/%s", ns, name)
case "daemonsets", "daemonset":
apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/daemonsets/%s", ns, name)
case "replicasets", "replicaset":
apiPath = fmt.Sprintf("/apis/apps/v1/namespaces/%s/replicasets/%s", ns, name)
case "pods", "pod":
apiPath = fmt.Sprintf("/api/v1/namespaces/%s/pods/%s", ns, name)
case "services", "service":
apiPath = fmt.Sprintf("/api/v1/namespaces/%s/services/%s", ns, name)
case "ingresses", "ingress":
apiPath = fmt.Sprintf("/apis/networking.k8s.io/v1/namespaces/%s/ingresses/%s", ns, name)
case "configmaps", "configmap":
apiPath = fmt.Sprintf("/api/v1/namespaces/%s/configmaps/%s", ns, name)
case "secrets", "secret":
apiPath = fmt.Sprintf("/api/v1/namespaces/%s/secrets/%s", ns, name)
case "pvcs", "persistentvolumeclaim":
apiPath = fmt.Sprintf("/api/v1/namespaces/%s/persistentvolumeclaims/%s", ns, name)
case "hpas", "hpa", "horizontalpodautoscaler":
apiPath = fmt.Sprintf("/apis/autoscaling/v2/namespaces/%s/horizontalpodautoscalers/%s", ns, name)
case "cronjobs", "cronjob":
apiPath = fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", ns, name)
default:
c.JSON(400, models.Response{Code: 400, Message: "unsupported resource type: " + resource}); return
}
if err := client.UpdateResourceYAML(apiPath, yamlData); err != nil {
c.JSON(500, models.Response{Code: 500, Message: "更新失败: " + err.Error()}); return
}
c.JSON(200, models.Response{Code: 200, Message: "YAML已应用"})
}
// getK8sClient builds a K8s API client from cluster's stored kubeconfig
func (h *KubernetesHandler) getK8sClient(clusterID int) (*k8sclient.Client, *models.KubernetesCluster, error) {
var cluster models.KubernetesCluster
if err := h.db.First(&cluster, clusterID).Error; err != nil {
return nil, nil, fmt.Errorf("cluster not found")
}
if cluster.KubeConfig == "" {
return nil, nil, fmt.Errorf("no kubeconfig")
}
client, err := k8sclient.NewClient(cluster.KubeConfig)
if err != nil {
return nil, nil, err
}
return client, &cluster, nil
}
// ============================================================================
// PVC (存储) handlers
// ============================================================================
// ListPVCsNS lists PersistentVolumeClaims in a namespace (code-based)
func (h *KubernetesHandler) ListPVCsNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
ns := c.Param("namespace")
pvcs, err := client.ListPVCs(ns)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type PVCSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Phase string `json:"phase"`
Size string `json:"size"`
StorageClass string `json:"storage_class"`
}
result := []PVCSummary{}
for _, p := range pvcs.Items {
size := ""
if p.Status.Capacity != nil {
if s, ok := p.Status.Capacity["storage"]; ok { size = s }
}
result = append(result, PVCSummary{
Namespace: p.Metadata.Namespace,
Name: p.Metadata.Name,
Phase: p.Status.Phase,
Size: size,
StorageClass: p.Spec.StorageClassName,
})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
// DeletePVCNS deletes a PVC (code-based)
func (h *KubernetesHandler) DeletePVCNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.DeletePVC(c.Param("namespace"), c.Param("name")); err != nil {
c.JSON(500, models.Response{Code: 500, Message: "删除失败: " + err.Error()}); return
}
c.JSON(200, models.Response{Code: 200, Message: "已删除"})
}
// ============================================================================
// StatefulSet / DaemonSet handlers (code-based)
// ============================================================================
// ListStatefulSetsNS lists StatefulSets in a namespace
func (h *KubernetesHandler) ListStatefulSetsNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
sts, err := client.ListStatefulSets()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
ns := c.Param("namespace")
type StsSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Replicas int `json:"replicas"`
Ready int `json:"ready"`
Image string `json:"image"`
}
result := []StsSummary{}
for _, s := range sts.Items {
if s.Metadata.Namespace != ns { continue }
image := ""; if len(s.Spec.Template.Spec.Containers) > 0 { image = s.Spec.Template.Spec.Containers[0].Image }
result = append(result, StsSummary{
Namespace: s.Metadata.Namespace,
Name: s.Metadata.Name,
Replicas: s.Spec.Replicas,
Ready: s.Status.ReadyReplicas,
Image: image,
})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
// ListDaemonSetsNS lists DaemonSets in a namespace
func (h *KubernetesHandler) ListDaemonSetsNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
ds, err := client.ListDaemonSets()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
ns := c.Param("namespace")
type DsSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Desired int `json:"desired"`
Current int `json:"current"`
Ready int `json:"ready"`
Image string `json:"image"`
}
result := []DsSummary{}
for _, d := range ds.Items {
if d.Metadata.Namespace != ns { continue }
image := ""; if len(d.Spec.Template.Spec.Containers) > 0 { image = d.Spec.Template.Spec.Containers[0].Image }
result = append(result, DsSummary{
Namespace: d.Metadata.Namespace,
Name: d.Metadata.Name,
Desired: d.Status.DesiredNumberScheduled,
Current: d.Status.CurrentNumberScheduled,
Ready: d.Status.NumberReady,
Image: image,
})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
// ============================================================================
// Pod Metrics (kubectl top pods) — per-container CPU/Memory usage via metrics-server
// ============================================================================
// GetPodMetrics returns pod-level CPU/Memory metrics
func (h *KubernetesHandler) GetPodMetrics(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
ns := c.Query("namespace")
metrics, err := client.ListPodMetrics(ns)
if err != nil { c.JSON(500, models.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, models.Response{Code: 200, Data: result})
}
// ============================================================================
// Events (kubectl get events)
// ============================================================================
// ListEvents returns cluster events (optionally filtered by namespace and type)
func (h *KubernetesHandler) ListEvents(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
ns := c.Query("namespace")
eventType := c.Query("type") // Normal or Warning
events, err := client.ListEvents(ns)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type EventSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Type string `json:"type"`
Reason string `json:"reason"`
Message string `json:"message"`
Object string `json:"object"`
Count int `json:"count"`
FirstTime string `json:"first_time"`
LastTime string `json:"last_time"`
}
result := []EventSummary{}
for _, e := range events.Items {
if eventType != "" && e.Type != eventType { continue }
object := fmt.Sprintf("%s/%s", e.InvolvedObject.Kind, e.InvolvedObject.Name)
if e.InvolvedObject.Namespace != "" {
object = fmt.Sprintf("%s/%s/%s", e.InvolvedObject.Namespace, e.InvolvedObject.Kind, e.InvolvedObject.Name)
}
firstTime := ""; if !e.FirstTimestamp.IsZero() { firstTime = e.FirstTimestamp.Format(time.RFC3339) }
lastTime := ""; if !e.LastTimestamp.IsZero() { lastTime = e.LastTimestamp.Format(time.RFC3339) }
result = append(result, EventSummary{
Namespace: e.Metadata.Namespace, Name: e.Metadata.Name, Type: e.Type, Reason: e.Reason,
Message: e.Message, Object: object, Count: int(e.Count),
FirstTime: firstTime, LastTime: lastTime,
})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
// ============================================================================
// Capacity Analysis — CPU/Memory request/limit per Deployment
// ============================================================================
// GetCapacityAnalysis returns resource analysis for all deployments
func (h *KubernetesHandler) GetCapacityAnalysis(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
ns := c.Query("namespace")
deps, err := client.ListDeployments()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type ContainerResource struct {
Name string `json:"name"`
CPUReq string `json:"cpu_request"`
CPULim string `json:"cpu_limit"`
MemReq string `json:"mem_request"`
MemLim string `json:"mem_limit"`
}
type CapacityRow struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Replicas int `json:"replicas"`
Containers []ContainerResource `json:"containers"`
}
result := []CapacityRow{}
for _, d := range deps.Items {
if ns != "" && d.Metadata.Namespace != ns { continue }
containers := []ContainerResource{}
for _, c := range d.Spec.Template.Spec.Containers {
cpuReq := c.Resources.Requests["cpu"]
cpuLim := c.Resources.Limits["cpu"]
memReq := c.Resources.Requests["memory"]
memLim := c.Resources.Limits["memory"]
containers = append(containers, ContainerResource{c.Name, cpuReq, cpuLim, memReq, memLim})
}
result = append(result, CapacityRow{d.Metadata.Namespace, d.Metadata.Name, d.Spec.Replicas, containers})
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
// ============================================================================
// YAML Apply (kubectl apply -f)
// ============================================================================
// ApplyYAML applies a YAML manifest to create or update a resource
func (h *KubernetesHandler) ApplyYAML(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 { c.JSON(500, models.Response{Code: 500, Message: "cluster not found"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
var req struct {
YAML string `json:"yaml"`
}
if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误: " + err.Error()}); return }
if req.YAML == "" { c.JSON(400, models.Response{Code: 400, Message: "YAML 内容不能为空"}); return }
result, err := client.ApplyYAML(req.YAML)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "Apply 失败: " + err.Error()}); return }
c.JSON(200, models.Response{Code: 200, Message: "资源已应用", Data: result})
}
// ============================================================================
// ReplicaSet 列表 / Deployment 回滚 / 历史版本 // Logan.Li
// ============================================================================
// ListReplicaSetsNS 列出 namespace 下的 ReplicaSet
func (h *KubernetesHandler) ListReplicaSetsNS(c *gin.Context) {
h.listReplicaSetsWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) listReplicaSetsWithNS(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, models.Response{Code: 500, Message: err.Error()}); return }
items, err := client.ListReplicaSetsRaw(ns)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type RSSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Replicas int `json:"replicas"`
AvailableReplicas int `json:"available_replicas"`
ReadyReplicas int `json:"ready_replicas"`
Image string `json:"image"`
CreatedAt string `json:"created_at"`
Revision string `json:"revision"`
OwnerDeployment string `json:"owner_deployment"`
}
result := []RSSummary{}
for _, rs := range items {
image := ""
if len(rs.Spec.Template.Spec.Containers) > 0 { image = rs.Spec.Template.Spec.Containers[0].Image }
revision := rs.Metadata.Annotations["deployment.kubernetes.io/revision"]
owner := ""
for _, o := range rs.Metadata.OwnerReferences {
if o.Kind == "Deployment" { owner = o.Name; break }
}
result = append(result, RSSummary{
Namespace: rs.Metadata.Namespace,
Name: rs.Metadata.Name,
Replicas: rs.Spec.Replicas,
AvailableReplicas: rs.Status.AvailableReplicas,
ReadyReplicas: rs.Status.ReadyReplicas,
Image: image,
CreatedAt: rs.Metadata.CreationTimestamp,
Revision: revision,
OwnerDeployment: owner,
})
}
c.JSON(200, models.Response{Code: 200, Message: "success", Data: result})
}
// ListDeploymentRevisionsNS Deployment 发布历史（按 revision 倒序）
func (h *KubernetesHandler) ListDeploymentRevisionsNS(c *gin.Context) {
h.listDeploymentRevisionsWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) listDeploymentRevisionsWithNS(c *gin.Context, ns string) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) }
deploymentName := c.Param("name")
if deploymentName == "" { c.JSON(400, models.Response{Code: 400, Message: "缺少 deployment name"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
items, err := client.ListReplicaSetsRaw(ns)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type Revision struct {
Revision string `json:"revision"`
ReplicaSet string `json:"replica_set"`
Replicas int `json:"replicas"`
Image string `json:"image"`
CreatedAt string `json:"created_at"`
Current bool `json:"current"`
}
result := []Revision{}
for _, rs := range items {
isOwner := false
for _, o := range rs.Metadata.OwnerReferences {
if o.Kind == "Deployment" && o.Name == deploymentName { isOwner = true; break }
}
if !isOwner { continue }
image := ""
if len(rs.Spec.Template.Spec.Containers) > 0 { image = rs.Spec.Template.Spec.Containers[0].Image }
result = append(result, Revision{
Revision: rs.Metadata.Annotations["deployment.kubernetes.io/revision"],
ReplicaSet: rs.Metadata.Name,
Replicas: rs.Spec.Replicas,
Image: image,
CreatedAt: rs.Metadata.CreationTimestamp,
Current: rs.Spec.Replicas > 0,
})
}
// 按 revision 数字倒序（最新在前）
sort.Slice(result, func(i, j int) bool {
ri, _ := strconv.Atoi(result[i].Revision)
rj, _ := strconv.Atoi(result[j].Revision)
return ri > rj
})
c.JSON(200, models.Response{Code: 200, Message: "success", Data: result})
}
// RollbackDeploymentNS 回滚 Deployment 到指定 revision 或指定 ReplicaSet
func (h *KubernetesHandler) RollbackDeploymentNS(c *gin.Context) {
h.rollbackDeploymentWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) rollbackDeploymentWithNS(c *gin.Context, ns string) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) }
deploymentName := c.Param("name")
if deploymentName == "" { c.JSON(400, models.Response{Code: 400, Message: "缺少 deployment name"}); return }
var req struct {
Revision int `json:"revision"`
ReplicaSet string `json:"replicaSet"`
}
_ = c.ShouldBindJSON(&req) // 允许空 body（但下面会校验）
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
targetRS := req.ReplicaSet
// 通过 revision 定位 ReplicaSet
if targetRS == "" && req.Revision > 0 {
items, err := client.ListReplicaSetsRaw(ns)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "获取 ReplicaSet 列表失败: " + err.Error()}); return }
for _, rs := range items {
isOwner := false
for _, o := range rs.Metadata.OwnerReferences {
if o.Kind == "Deployment" && o.Name == deploymentName { isOwner = true; break }
}
if !isOwner { continue }
if rs.Metadata.Annotations["deployment.kubernetes.io/revision"] == strconv.Itoa(req.Revision) {
targetRS = rs.Metadata.Name
break
}
}
if targetRS == "" {
c.JSON(404, models.Response{Code: 404, Message: fmt.Sprintf("revision %d 的 ReplicaSet 未找到", req.Revision)})
return
}
}
if targetRS == "" { c.JSON(400, models.Response{Code: 400, Message: "需要 revision 或 replicaSet 参数"}); return }
// 取 ReplicaSet 的 template，patch 到 Deployment.spec.template 实现回滚
rsJSON, err := client.GetReplicaSet(ns, targetRS)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "获取 ReplicaSet 失败: " + err.Error()}); return }
specMap, ok := rsJSON["spec"].(map[string]interface{})
if !ok { c.JSON(500, models.Response{Code: 500, Message: "ReplicaSet 缺少 spec"}); return }
template, ok := specMap["template"].(map[string]interface{})
if !ok { c.JSON(500, models.Response{Code: 500, Message: "ReplicaSet 缺少 template"}); return }
if err := client.PatchDeploymentSpecTemplate(ns, deploymentName, template); err != nil {
c.JSON(500, models.Response{Code: 500, Message: "回滚失败: " + err.Error()}); return
}
c.JSON(200, models.Response{
Code: 200,
Message: fmt.Sprintf("Deployment %s 已回滚到 %s", deploymentName, targetRS),
Data: gin.H{"replicaSet": targetRS, "revision": req.Revision},
})
}
// ============================================================================
// Pod 删除 // Logan.Li
// ============================================================================
// DeletePodNS 删除指定 Pod
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, models.Response{Code: 400, Message: "缺少 podName"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
if err := client.DeletePod(ns, podName); err != nil {
c.JSON(500, models.Response{Code: 500, Message: "删除 Pod 失败: " + err.Error()}); return
}
c.JSON(200, models.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 内指定路径的文件
func (h *KubernetesHandler) ListPodFilesNS(c *gin.Context) {
h.listPodFilesWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) listPodFilesWithNS(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 {
Path string `json:"path"`
Container string `json:"container"`
}
if err := c.ShouldBindJSON(&req); err != nil { c.JSON(400, models.Response{Code: 400, Message: "参数错误: " + err.Error()}); return }
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 — 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 }
type FileInfo struct {
Name string `json:"name"`
Size string `json:"size"`
Mode string `json:"mode"`
ModTime string `json:"mod_time"`
IsDir bool `json:"is_dir"`
}
files := []FileInfo{}
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 }
fields := strings.Fields(line)
if len(fields) < 7 { continue }
mode := fields[0]
if len(mode) == 0 || (mode[0] != 'd' && mode[0] != '-' && mode[0] != 'l') { continue }
size := fields[4]
// 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 == "." || name == ".." { continue }
files = append(files, FileInfo{
Name: name,
Size: size,
Mode: mode,
ModTime: modTime,
IsDir: mode[0] == 'd',
})
}
c.JSON(200, models.Response{Code: 200, Message: "success", Data: gin.H{"path": req.Path, "files": files}})
}
// DownloadPodFileNS 下载 Pod 内指定路径的文件（通过 exec cat）
func (h *KubernetesHandler) DownloadPodFileNS(c *gin.Context) {
h.downloadPodFileWithNS(c, c.Param("namespace"))
}
func (h *KubernetesHandler) downloadPodFileWithNS(c *gin.Context, ns string) {
clusterID, _ := strconv.Atoi(c.Param("id"))
if clusterID == 0 { clusterID = h.resolveClusterCode(c.Param("clusterCode")) }
podName := c.Param("podName")
filePath := c.Query("path")
container := c.Query("container")
if filePath == "" { c.JSON(400, models.Response{Code: 400, Message: "path 参数不能为空"}); return }
client, _, err := h.getK8sClient(clusterID)
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
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 := 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,
},
})
}
Reference in new issue View Git Blame Copy Permalink