pcm-participant/client/initialize/cloudCluster.go

package initialize

import (
	"fmt"
	"github.com/go-resty/resty/v2"
	"gitlink.org.cn/JointCloud/pcm-participant-client/common/types"
	"gitlink.org.cn/JointCloud/pcm-participant-client/common/utils"
	"gitlink.org.cn/JointCloud/pcm-participant-client/config"
	"gitlink.org.cn/JointCloud/pcm-participant-cloud/platform"
	"gitlink.org.cn/JointCloud/pcm-participant-cloud/service"
	eci "gitlink.org.cn/JointCloud/pcm-participant-eci"
	k8s "gitlink.org.cn/JointCloud/pcm-participant-k8s"
	"strconv"

	"go.uber.org/zap"
)

// 获取所有集群信息
func GetAllCloudClusterInfos() map[string]types.ClusterInfo {
	result := make(map[string]types.ClusterInfo)
	clusterInfos.Range(func(key, value interface{}) bool {
		result[key.(string)] = value.(types.ClusterInfo)
		return true
	})
	return result
}

func InitCloudCluster(cfg *config.Server) (*service.Service, error) {
	client := utils.InitClient(cfg.PcmCore.CoordinatorHost, "")
	return initCloudSvc(client, cfg.PcmCore)
}

// 初始化智算集群连接池
func initCloudSvc(client *utils.RestyClient, core config.PcmCore) (*service.Service, error) {
	resp := types.ResultResp{}
	token := "Bearer " + core.Token
	_, err := client.Request(core.CoordinatorHost+core.CloudClusterList, "GET", func(req *resty.Request) {
		req.SetHeader("Authorization", token).SetResult(&resp)
	})
	if err != nil {
		return nil, fmt.Errorf("获取集群列表失败: %w", err)
	}

	if resp.Code != 200 {
		return nil, fmt.Errorf("API返回错误: %d, 消息: %s", resp.Code, resp.Msg)
	}

	var platforms []platform.IPlatform
	var svc *service.Service
	//var octopus octopus.Octopus

	for _, cluster := range resp.Data.List {
		if cluster.Status == "offline" {

			// 修改集群的server地址为本服务的地址，修改状态为在线
			cluster.Server = core.ParticipantHost
			cluster.Status = "online"

			updateClusterReq := types.ClusterCreateReq{
				Id:            cluster.Id,
				AdapterId:     strconv.FormatInt(cluster.AdapterId, 10),
				Name:          cluster.Name,
				Nickname:      cluster.Nickname,
				Description:   cluster.Description,
				Server:        cluster.Server,
				MonitorServer: cluster.MonitorServer,
				Username:      cluster.Username,
				Password:      cluster.Password,
				Token:         cluster.Token,
				Ak:            cluster.Ak,
				Sk:            cluster.Sk,
				RegionName:    cluster.Region,
				ProjectId:     cluster.ProjectId,
				Version:       cluster.Version,
				Label:         cluster.Label,
				OwnerId:       cluster.OwnerId,
				AuthType:      cluster.AuthType,
				ProducerDict:  cluster.ProducerDict,
				RegionDict:    cluster.RegionDict,
				Status:        cluster.Status,
			}

			updateResp := types.ResultResp{}
			_, err := client.Request(core.CoordinatorHost+"/pcm/v1/adapter/cluster/update", "PUT", func(req *resty.Request) {
				req.SetBody(updateClusterReq).SetResult(&updateResp) // 添加请求体
			})

			if err != nil {
				zap.L().Error("更新集群状态失败", zap.Error(err))
				continue
			}
			if updateResp.Code != 200 {
				zap.L().Error("更新集群状态API返回错误",
					zap.Int("code", updateResp.Code),
					zap.String("message", updateResp.Msg))
				continue
			}

		} else if cluster.Status == "online" {
			if cluster.Server != core.ParticipantHost {
				zap.L().Warn("集群已被其他服务代理",
					zap.String("cluster_id", cluster.Id),
					zap.String("当前服务地址", core.ParticipantHost),
					zap.String("集群记录地址", cluster.Server))
				continue
			} else {
				// 更新集群信息的修改时间
				_, err := client.Request(core.CoordinatorHost+"/pcm/v1/adapter/cluster/update", "PUT", func(req *resty.Request) {
					req.SetBody(cluster)
				})
				if err != nil {
					zap.L().Error("刷新集群时间失败", zap.Error(err))
				}
			}
		}

		if cluster.Id == "" {
			zap.L().Warn("跳过无效集群条目: 缺少集群ID")
			continue
		}
		switch cluster.Label {
		case "kubernetes":
			k8s, err := k8s.New(cluster.Token, cluster.Address, platform.Id(cluster.Id))
			if err != nil {
				Error("初始化失败", zap.Error(err))
				continue
			}
			platforms = append(platforms, k8s)
			//更新C端集群状态
		case "eci":
			eci, err := eci.New(cluster.Ak, cluster.Sk, cluster.Password, platform.Id(cluster.Id))
			if err != nil {
				Error("初始化失败", zap.Error(err))
				continue
			}
			platforms = append(platforms, eci)

		}
	}
	if len(platforms) == 0 {
		return nil, fmt.Errorf("注册集群列表为空")
	}

	for _, p := range platforms {
		Info("注册集群列表:", zap.Any("id", p.Id()), zap.Any("type", p.Type()), zap.Any("name", p.Name()))
	}
	svc, err = service.NewService(platforms...)
	if err != nil {
		return nil, err
	}
	return svc, nil
}