• kube-scheduler源码分析（五）之 PrioritizeNodes
  • 1. 调用入口
  • 2. PrioritizeNodes
  • 3. EqualPriorityMap
  • 4. processNode
  • 5. PriorityMapFunction
  • 6. PriorityReduceFunction
  • 7. Summarize all scores
  • 8. NewSelectorSpreadPriority
    • 8.1. CalculateSpreadPriorityMap
    • 8.2. CalculateSpreadPriorityReduce
  • 9. 总结
    • 9.1. PrioritizeNodes
    • 9.2. NewSelectorSpreadPriority
      • 9.2.1. CalculateSpreadPriorityMap
      • 9.2.2. CalculateSpreadPriorityReduce

    kube-scheduler源码分析（五）之 PrioritizeNodes

    以下代码分析基于 kubernetes v1.12.0 版本。

    本文主要分析优选策略逻辑，即从预选的节点中选择出最优的节点。优选策略的具体实现函数为PrioritizeNodes。PrioritizeNodes最终返回是一个记录了各个节点分数的列表。

    1. 调用入口

    genericScheduler.Schedule中对PrioritizeNodes的调用过程如下：

    此部分代码位于pkg/scheduler/core/generic_scheduler.go

    1. func (g *genericScheduler) Schedule(pod *v1.Pod, nodeLister algorithm.NodeLister) (string, error) {
    2.   ...
    3.     trace.Step("Prioritizing")
    4.     startPriorityEvalTime := time.Now()
    5.     // When only one node after predicate, just use it.
    6.     if len(filteredNodes) == 1 {
    7.         metrics.SchedulingAlgorithmPriorityEvaluationDuration.Observe(metrics.SinceInMicroseconds(startPriorityEvalTime))
    8.         return filteredNodes[0].Name, nil
    9.     }
    11.     metaPrioritiesInterface := g.priorityMetaProducer(pod, g.cachedNodeInfoMap)
    12.   // 执行优选逻辑的操作，返回记录各个节点分数的列表
    13.     priorityList, err := PrioritizeNodes(pod, g.cachedNodeInfoMap, metaPrioritiesInterface, g.prioritizers, filteredNodes, g.extenders)
    14.     if err != nil {
    15.         return "", err
    16.     }
    17.     metrics.SchedulingAlgorithmPriorityEvaluationDuration.Observe(metrics.SinceInMicroseconds(startPriorityEvalTime))
    18.     metrics.SchedulingLatency.WithLabelValues(metrics.PriorityEvaluation).Observe(metrics.SinceInSeconds(startPriorityEvalTime))
    19.   ...
    20. }

    核心代码：

    1. // 基于预选节点filteredNodes进一步筛选优选的节点，返回记录各个节点分数的列表
    2. priorityList, err := PrioritizeNodes(pod, g.cachedNodeInfoMap, metaPrioritiesInterface, g.prioritizers, filteredNodes, g.extenders)

    2. PrioritizeNodes

    优选，从满足的节点中选择出最优的节点。PrioritizeNodes最终返回是一个记录了各个节点分数的列表。

    具体操作如下：

    • PrioritizeNodes通过并行运行各个优先级函数来对节点进行优先级排序。
    • 每个优先级函数会给节点打分，打分范围为0-10分。
    • 0 表示优先级最低的节点，10表示优先级最高的节点。
    • 每个优先级函数也有各自的权重。
    • 优先级函数返回的节点分数乘以权重以获得加权分数。
    • 最后组合（添加）所有分数以获得所有节点的总加权分数。

    PrioritizeNodes主要流程如下：

    1. 如果没有设置优选函数和拓展函数，则全部节点设置相同的分数，直接返回。
    2. 依次给node执行map函数进行打分。
    3. 再对上述map函数的执行结果执行reduce函数计算最终得分。
    4. 最后根据不同优先级函数的权重对得分取加权平均数。

    入参：

    • pod
    • nodeNameToInfo
    • meta interface{},
    • priorityConfigs
    • nodes
    • extenders

    出参：

    • HostPriorityList：记录节点分数的列表。

    HostPriority定义如下:

    1. // HostPriority represents the priority of scheduling to a particular host, higher priority is better.
    2. type HostPriority struct {
    3.     // Name of the host
    4.     Host string
    5.     // Score associated with the host
    6.     Score int
    7. }

    PrioritizeNodes完整代码如下：

    此部分代码位于pkg/scheduler/core/generic_scheduler.go

    1. // PrioritizeNodes prioritizes the nodes by running the individual priority functions in parallel.
    2. // Each priority function is expected to set a score of 0-10
    3. // 0 is the lowest priority score (least preferred node) and 10 is the highest
    4. // Each priority function can also have its own weight
    5. // The node scores returned by the priority function are multiplied by the weights to get weighted scores
    6. // All scores are finally combined (added) to get the total weighted scores of all nodes
    7. func PrioritizeNodes(
    8.     pod *v1.Pod,
    9.     nodeNameToInfo map[string]*schedulercache.NodeInfo,
    10.     meta interface{},
    11.     priorityConfigs []algorithm.PriorityConfig,
    12.     nodes []*v1.Node,
    13.     extenders []algorithm.SchedulerExtender,
    14. ) (schedulerapi.HostPriorityList, error) {
    15.     // If no priority configs are provided, then the EqualPriority function is applied
    16.     // This is required to generate the priority list in the required format
    17.     if len(priorityConfigs) == 0 && len(extenders) == 0 {
    18.         result := make(schedulerapi.HostPriorityList, 0, len(nodes))
    19.         for i := range nodes {
    20.             hostPriority, err := EqualPriorityMap(pod, meta, nodeNameToInfo[nodes[i].Name])
    21.             if err != nil {
    22.                 return nil, err
    23.             }
    24.             result = append(result, hostPriority)
    25.         }
    26.         return result, nil
    27.     }
    29.     var (
    30.         mu   = sync.Mutex{}
    31.         wg   = sync.WaitGroup{}
    32.         errs []error
    33.     )
    34.     appendError := func(err error) {
    35.         mu.Lock()
    36.         defer mu.Unlock()
    37.         errs = append(errs, err)
    38.     }
    40.     results := make([]schedulerapi.HostPriorityList, len(priorityConfigs), len(priorityConfigs))
    42.     for i, priorityConfig := range priorityConfigs {
    43.         if priorityConfig.Function != nil {
    44.             // DEPRECATED
    45.             wg.Add(1)
    46.             go func(index int, config algorithm.PriorityConfig) {
    47.                 defer wg.Done()
    48.                 var err error
    49.                 results[index], err = config.Function(pod, nodeNameToInfo, nodes)
    50.                 if err != nil {
    51.                     appendError(err)
    52.                 }
    53.             }(i, priorityConfig)
    54.         } else {
    55.             results[i] = make(schedulerapi.HostPriorityList, len(nodes))
    56.         }
    57.     }
    58.     processNode := func(index int) {
    59.         nodeInfo := nodeNameToInfo[nodes[index].Name]
    60.         var err error
    61.         for i := range priorityConfigs {
    62.             if priorityConfigs[i].Function != nil {
    63.                 continue
    64.             }
    65.             results[i][index], err = priorityConfigs[i].Map(pod, meta, nodeInfo)
    66.             if err != nil {
    67.                 appendError(err)
    68.                 return
    69.             }
    70.         }
    71.     }
    72.     workqueue.Parallelize(16, len(nodes), processNode)
    73.     for i, priorityConfig := range priorityConfigs {
    74.         if priorityConfig.Reduce == nil {
    75.             continue
    76.         }
    77.         wg.Add(1)
    78.         go func(index int, config algorithm.PriorityConfig) {
    79.             defer wg.Done()
    80.             if err := config.Reduce(pod, meta, nodeNameToInfo, results[index]); err != nil {
    81.                 appendError(err)
    82.             }
    83.             if glog.V(10) {
    84.                 for _, hostPriority := range results[index] {
    85.                     glog.Infof("%v -> %v: %v, Score: (%d)", pod.Name, hostPriority.Host, config.Name, hostPriority.Score)
    86.                 }
    87.             }
    88.         }(i, priorityConfig)
    89.     }
    90.     // Wait for all computations to be finished.
    91.     wg.Wait()
    92.     if len(errs) != 0 {
    93.         return schedulerapi.HostPriorityList{}, errors.NewAggregate(errs)
    94.     }
    96.     // Summarize all scores.
    97.     result := make(schedulerapi.HostPriorityList, 0, len(nodes))
    99.     for i := range nodes {
    100.         result = append(result, schedulerapi.HostPriority{Host: nodes[i].Name, Score: 0})
    101.         for j := range priorityConfigs {
    102.             result[i].Score += results[j][i].Score * priorityConfigs[j].Weight
    103.         }
    104.     }
    106.     if len(extenders) != 0 && nodes != nil {
    107.         combinedScores := make(map[string]int, len(nodeNameToInfo))
    108.         for _, extender := range extenders {
    109.             if !extender.IsInterested(pod) {
    110.                 continue
    111.             }
    112.             wg.Add(1)
    113.             go func(ext algorithm.SchedulerExtender) {
    114.                 defer wg.Done()
    115.                 prioritizedList, weight, err := ext.Prioritize(pod, nodes)
    116.                 if err != nil {
    117.                     // Prioritization errors from extender can be ignored, let k8s/other extenders determine the priorities
    118.                     return
    119.                 }
    120.                 mu.Lock()
    121.                 for i := range *prioritizedList {
    122.                     host, score := (*prioritizedList)[i].Host, (*prioritizedList)[i].Score
    123.                     combinedScores[host] += score * weight
    124.                 }
    125.                 mu.Unlock()
    126.             }(extender)
    127.         }
    128.         // wait for all go routines to finish
    129.         wg.Wait()
    130.         for i := range result {
    131.             result[i].Score += combinedScores[result[i].Host]
    132.         }
    133.     }
    135.     if glog.V(10) {
    136.         for i := range result {
    137.             glog.V(10).Infof("Host %s => Score %d", result[i].Host, result[i].Score)
    138.         }
    139.     }
    140.     return result, nil
    141. }

    以下对PrioritizeNodes分段进行分析。

    3. EqualPriorityMap

    如果没有提供优选函数和拓展函数，则将所有的节点设置为相同的优先级，即节点的score都为1，然后直接返回结果。(但一般情况下优选函数列表都不为空)

    1. // If no priority configs are provided, then the EqualPriority function is applied
    2. // This is required to generate the priority list in the required format
    3. if len(priorityConfigs) == 0 && len(extenders) == 0 {
    4.     result := make(schedulerapi.HostPriorityList, 0, len(nodes))
    5.     for i := range nodes {
    6.         hostPriority, err := EqualPriorityMap(pod, meta, nodeNameToInfo[nodes[i].Name])
    7.         if err != nil {
    8.             return nil, err
    9.         }
    10.         result = append(result, hostPriority)
    11.     }
    12.     return result, nil
    13. }

    EqualPriorityMap具体实现如下：

    1. // EqualPriorityMap is a prioritizer function that gives an equal weight of one to all nodes
    2. func EqualPriorityMap(_ *v1.Pod, _ interface{}, nodeInfo *schedulercache.NodeInfo) (schedulerapi.HostPriority, error) {
    3.     node := nodeInfo.Node()
    4.     if node == nil {
    5.         return schedulerapi.HostPriority{}, fmt.Errorf("node not found")
    6.     }
    7.     return schedulerapi.HostPriority{
    8.         Host:  node.Name,
    9.         Score: 1,
    10.     }, nil
    11. }

    4. processNode

    processNode就是基于index拿出node的信息，调用之前注册的各种优选函数（此处是mapFunction），通过优选函数对node和pod进行处理，最后返回一个记录node分数的列表result。processNode同样也使用workqueue.Parallelize来进行并行处理。(processNode类似于预选逻辑findNodesThatFit中使用到的checkNode的作用)

    其中优选函数是通过priorityConfigs来记录，每类优选函数包括PriorityMapFunction和PriorityReduceFunction两种函数。优选函数的注册部分可参考registerAlgorithmProvider。

    1. processNode := func(index int) {
    2.     nodeInfo := nodeNameToInfo[nodes[index].Name]
    3.     var err error
    4.     for i := range priorityConfigs {
    5.         if priorityConfigs[i].Function != nil {
    6.             continue
    7.         }
    8.         results[i][index], err = priorityConfigs[i].Map(pod, meta, nodeInfo)
    9.         if err != nil {
    10.             appendError(err)
    11.             return
    12.         }
    13.     }
    14. }
    15. // 并行执行processNode
    16. workqueue.Parallelize(16, len(nodes), processNode)

    priorityConfigs定义如下：

    核心属性：

    • Map ：PriorityMapFunction
    • Reduce：PriorityReduceFunction
    1. // PriorityConfig is a config used for a priority function.
    2. type PriorityConfig struct {
    3.     Name   string
    4.     Map    PriorityMapFunction   
    5.     Reduce PriorityReduceFunction
    6.     // TODO: Remove it after migrating all functions to
    7.     // Map-Reduce pattern.
    8.     Function PriorityFunction
    9.     Weight   int
    10. }

    具体的优选函数处理逻辑待下文分析，本文会以NewSelectorSpreadPriority函数为例。

    5. PriorityMapFunction

    PriorityMapFunction是一个计算给定节点的每个节点结果的函数。

    PriorityMapFunction定义如下：

    1. // PriorityMapFunction is a function that computes per-node results for a given node.
    2. // TODO: Figure out the exact API of this method.
    3. // TODO: Change interface{} to a specific type.
    4. type PriorityMapFunction func(pod *v1.Pod, meta interface{}, nodeInfo *schedulercache.NodeInfo) (schedulerapi.HostPriority, error)

    PriorityMapFunction是在processNode中调用的，代码如下：

    1. results[i][index], err = priorityConfigs[i].Map(pod, meta, nodeInfo)

    下文会分析NewSelectorSpreadPriority在的map函数CalculateSpreadPriorityMap。

    6. PriorityReduceFunction

    PriorityReduceFunction是一个聚合每个节点结果并计算所有节点的最终得分的函数。

    PriorityReduceFunction定义如下：

    1. // PriorityReduceFunction is a function that aggregated per-node results and computes
    2. // final scores for all nodes.
    3. // TODO: Figure out the exact API of this method.
    4. // TODO: Change interface{} to a specific type.
    5. type PriorityReduceFunction func(pod *v1.Pod, meta interface{}, nodeNameToInfo map[string]*schedulercache.NodeInfo, result schedulerapi.HostPriorityList) error

    PrioritizeNodes中对reduce函数调用部分如下：

    1. for i, priorityConfig := range priorityConfigs {
    2.     if priorityConfig.Reduce == nil {
    3.         continue
    4.     }
    5.     wg.Add(1)
    6.     go func(index int, config algorithm.PriorityConfig) {
    7.         defer wg.Done()
    8.         if err := config.Reduce(pod, meta, nodeNameToInfo, results[index]); err != nil {
    9.             appendError(err)
    10.         }
    11.         if glog.V(10) {
    12.             for _, hostPriority := range results[index] {
    13.                 glog.Infof("%v -> %v: %v, Score: (%d)", pod.Name, hostPriority.Host, config.Name, hostPriority.Score)
    14.             }
    15.         }
    16.     }(i, priorityConfig)
    17. }

    下文会分析NewSelectorSpreadPriority在的reduce函数CalculateSpreadPriorityReduce。

    7. Summarize all scores

    先等待计算完成再计算加权平均数。

    1. // Wait for all computations to be finished.
    2. wg.Wait()
    3. if len(errs) != 0 {
    4.     return schedulerapi.HostPriorityList{}, errors.NewAggregate(errs)
    5. }

    计算所有节点的加权平均数。

    1. // Summarize all scores.
    2. result := make(schedulerapi.HostPriorityList, 0, len(nodes))
    4. for i := range nodes {
    5.     result = append(result, schedulerapi.HostPriority{Host: nodes[i].Name, Score: 0})
    6.     for j := range priorityConfigs {
    7.         result[i].Score += results[j][i].Score * priorityConfigs[j].Weight
    8.     }
    9. }

    当设置了拓展的计算方式，则增加拓展计算方式的加权平均数。

    1. if len(extenders) != 0 && nodes != nil {
    2.     combinedScores := make(map[string]int, len(nodeNameToInfo))
    3.     for _, extender := range extenders {
    4.         if !extender.IsInterested(pod) {
    5.             continue
    6.         }
    7.         wg.Add(1)
    8.         go func(ext algorithm.SchedulerExtender) {
    9.             defer wg.Done()
    10.             prioritizedList, weight, err := ext.Prioritize(pod, nodes)
    11.             if err != nil {
    12.                 // Prioritization errors from extender can be ignored, let k8s/other extenders determine the priorities
    13.                 return
    14.             }
    15.             mu.Lock()
    16.             for i := range *prioritizedList {
    17.                 host, score := (*prioritizedList)[i].Host, (*prioritizedList)[i].Score
    18.                 combinedScores[host] += score * weight
    19.             }
    20.             mu.Unlock()
    21.         }(extender)
    22.     }
    23.     // wait for all go routines to finish
    24.     wg.Wait()
    25.     for i := range result {
    26.         result[i].Score += combinedScores[result[i].Host]
    27.     }
    28. }

    8. NewSelectorSpreadPriority

    以下以NewSelectorSpreadPriority这个优选函数来做分析，其他重要的优选函数待后续专门分析。

    NewSelectorSpreadPriority主要的功能是将属于相同service和rs下的pod尽量分布在不同的node上。

    该函数的注册代码如下：

    此部分代码位于pkg/scheduler/algorithmprovider/defaults/defaults.go

    1. // ServiceSpreadingPriority is a priority config factory that spreads pods by minimizing
    2. // the number of pods (belonging to the same service) on the same node.
    3. // Register the factory so that it's available, but do not include it as part of the default priorities
    4. // Largely replaced by "SelectorSpreadPriority", but registered for backward compatibility with 1.0
    5. factory.RegisterPriorityConfigFactory(
    6.     "ServiceSpreadingPriority",
    7.     factory.PriorityConfigFactory{
    8.         MapReduceFunction: func(args factory.PluginFactoryArgs) (algorithm.PriorityMapFunction, algorithm.PriorityReduceFunction) {
    9.             return priorities.NewSelectorSpreadPriority(args.ServiceLister, algorithm.EmptyControllerLister{}, algorithm.EmptyReplicaSetLister{}, algorithm.EmptyStatefulSetLister{})
    10.         },
    11.         Weight: 1,
    12.     },
    13. )

    NewSelectorSpreadPriority的具体实现如下：

    此部分代码位于pkg/scheduler/algorithm/priorities/selector_spreading.go

    1. // NewSelectorSpreadPriority creates a SelectorSpread.
    2. func NewSelectorSpreadPriority(
    3.     serviceLister algorithm.ServiceLister,
    4.     controllerLister algorithm.ControllerLister,
    5.     replicaSetLister algorithm.ReplicaSetLister,
    6.     statefulSetLister algorithm.StatefulSetLister) (algorithm.PriorityMapFunction, algorithm.PriorityReduceFunction) {
    7.     selectorSpread := &SelectorSpread{
    8.         serviceLister:     serviceLister,
    9.         controllerLister:  controllerLister,
    10.         replicaSetLister:  replicaSetLister,
    11.         statefulSetLister: statefulSetLister,
    12.     }
    13.     return selectorSpread.CalculateSpreadPriorityMap, selectorSpread.CalculateSpreadPriorityReduce
    14. }

    NewSelectorSpreadPriority主要包括map和reduce两种函数，分别对应CalculateSpreadPriorityMap，CalculateSpreadPriorityReduce。

    8.1. CalculateSpreadPriorityMap

    CalculateSpreadPriorityMap的主要作用是将相同service、RC、RS或statefulset的pod分布在不同的节点上。当调度一个pod的时候，先寻找与该pod匹配的service、RS、RC或statefulset，然后寻找与其selector匹配的已存在的pod，寻找存在这类pod最少的节点。

    基本流程如下：

    1. 寻找与该pod对应的service、RS、RC、statefulset匹配的selector。
    2. 遍历当前节点的所有pod，将该节点上已存在的selector匹配到的pod的个数作为该节点的分数（此时，分数大的表示匹配到的pod越多，越不符合被调度的条件，该分数在reduce阶段会被按10分制处理成分数大的越符合被调度的条件）。

    此部分代码位于pkg/scheduler/algorithm/priorities/selector_spreading.go

    1. // CalculateSpreadPriorityMap spreads pods across hosts, considering pods
    2. // belonging to the same service,RC,RS or StatefulSet.
    3. // When a pod is scheduled, it looks for services, RCs,RSs and StatefulSets that match the pod,
    4. // then finds existing pods that match those selectors.
    5. // It favors nodes that have fewer existing matching pods.
    6. // i.e. it pushes the scheduler towards a node where there's the smallest number of
    7. // pods which match the same service, RC,RSs or StatefulSets selectors as the pod being scheduled.
    8. func (s *SelectorSpread) CalculateSpreadPriorityMap(pod *v1.Pod, meta interface{}, nodeInfo *schedulercache.NodeInfo) (schedulerapi.HostPriority, error) {
    9.     var selectors []labels.Selector
    10.     node := nodeInfo.Node()
    11.     if node == nil {
    12.         return schedulerapi.HostPriority{}, fmt.Errorf("node not found")
    13.     }
    15.     priorityMeta, ok := meta.(*priorityMetadata)
    16.     if ok {
    17.         selectors = priorityMeta.podSelectors
    18.     } else {
    19.         selectors = getSelectors(pod, s.serviceLister, s.controllerLister, s.replicaSetLister, s.statefulSetLister)
    20.     }
    22.     if len(selectors) == 0 {
    23.         return schedulerapi.HostPriority{
    24.             Host:  node.Name,
    25.             Score: int(0),
    26.         }, nil
    27.     }
    29.     count := int(0)
    30.     for _, nodePod := range nodeInfo.Pods() {
    31.         if pod.Namespace != nodePod.Namespace {
    32.             continue
    33.         }
    34.         // When we are replacing a failed pod, we often see the previous
    35.         // deleted version while scheduling the replacement.
    36.         // Ignore the previous deleted version for spreading purposes
    37.         // (it can still be considered for resource restrictions etc.)
    38.         if nodePod.DeletionTimestamp != nil {
    39.             glog.V(4).Infof("skipping pending-deleted pod: %s/%s", nodePod.Namespace, nodePod.Name)
    40.             continue
    41.         }
    42.         for _, selector := range selectors {
    43.             if selector.Matches(labels.Set(nodePod.ObjectMeta.Labels)) {
    44.                 count++
    45.                 break
    46.             }
    47.         }
    48.     }
    49.     return schedulerapi.HostPriority{
    50.         Host:  node.Name,
    51.         Score: int(count),
    52.     }, nil
    53. }

    以下分段分析：

    先获得selector。

    1. selectors = getSelectors(pod, s.serviceLister, s.controllerLister, s.replicaSetLister, s.statefulSetLister)

    计算节点上匹配selector的pod的个数，作为该节点分数，该分数并不是最终节点的分数，只是中间过渡的记录状态。

    1. count := int(0)
    2. for _, nodePod := range nodeInfo.Pods() {
    3.     ...
    4.     for _, selector := range selectors {
    5.         if selector.Matches(labels.Set(nodePod.ObjectMeta.Labels)) {
    6.             count++
    7.             break
    8.         }
    9.     }
    10. }

    8.2. CalculateSpreadPriorityReduce

    CalculateSpreadPriorityReduce根据节点上现有匹配pod的数量计算每个节点的十分制的分数，具有较少现有匹配pod的节点的分数越高，表示节点越可能被调度到。

    基本流程如下：

    1. 记录所有节点中匹配到pod个数最多的节点的分数（即匹配到的pod最多的个数）。
    2. 遍历所有的节点，按比例取十分制的得分，计算方式为：(节点中最多匹配pod的个数-当前节点pod的个数)/节点中最多匹配pod的个数。此时，分数越高表示该节点上匹配到的pod的个数越少，越可能被调度到，即满足把相同selector的pod分散到不同节点的需求。

    此部分代码位于pkg/scheduler/algorithm/priorities/selector_spreading.go

    1. // CalculateSpreadPriorityReduce calculates the source of each node
    2. // based on the number of existing matching pods on the node
    3. // where zone information is included on the nodes, it favors nodes
    4. // in zones with fewer existing matching pods.
    5. func (s *SelectorSpread) CalculateSpreadPriorityReduce(pod *v1.Pod, meta interface{}, nodeNameToInfo map[string]*schedulercache.NodeInfo, result schedulerapi.HostPriorityList) error {
    6.     countsByZone := make(map[string]int, 10)
    7.     maxCountByZone := int(0)
    8.     maxCountByNodeName := int(0)
    10.     for i := range result {
    11.         if result[i].Score > maxCountByNodeName {
    12.             maxCountByNodeName = result[i].Score
    13.         }
    14.         zoneID := utilnode.GetZoneKey(nodeNameToInfo[result[i].Host].Node())
    15.         if zoneID == "" {
    16.             continue
    17.         }
    18.         countsByZone[zoneID] += result[i].Score
    19.     }
    21.     for zoneID := range countsByZone {
    22.         if countsByZone[zoneID] > maxCountByZone {
    23.             maxCountByZone = countsByZone[zoneID]
    24.         }
    25.     }
    27.     haveZones := len(countsByZone) != 0
    29.     maxCountByNodeNameFloat64 := float64(maxCountByNodeName)
    30.     maxCountByZoneFloat64 := float64(maxCountByZone)
    31.     MaxPriorityFloat64 := float64(schedulerapi.MaxPriority)
    33.     for i := range result {
    34.         // initializing to the default/max node score of maxPriority
    35.         fScore := MaxPriorityFloat64
    36.         if maxCountByNodeName > 0 {
    37.             fScore = MaxPriorityFloat64 * (float64(maxCountByNodeName-result[i].Score) / maxCountByNodeNameFloat64)
    38.         }
    39.         // If there is zone information present, incorporate it
    40.         if haveZones {
    41.             zoneID := utilnode.GetZoneKey(nodeNameToInfo[result[i].Host].Node())
    42.             if zoneID != "" {
    43.                 zoneScore := MaxPriorityFloat64
    44.                 if maxCountByZone > 0 {
    45.                     zoneScore = MaxPriorityFloat64 * (float64(maxCountByZone-countsByZone[zoneID]) / maxCountByZoneFloat64)
    46.                 }
    47.                 fScore = (fScore * (1.0 - zoneWeighting)) + (zoneWeighting * zoneScore)
    48.             }
    49.         }
    50.         result[i].Score = int(fScore)
    51.         if glog.V(10) {
    52.             glog.Infof(
    53.                 "%v -> %v: SelectorSpreadPriority, Score: (%d)", pod.Name, result[i].Host, int(fScore),
    54.             )
    55.         }
    56.     }
    57.     return nil
    58. }

    以下分段分析：

    先获取所有节点中匹配到的pod最多的个数。

    1. for i := range result {
    2.     if result[i].Score > maxCountByNodeName {
    3.         maxCountByNodeName = result[i].Score
    4.     }
    5.     zoneID := utilnode.GetZoneKey(nodeNameToInfo[result[i].Host].Node())
    6.     if zoneID == "" {
    7.         continue
    8.     }
    9.     countsByZone[zoneID] += result[i].Score
    10. }

    遍历所有的节点，按比例取十分制的得分。

    1. for i := range result {
    2.     // initializing to the default/max node score of maxPriority
    3.     fScore := MaxPriorityFloat64
    4.     if maxCountByNodeName > 0 {
    5.         fScore = MaxPriorityFloat64 * (float64(maxCountByNodeName-result[i].Score) / maxCountByNodeNameFloat64)
    6.     }
    7.   ...
    8. }

    9. 总结

    优选，从满足的节点中选择出最优的节点。PrioritizeNodes最终返回是一个记录了各个节点分数的列表。

    9.1. PrioritizeNodes

    主要流程如下：

    1. 如果没有设置优选函数和拓展函数，则全部节点设置相同的分数，直接返回。
    2. 依次给node执行map函数进行打分。
    3. 再对上述map函数的执行结果执行reduce函数计算最终得分。
    4. 最后根据不同优先级函数的权重对得分取加权平均数。

    其中每类优选函数会包含map函数和reduce函数两种。

    9.2. NewSelectorSpreadPriority

    其中以NewSelectorSpreadPriority这个优选函数为例作分析，该函数的功能是将相同service、RS、RC或statefulset下pod尽量分散到不同的节点上。包括map函数和reduce函数两部分，具体如下。

    9.2.1. CalculateSpreadPriorityMap

    基本流程如下：

    1. 寻找与该pod对应的service、RS、RC、statefulset匹配的selector。
    2. 遍历当前节点的所有pod，将该节点上已存在的selector匹配到的pod的个数作为该节点的分数（此时，分数大的表示匹配到的pod越多，越不符合被调度的条件，该分数在reduce阶段会被按10分制处理成分数大的越符合被调度的条件）。

    9.2.2. CalculateSpreadPriorityReduce

    基本流程如下：

    1. 记录所有节点中匹配到pod个数最多的节点的分数（即匹配到的pod最多的个数）。
    2. 遍历所有的节点，按比例取十分制的得分，计算方式为：(节点中最多匹配pod的个数-当前节点pod的个数)/节点中最多匹配pod的个数。此时，分数越高表示该节点上匹配到的pod的个数越少，越可能被调度到，即满足把相同selector的pod分散到不同节点的需求。

    参考：

    • https://github.com/kubernetes/kubernetes/blob/v1.12.0/pkg/scheduler/core/generic_scheduler.go
    • https://github.com/kubernetes/kubernetes/blob/v1.12.0/pkg/scheduler/algorithm/priorities/selector_spreading.go