aiforge/services/cloudbrain/resource/resource_queue.go

package resource

import (
	"code.gitea.io/gitea/models"
	"code.gitea.io/gitea/modules/grampus"
	"code.gitea.io/gitea/modules/log"
	"fmt"
	"strings"
)

func AddResourceQueue(req models.ResourceQueueReq) error {
	if _, err := models.InsertResourceQueue(req.ToDTO()); err != nil {
		return err
	}
	return nil
}

func UpdateResourceQueue(queueId int64, req models.ResourceQueueReq) error {
	if _, err := models.UpdateResourceCardsTotalNum(queueId, models.ResourceQueue{
		CardsTotalNum: req.CardsTotalNum,
		Remark:        req.Remark,
	}); err != nil {
		return err
	}
	return nil
}

func GetResourceQueueList(opts models.SearchResourceQueueOptions) (*models.ResourceQueueListRes, error) {
	n, r, err := models.SearchResourceQueue(opts)
	if err != nil {
		return nil, err
	}

	return models.NewResourceQueueListRes(n, r), nil
}

func GetResourceQueueCodes(opts models.GetQueueCodesOptions) ([]*models.ResourceQueueCodesRes, error) {
	r, err := models.GetResourceQueueCodes(opts)
	if err != nil {
		return nil, err
	}

	return r, nil
}

func GetResourceAiCenters() ([]models.ResourceAiCenterRes, error) {
	r, err := models.GetResourceAiCenters()
	if err != nil {
		return nil, err
	}

	return r, nil
}

func SyncGrampusQueue(doerId int64) error {
	r, err := grampus.GetAiCenters(1, 100)
	if err != nil {
		return err
	}
	log.Info("SyncGrampusQueue result = %+v", r)
	queueUpdateList := make([]models.ResourceQueue, 0)
	queueInsertList := make([]models.ResourceQueue, 0)
	existIds := make([]int64, 0)

	for _, center := range r.Infos {
		for _, device := range center.AccDevices {
			computeResource := models.ParseComputeResourceFormGrampus(device.Kind)
			accCardType := strings.ToUpper(device.Model)
			if computeResource == "" {
				continue
			}
			//Determine if this quque already exists.if exist,update params
			//if not exist,insert a new record
			oldQueue, err := models.GetResourceQueue(&models.ResourceQueue{
				Cluster:         models.C2NetCluster,
				AiCenterCode:    center.Id,
				ComputeResource: computeResource,
				AccCardType:     accCardType,
			})
			if err != nil {
				return err
			}

			if oldQueue == nil {
				queueInsertList = append(queueInsertList, models.ResourceQueue{
					Cluster:         models.C2NetCluster,
					AiCenterCode:    center.Id,
					AiCenterName:    center.Name,
					ComputeResource: computeResource,
					AccCardType:     accCardType,
					IsAutomaticSync: true,
					CreatedBy:       doerId,
					UpdatedBy:       doerId,
				})
			} else {
				existIds = append(existIds, oldQueue.ID)
				queueUpdateList = append(queueUpdateList, models.ResourceQueue{
					ID:              oldQueue.ID,
					ComputeResource: computeResource,
					AiCenterName:    center.Name,
					AccCardType:     accCardType,
					UpdatedBy:       doerId,
				})
			}

		}
	}
	return models.SyncGrampusQueues(queueUpdateList, queueInsertList, existIds)
}

func SyncGrampusQueueAndSpecs() {
	defer func() {
		if err := recover(); err != nil {
			combinedErr := fmt.Errorf("%s\n%s", err, log.Stack(2))
			log.Error("PANIC:", combinedErr)
		}
	}()
	log.Info("start to sync grampus queue and specs")
	SyncGrampusQueue(0)
	SyncGrampusSpecs(0)
	log.Info("sync grampus queue and specs finished")
}