Files
2026-07-09 14:16:56 +08:00

662 lines
20 KiB
Go

package controller
import (
"encoding/json"
"fmt"
"log"
"net/http"
"strconv"
"time"
"ezops/model"
k8sclient "ezops/pkg/kubernetesclient"
"github.com/gin-gonic/gin"
)
func (h *KubernetesHandler) ListClusters(c *gin.Context) {
var query model.PageQuery
c.ShouldBindQuery(&query)
var total int64
h.db.Model(&model.KubernetesCluster{}).Count(&total)
var clusters []model.KubernetesCluster
h.db.Offset(query.GetOffset()).Limit(query.GetLimit()).Find(&clusters)
// 列表页资源计数与详情页对齐(实时拉取 Ingress 等)// Logan.Li
h.enrichClustersForList(clusters)
c.JSON(http.StatusOK, model.Response{
Code: 200,
Message: "success",
Data: model.ListResponse{
List: clusters,
Total: total,
},
})
}
// GetCluster 获取集群详情
// @Summary 获取集群详情
// @Description 获取指定K8s集群的详细信息
// @Tags K8s集群
// @Produce json
// @Security Bearer
// @Param id path int true "集群ID"
// @Success 200 {object} model.Response{data=model.KubernetesCluster}
// @Router /kubernetes-clusters/{id} [get]
func (h *KubernetesHandler) GetCluster(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
var cluster model.KubernetesCluster
if err := h.db.First(&cluster, id).Error; err != nil {
c.JSON(http.StatusNotFound, model.Response{
Code: 404,
Message: "集群不存在",
})
return
}
c.JSON(http.StatusOK, model.Response{
Code: 200,
Message: "success",
Data: cluster,
})
}
// CreateCluster 创建集群
// @Summary 创建集群
// @Description 创建新K8s集群
// @Tags K8s集群
// @Accept json
// @Produce json
// @Security Bearer
// @Param request body model.CreateKubernetesClusterRequest true "创建请求"
// @Success 200 {object} model.Response{data=model.KubernetesCluster}
// @Router /kubernetes-clusters [post]
func (h *KubernetesHandler) CreateCluster(c *gin.Context) {
var req model.CreateKubernetesClusterRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, model.Response{
Code: 400,
Message: "请求参数错误: " + err.Error(),
})
return
}
cluster := model.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, model.Response{
Code: 500,
Message: "创建失败",
})
return
}
c.JSON(http.StatusOK, model.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 model.CreateKubernetesClusterRequest true "更新请求"
// @Success 200 {object} model.Response{data=model.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, model.Response{
Code: 400,
Message: "请求参数错误: " + err.Error(),
})
return
}
var cluster model.KubernetesCluster
if err := h.db.First(&cluster, id).Error; err != nil {
c.JSON(http.StatusNotFound, model.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, model.Response{
Code: 200,
Message: "无更新内容",
Data: cluster,
})
return
}
if err := h.db.Model(&cluster).Updates(updates).Error; err != nil {
c.JSON(http.StatusInternalServerError, model.Response{
Code: 500,
Message: "更新失败",
})
return
}
h.db.First(&cluster, id)
c.JSON(http.StatusOK, model.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 model.KubernetesCluster
if err := h.db.First(&cluster, id).Error; err != nil {
c.JSON(http.StatusNotFound, model.Response{
Code: 404,
Message: "集群不存在",
})
return
}
// 返回时手动脱敏:token/kube_config 只返回是否存在标记,不返回原文
c.JSON(http.StatusOK, model.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,
"service_count": cluster.ServiceCount,
"deployment_count": cluster.DeploymentCount,
"ingress_count": cluster.IngressCount,
"cpu_usage": cluster.CPUUsage,
"memory_usage": cluster.MemoryUsage,
"labels": cluster.Labels,
"last_sync_at": cluster.LastSyncAt,
"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} model.Response
// @Router /kubernetes-clusters/{id} [delete]
func (h *KubernetesHandler) DeleteCluster(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
if err := h.db.Delete(&model.KubernetesCluster{}, id).Error; err != nil {
c.JSON(http.StatusInternalServerError, model.Response{
Code: 500,
Message: "删除失败",
})
return
}
c.JSON(http.StatusOK, model.Response{
Code: 200,
Message: "删除成功",
})
}
// ListNodes 获取节点列表
// @Summary 获取节点列表
// @Description 获取指定集群的节点列表
// @Tags K8s集群
// @Produce json
// @Security Bearer
// @Param id path int true "集群ID"
// @Success 200 {object} model.Response{data=[]model.KubernetesNode}
// @Router /kubernetes-clusters/{id}/nodes [get]
func (h *KubernetesHandler) SyncCluster(c *gin.Context) {
id, _ := strconv.Atoi(c.Param("id"))
var cluster model.KubernetesCluster
if err := h.db.First(&cluster, id).Error; err != nil {
c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "集群不存在"})
return
}
if cluster.KubeConfig == "" && cluster.Token == "" {
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "集群未配置凭证"})
return
}
client, err := buildK8sClient(&cluster)
if err != nil {
c.JSON(http.StatusInternalServerError, model.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 k8s_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 k8s_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)
result.Errors = append(result.Errors, "nodes: "+err.Error())
}
// Sync pods
if pods, err := client.ListPods(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&model.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(&model.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)
result.Errors = append(result.Errors, "pods: "+err.Error())
}
// Sync deployments
if deps, err := client.ListDeployments(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&model.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(&model.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)
result.Errors = append(result.Errors, "deployments: "+err.Error())
}
// Sync services
if svcs, err := client.ListServices(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&model.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(&model.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++
}
} else {
log.Printf("[K8s] sync services error: %v", err)
result.Errors = append(result.Errors, "services: "+err.Error())
}
// Sync ingresses
if ings, err := client.ListIngresses(); err == nil {
h.db.Where("cluster_id = ?", cluster.ID).Delete(&model.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(&model.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++
}
} else {
log.Printf("[K8s] sync ingresses error: %v", err)
result.Errors = append(result.Errors, "ingresses: "+err.Error())
}
// 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,
"ingress_count": result.Ingresses,
"last_sync_at": &now,
})
h.updateClusterUsageAfterSync(cluster.ID)
msg := fmt.Sprintf("同步完成: %d nodes, %d pods, %d deployments, %d services, %d ingresses",
result.Nodes, result.Pods, result.Deployments, result.Services, result.Ingresses)
if len(result.Errors) > 0 {
msg += " (部分资源同步失败,详见 errors 字段)"
}
c.JSON(http.StatusOK, model.Response{
Code: 200,
Message: msg,
Data: result,
})
}
// ============================================================================
// Pod Logs
// ============================================================================
// GetPodLogs 获取 Pod 日志(支持 stream=true 实时流)
func (h *KubernetesHandler) ListServiceAccountsNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 {
c.JSON(500, model.Response{Code: 500, Message: "cluster not found"})
return
}
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(500, model.Response{Code: 500, Message: err.Error()})
return
}
sas, err := client.ListServiceAccounts(c.Param("namespace"))
if err != nil {
c.JSON(500, model.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
type SASummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
SecretCount int `json:"secret_count"`
ImagePullSecrets []string `json:"image_pull_secrets"`
Age string `json:"age"`
}
result := make([]SASummary, 0, len(sas))
for _, sa := range sas {
ipsNames := make([]string, 0, len(sa.ImagePullSecrets))
for _, ips := range sa.ImagePullSecrets {
ipsNames = append(ipsNames, ips.Name)
}
result = append(result, SASummary{
Namespace: sa.Metadata.Namespace,
Name: sa.Metadata.Name,
SecretCount: len(sa.Secrets),
ImagePullSecrets: ipsNames,
Age: formatAge(sa.Metadata.CreationTimestamp),
})
}
c.JSON(200, model.Response{Code: 200, Data: result})
}
// ListRolesNS lists Roles in a namespace
func (h *KubernetesHandler) ListRolesNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 {
c.JSON(500, model.Response{Code: 500, Message: "cluster not found"})
return
}
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(500, model.Response{Code: 500, Message: err.Error()})
return
}
roles, err := client.ListRoles(c.Param("namespace"))
if err != nil {
c.JSON(500, model.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
type RoleSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
RuleCount int `json:"rule_count"`
Verbs []string `json:"verbs"`
Resources []string `json:"resources"`
Age string `json:"age"`
}
result := make([]RoleSummary, 0, len(roles.Items))
for _, r := range roles.Items {
allVerbs := []string{}
allResources := []string{}
seen := map[string]bool{}
for _, rule := range r.Rules {
for _, v := range rule.Verbs {
if !seen[v] {
allVerbs = append(allVerbs, v)
seen[v] = true
}
}
for _, res := range rule.Resources {
if !seen[res] {
allResources = append(allResources, res)
seen[res] = true
}
}
}
result = append(result, RoleSummary{
Namespace: r.Metadata.Namespace,
Name: r.Metadata.Name,
RuleCount: len(r.Rules),
Verbs: allVerbs,
Resources: allResources,
Age: formatAge(r.Metadata.CreationTimestamp),
})
}
c.JSON(200, model.Response{Code: 200, Data: result})
}
// ListRoleBindingsNS lists RoleBindings in a namespace
func (h *KubernetesHandler) ListRoleBindingsNS(c *gin.Context) {
clusterID := h.resolveClusterCode(c.Param("clusterCode"))
if clusterID == 0 {
c.JSON(500, model.Response{Code: 500, Message: "cluster not found"})
return
}
client, _, err := h.getK8sClient(clusterID)
if err != nil {
c.JSON(500, model.Response{Code: 500, Message: err.Error()})
return
}
rbs, err := client.ListRoleBindings(c.Param("namespace"))
if err != nil {
c.JSON(500, model.Response{Code: 500, Message: "K8s API error: " + err.Error()})
return
}
type RBSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
RoleRefKind string `json:"role_ref_kind"`
RoleRefName string `json:"role_ref_name"`
Subjects []string `json:"subjects"`
Age string `json:"age"`
}
result := make([]RBSummary, 0, len(rbs.Items))
for _, rb := range rbs.Items {
subjStrs := make([]string, 0, len(rb.Subjects))
for _, s := range rb.Subjects {
if s.Namespace != "" && s.Namespace != rb.Metadata.Namespace {
subjStrs = append(subjStrs, fmt.Sprintf("%s/%s (%s)", s.Kind, s.Name, s.Namespace))
} else {
subjStrs = append(subjStrs, fmt.Sprintf("%s/%s", s.Kind, s.Name))
}
}
result = append(result, RBSummary{
Namespace: rb.Metadata.Namespace,
Name: rb.Metadata.Name,
RoleRefKind: rb.RoleRef.Kind,
RoleRefName: rb.RoleRef.Name,
Subjects: subjStrs,
Age: formatAge(rb.Metadata.CreationTimestamp),
})
}
c.JSON(200, model.Response{Code: 200, Data: result})
}