Koodinator 混合工作负载调度系统

[root@master-01 ~]# kubectl get nodemetrics.slo.koordinator.sh master-01 -o yaml
apiVersion: slo.koordinator.sh/v1alpha1
kind: NodeMetric
metadata:
  creationTimestamp: "2025-05-24T11:31:54Z"
  generation: 1
  name: master-01
  resourceVersion: "15281512"
  uid: 21d81aaf-0338-4e94-bccc-07540fec3575
spec:
  metricCollectPolicy:
    aggregateDurationSeconds: 300
    nodeAggregatePolicy:
      durations:
      - 5m0s
      - 10m0s
      - 30m0s
    nodeMemoryCollectPolicy: usageWithoutPageCache
    reportIntervalSeconds: 60
status:
  nodeMetric:
    aggregatedNodeUsages:
    - duration: 5m0s
      usage:
        p50:
          resources:
            cpu: 429m
            memory: 5712464Ki
        p90:
          resources:
            cpu: 886m
            memory: 5750116Ki
        p95:
          resources:
            cpu: 1097m
            memory: 5833412Ki
        p99:
          resources:
            cpu: 1366m
            memory: 5938812Ki
    #     ...
    - duration: 30m0s
      usage:
        p50:
          resources:
            cpu: 429m
            memory: 5723620Ki
        p90:
          resources:
            cpu: 858m
            memory: 5823232Ki
        p95:
          resources:
            cpu: 1113m
            memory: 5867972Ki
        p99:
          resources:
            cpu: 1467m
            memory: 5977372Ki
    aggregatedSystemUsages:
    - duration: 5m0s
      usage:
        p50:
          resources:
            cpu: 153m
            memory: 3078852Ki
        p90:
          resources:
            cpu: 407m
            memory: 3085764Ki
        p95:
          resources:
            cpu: 539m
            memory: 3088652Ki
        p99:
          resources:
            cpu: 795m
            memory: 3106312Ki
    # ...
    - duration: 30m0s
      usage:
        p50:
          resources:
            cpu: 156m
            memory: 3082092Ki
        p90:
          resources:
            cpu: 424m
            memory: 3179492Ki
        p95:
          resources:
            cpu: 569m
            memory: 3184832Ki
        p99:
          resources:
            cpu: 853m
            memory: 3203756Ki
    nodeUsage:
      resources:
        cpu: 515m
        memory: "5862330450"
    systemUsage:
      resources:
        cpu: 196m
        memory: "3150994008"
  podsMetric:
  - name: my-clickhouse-zookeeper-0
    namespace: clickhouse
    podUsage:
      resources:
        cpu: 26m
        memory: "396051912"
    priority: koord-prod
    qos: LS
  # ....
  - name: mysql-7474b86d4f-52p5n
    namespace: mysql
    podUsage:
      resources:
        cpu: 21m
        memory: "393941019"
    priority: koord-prod
    qos: LSR
  prodReclaimableMetric:
    resource:
      resources:
        cpu: 2380m
        memory: "0"
  updateTime: "2025-05-25T03:36:38Z"

// https://github.com/koordinator-sh/koordinator/blob/76436ec089110311b8a8a359918e1c7b0942ed67/cmd/koord-scheduler/main.go
var koordinatorPlugins = map[string]frameworkruntime.PluginFactory{
	loadaware.Name:               loadaware.New, // 负载感知
	nodenumaresource.Name:        nodenumaresource.New, // numa 感知
	reservation.Name:             reservation.New, // 资源预留
	coscheduling.Name:            coscheduling.New,
	deviceshare.Name:             deviceshare.New,
	elasticquota.Name:            elasticquota.New,
	defaultprebind.Name:          defaultprebind.New,
	noderesourcesfitplus.Name:    noderesourcesfitplus.New,
	scarceresourceavoidance.Name: scarceresourceavoidance.New,
	schedulinghint.Name:          schedulinghint.New,
}

func (p *Plugin) Filter(ctx context.Context, state *framework.CycleState, pod *corev1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
	node := nodeInfo.Node()
    
	// ...
	
	return p.filterNodeUsage(node.Name, pod, usageThresholds, estimated, allocatable, isAgg)
}

func (p *Plugin) filterNodeUsage(nodeName string, pod *corev1.Pod, usageThresholds, estimatedUsed, allocatable ResourceVector, isAgg bool) *framework.Status {
	for i, value := range usageThresholds {
		if value == 0 {
			continue
		}
		total := allocatable[i]
		if total == 0 {
			continue
		}
		// 按利用率阈值过滤节点
		estimated := estimatedUsed[i]
		usage := int64(math.Round(float64(estimated) / float64(total) * 100))
		if usage <= value { // 使用率满足要求
			continue
		}

		// 使用率不满足需求
		reason := ErrReasonUsageExceedThreshold
		if isAgg {
			reason = ErrReasonAggregatedUsageExceedThreshold
		}
		resourceName := p.vectorizer[i]
		if klog.V(5).Enabled() {
			klog.InfoS("Node is unschedulable since usage exceeds threshold", "pod", klog.KObj(pod), "node", nodeName,
				"resource", resourceName, "usage", usage, "threshold", value,
				"estimated", getResourceQuantity(resourceName, estimated),
				"total", getResourceQuantity(resourceName, total))
		}
		return framework.NewStatus(framework.Unschedulable, fmt.Sprintf(reason, resourceName))
	}
	return nil
}

func (p *Plugin) Score(ctx context.Context, state *framework.CycleState, pod *corev1.Pod, nodeName string) (int64, *framework.Status) {
    // ...
	
	if klog.V(6).Enabled() {
		klog.InfoS("Estimate node usage for scoring", "pod", klog.KObj(pod), "node", nodeMetric.Name,
			"estimated", klog.Format(p.vectorizer.ToList(estimated)),
			"estimatedExistingPods", klog.KObjSlice(estimatedPods))
	}
	score := loadAwareSchedulingScorer(p.args.DominantResourceWeight, p.scoreWeights, estimated, allocatable)
	return score, nil
}

func loadAwareSchedulingScorer(dominantWeight int64, resToWeightMap, used, allocatable ResourceVector) int64 {
	var nodeScore, dominantScore, weightSum int64
	if dominantWeight != 0 {
		dominantScore, weightSum = framework.MaxNodeScore, dominantWeight
	}
	for i, weight := range resToWeightMap {
		score := leastUsedScore(used[i], allocatable[i])
		nodeScore += score * weight
		weightSum += weight
		if dominantScore > score {
			dominantScore = score
		}
	}
	nodeScore += dominantScore * dominantWeight
	if weightSum <= 0 {
		return 0
	}
	return nodeScore / weightSum
}


func leastUsedScore(used, capacity int64) int64 {
	if capacity == 0 {
		return 0
	}
	if used > capacity {
		return 0
	}
    // 计算规则
	return ((capacity - used) * framework.MaxNodeScore) / capacity
}