pcm-coordinator/internal/cron/hpc_cron_task.go

package cron

import (
	"errors"
	"fmt"
	"github.com/zeromicro/go-zero/core/logx"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/types"
)

// GetHpcTaskList get hpc task list
func GetHpcTaskList(svc *svc.ServiceContext) ([]*types.TaskModel, error) {
	limit := 10
	offset := 0
	var list []*types.TaskModel
	db := svc.DbEngin.Model(&types.TaskModel{}).Table("task").
		Joins("join task_hpc hpc on task.id = hpc.task_id").
		Select("task.* ,hpc.job_id, hpc.work_dir, hpc.status, hpc.updated_time").
		Where("task.adapter_type_dict = 2  AND task.status NOT IN ('Succeeded', 'Failed') and task.deleted_at is null")

	//count total
	var total int64
	err := db.Count(&total).Error
	if err != nil {
		return nil, err
	}

	db.Limit(limit).Offset(offset)
	err = db.Order("created_time desc").Scan(&list).Error
	if err != nil {
		return nil, err
	}
	return list, nil
}

func UpdateHpcAdapterMaps(svc *svc.ServiceContext) {
	var hpcType = "2"
	adapterIds, err := svc.Scheduler.HpcStorages.GetAdapterIdsByType(hpcType)
	if err != nil {
		msg := fmt.Sprintf("###UpdateHpcAdapterMaps###, error: %v \n", err.Error())
		logx.Errorf(errors.New(msg).Error())
		return
	}
	if len(adapterIds) == 0 {
		return
	}

	for _, id := range adapterIds {
		clusters, err := svc.Scheduler.HpcStorages.GetClustersByAdapterId(id)
		if err != nil {
			msg := fmt.Sprintf("###UpdateHpcAdapterMaps###, error: %v \n", err.Error())
			logx.Errorf(errors.New(msg).Error())
			return
		}
		if len(clusters.List) == 0 {
			continue
		}
		if hpcAdapterExist(svc, id, len(clusters.List)) {
			continue
		} else {
			if hpcAdapterEmpty(svc, id) {
				exeClusterMap := service.InitHpcClusterMap(&svc.Config, clusters.List)
				svc.Scheduler.HpcService.HpcExecutorAdapterMap[id] = exeClusterMap
			} else {
				svc.Scheduler.HpcService.UpdateHpcClusterMaps(&svc.Config, id, clusters.List)
			}
		}
	}
}

func hpcAdapterExist(svc *svc.ServiceContext, id string, clusterNum int) bool {
	emap, ok := svc.Scheduler.HpcService.HpcExecutorAdapterMap[id]

	if ok {
		if len(emap) == clusterNum {
			return true
		}
	}
	return false
}

func hpcAdapterEmpty(svc *svc.ServiceContext, id string) bool {
	_, ok := svc.Scheduler.HpcService.HpcExecutorAdapterMap[id]
	if !ok {
		return true
	}
	return false
}