pcm-coordinator/internal/scheduler/strategy/leastLoadFirst.go

package strategy

import (
	"errors"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/common"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/collector"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/storeLink"
	"strings"
)

const (
	RUNNING_TASK_WEIGHT     = 300
	TASK_PREDICTED_WEIGHT   = 0.03
	UNLIMITED_PREDICTED_NUM = 10000
)

type LeastLoadFirst struct {
	replicas int32
	clusters []*CLusterLoad
}

type CLusterLoad struct {
	ClusterId        string
	TaskRunningNum   int64
	TaskPredictedNum int64
}

func NewLeastLoadFirst(replicas int32, resources []*collector.ResourceSpec) *LeastLoadFirst {
	var clusters []*CLusterLoad

	for _, resource := range resources {
		if resource.ClusterId == "" {
			continue
		}
		cluster := &CLusterLoad{
			ClusterId: resource.ClusterId,
		}
		var balance float64
		var rate float64
		for _, res := range resource.Resources {
			r, ok := res.(*collector.ClusterResource)
			if !ok {
				continue
			}
			switch r.Resource.Type {
			case storeLink.RUNNINGTASK:
				num, ok := r.Resource.Total.Value.(int64)
				if !ok {
					continue
				}
				cluster.TaskRunningNum = num
			case strings.ToUpper(storeLink.BALANCE):
				b, ok := r.Resource.Total.Value.(float64)
				if !ok {
					continue
				}
				balance = b
			case strings.ToUpper(storeLink.RATE):
				rt, ok := r.Resource.Total.Value.(float64)
				if !ok {
					continue
				}
				rate = rt
			default:
				continue
			}
		}

		if balance == 0 || rate == 0 {
			continue
		}

		var predictedNum int64
		if balance == -1 {
			// unlimited
			predictedNum = UNLIMITED_PREDICTED_NUM
		} else {
			predictedNum = int64(common.RoundFloat(balance/rate, 0))
		}

		cluster.TaskPredictedNum = predictedNum
		clusters = append(clusters, cluster)
	}

	return &LeastLoadFirst{replicas: replicas, clusters: clusters}
}

func (l *LeastLoadFirst) Schedule() ([]*AssignedCluster, error) {
	if l.replicas < 1 {
		return nil, errors.New("replicas must be greater than 0")
	}

	swMap := make(map[string]int32, 0)
	for _, cluster := range l.clusters {
		var runningTaskWeight int64
		if cluster.TaskRunningNum != 0 {
			runningTaskWeight = RUNNING_TASK_WEIGHT / cluster.TaskRunningNum
		} else {
			runningTaskWeight = RUNNING_TASK_WEIGHT * 2
		}
		weight := runningTaskWeight + int64(float64(cluster.TaskPredictedNum)*TASK_PREDICTED_WEIGHT)
		swMap[cluster.ClusterId] = int32(weight)
	}

	sw := NewStaticWeightStrategy(swMap, l.replicas)
	clusters, err := sw.Schedule()
	if err != nil {
		return nil, err
	}

	return clusters, nil
}