LokiStack/backend/services/cluster.go

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,
	})
}