package core import ( "context" "fmt" "github.com/mitchellh/mapstructure" "github.com/rs/zerolog/log" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/logic/schedule" "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models" "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils" "gorm.io/gorm" "time" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types" "github.com/zeromicro/go-zero/core/logx" ) type CompareResourceSpecLogic struct { logx.Logger ctx context.Context svcCtx *svc.ServiceContext } const ( ChangeTypeNormal = 0 // 资源规格正常 ChangeTypeModified = 1 // 资源规格变更 ChangeTypeDeleted = 2 // 资源被删除 ) type APIResponse struct { ClusterId string `json:"ClusterId"` ClusterType string `json:"clusterType"` Region string `json:"region"` Resources []Resource `json:"resources"` Msg string `json:"msg"` } type Resource struct { Resource ResourceDetail `json:"resource"` BaseResources []ResourceDetail `json:"baseResources"` } type ResourceDetail struct { Type string `json:"type"` Name string `json:"name"` Total Metric `json:"total"` Available Metric `json:"available"` } type Metric struct { Unit string `json:"unit"` Value float64 `json:"value"` } func NewCompareResourceSpecLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CompareResourceSpecLogic { return &CompareResourceSpecLogic{ Logger: logx.WithContext(ctx), ctx: ctx, svcCtx: svcCtx, } } func (l *CompareResourceSpecLogic) CompareResourceSpec(req *types.ResourceSpecReq) (resp *types.PageResult, err error) { log.Debug().Msgf("开始比较资源规格,ClusterId: %s, PageNum: %d, PageSize: %d", req.ClusterId, req.PageNum, req.PageSize) if req.ClusterId == "" { return resp, nil } // 获取集群资源数据 startTime := time.Now() apiResources, err := l.FetchClusterResources(req.ClusterId) log.Debug().Msgf("调用获取ai训练资源接口耗时: %v", time.Since(startTime)) if err != nil { return nil, fmt.Errorf("failed to fetch cluster resources: %w", err) } // 同步资源到数据库 if err := l.syncResourcesToDB(apiResources); err != nil { return nil, fmt.Errorf("failed to sync resources: %w", err) } // 查询数据库结果 return l.queryDatabaseResults(req) } func (l *CompareResourceSpecLogic) FetchClusterResources(ClusterId string) ([]APIResponse, error) { queryLogic := schedule.NewQueryResourcesLogic(l.ctx, l.svcCtx) resources, err := queryLogic.QueryResources(&types.QueryResourcesReq{ ClusterIDs: []string{ClusterId}, }) if err != nil { return nil, fmt.Errorf("query resources failed: %w", err) } var apiResponses []APIResponse if err := decodeAPIResponse(resources.Data, &apiResponses); err != nil { return nil, fmt.Errorf("decode response failed: %w", err) } for _, resource := range apiResponses { if resource.Resources == nil { return nil, fmt.Errorf(resource.Msg) } } return apiResponses, nil } func decodeAPIResponse(input interface{}, output *[]APIResponse) error { config := &mapstructure.DecoderConfig{ Result: output, TagName: "json", ErrorUnused: true, DecodeHook: mapstructure.ComposeDecodeHookFunc( mapstructure.StringToTimeHookFunc(time.RFC3339), mapstructure.StringToSliceHookFunc(","), ), } decoder, err := mapstructure.NewDecoder(config) if err != nil { return fmt.Errorf("failed to create decoder: %w", err) } if err := decoder.Decode(input); err != nil { return fmt.Errorf("decoding error: %w", err) } return nil } func (l *CompareResourceSpecLogic) syncResourcesToDB(apiResponses []APIResponse) error { for _, response := range apiResponses { // 转换API响应到数据库模型 dbSpecs, apiSpecs, err := l.processAPIResponse(response) if err != nil { return err } // 处理资源变更 if err := l.handleResourceChanges(dbSpecs, apiSpecs); err != nil { return fmt.Errorf("failed to handle resource changes: %w", err) } } return nil } func (l *CompareResourceSpecLogic) processAPIResponse(response APIResponse) ([]models.TResourceSpec, []models.TResourceSpec, error) { ClusterId := utils.StringToInt64(response.ClusterId) var dbSpecs []models.TResourceSpec if err := l.svcCtx.DbEngin.Model(models.TResourceSpec{}).Preload("BaseResourceSpecs"). Where("cluster_id = ?", ClusterId). Find(&dbSpecs).Error; err != nil { return nil, nil, fmt.Errorf("database query failed: %w", err) } var apiSpecs []models.TResourceSpec for _, res := range response.Resources { // 检查资源类型和名称是否存在 if res.Resource.Name == "" || res.Resource.Type == "" { continue } spec := l.convertToResourceSpec(ClusterId, res) apiSpecs = append(apiSpecs, spec) } return dbSpecs, apiSpecs, nil } func (l *CompareResourceSpecLogic) handleResourceChanges(dbSpecs, apiSpecs []models.TResourceSpec) error { // 创建资源映射 dbMap := make(map[string]models.TResourceSpec) for _, spec := range dbSpecs { key := spec.SourceKey dbMap[key] = spec } apiMap := make(map[string]models.TResourceSpec) for _, spec := range apiSpecs { key := resourceKey(spec.Type, spec.Name) apiMap[key] = spec } // 处理新增或更新的资源 for key, apiSpec := range apiMap { dbSpec, exists := dbMap[key] if !exists { // 资源不存在于数据库,创建新资源 if err := l.createNewResource(&apiSpec); err != nil { return err } continue } // 资源已存在于数据库,检查是否有变更,资源规格有变更则更新changeType为变更 if l.isSpecChanged(dbSpec, apiSpec) { if err := l.updateResource(&dbSpec, apiSpec, ChangeTypeModified); err != nil { return err } } else { // 如果资源存在且没有变更,更新changeType为正常 if err := l.updateResource(&dbSpec, apiSpec, ChangeTypeNormal); err != nil { return err } } } // 处理删除的资源 for key, dbSpec := range dbMap { if _, exists := apiMap[key]; !exists { if err := l.markResourceDeleted(dbSpec.Id); err != nil { return err } } } return nil } func resourceKey(resType, name string) string { return fmt.Sprintf("%s::%s", resType, name) } func (l *CompareResourceSpecLogic) createNewResource(spec *models.TResourceSpec) error { return l.svcCtx.DbEngin.Transaction(func(tx *gorm.DB) error { if err := tx.Create(spec).Error; err != nil { return fmt.Errorf("failed to create resource: %w", err) } return nil }) } // 标识资源规格变更 func (l *CompareResourceSpecLogic) updateResource(existing *models.TResourceSpec, newSpec models.TResourceSpec, changeType int) error { return l.svcCtx.DbEngin.Transaction(func(tx *gorm.DB) error { updates := map[string]interface{}{ "change_type": changeType, "update_time": time.Now(), } if err := tx.Model(existing).Updates(updates).Error; err != nil { return fmt.Errorf("failed to update resource: %w", err) } return l.syncBaseResources(tx, existing.Id, newSpec.BaseResourceSpecs) }) } func (l *CompareResourceSpecLogic) syncBaseResources(tx *gorm.DB, specID int64, newResources []models.TBaseResourceSpec) error { // 处理基础资源更新 var existingResources []models.TBaseResourceSpec if err := tx.Where("resource_spec_id = ?", specID).Find(&existingResources).Error; err != nil { return fmt.Errorf("failed to query base resources: %w", err) } existingMap := make(map[string]models.TBaseResourceSpec) for _, r := range existingResources { key := resourceKey(r.Type, r.Name) existingMap[key] = r } // 处理更新和新增 for i, newRes := range newResources { newRes.ResourceSpecId = specID key := resourceKey(newRes.Type, newRes.Name) if existing, exists := existingMap[key]; exists { newRes.Id = existing.Id newRes.CreateTime = existing.CreateTime if err := tx.Save(&newRes).Error; err != nil { return fmt.Errorf("failed to update base resource: %w", err) } } else { if err := tx.Create(&newRes).Error; err != nil { return fmt.Errorf("failed to create base resource: %w", err) } } newResources[i] = newRes } // 处理删除 currentIDs := make(map[int64]struct{}) for _, r := range newResources { currentIDs[r.Id] = struct{}{} } for _, existing := range existingResources { if _, exists := currentIDs[existing.Id]; !exists { if err := tx.Delete(&existing).Error; err != nil { return fmt.Errorf("failed to delete base resource: %w", err) } } } return nil } func (l *CompareResourceSpecLogic) markResourceDeleted(id int64) error { return l.svcCtx.DbEngin.Model(&models.TResourceSpec{}). Where("id = ?", id). Update("change_type", ChangeTypeDeleted). Error } func (l *CompareResourceSpecLogic) isSpecChanged(old, new models.TResourceSpec) bool { if old.TotalCount != new.TotalCount || old.AvailableCount != new.AvailableCount || old.Region != new.Region { return true } // 比较基础资源 oldBaseMap := make(map[string]models.TBaseResourceSpec) for _, br := range old.BaseResourceSpecs { oldBaseMap[resourceKey(br.Type, br.Name)] = br } for _, newBr := range new.BaseResourceSpecs { key := resourceKey(newBr.Type, newBr.Name) oldBr, exists := oldBaseMap[key] if !exists || oldBr.TotalValue != newBr.TotalValue || oldBr.AvailableValue != newBr.AvailableValue { return true } delete(oldBaseMap, key) } return len(oldBaseMap) > 0 } func (l *CompareResourceSpecLogic) convertToResourceSpec(ClusterId int64, res Resource) models.TResourceSpec { spec := models.TResourceSpec{ SourceKey: resourceKey(res.Resource.Type, res.Resource.Name), Type: res.Resource.Type, Name: res.Resource.Name, TotalCount: int64(res.Resource.Total.Value), AvailableCount: int64(res.Resource.Available.Value), ClusterId: ClusterId, CreateTime: time.Now(), UpdateTime: time.Now(), ChangeType: ChangeTypeNormal, } for _, br := range res.BaseResources { spec.BaseResourceSpecs = append(spec.BaseResourceSpecs, models.TBaseResourceSpec{ Type: br.Type, Name: br.Name, TotalValue: br.Total.Value, TotalUnit: br.Total.Unit, AvailableValue: br.Available.Value, AvailableUnit: br.Available.Unit, CreateTime: time.Now(), UpdateTime: time.Now(), }) } return spec } func (l *CompareResourceSpecLogic) queryDatabaseResults(req *types.ResourceSpecReq) (*types.PageResult, error) { result := &types.PageResult{ PageNum: req.PageNum, PageSize: req.PageSize, } query := l.buildBaseQuery(req) if err := query.Count(&result.Total).Error; err != nil { return nil, fmt.Errorf("failed to count records: %w", err) } var specs []*models.TResourceSpec if err := query.Model(models.TResourceSpec{}).Preload("BaseResourceSpecs"). Scopes(paginate(req.PageNum, req.PageSize)). Find(&specs).Error; err != nil { return nil, fmt.Errorf("failed to query resources: %w", err) } result.List = specs return result, nil } func (l *CompareResourceSpecLogic) buildBaseQuery(req *types.ResourceSpecReq) *gorm.DB { query := l.svcCtx.DbEngin.Model(&models.TResourceSpec{}). Where("cluster_id = ?", utils.StringToInt64(req.ClusterId)). Where("deleted_at IS NULL") if req.Status != "" { query = query.Where("status = ?", req.Status) } if req.ChangeType != "" { query = query.Where("change_type = ?", req.ChangeType) } if req.Type != "" { query = query.Where("type = ?", req.Type) } if req.Name != "" { query = query.Where("name LIKE ?", "%"+req.Name+"%") } return query } func paginate(pageNum, pageSize int) func(db *gorm.DB) *gorm.DB { return func(db *gorm.DB) *gorm.DB { offset := (pageNum - 1) * pageSize return db.Offset(offset).Limit(pageSize).Order("create_time DESC") } }