Volcano

// 新增了一个 pod
func (sc *SchedulerCache) addPod(pod *v1.Pod) error {
	pi, err := sc.NewTaskInfo(pod)
	if err != nil {
		klog.Errorf("generate taskInfo for pod(%s) failed: %v", pod.Name, err)
		sc.resyncTask(pi)
	}

	return sc.addTask(pi)
}

// https://github.com/volcano-sh/volcano/blob/8944bfd7bc48cb3e1d6b3c047ef2fa52fec4c276/pkg/scheduler/plugins/binpack/binpack.go
func BinPackingScore(task *api.TaskInfo, node *api.NodeInfo, weight priorityWeight) float64 {
	score := 0.0
	weightSum := 0
	requested := task.Resreq
	allocatable := node.Allocatable
	used := node.Used

	for _, resource := range requested.ResourceNames() {
		request := requested.Get(resource)
		if request == 0 {
			continue
		}
		allocate := allocatable.Get(resource)
		nodeUsed := used.Get(resource)

		resourceWeight, found := weight.BinPackingResources[resource]
		if !found {
			continue
		}

		// 计算得分
		resourceScore, err := ResourceBinPackingScore(request, allocate, nodeUsed, resourceWeight)
		if err != nil {
			klog.V(4).Infof("task %s/%s cannot binpack node %s: resource: %s is %s, need %f, used %f, allocatable %f",
				task.Namespace, task.Name, node.Name, resource, err.Error(), request, nodeUsed, allocate)
			return 0
		}
		klog.V(5).Infof("task %s/%s on node %s resource %s, need %f, used %f, allocatable %f, weight %d, score %f",
			task.Namespace, task.Name, node.Name, resource, request, nodeUsed, allocate, resourceWeight, resourceScore)

		score += resourceScore
		weightSum += resourceWeight
	}

	// mapping the result from [0, weightSum] to [0, 10(MaxPriority)]
	if weightSum > 0 {
		// 总得分
		score /= float64(weightSum)
	}
	score *= float64(k8sFramework.MaxNodeScore * int64(weight.BinPackingWeight))

	return score
}

// 计算公式  Resource.weight * (request + used) / allocatable
func ResourceBinPackingScore(requested, capacity, used float64, weight int) (float64, error) {
	if capacity == 0 || weight == 0 {
		return 0, nil
	}

	usedFinally := requested + used
	if usedFinally > capacity {
		return 0, fmt.Errorf("not enough")
	}

	score := usedFinally * float64(weight) / capacity
	return score, nil
}

# CPU 权重值越高，得分越高，节点资源使用量越满，得分越高。Memory、GPU 等资源原理类似
CPU.weight * (request + used) / allocatable

(⎈|kubeasz-test:volcano-system)➜  ~ kubectl api-resources| head -1;kubectl api-resources |grep volcano
NAME                                SHORTNAMES            APIVERSION                             NAMESPACED   KIND
jobs                                vcjob,vj              batch.volcano.sh/v1alpha1              true         Job
commands                                                  bus.volcano.sh/v1alpha1                true         Command
jobflows                            jf                    flow.volcano.sh/v1alpha1               true         JobFlow
jobtemplates                        jt                    flow.volcano.sh/v1alpha1               true         JobTemplate
numatopologies                      numatopo              nodeinfo.volcano.sh/v1alpha1           false        Numatopology
podgroups                           pg,podgroup-v1beta1   scheduling.volcano.sh/v1beta1          true         PodGroup
queues                              q,queue-v1beta1       scheduling.volcano.sh/v1beta1          false        Queue
hypernodes                          hn                    topology.volcano.sh/v1alpha1           false        HyperNode

# default queue
(⎈|kubeasz-test:volcano-system)➜  ~ kubectl get queue default -o yaml
apiVersion: scheduling.volcano.sh/v1beta1
kind: Queue
metadata:
  creationTimestamp: "2025-07-26T02:38:24Z"
  generation: 2
  name: default
  resourceVersion: "16927137"
  uid: b3816aae-d77e-4576-8699-eed4d0324848
spec:
  guarantee: {}
  parent: root
  reclaimable: true
  weight: 1
status:
  allocated:
    cpu: "3"
    memory: "0"
    pods: "3"
  reservation: {}
  state: Open

# root的queue，该queue为开启层级队列功能时使用，作为所有队列的根队列，default queue为root queue的子队列
(⎈|kubeasz-test:volcano-system)➜  ~ kubectl get queue root -o yaml
apiVersion: scheduling.volcano.sh/v1beta1
kind: Queue
metadata:
  creationTimestamp: "2025-07-26T02:38:24Z"
  generation: 786
  name: root
  resourceVersion: "17202849"
  uid: 8399648f-7082-49a1-98ec-7fb9c1138172
spec:
  deserved:
    attachable-volumes-csi-local.csi.openebs.io: 12884901882m
    cpu: 92400m
    ephemeral-storage: "381412259579"
    hugepages-2Mi: "0"
    kubernetes.io/batch-cpu: "46335"
    kubernetes.io/batch-memory: "38576847428"
    kubernetes.io/mid-cpu: "11076"
    kubernetes.io/mid-memory: "10040600338"
    memory: 96480436Ki
    pods: "660"
  guarantee:
    resource:
      cpu: "0"
      memory: "0"
  reclaimable: false
  weight: 1
status:
  allocated:
    cpu: "3"
    memory: "0"
    pods: "3"
  reservation: {}
  state: Open

// https://github.com/volcano-sh/volcano/blob/b27b4bbe7d19e225e75a11e424bee38ec29a4041/pkg/scheduler/actions/factory.go
func init() {
	// 注册 action
	framework.RegisterAction(reclaim.New())
	framework.RegisterAction(allocate.New())
	framework.RegisterAction(backfill.New())
	framework.RegisterAction(preempt.New())
	framework.RegisterAction(enqueue.New())
	framework.RegisterAction(shuffle.New())
}

type Action interface {
	// The unique name of Action.
	Name() string

	// Initialize initializes the allocator plugins.
	Initialize()

	// Execute allocates the cluster's resources into each queue.
	Execute(ssn *Session)

	// UnIntialize un-initializes the allocator plugins.
	UnInitialize()
}

func (pc *Scheduler) runOnce() {
	klog.V(4).Infof("Start scheduling ...")
	scheduleStartTime := time.Now()
	defer klog.V(4).Infof("End scheduling ...")

	pc.mutex.Lock()
	actions := pc.actions
	plugins := pc.plugins
	configurations := pc.configurations
	pc.mutex.Unlock()

	// Load ConfigMap to check which action is enabled.
	conf.EnabledActionMap = make(map[string]bool)
	for _, action := range actions {
		conf.EnabledActionMap[action.Name()] = true
	}

	ssn := framework.OpenSession(pc.cache, plugins, configurations)
	defer func() {
		framework.CloseSession(ssn)
		metrics.UpdateE2eDuration(metrics.Duration(scheduleStartTime))
	}()

	// 遍历 actions, 执行 action
	for _, action := range actions {
		actionStartTime := time.Now()
		action.Execute(ssn) // 传递了一个 ssn（*Session 类型）对象进去
		metrics.UpdateActionDuration(action.Name(), metrics.Duration(actionStartTime))
	}
}