package services import ( "context" "fmt" "opsxloki/pkg/crypto" "os" "path/filepath" "strings" "time" "opsxloki/config" "opsxloki/models" "opsxloki/repository" "opsxloki/utils" ) type ClusterService struct{} func NewClusterService() *ClusterService { return &ClusterService{} } func SaveKubeConfig(name, content string) (string, error) { dir := config.GetConfig().KubeDir if err := os.MkdirAll(dir, 0700); err != nil { return "", err } path := filepath.Join(dir, fmt.Sprintf("%s.yaml", name)) if err := os.WriteFile(path, []byte(content), 0600); err != nil { return "", err } return path, nil } // MigrateClusterKubeconfigs 启动时把历史明文 kubeconfig 加密回 DB,并删除明文文件。 // 幂等:已加密的直接跳过。 func MigrateClusterKubeconfigs() { db := repository.GetDB() if db == nil { return } kubeDir := filepath.Clean(config.GetConfig().KubeDir) db.Lock() changed := false for i := range db.Clusters { c := &db.Clusters[i] if c.KubeConfig == "" || c.IsKubeconfigEncrypted() { continue } if err := c.EncryptAndSetKubeconfig(c.KubeConfig); err != nil { continue } if c.KubeConfigPath != "" { p := filepath.Clean(c.KubeConfigPath) if strings.HasPrefix(p, kubeDir+string(os.PathSeparator)) { _ = os.Remove(p) } c.KubeConfigPath = "" } changed = true } db.Unlock() if changed { _ = db.RequestSaveSync() } } // Create 创建集群。token 接入时自动构造等价 kubeconfig,统一存储,使部署模块同样可用。 func (s *ClusterService) Create(name, description, configType, apiServer, kubeconfig, token, contextName string) (*models.Cluster, error) { if configType == "" { configType = "kubeconfig" } finalKubeconfig := kubeconfig if configType == "token" { // Token 模式通常用于内网/自签证书集群:默认跳过 TLS 校验。 // 如需严格校验,请使用 kubeconfig 模式(其中包含 CA 证书)。 finalKubeconfig = BuildKubeconfigFromToken(apiServer, token, "", true) // Token 模式默认跳过 TLS 校验,兼容内网自签证书 } // 展示用 APIServer:token 模式取入参,kubeconfig 模式从内容提取 if apiServer == "" { apiServer = ExtractAPIServer(finalKubeconfig) } // 安全加固:kubeconfig 只加密存 DB,不再落明文文件 cluster := models.Cluster{ ID: repository.GetDB().NextID("clusters"), Name: name, Description: description, ConfigType: configType, APIServer: apiServer, KubeConfig: "", KubeConfigPath: "", Context: contextName, Status: "pending", CreatedAt: time.Now(), UpdatedAt: time.Now(), } if err := cluster.EncryptAndSetKubeconfig(finalKubeconfig); err != nil { return nil, fmt.Errorf("kubeconfig 加密失败: %w", err) } db := repository.GetDB() db.Lock() db.Clusters = append(db.Clusters, cluster) db.Unlock() db.RequestSave() go s.Validate(cluster.ID) return &cluster, nil } func (s *ClusterService) Get(id uint) (*models.Cluster, error) { db := repository.GetDB() db.RLock() defer db.RUnlock() for i := range db.Clusters { if db.Clusters[i].ID == id { c := db.Clusters[i] return &c, nil } } return nil, utils.ErrNotFound } func (s *ClusterService) List(page, pageSize int) ([]models.Cluster, int64, error) { db := repository.GetDB() db.RLock() defer db.RUnlock() total := int64(len(db.Clusters)) start := (page - 1) * pageSize if start < 0 || start >= len(db.Clusters) { return []models.Cluster{}, total, nil } end := start + pageSize if end > len(db.Clusters) { end = len(db.Clusters) } result := make([]models.Cluster, end-start) copy(result, db.Clusters[start:end]) return result, total, nil } func (s *ClusterService) Update(id uint, updates map[string]interface{}) error { db := repository.GetDB() db.Lock() defer db.Unlock() for i := range db.Clusters { if db.Clusters[i].ID == id { if v, ok := updates["name"].(string); ok && v != "" { db.Clusters[i].Name = v } if v, ok := updates["description"].(string); ok { db.Clusters[i].Description = v } if v, ok := updates["status"].(string); ok && v != "" { db.Clusters[i].Status = v } if v, ok := updates["version"].(string); ok { db.Clusters[i].Version = v } if v, ok := updates["nodesCount"].(int); ok { db.Clusters[i].NodesCount = v } if v, ok := updates["context"].(string); ok { db.Clusters[i].Context = v } if v, ok := updates["lastConnected"].(time.Time); ok { db.Clusters[i].LastConnected = v } // 接入方式与展示字段 if v, ok := updates["configType"].(string); ok && v != "" { db.Clusters[i].ConfigType = v } if v, ok := updates["apiServer"].(string); ok { db.Clusters[i].APIServer = v } // 凭据:kubeconfig 或 token 二选一,非空才更新,统一加密存 DB if v, ok := updates["kubeconfig"].(string); ok && v != "" { if enc, err := crypto.EncryptField(v); err == nil { db.Clusters[i].KubeConfig = enc db.Clusters[i].KubeConfigPath = "" } } else if tok, ok := updates["token"].(string); ok && tok != "" { apiServer := db.Clusters[i].APIServer if v, ok := updates["apiServer"].(string); ok && v != "" { apiServer = v } if apiServer == "" { return fmt.Errorf("Token 接入需要 API Server") } kc := BuildKubeconfigFromToken(apiServer, tok, "", true) // Token 模式默认跳过 TLS 校验 if enc, err := crypto.EncryptField(kc); err == nil { db.Clusters[i].KubeConfig = enc db.Clusters[i].KubeConfigPath = "" } } db.Clusters[i].UpdatedAt = time.Now() db.RequestSave() return nil } } return utils.ErrNotFound } func (s *ClusterService) Delete(id uint) error { db := repository.GetDB() db.Lock() defer db.Unlock() for i, c := range db.Clusters { if c.ID == id { if c.KubeConfigPath != "" { _ = os.Remove(c.KubeConfigPath) } db.Clusters = append(db.Clusters[:i], db.Clusters[i+1:]...) db.RequestSave() return nil } } return utils.ErrNotFound } func (s *ClusterService) Validate(id uint) { cluster, err := s.Get(id) if err != nil { return } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() info, err := GetClusterInfo(ctx, cluster) if err != nil { s.Update(id, map[string]interface{}{"status": "error"}) return } s.Update(id, map[string]interface{}{ "status": "active", "version": info.Version, "nodesCount": len(info.Nodes), "lastConnected": time.Now(), "context": info.CurrentCtx, }) }