版本1.18.1 代码路径：kubernetes/pkg/scheduler/framework/plugins PS：代码是经过我整理、删减，主要是为了自己阅读方便，但是不影响主要逻辑

1、noderesources(关于node资源相关的调度算法)

1.1、前置解析：resourceAllocationScorer

1.1.1、描述

此结构体及相关方法是公用的，NodeResourcesBalancedAllocation、NodeResourcesLeastAllocated、NodeResourcesMostAllocated等均会调用进行资源计算和最终的分值计算

1.1.2、代码解析

  1. var defaultRequestedRatioResources = resourceToWeightMap{v1.ResourceMemory: 1, v1.ResourceCPU: 1}
  3. type resourceToWeightMap map[v1.ResourceName]int64
  5. type resourceAllocationScorer struct {
  6.     Name                string
  7.     scorer              func(requested, allocable resourceToValueMap, includeVolumes bool, requestedVolumes int, allocatableVolumes int) int64
  8.     resourceToWeightMap resourceToWeightMap
  9. }
  11. type resourceToValueMap map[v1.ResourceName]int64
  13. func (r *resourceAllocationScorer) score(
  14.     pod *v1.Pod,
  15.     nodeInfo *schedulernodeinfo.NodeInfo) (int64, *framework.Status) {
  16.     node := nodeInfo.Node()
  17.     if node == nil {
  18.         return 0, framework.NewStatus(framework.Error, "node not found")
  19.     }
  20.     if r.resourceToWeightMap == nil {
  21.         return 0, framework.NewStatus(framework.Error, "resources not found")
  22.     }
  23.     requested := make(resourceToValueMap, len(r.resourceToWeightMap))
  24.     allocatable := make(resourceToValueMap, len(r.resourceToWeightMap))
  25.     for resource := range r.resourceToWeightMap {
  26.         allocatable[resource], requested[resource] = calculateResourceAllocatableRequest(nodeInfo, pod, resource)
  27.     }
  29.     var score int64
  30.     if len(pod.Spec.Volumes) >= 0 && utilfeature.DefaultFeatureGate.Enabled(features.BalanceAttachedNodeVolumes) && nodeInfo.TransientInfo != nil {
  31.         score = r.scorer(requested, allocatable, true, nodeInfo.TransientInfo.TransNodeInfo.RequestedVolumes, nodeInfo.TransientInfo.TransNodeInfo.AllocatableVolumesCount)
  32.     } else {
  33.         score = r.scorer(requested, allocatable, false, 0, 0)
  34.     }
  35.     return score, nil
  36. }
  38. func calculateResourceAllocatableRequest(nodeInfo *schedulernodeinfo.NodeInfo, pod *v1.Pod, resource v1.ResourceName) (int64, int64) {
  39.     allocatable := nodeInfo.AllocatableResource()
  40.     requested := nodeInfo.RequestedResource()
  41.     podRequest := calculatePodResourceRequest(pod, resource)
  42.     switch resource {
  43.     case v1.ResourceCPU:
  44.         return allocatable.MilliCPU, nodeInfo.NonZeroRequest().MilliCPU + podRequest
  45.     case v1.ResourceMemory:
  46.         return allocatable.Memory, nodeInfo.NonZeroRequest().Memory + podRequest
  47.     case v1.ResourceEphemeralStorage:
  48.         return allocatable.EphemeralStorage, requested.EphemeralStorage + podRequest
  49.     default:
  50.         if v1helper.IsScalarResourceName(resource) {
  51.             return allocatable.ScalarResources[resource], requested.ScalarResources[resource] + podRequest
  52.         }
  53.     }
  54.     return 0, 0
  55. }
  57. func calculatePodResourceRequest(pod *v1.Pod, resource v1.ResourceName) int64 {
  58.     var podRequest int64
  59.     for i := range pod.Spec.Containers {
  60.         container := &pod.Spec.Containers[i]
  61.         value := schedutil.GetNonzeroRequestForResource(resource, &container.Resources.Requests)
  62.         podRequest += value
  63.     }
  65.     for i := range pod.Spec.InitContainers {
  66.         initContainer := &pod.Spec.InitContainers[i]
  67.         value := schedutil.GetNonzeroRequestForResource(resource, &initContainer.Resources.Requests)
  68.         if podRequest < value {
  69.             podRequest = value
  70.         }
  71.     }
  73.     if pod.Spec.Overhead != nil && utilfeature.DefaultFeatureGate.Enabled(features.PodOverhead) {
  74.         if quantity, found := pod.Spec.Overhead[resource]; found {
  75.             podRequest += quantity.Value()
  76.         }
  77.     }
  79.     return podRequest
  80. }

1.2、NodeResourcesFit(PreFilter、Filter扩展点)

1.2.1、算法功能

该算法主要是检验给node是否能够满足pod的资源要求，只出现在预选阶段

1.2.2、代码解析

  1. const (
  2.     FitName           = "NodeResourcesFit"
  3.     preFilterStateKey = "PreFilter" + FitName
  4. )
  6. type Fit struct {
  7.     ignoredResources sets.String
  8. }
  10. type FitArgs struct {
  11.     // 用以忽略一些资源的校验
  12.     IgnoredResources []string `json:"ignoredResources,omitempty"`
  13. }
  15. func NewFit(plArgs *runtime.Unknown, _ framework.FrameworkHandle) (framework.Plugin, error) {
  16.     args := &FitArgs{}
  17.     if err := framework.DecodeInto(plArgs, args); err != nil {
  18.         return nil, err
  19.     }
  20.     fit := &Fit{}
  21.     fit.ignoredResources = sets.NewString(args.IgnoredResources...)
  22.     return fit, nil
  23. }
  25. func (f *Fit) Name() string {
  26.     return FitName
  27. }
  29. // 将pod的资源申请写入CycleState供Filter扩展点使用
  30. func (f *Fit) PreFilter(_ context.Context, cycleState *framework.CycleState, pod *v1.Pod) *framework.Status {
  31.     cycleState.Write(preFilterStateKey, computePodResourceRequest(pod))
  32.     return nil
  33. }
  35. func (f *Fit) PreFilterExtensions() framework.PreFilterExtensions {
  36.     return nil
  37. }
  39. func (f *Fit) Filter(_ context.Context, cycleState *framework.CycleState, _ *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *framework.Status {
  40.     // 获取pod资源数据
  41.     s, err := getPreFilterState(cycleState)
  42.     if err != nil {
  43.         return framework.NewStatus(framework.Error, err.Error())
  44.     }
  46.     // 判断资源是否满足
  47.     insufficientResources := fitsRequest(s, nodeInfo, f.ignoredResources)
  49.     // 返回结果
  50.     if len(insufficientResources) != 0 {
  51.         failureReasons := make([]string, 0, len(insufficientResources))
  52.         for _, r := range insufficientResources {
  53.             failureReasons = append(failureReasons, r.Reason)
  54.         }
  55.         return framework.NewStatus(framework.Unschedulable, failureReasons...)
  56.     }
  57.     return nil
  58. }
  60. type preFilterState struct {
  61.     schedulernodeinfo.Resource
  62. }
  64. func (s *preFilterState) Clone() framework.StateData {
  65.     return s
  66. }
  68. // 何种资源导致filter失败
  69. type InsufficientResource struct {
  70.     ResourceName v1.ResourceName
  71.     Reason       string
  72.     Requested    int64
  73.     Used         int64
  74.     Capacity     int64
  75. }
  77. func Fits(pod *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo, ignoredExtendedResources sets.String) []InsufficientResource {
  78.     return fitsRequest(computePodResourceRequest(pod), nodeInfo, ignoredExtendedResources)
  79. }
  81. func fitsRequest(podRequest *preFilterState, nodeInfo *schedulernodeinfo.NodeInfo, ignoredExtendedResources sets.String) []InsufficientResource {
  83.     // 此处cap设置为4应该是默认一般不包含扩展资源
  84.     // 因此极大概率不会发送切片扩容
  85.     insufficientResources := make([]InsufficientResource, 0, 4)
  87.     allowedPodNumber := nodeInfo.AllowedPodNumber()
  89.     // pod太多
  90.     if len(nodeInfo.Pods())+1 > allowedPodNumber {
  91.         insufficientResources = append(insufficientResources, InsufficientResource{
  92.             v1.ResourcePods,
  93.             "Too many pods",
  94.             1,
  95.             int64(len(nodeInfo.Pods())),
  96.             int64(allowedPodNumber),
  97.         })
  98.     }
  100.     if ignoredExtendedResources == nil {
  101.         ignoredExtendedResources = sets.NewString()
  102.     }
  104.     // pod未申请任何资源
  105.     if podRequest.MilliCPU == 0 && podRequest.Memory == 0 && podRequest.EphemeralStorage == 0 && len(podRequest.ScalarResources) == 0 {
  106.         return insufficientResources
  107.     }
  109.     allocatable := nodeInfo.AllocatableResource()
  111.     // 判断可分配cpu是否满足
  112.     if allocatable.MilliCPU < podRequest.MilliCPU+nodeInfo.RequestedResource().MilliCPU {
  113.         insufficientResources = append(insufficientResources, InsufficientResource{
  114.             v1.ResourceCPU,
  115.             "Insufficient cpu",
  116.             podRequest.MilliCPU,
  117.             nodeInfo.RequestedResource().MilliCPU,
  118.             allocatable.MilliCPU,
  119.         })
  120.     }
  122.     // 判断可分配内存是否满足
  123.     if allocatable.Memory < podRequest.Memory+nodeInfo.RequestedResource().Memory {
  124.         insufficientResources = append(insufficientResources, InsufficientResource{
  125.             v1.ResourceMemory,
  126.             "Insufficient memory",
  127.             podRequest.Memory,
  128.             nodeInfo.RequestedResource().Memory,
  129.             allocatable.Memory,
  130.         })
  131.     }
  133.     // 判断可分配临时存储是否满足
  134.     if allocatable.EphemeralStorage < podRequest.EphemeralStorage+nodeInfo.RequestedResource().EphemeralStorage {
  135.         insufficientResources = append(insufficientResources, InsufficientResource{
  136.             v1.ResourceEphemeralStorage,
  137.             "Insufficient ephemeral-storage",
  138.             podRequest.EphemeralStorage,
  139.             nodeInfo.RequestedResource().EphemeralStorage,
  140.             allocatable.EphemeralStorage,
  141.         })
  142.     }
  144.     // 判断扩展资源，诸如gpu
  145.     for rName, rQuant := range podRequest.ScalarResources {
  146.         if v1helper.IsExtendedResourceName(rName) {
  147.             if ignoredExtendedResources.Has(string(rName)) {
  148.                 continue
  149.             }
  150.         }
  151.         if allocatable.ScalarResources[rName] < rQuant+nodeInfo.RequestedResource().ScalarResources[rName] {
  152.             insufficientResources = append(insufficientResources, InsufficientResource{
  153.                 rName,
  154.                 fmt.Sprintf("Insufficient %v", rName),
  155.                 podRequest.ScalarResources[rName],
  156.                 nodeInfo.RequestedResource().ScalarResources[rName],
  157.                 allocatable.ScalarResources[rName],
  158.             })
  159.         }
  160.     }
  161.     return insufficientResources
  162. }
  164. // 计算pod所需资源
  165. // 首先计算所有容器需要的资源
  166. // 然后根据init容器所需资源计算最大值
  167. // 最后计算额外资源(类似于pause容器等资源的消耗)
  168. func computePodResourceRequest(pod *v1.Pod) *preFilterState {
  169.     result := &preFilterState{}
  170.     for _, container := range pod.Spec.Containers {
  171.         result.Add(container.Resources.Requests)
  172.     }
  174.     for _, container := range pod.Spec.InitContainers {
  175.         result.SetMaxResource(container.Resources.Requests)
  176.     }
  178.     if pod.Spec.Overhead != nil && utilfeature.DefaultFeatureGate.Enabled(features.PodOverhead) {
  179.         result.Add(pod.Spec.Overhead)
  180.     }
  182.     return result
  183. }
  185. // 获取preFilter填充的状态
  186. func getPreFilterState(cycleState *framework.CycleState) (*preFilterState, error) {
  187.     c, err := cycleState.Read(preFilterStateKey)
  188.     if err != nil {
  189.         return nil, fmt.Errorf("error reading %q from cycleState: %v", preFilterStateKey, err)
  190.     }
  192.     s, ok := c.(*preFilterState)
  193.     if !ok {
  194.         return nil, fmt.Errorf("%+v  convert to NodeResourcesFit.preFilterState error", c)
  195.     }
  196.     return s, nil
  197. }

1.3、NodeResourcesLeastAllocated(Score扩展点)

1.3.1、算法功能

该算法主要是用于优选当前资源利用低的node节点且一般和NodeResourcesBalancedAllocation一起使用

1.3.2、代码解析

  1. const LeastAllocatedName = "NodeResourcesLeastAllocated"
  3. type LeastAllocated struct {
  4.     resourceAllocationScorer
  5.     handle framework.FrameworkHandle
  6. }
  8. func NewLeastAllocated(_ *runtime.Unknown, h framework.FrameworkHandle) (framework.Plugin, error) {
  9.     return &LeastAllocated{
  10.         handle: h,
  11.         resourceAllocationScorer: resourceAllocationScorer{
  12.             LeastAllocatedName,
  13.             // 核心函数，用以计算node分值
  14.             leastResourceScorer,
  15.             defaultRequestedRatioResources,
  16.         },
  17.     }, nil
  18. }
  20. func (la *LeastAllocated) Name() string {
  21.     return LeastAllocatedName
  22. }
  24. func (la *LeastAllocated) Score(_ context.Context, _ *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
  25.     nodeInfo, err := la.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
  26.     if err != nil {
  27.         return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v", nodeName, err))
  28.     }
  29.     return la.score(pod, nodeInfo)
  30. }
  32. func (la *LeastAllocated) ScoreExtensions() framework.ScoreExtensions {
  33.     return nil
  34. }
  36. // 计算方式为每种资源得分乘以权重并进行累加
  37. // 然后再根据权重综合进行除法得到最终得分
  38. // 假设:
  39. //     node机器内存资源得分为50，权重为1
  40. //     node机器cpu资源得分为60，权重为1
  41. // 则机器总分为50 * 1 + 60 * 1 = 110
  42. // 最终得分为110 / 2 = 55分
  43. // 相当于根据各个资源的得分进行总体平局得分计算
  44. func leastResourceScorer(requested, allocable resourceToValueMap, _ bool, _ int, _ int) int64 {
  45.     var nodeScore, weightSum int64
  46.     for resource, weight := range defaultRequestedRatioResources {
  47.         resourceScore := leastRequestedScore(requested[resource], allocable[resource])
  48.         nodeScore += resourceScore * weight
  49.         weightSum += weight
  50.     }
  51.     return nodeScore / weightSum
  52. }
  54. // 算法计算方式
  55. // 剩余资源越多，得分越高
  56. // 以内存为例：
  57. // 假设当前可用内存总量8G，POD申请4G，则Node得分50
  58. // 假设当前可用内存总量8G，POD申请2G，则Node得分75分
  59. // 资源利用低的node更容器被选择
  60. func leastRequestedScore(requested, capacity int64) int64 {
  61.     if capacity == 0 {
  62.         return 0
  63.     }
  64.     if requested > capacity {
  65.         return 0
  66.     }
  67.     return ((capacity - requested) * framework.MaxNodeScore) / capacity
  68. }

1.4、NodeResourcesBalancedAllocation(Score扩展点)

1.4.1、算法功能

该算法主要是用于优选当前cpu利用、内存利用率相对均衡的机器。且一般和NodeResourcesLeastAllocated使用
假设：
机器A cpu利用率80%，内存利用率20%
机器B cpu利用率60%，内存利用率50%
则机器B更容器被选择

1.4.2、代码解析

  1. const BalancedAllocationName = "NodeResourcesBalancedAllocation"
  3. type BalancedAllocation struct {
  4.     resourceAllocationScorer
  5.     handle framework.FrameworkHandle
  6. }
  8. func NewBalancedAllocation(_ *runtime.Unknown, h framework.FrameworkHandle) (framework.Plugin, error) {
  9.     return &BalancedAllocation{
  10.         handle: h,
  11.         resourceAllocationScorer: resourceAllocationScorer{
  12.             BalancedAllocationName,
  13.             // 核心函数，用以计算node分值
  14.             balancedResourceScorer,
  15.             defaultRequestedRatioResources,
  16.         },
  17.     }, nil
  18. }
  20. func (ba *BalancedAllocation) Name() string {
  21.     return BalancedAllocationName
  22. }
  24. func (ba *BalancedAllocation) Score(_ context.Context, _ *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
  25.     nodeInfo, err := ba.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
  26.     if err != nil {
  27.         return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v", nodeName, err))
  28.     }
  29.     return ba.score(pod, nodeInfo)
  30. }
  32. func (ba *BalancedAllocation) ScoreExtensions() framework.ScoreExtensions {
  33.     return nil
  34. }
  36. // 核心函数
  37. func balancedResourceScorer(requested, allocable resourceToValueMap, includeVolumes bool, requestedVolumes int, allocatableVolumes int) int64 {
  39.     // 计算cpu/mem各自的申请比例
  40.     cpuFraction := fractionOfCapacity(requested[v1.ResourceCPU], allocable[v1.ResourceCPU])
  41.     memoryFraction := fractionOfCapacity(requested[v1.ResourceMemory], allocable[v1.ResourceMemory])
  42.     // 如果有一个大于或者等于1，则得分为0
  43.     if cpuFraction >= 1 || memoryFraction >= 1 {
  44.         return 0
  45.     }
  47.     // 包含volume的判断
  48.     if includeVolumes && utilfeature.DefaultFeatureGate.Enabled(features.BalanceAttachedNodeVolumes) && allocatableVolumes > 0 {
  49.         // 计算volume比例
  50.         volumeFraction := float64(requestedVolumes) / float64(allocatableVolumes)
  51.         // 大于等于1，则得分为0
  52.         if volumeFraction >= 1 {
  53.             return 0
  54.         }
  55.         // 计算3种资源的变化幅度
  56.         mean := (cpuFraction + memoryFraction + volumeFraction) / float64(3)
  57.         // 本质上是通过方差来获取一个离散程度描述
  58.         // 假设：
  59.         //    机器A：cpu申请比例0.5，mem申请0.6，volume申请0.1，则mean为0.4
  60.         //    机器B：cpu申请比例0.8，mem申请0.2，volume申请0.5，则mean为0.5
  61.         // 则机器A variance = (0.1*0.1 + 0.2*0.2 + 0.3*0.3)/3=0.5
  62.         // 则机器B variance = (0.3*0.3 + 0.3*0.3 + 0*0)/3=0.6
  63.         // 则机器A更容器被选择，因此机器A各个资源偏离mean相对均衡
  64.         // 而机器B有一个极端值0，导致机器B的偏离程度不均衡
  65.         variance := (((cpuFraction - mean) * (cpuFraction - mean)) + ((memoryFraction - mean) * (memoryFraction - mean)) + ((volumeFraction - mean) * (volumeFraction - mean))) / float64(3)
  66.         return int64((1 - variance) * float64(framework.MaxNodeScore))
  67.     }
  69.     // diff越大代表，cpu和mem本次资源申请占用比例越不均衡
  70.     // 假设：
  71.     //    机器A：cpu申请比例0.5，mem申请0.6，则diff为0.2
  72.     //    机器B：cpu申请比例0.8，mem申请0.2，则diff为0.6
  73.     // 则机器A得分：(1 - 0.2) * 100 = 80
  74.     // 则机器B得分：(1 - 0.6) * 100 = 40
  75.     // 显然，资源申请之后，cpu和mem剩余资源越均衡，越容易被选择
  76.     diff := math.Abs(cpuFraction - memoryFraction)
  77.     return int64((1 - diff) * float64(framework.MaxNodeScore))
  78. }
  80. // 计算申请和可用的比例
  81. func fractionOfCapacity(requested, capacity int64) float64 {
  82.     if capacity == 0 {
  83.         return 1
  84.     }
  85.     return float64(requested) / float64(capacity)
  86. }

1.5、NodeResourcesMostAllocated(Score扩展点)

1.5.1、算法功能

该算法主要是用于优选当前资源利用高的node节点，这样做的选择是为了给大资源申请对象预留。比如ai场景。如果资源分布比较平均，可能总可分配资源可以达到要求，但是单台node可能资源不能满足，导致无法调度

1.5.2、代码解析

  1. const MostAllocatedName = "NodeResourcesMostAllocated"
  3. type MostAllocated struct {
  4.     resourceAllocationScorer
  5.     handle framework.FrameworkHandle
  6. }
  8. func NewMostAllocated(_ *runtime.Unknown, h framework.FrameworkHandle) (framework.Plugin, error) {
  9.     return &MostAllocated{
  10.         handle: h,
  11.         resourceAllocationScorer: resourceAllocationScorer{
  12.             MostAllocatedName,
  13.             // 核心函数，用以计算node分值
  14.             mostResourceScorer,
  15.             defaultRequestedRatioResources,
  16.         },
  17.     }, nil
  18. }
  20. func (ma *MostAllocated) Name() string {
  21.     return MostAllocatedName
  22. }
  24. func (ma *MostAllocated) Score(_ context.Context, _ *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
  25.     nodeInfo, err := ma.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
  26.     if err != nil || nodeInfo.Node() == nil {
  27.         return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v, node is nil: %v", nodeName, err, nodeInfo.Node() == nil))
  28.     }
  29.     return ma.score(pod, nodeInfo)
  30. }
  32. func (ma *MostAllocated) ScoreExtensions() framework.ScoreExtensions {
  33.     return nil
  34. }
  36. // 计算方式为每种资源得分乘以权重并进行累加
  37. // 然后再根据权重综合进行除法得到最终得分
  38. // 假设:
  39. //     node机器内存资源得分为50，权重为1
  40. //     node机器cpu资源得分为60，权重为1
  41. // 则机器总分为50 * 1 + 60 * 1 = 110
  42. // 最终得分为110 / 2 = 55分
  43. // 相当于根据各个资源的得分进行总体平局得分计算
  44. func mostResourceScorer(requested, allocable resourceToValueMap, _ bool, _ int, _ int) int64 {
  45.     var nodeScore, weightSum int64
  46.     for resource, weight := range defaultRequestedRatioResources {
  47.         resourceScore := mostRequestedScore(requested[resource], allocable[resource])
  48.         nodeScore += resourceScore * weight
  49.         weightSum += weight
  50.     }
  51.     return nodeScore / weightSum
  52. }
  54. // 算法计算方式
  55. // 剩余资源越少，得分越高
  56. // 以内存为例：
  57. // 假设当前可用内存总量8G，POD申请4G，则Node得分50
  58. // 假设当前可用内存总量8G，POD申请2G，则Node得分25分
  59. // 资源利用低的node更容器被选择
  60. func mostRequestedScore(requested, capacity int64) int64 {
  61.     if capacity == 0 {
  62.         return 0
  63.     }
  64.     if requested > capacity {
  65.         return 0
  66.     }
  67.     return (requested * framework.MaxNodeScore) / capacity
  68. }

1.6、NodeResourceLimits(PreScore/Score扩展点)

1.6.1、算法功能

该算法主要是用于优选pod配置的limit小于node可分配资源的节点

1.6.2、代码解析

  1. const (
  2.     ResourceLimitsName = "NodeResourceLimits"
  3.     preScoreStateKey   = "PreScore" + ResourceLimitsName
  4. )
  6. type ResourceLimits struct {
  7.     handle framework.FrameworkHandle
  8. }
  10. func NewResourceLimits(_ *runtime.Unknown, h framework.FrameworkHandle) (framework.Plugin, error) {
  11.     return &ResourceLimits{handle: h}, nil
  12. }
  14. func (rl *ResourceLimits) Name() string {
  15.     return ResourceLimitsName
  16. }
  18. func (rl *ResourceLimits) PreScore(
  19.     _ context.Context,
  20.     cycleState *framework.CycleState,
  21.     pod *v1.Pod,
  22.     nodes []*v1.Node,
  23. ) *framework.Status {
  24.     if len(nodes) == 0 {
  25.         return nil
  26.     }
  28.     if rl.handle.SnapshotSharedLister() == nil {
  29.         return framework.NewStatus(framework.Error, fmt.Sprintf("empty shared lister"))
  30.     }
  31.     s := &preScoreState{
  32.         podResourceRequest: getResourceLimits(pod),
  33.     }
  34.     cycleState.Write(preScoreStateKey, s)
  35.     return nil
  36. }
  38. func (rl *ResourceLimits) Score(_ context.Context, state *framework.CycleState, _ *v1.Pod, nodeName string) (int64, *framework.Status) {
  39.     nodeInfo, err := rl.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
  40.     if err != nil || nodeInfo.Node() == nil {
  41.         return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v, node is nil: %v", nodeName, err, nodeInfo.Node() == nil))
  42.     }
  43.     allocatableResources := nodeInfo.AllocatableResource()
  44.     podLimits, err := getPodResource(state)
  45.     if err != nil {
  46.         return 0, framework.NewStatus(framework.Error, err.Error())
  47.     }
  49.     cpuScore := computeScore(podLimits.MilliCPU, allocatableResources.MilliCPU)
  50.     memScore := computeScore(podLimits.Memory, allocatableResources.Memory)
  52.     // 只要cpu或者mem满足，就得一分
  53.     score := int64(0)
  54.     if cpuScore == 1 || memScore == 1 {
  55.         score = 1
  56.     }
  57.     return score, nil
  58. }
  60. func (rl *ResourceLimits) ScoreExtensions() framework.ScoreExtensions {
  61.     return nil
  62. }
  64. type preScoreState struct {
  65.     podResourceRequest *schedulernodeinfo.Resource
  66. }
  68. func (s *preScoreState) Clone() framework.StateData {
  69.     return s
  70. }
  72. func getPodResource(cycleState *framework.CycleState) (*schedulernodeinfo.Resource, error) {
  73.     c, err := cycleState.Read(preScoreStateKey)
  74.     if err != nil {
  75.         return nil, fmt.Errorf("Error reading %q from cycleState: %v", preScoreStateKey, err)
  76.     }
  78.     s, ok := c.(*preScoreState)
  79.     if !ok {
  80.         return nil, fmt.Errorf("%+v  convert to ResourceLimits.preScoreState error", c)
  81.     }
  82.     return s.podResourceRequest, nil
  83. }
  85. // 获取Containers的limit
  86. // 并根据InitContainers计算最大值
  87. func getResourceLimits(pod *v1.Pod) *schedulernodeinfo.Resource {
  88.     result := &schedulernodeinfo.Resource{}
  89.     for _, container := range pod.Spec.Containers {
  90.         result.Add(container.Resources.Limits)
  91.     }
  93.     for _, container := range pod.Spec.InitContainers {
  94.         result.SetMaxResource(container.Resources.Limits)
  95.     }
  96.     return result
  97. }
  99. // 若limit配置小于可分配资源，则得一分
  100. func computeScore(limit, allocatable int64) int64 {
  101.     if limit != 0 && allocatable != 0 && limit <= allocatable {
  102.         return 1
  103.     }
  104.     return 0
  105. }

1.7、RequestedToCapacityRatio(Score扩展点)

1.7.1、算法功能

1.7.2、代码解析

2、nodeaffinity(关于node亲和性相关的调度算法)

2.1、NodeAffinity(Filter/Score扩展点)

2.1.1、算法功能

该算法主要描述node亲和性，主要用来解决pod绑定到给定条件的node上面，实现pod和node的亲和性调度。比如：pod1需要调度到有ssd盘的node上面，pod2需要调度到cpu配置较好的node上面等

2.1.2、代码解析

  1. const (
  2.     Name      = "NodeAffinity"
  3.     ErrReason = "node(s) didn't match node selector"
  4. )
  6. type NodeAffinity struct {
  7.     handle framework.FrameworkHandle
  8. }
  10. func New(_ *runtime.Unknown, h framework.FrameworkHandle) (framework.Plugin, error) {
  11.     return &NodeAffinity{handle: h}, nil
  12. }
  14. func (pl *NodeAffinity) Name() string {
  15.     return Name
  16. }
  18. // 预选插件
  19. func (pl *NodeAffinity) Filter(_ context.Context, _ *framework.CycleState, pod *v1.Pod, nodeInfo *nodeinfo.NodeInfo) *framework.Status {
  20.     node := nodeInfo.Node()
  21.     if node == nil {
  22.         return framework.NewStatus(framework.Error, "node not found")
  23.     }
  24.     if !pluginhelper.PodMatchesNodeSelectorAndAffinityTerms(pod, node) {
  25.         return framework.NewStatus(framework.UnschedulableAndUnresolvable, ErrReason)
  26.     }
  27.     return nil
  28. }
  30. // 优选插件
  31. func (pl *NodeAffinity) Score(_ context.Context, _ *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
  32.     nodeInfo, err := pl.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
  33.     if err != nil {
  34.         return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v", nodeName, err))
  35.     }
  37.     node := nodeInfo.Node()
  38.     if node == nil {
  39.         return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v", nodeName, err))
  40.     }
  42.     affinity := pod.Spec.Affinity
  44.     var count int64
  45.     if affinity != nil && affinity.NodeAffinity != nil && affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution != nil {
  46.         // 依次遍历所有的优选配置
  47.         // 若符合则累加相应的权重
  48.         // 最后返回得分
  49.         for i := range affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution {
  50.             preferredSchedulingTerm := &affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution[i]
  51.             // 权重为0的不处理
  52.             if preferredSchedulingTerm.Weight == 0 {
  53.                 continue
  54.             }
  55.             nodeSelector, err := v1helper.NodeSelectorRequirementsAsSelector(preferredSchedulingTerm.Preference.MatchExpressions)
  56.             if err != nil {
  57.                 return 0, framework.NewStatus(framework.Error, err.Error())
  58.             }
  59.             if nodeSelector.Matches(labels.Set(node.Labels)) {
  60.                 count += int64(preferredSchedulingTerm.Weight)
  61.             }
  62.         }
  63.     }
  64.     return count, nil
  65. }
  67. func (pl *NodeAffinity) NormalizeScore(_ context.Context, _ *framework.CycleState, _ *v1.Pod, scores framework.NodeScoreList) *framework.Status {
  68.     return pluginhelper.DefaultNormalizeScore(framework.MaxNodeScore, false, scores)
  69. }
  71. func (pl *NodeAffinity) ScoreExtensions() framework.ScoreExtensions {
  72.     return pl
  73. }
  75. // 预选阶段判断node亲和性
  76. // 1、node的label必须与pod的NodeSelector一致
  77. // 2、在1的基础上再判断node亲和性
  78. func PodMatchesNodeSelectorAndAffinityTerms(pod *v1.Pod, node *v1.Node) bool {
  80.     // 判断label
  81.     if len(pod.Spec.NodeSelector) > 0 {
  82.         selector := labels.SelectorFromSet(pod.Spec.NodeSelector)
  83.         if !selector.Matches(labels.Set(node.Labels)) {
  84.             return false
  85.         }
  86.     }
  88.     // 只要有一个NodeSelectorTerm符合条件，就return true
  89.     // 且MatchFields目前只支持node的metadata.name
  90.     nodeAffinityMatches := true
  91.     affinity := pod.Spec.Affinity
  92.     if affinity != nil && affinity.NodeAffinity != nil {
  93.         nodeAffinity := affinity.NodeAffinity
  94.         // 若没有设置RequiredDuringSchedulingIgnoredDuringExecution，直接返回
  95.         if nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution == nil {
  96.             return true
  97.         }
  98.         // 查看node的label是否符合pod配置的亲和性
  99.         nodeSelectorTerms := nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms
  100.         nodeAffinityMatches = nodeAffinityMatches && nodeMatchesNodeSelectorTerms(node, nodeSelectorTerms)
  101.     }
  102.     return nodeAffinityMatches
  103. }
  105. func nodeMatchesNodeSelectorTerms(node *v1.Node, nodeSelectorTerms []v1.NodeSelectorTerm) bool {
  106.     return v1helper.MatchNodeSelectorTerms(nodeSelectorTerms, node.Labels, fields.Set{
  107.         "metadata.name": node.Name,
  108.     })
  109. }
  111. // DefaultNormalizeScore generates a Normalize Score function that can normalize the
  112. // scores to [0, maxPriority]. If reverse is set to true, it reverses the scores by
  113. // subtracting it from maxPriority.
  114. func DefaultNormalizeScore(maxPriority int64, reverse bool, scores framework.NodeScoreList) *framework.Status {
  115.     var maxCount int64
  116.     for i := range scores {
  117.         if scores[i].Score > maxCount {
  118.             maxCount = scores[i].Score
  119.         }
  120.     }
  122.     if maxCount == 0 {
  123.         if reverse {
  124.             for i := range scores {
  125.                 scores[i].Score = maxPriority
  126.             }
  127.         }
  128.         return nil
  129.     }
  131.     for i := range scores {
  132.         score := scores[i].Score
  134.         score = maxPriority * score / maxCount
  135.         if reverse {
  136.             score = maxPriority - score
  137.         }
  139.         scores[i].Score = score
  140.     }
  141.     return nil
  142. }