• kube-scheduler源码分析（四）之 findNodesThatFit
  • 1. 调用入口
  • 2. findNodesThatFit
  • 3. numFeasibleNodesToFind
  • 4. checkNode
  • 5. podFitsOnNode
    • 5.1. predicateFuncs
  • 6. PodFitsResources
    • 6.1. NodeInfo
    • 6.2. Resource
    • 6.3. allowedPodNumber
    • 6.4. podRequest
    • 6.5. AllocatableResource
    • 6.6. ScalarResources
  • 7. 总结
    • 7.1. checkNode
    • 7.2. podFitsOnNode
    • 7.3. PodFitsResources

    kube-scheduler源码分析（四）之 findNodesThatFit

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

    本文主要分析调度逻辑中的预选策略，即第一步筛选出符合pod调度条件的节点。

    1. 调用入口

    预选，通过预选函数来判断每个节点是否适合被该Pod调度。

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

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

    1. func (g *genericScheduler) Schedule(pod *v1.Pod, nodeLister algorithm.NodeLister) (string, error) {
    2.     ...
    3.   // 列出所有的节点
    4.     nodes, err := nodeLister.List()
    5.     if err != nil {
    6.         return "", err
    7.     }
    8.     if len(nodes) == 0 {
    9.         return "", ErrNoNodesAvailable
    10.     }
    12.     // Used for all fit and priority funcs.
    13.     err = g.cache.UpdateNodeNameToInfoMap(g.cachedNodeInfoMap)
    14.     if err != nil {
    15.         return "", err
    16.     }
    18.     trace.Step("Computing predicates")
    19.     startPredicateEvalTime := time.Now()
    20.   // 调用findNodesThatFit过滤出预选节点
    21.     filteredNodes, failedPredicateMap, err := g.findNodesThatFit(pod, nodes)
    22.     if err != nil {
    23.         return "", err
    24.     }
    26.     if len(filteredNodes) == 0 {
    27.         return "", &FitError{
    28.             Pod:              pod,
    29.             NumAllNodes:      len(nodes),
    30.             FailedPredicates: failedPredicateMap,
    31.         }
    32.     }
    33. // metrics
    34.   metrics.SchedulingAlgorithmPredicateEvaluationDuration.Observe(metrics.SinceInMicroseconds(startPredicateEvalTime))
    35.               metrics.SchedulingLatency.WithLabelValues(metrics.PredicateEvaluation).Observe(metrics.SinceInSeconds(startPredicateEvalTime))
    36.     ...
    37. }

    核心代码：

    1. // 调用findNodesThatFit过滤出预选节点
    2. filteredNodes, failedPredicateMap, err := g.findNodesThatFit(pod, nodes)

    2. findNodesThatFit

    findNodesThatFit基于给定的预选函数过滤node，每个node传入到预选函数中来确实该节点是否符合要求。

    findNodesThatFit的入参是被调度的pod和当前的节点列表，返回预选节点列表和错误。

    findNodesThatFit基本流程如下：

    1. 设置可行节点的总数，作为预选节点数组的容量，避免总节点过多需要筛选的节点过多。
    2. 通过NodeTree不断获取下一个节点来判断该节点是否满足pod的调度条件。
    3. 通过之前注册的各种预选函数来判断当前节点是否符合pod的调度条件。
    4. 最后返回满足调度条件的node列表，供下一步的优选操作。

    findNodesThatFit完整代码如下：

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

    1. // Filters the nodes to find the ones that fit based on the given predicate functions
    2. // Each node is passed through the predicate functions to determine if it is a fit
    3. func (g *genericScheduler) findNodesThatFit(pod *v1.Pod, nodes []*v1.Node) ([]*v1.Node, FailedPredicateMap, error) {
    4.    var filtered []*v1.Node
    5.    failedPredicateMap := FailedPredicateMap{}
    7.    if len(g.predicates) == 0 {
    8.       filtered = nodes
    9.    } else {
    10.       allNodes := int32(g.cache.NodeTree().NumNodes)
    11.       numNodesToFind := g.numFeasibleNodesToFind(allNodes)
    13.       // Create filtered list with enough space to avoid growing it
    14.       // and allow assigning.
    15.       filtered = make([]*v1.Node, numNodesToFind)
    16.       errs := errors.MessageCountMap{}
    17.       var (
    18.          predicateResultLock sync.Mutex
    19.          filteredLen         int32
    20.          equivClass          *equivalence.Class
    21.       )
    23.       ctx, cancel := context.WithCancel(context.Background())
    25.       // We can use the same metadata producer for all nodes.
    26.       meta := g.predicateMetaProducer(pod, g.cachedNodeInfoMap)
    28.       if g.equivalenceCache != nil {
    29.          // getEquivalenceClassInfo will return immediately if no equivalence pod found
    30.          equivClass = equivalence.NewClass(pod)
    31.       }
    33.       checkNode := func(i int) {
    34.          var nodeCache *equivalence.NodeCache
    35.          nodeName := g.cache.NodeTree().Next()
    36.          if g.equivalenceCache != nil {
    37.             nodeCache, _ = g.equivalenceCache.GetNodeCache(nodeName)
    38.          }
    39.          fits, failedPredicates, err := podFitsOnNode(
    40.             pod,
    41.             meta,
    42.             g.cachedNodeInfoMap[nodeName],
    43.             g.predicates,
    44.             g.cache,
    45.             nodeCache,
    46.             g.schedulingQueue,
    47.             g.alwaysCheckAllPredicates,
    48.             equivClass,
    49.          )
    50.          if err != nil {
    51.             predicateResultLock.Lock()
    52.             errs[err.Error()]++
    53.             predicateResultLock.Unlock()
    54.             return
    55.          }
    56.          if fits {
    57.             length := atomic.AddInt32(&filteredLen, 1)
    58.             if length > numNodesToFind {
    59.                cancel()
    60.                atomic.AddInt32(&filteredLen, -1)
    61.             } else {
    62.                filtered[length-1] = g.cachedNodeInfoMap[nodeName].Node()
    63.             }
    64.          } else {
    65.             predicateResultLock.Lock()
    66.             failedPredicateMap[nodeName] = failedPredicates
    67.             predicateResultLock.Unlock()
    68.          }
    69.       }
    71.       // Stops searching for more nodes once the configured number of feasible nodes
    72.       // are found.
    73.       workqueue.ParallelizeUntil(ctx, 16, int(allNodes), checkNode)
    75.       filtered = filtered[:filteredLen]
    76.       if len(errs) > 0 {
    77.          return []*v1.Node{}, FailedPredicateMap{}, errors.CreateAggregateFromMessageCountMap(errs)
    78.       }
    79.    }
    81.    if len(filtered) > 0 && len(g.extenders) != 0 {
    82.       for _, extender := range g.extenders {
    83.          if !extender.IsInterested(pod) {
    84.             continue
    85.          }
    86.          filteredList, failedMap, err := extender.Filter(pod, filtered, g.cachedNodeInfoMap)
    87.          if err != nil {
    88.             if extender.IsIgnorable() {
    89.                glog.Warningf("Skipping extender %v as it returned error %v and has ignorable flag set",
    90.                   extender, err)
    91.                continue
    92.             } else {
    93.                return []*v1.Node{}, FailedPredicateMap{}, err
    94.             }
    95.          }
    97.          for failedNodeName, failedMsg := range failedMap {
    98.             if _, found := failedPredicateMap[failedNodeName]; !found {
    99.                failedPredicateMap[failedNodeName] = []algorithm.PredicateFailureReason{}
    100.             }
    101.             failedPredicateMap[failedNodeName] = append(failedPredicateMap[failedNodeName], predicates.NewFailureReason(failedMsg))
    102.          }
    103.          filtered = filteredList
    104.          if len(filtered) == 0 {
    105.             break
    106.          }
    107.       }
    108.    }
    109.    return filtered, failedPredicateMap, nil
    110. }

    以下对findNodesThatFit分段分析。

    3. numFeasibleNodesToFind

    findNodesThatFit先基于所有的节点找出可行的节点是总数。numFeasibleNodesToFind的作用主要是避免当节点过多（超过100）影响调度的效率。

    1. allNodes := int32(g.cache.NodeTree().NumNodes)
    2. numNodesToFind := g.numFeasibleNodesToFind(allNodes)
    4. // Create filtered list with enough space to avoid growing it
    5. // and allow assigning.
    6. filtered = make([]*v1.Node, numNodesToFind)

    numFeasibleNodesToFind基本流程如下：

    • 如果所有的node节点小于minFeasibleNodesToFind(当前默认为100)则返回节点数。
    • 如果节点数超100，则取指定计分的百分比的节点数，当该百分比后的数目仍小于minFeasibleNodesToFind，则返回minFeasibleNodesToFind。
    • 如果百分比后的数目大于minFeasibleNodesToFind，则返回该百分比。
    1. // numFeasibleNodesToFind returns the number of feasible nodes that once found, the scheduler stops
    2. // its search for more feasible nodes.
    3. func (g *genericScheduler) numFeasibleNodesToFind(numAllNodes int32) int32 {
    4.     if numAllNodes < minFeasibleNodesToFind || g.percentageOfNodesToScore <= 0 ||
    5.         g.percentageOfNodesToScore >= 100 {
    6.         return numAllNodes
    7.     }
    8.     numNodes := numAllNodes * g.percentageOfNodesToScore / 100
    9.     if numNodes < minFeasibleNodesToFind {
    10.         return minFeasibleNodesToFind
    11.     }
    12.     return numNodes
    13. }

    4. checkNode

    checkNode是一个校验node是否符合要求的函数，其中实际调用到的核心函数是podFitsOnNode。再通过workqueue并发执行checkNode操作。

    checkNode主要流程如下：

    1. 通过cache中的nodeTree不断获取下一个node。
    2. 将当前node和pod传入podFitsOnNode判断当前node是否符合要求。
    3. 如果当前node符合要求就将当前node加入预选节点的数组中filtered。
    4. 如果当前node不满足要求，则加入到失败的数组中，并记录原因。
    5. 通过workqueue.ParallelizeUntil并发执行checkNode函数，一旦找到配置的可行节点数，就停止搜索更多节点。
    1. checkNode := func(i int) {
    2.     var nodeCache *equivalence.NodeCache
    3.     nodeName := g.cache.NodeTree().Next()
    4.     if g.equivalenceCache != nil {
    5.         nodeCache, _ = g.equivalenceCache.GetNodeCache(nodeName)
    6.     }
    7.     fits, failedPredicates, err := podFitsOnNode(
    8.         pod,
    9.         meta,
    10.         g.cachedNodeInfoMap[nodeName],
    11.         g.predicates,
    12.         g.cache,
    13.         nodeCache,
    14.         g.schedulingQueue,
    15.         g.alwaysCheckAllPredicates,
    16.         equivClass,
    17.     )
    18.     if err != nil {
    19.         predicateResultLock.Lock()
    20.         errs[err.Error()]++
    21.         predicateResultLock.Unlock()
    22.         return
    23.     }
    24.     if fits {
    25.         length := atomic.AddInt32(&filteredLen, 1)
    26.         if length > numNodesToFind {
    27.             cancel()
    28.             atomic.AddInt32(&filteredLen, -1)
    29.         } else {
    30.             filtered[length-1] = g.cachedNodeInfoMap[nodeName].Node()
    31.         }
    32.     } else {
    33.         predicateResultLock.Lock()
    34.         failedPredicateMap[nodeName] = failedPredicates
    35.         predicateResultLock.Unlock()
    36.     }
    37. }

    workqueue的并发操作：

    1. // Stops searching for more nodes once the configured number of feasible nodes
    2. // are found.
    3. workqueue.ParallelizeUntil(ctx, 16, int(allNodes), checkNode)

    ParallelizeUntil具体代码如下：

    1. // ParallelizeUntil is a framework that allows for parallelizing N
    2. // independent pieces of work until done or the context is canceled.
    3. func ParallelizeUntil(ctx context.Context, workers, pieces int, doWorkPiece DoWorkPieceFunc) {
    4.     var stop <-chan struct{}
    5.     if ctx != nil {
    6.         stop = ctx.Done()
    7.     }
    9.     toProcess := make(chan int, pieces)
    10.     for i := 0; i < pieces; i++ {
    11.         toProcess <- i
    12.     }
    13.     close(toProcess)
    15.     if pieces < workers {
    16.         workers = pieces
    17.     }
    19.     wg := sync.WaitGroup{}
    20.     wg.Add(workers)
    21.     for i := 0; i < workers; i++ {
    22.         go func() {
    23.             defer utilruntime.HandleCrash()
    24.             defer wg.Done()
    25.             for piece := range toProcess {
    26.                 select {
    27.                 case <-stop:
    28.                     return
    29.                 default:
    30.                     doWorkPiece(piece)
    31.                 }
    32.             }
    33.         }()
    34.     }
    35.     wg.Wait()
    36. }

    5. podFitsOnNode

    podFitsOnNode主要内容如下：

    • podFitsOnNode会检查给定的某个Node是否满足预选的函数。

    • 对于给定的pod，podFitsOnNode会检查是否有相同的pod存在，尽量复用缓存过的预选结果。

    podFitsOnNode主要在Schedule（调度）和Preempt（抢占）的时候被调用。

    当在Schedule中被调用的时候，主要判断是否可以被调度到当前节点，依据为当前节点上所有已存在的pod及被提名要运行到该节点的具有相等或更高优先级的pod。

    当在Preempt中被调用的时候，即发生抢占的时候，通过SelectVictimsOnNode函数选出需要被移除的pod，移除后然后将预调度的pod调度到该节点上。

    podFitsOnNode基本流程如下：

    1. 遍历之前注册好的预选策略predicates.Ordering，并获取预选策略的执行函数。
    2. 遍历执行每个预选函数，并返回是否合适，预选失败的原因和错误。
    3. 如果预选函数执行的结果不合适，则加入预选失败的数组中。
    4. 最后返回预选失败的个数是否为0，和预选失败的原因。

    入参：

    • pod
    • PredicateMetadata
    • NodeInfo
    • predicateFuncs
    • schedulercache.Cache
    • nodeCache
    • SchedulingQueue
    • alwaysCheckAllPredicates
    • equivClass

    出参：

    • fit
    • PredicateFailureReason

    完整代码如下：

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

    1. // podFitsOnNode checks whether a node given by NodeInfo satisfies the given predicate functions.
    2. // For given pod, podFitsOnNode will check if any equivalent pod exists and try to reuse its cached
    3. // predicate results as possible.
    4. // This function is called from two different places: Schedule and Preempt.
    5. // When it is called from Schedule, we want to test whether the pod is schedulable
    6. // on the node with all the existing pods on the node plus higher and equal priority
    7. // pods nominated to run on the node.
    8. // When it is called from Preempt, we should remove the victims of preemption and
    9. // add the nominated pods. Removal of the victims is done by SelectVictimsOnNode().
    10. // It removes victims from meta and NodeInfo before calling this function.
    11. func podFitsOnNode(
    12.     pod *v1.Pod,
    13.     meta algorithm.PredicateMetadata,
    14.     info *schedulercache.NodeInfo,
    15.     predicateFuncs map[string]algorithm.FitPredicate,
    16.     cache schedulercache.Cache,
    17.     nodeCache *equivalence.NodeCache,
    18.     queue SchedulingQueue,
    19.     alwaysCheckAllPredicates bool,
    20.     equivClass *equivalence.Class,
    21. ) (bool, []algorithm.PredicateFailureReason, error) {
    22.     var (
    23.         eCacheAvailable  bool
    24.         failedPredicates []algorithm.PredicateFailureReason
    25.     )
    27.     podsAdded := false
    28.     // We run predicates twice in some cases. If the node has greater or equal priority
    29.     // nominated pods, we run them when those pods are added to meta and nodeInfo.
    30.     // If all predicates succeed in this pass, we run them again when these
    31.     // nominated pods are not added. This second pass is necessary because some
    32.     // predicates such as inter-pod affinity may not pass without the nominated pods.
    33.     // If there are no nominated pods for the node or if the first run of the
    34.     // predicates fail, we don't run the second pass.
    35.     // We consider only equal or higher priority pods in the first pass, because
    36.     // those are the current "pod" must yield to them and not take a space opened
    37.     // for running them. It is ok if the current "pod" take resources freed for
    38.     // lower priority pods.
    39.     // Requiring that the new pod is schedulable in both circumstances ensures that
    40.     // we are making a conservative decision: predicates like resources and inter-pod
    41.     // anti-affinity are more likely to fail when the nominated pods are treated
    42.     // as running, while predicates like pod affinity are more likely to fail when
    43.     // the nominated pods are treated as not running. We can't just assume the
    44.     // nominated pods are running because they are not running right now and in fact,
    45.     // they may end up getting scheduled to a different node.
    46.     for i := 0; i < 2; i++ {
    47.         metaToUse := meta
    48.         nodeInfoToUse := info
    49.         if i == 0 {
    50.             podsAdded, metaToUse, nodeInfoToUse = addNominatedPods(util.GetPodPriority(pod), meta, info, queue)
    51.         } else if !podsAdded || len(failedPredicates) != 0 {
    52.             break
    53.         }
    54.         // Bypass eCache if node has any nominated pods.
    55.         // TODO(bsalamat): consider using eCache and adding proper eCache invalidations
    56.         // when pods are nominated or their nominations change.
    57.         eCacheAvailable = equivClass != nil && nodeCache != nil && !podsAdded
    58.         for _, predicateKey := range predicates.Ordering() {
    59.             var (
    60.                 fit     bool
    61.                 reasons []algorithm.PredicateFailureReason
    62.                 err     error
    63.             )
    64.             //TODO (yastij) : compute average predicate restrictiveness to export it as Prometheus metric
    65.             if predicate, exist := predicateFuncs[predicateKey]; exist {
    66.                 if eCacheAvailable {
    67.                     fit, reasons, err = nodeCache.RunPredicate(predicate, predicateKey, pod, metaToUse, nodeInfoToUse, equivClass, cache)
    68.                 } else {
    69.                     fit, reasons, err = predicate(pod, metaToUse, nodeInfoToUse)
    70.                 }
    71.                 if err != nil {
    72.                     return false, []algorithm.PredicateFailureReason{}, err
    73.                 }
    75.                 if !fit {
    76.                     // eCache is available and valid, and predicates result is unfit, record the fail reasons
    77.                     failedPredicates = append(failedPredicates, reasons...)
    78.                     // if alwaysCheckAllPredicates is false, short circuit all predicates when one predicate fails.
    79.                     if !alwaysCheckAllPredicates {
    80.                         glog.V(5).Infoln("since alwaysCheckAllPredicates has not been set, the predicate " +
    81.                             "evaluation is short circuited and there are chances " +
    82.                             "of other predicates failing as well.")
    83.                         break
    84.                     }
    85.                 }
    86.             }
    87.         }
    88.     }
    90.     return len(failedPredicates) == 0, failedPredicates, nil
    91. }

    5.1. predicateFuncs

    根据之前初注册好的预选策略函数来执行预选，判断节点是否符合调度。

    1. for _, predicateKey := range predicates.Ordering() {
    2.     if predicate, exist := predicateFuncs[predicateKey]; exist {
    3.         if eCacheAvailable {
    4.             fit, reasons, err = nodeCache.RunPredicate(predicate, predicateKey, pod, metaToUse, nodeInfoToUse, equivClass, cache)
    5.         } else {
    6.             fit, reasons, err = predicate(pod, metaToUse, nodeInfoToUse)
    7.         }

    预选策略如下：

    1. var (
    2.     predicatesOrdering = []string{CheckNodeConditionPred, CheckNodeUnschedulablePred,
    3.         GeneralPred, HostNamePred, PodFitsHostPortsPred,
    4.         MatchNodeSelectorPred, PodFitsResourcesPred, NoDiskConflictPred,
    5.         PodToleratesNodeTaintsPred, PodToleratesNodeNoExecuteTaintsPred, CheckNodeLabelPresencePred,
    6.         CheckServiceAffinityPred, MaxEBSVolumeCountPred, MaxGCEPDVolumeCountPred, MaxCSIVolumeCountPred,
    7.         MaxAzureDiskVolumeCountPred, CheckVolumeBindingPred, NoVolumeZoneConflictPred,
    8.         CheckNodeMemoryPressurePred, CheckNodePIDPressurePred, CheckNodeDiskPressurePred, MatchInterPodAffinityPred}
    9. )

    6. PodFitsResources

    以下以PodFitsResources这个预选函数为例做分析，其他重要的预选函数待后续单独分析。

    PodFitsResources用来检查一个节点是否有足够的资源来运行当前的pod，包括CPU、内存、GPU等。

    PodFitsResources基本流程如下：

    1. 判断当前节点上pod总数加上预调度pod个数是否大于node的可分配pod总数，若是则不允许调度。
    2. 判断pod的request值是否都为0，若是则允许调度。
    3. 判断pod的request值加上当前node上所有pod的request值总和是否大于node的可分配资源，若是则不允许调度。
    4. 判断pod的拓展资源request值加上当前node上所有pod对应的request值总和是否大于node对应的可分配资源，若是则不允许调度。

    PodFitsResources的注册代码如下：

    1. factory.RegisterFitPredicate(predicates.PodFitsResourcesPred, predicates.PodFitsResources)

    PodFitsResources入参：

    • pod

    • nodeInfo

    • PredicateMetadata

    PodFitsResources出参：

    • fit
    • PredicateFailureReason

    PodFitsResources完整代码：

    此部分的代码位于pkg/scheduler/algorithm/predicates/predicates.go

    1. // PodFitsResources checks if a node has sufficient resources, such as cpu, memory, gpu, opaque int resources etc to run a pod.
    2. // First return value indicates whether a node has sufficient resources to run a pod while the second return value indicates the
    3. // predicate failure reasons if the node has insufficient resources to run the pod.
    4. func PodFitsResources(pod *v1.Pod, meta algorithm.PredicateMetadata, nodeInfo *schedulercache.NodeInfo) (bool, []algorithm.PredicateFailureReason, error) {
    5.     node := nodeInfo.Node()
    6.     if node == nil {
    7.         return false, nil, fmt.Errorf("node not found")
    8.     }
    10.     var predicateFails []algorithm.PredicateFailureReason
    11.     allowedPodNumber := nodeInfo.AllowedPodNumber()
    12.     if len(nodeInfo.Pods())+1 > allowedPodNumber {
    13.         predicateFails = append(predicateFails, NewInsufficientResourceError(v1.ResourcePods, 1, int64(len(nodeInfo.Pods())), int64(allowedPodNumber)))
    14.     }
    16.     // No extended resources should be ignored by default.
    17.     ignoredExtendedResources := sets.NewString()
    19.     var podRequest *schedulercache.Resource
    20.     if predicateMeta, ok := meta.(*predicateMetadata); ok {
    21.         podRequest = predicateMeta.podRequest
    22.         if predicateMeta.ignoredExtendedResources != nil {
    23.             ignoredExtendedResources = predicateMeta.ignoredExtendedResources
    24.         }
    25.     } else {
    26.         // We couldn't parse metadata - fallback to computing it.
    27.         podRequest = GetResourceRequest(pod)
    28.     }
    29.     if podRequest.MilliCPU == 0 &&
    30.         podRequest.Memory == 0 &&
    31.         podRequest.EphemeralStorage == 0 &&
    32.         len(podRequest.ScalarResources) == 0 {
    33.         return len(predicateFails) == 0, predicateFails, nil
    34.     }
    36.     allocatable := nodeInfo.AllocatableResource()
    37.     if allocatable.MilliCPU < podRequest.MilliCPU+nodeInfo.RequestedResource().MilliCPU {
    38.         predicateFails = append(predicateFails, NewInsufficientResourceError(v1.ResourceCPU, podRequest.MilliCPU, nodeInfo.RequestedResource().MilliCPU, allocatable.MilliCPU))
    39.     }
    40.     if allocatable.Memory < podRequest.Memory+nodeInfo.RequestedResource().Memory {
    41.         predicateFails = append(predicateFails, NewInsufficientResourceError(v1.ResourceMemory, podRequest.Memory, nodeInfo.RequestedResource().Memory, allocatable.Memory))
    42.     }
    43.     if allocatable.EphemeralStorage < podRequest.EphemeralStorage+nodeInfo.RequestedResource().EphemeralStorage {
    44.         predicateFails = append(predicateFails, NewInsufficientResourceError(v1.ResourceEphemeralStorage, podRequest.EphemeralStorage, nodeInfo.RequestedResource().EphemeralStorage, allocatable.EphemeralStorage))
    45.     }
    47.     for rName, rQuant := range podRequest.ScalarResources {
    48.         if v1helper.IsExtendedResourceName(rName) {
    49.             // If this resource is one of the extended resources that should be
    50.             // ignored, we will skip checking it.
    51.             if ignoredExtendedResources.Has(string(rName)) {
    52.                 continue
    53.             }
    54.         }
    55.         if allocatable.ScalarResources[rName] < rQuant+nodeInfo.RequestedResource().ScalarResources[rName] {
    56.             predicateFails = append(predicateFails, NewInsufficientResourceError(rName, podRequest.ScalarResources[rName], nodeInfo.RequestedResource().ScalarResources[rName], allocatable.ScalarResources[rName]))
    57.         }
    58.     }
    60.     if glog.V(10) {
    61.         if len(predicateFails) == 0 {
    62.             // We explicitly don't do glog.V(10).Infof() to avoid computing all the parameters if this is
    63.             // not logged. There is visible performance gain from it.
    64.             glog.Infof("Schedule Pod %+v on Node %+v is allowed, Node is running only %v out of %v Pods.",
    65.                 podName(pod), node.Name, len(nodeInfo.Pods()), allowedPodNumber)
    66.         }
    67.     }
    68.     return len(predicateFails) == 0, predicateFails, nil
    69. }

    6.1. NodeInfo

    NodeInfo是node的聚合信息，主要包括：

    • node：k8s node的结构体
    • pods：当前node上pod的数量
    • requestedResource：当前node上所有pod的request总和
    • allocatableResource：node的实际所有的可分配资源(对应于Node.Status.Allocatable.*)，可理解为node的资源总量。

    此部分代码位于pkg/scheduler/cache/node_info.go

    1. // NodeInfo is node level aggregated information.
    2. type NodeInfo struct {
    3.     // Overall node information.
    4.     node *v1.Node
    6.     pods             []*v1.Pod
    7.     podsWithAffinity []*v1.Pod
    8.     usedPorts        util.HostPortInfo
    10.     // Total requested resource of all pods on this node.
    11.     // It includes assumed pods which scheduler sends binding to apiserver but
    12.     // didn't get it as scheduled yet.
    13.     requestedResource *Resource
    14.     nonzeroRequest    *Resource
    15.     // We store allocatedResources (which is Node.Status.Allocatable.*) explicitly
    16.     // as int64, to avoid conversions and accessing map.
    17.     allocatableResource *Resource
    19.     // Cached taints of the node for faster lookup.
    20.     taints    []v1.Taint
    21.     taintsErr error
    23.     // imageStates holds the entry of an image if and only if this image is on the node. The entry can be used for
    24.     // checking an image's existence and advanced usage (e.g., image locality scheduling policy) based on the image
    25.     // state information.
    26.     imageStates map[string]*ImageStateSummary
    28.     // TransientInfo holds the information pertaining to a scheduling cycle. This will be destructed at the end of
    29.     // scheduling cycle.
    30.     // TODO: @ravig. Remove this once we have a clear approach for message passing across predicates and priorities.
    31.     TransientInfo *transientSchedulerInfo
    33.     // Cached conditions of node for faster lookup.
    34.     memoryPressureCondition v1.ConditionStatus
    35.     diskPressureCondition   v1.ConditionStatus
    36.     pidPressureCondition    v1.ConditionStatus
    38.     // Whenever NodeInfo changes, generation is bumped.
    39.     // This is used to avoid cloning it if the object didn't change.
    40.     generation int64
    41. }

    6.2. Resource

    Resource是可计算资源的集合体。主要包括：

    • MilliCPU
    • Memory
    • EphemeralStorage
    • AllowedPodNumber：允许的pod总数(对应于Node.Status.Allocatable.Pods().Value())，一般为110。
    • ScalarResources
    1. // Resource is a collection of compute resource.
    2. type Resource struct {
    3.     MilliCPU         int64
    4.     Memory           int64
    5.     EphemeralStorage int64
    6.     // We store allowedPodNumber (which is Node.Status.Allocatable.Pods().Value())
    7.     // explicitly as int, to avoid conversions and improve performance.
    8.     AllowedPodNumber int
    9.     // ScalarResources
    10.     ScalarResources map[v1.ResourceName]int64
    11. }

    以下分析podFitsOnNode的具体流程。

    6.3. allowedPodNumber

    首先获取节点的信息，先判断如果该节点当前所有的pod的个数加上当前预调度的pod是否会大于该节点允许的pod的总数，一般为110个。如果超过，则predicateFails数组增加1，即当前节点不适合该pod。

    1. node := nodeInfo.Node()
    2. if node == nil {
    3.     return false, nil, fmt.Errorf("node not found")
    4. }
    6. var predicateFails []algorithm.PredicateFailureReason
    7. allowedPodNumber := nodeInfo.AllowedPodNumber()
    8. if len(nodeInfo.Pods())+1 > allowedPodNumber {
    9.     predicateFails = append(predicateFails, NewInsufficientResourceError(v1.ResourcePods, 1, int64(len(nodeInfo.Pods())), int64(allowedPodNumber)))
    10.     }

    6.4. podRequest

    如果podRequest都为0，则允许调度到该节点，直接返回结果。

    1. if podRequest.MilliCPU == 0 &&
    2.     podRequest.Memory == 0 &&
    3.     podRequest.EphemeralStorage == 0 &&
    4.     len(podRequest.ScalarResources) == 0 {
    5.     return len(predicateFails) == 0, predicateFails, nil
    6. }

    6.5. AllocatableResource

    如果当前预调度的pod的request资源加上当前node上所有pod的request总和大于该node的可分配资源总量，则不允许调度到该节点，直接返回结果。其中request资源包括CPU、内存、storage。

    1. allocatable := nodeInfo.AllocatableResource()
    2. if allocatable.MilliCPU < podRequest.MilliCPU+nodeInfo.RequestedResource().MilliCPU {
    3.     predicateFails = append(predicateFails, NewInsufficientResourceError(v1.ResourceCPU, podRequest.MilliCPU, nodeInfo.RequestedResource().MilliCPU, allocatable.MilliCPU))
    4. }
    5. if allocatable.Memory < podRequest.Memory+nodeInfo.RequestedResource().Memory {
    6.     predicateFails = append(predicateFails, NewInsufficientResourceError(v1.ResourceMemory, podRequest.Memory, nodeInfo.RequestedResource().Memory, allocatable.Memory))
    7. }
    8. if allocatable.EphemeralStorage < podRequest.EphemeralStorage+nodeInfo.RequestedResource().EphemeralStorage {
    9.     predicateFails = append(predicateFails, NewInsufficientResourceError(v1.ResourceEphemeralStorage, podRequest.EphemeralStorage, nodeInfo.RequestedResource().EphemeralStorage, allocatable.EphemeralStorage))
    10.     }

    6.6. ScalarResources

    判断其他拓展的标量资源，是否该pod的request值加上当前node上所有pod的对应资源的request总和大于该node上对应资源的可分配总量，如果是，则不允许调度到该节点。

    1. for rName, rQuant := range podRequest.ScalarResources {
    2.     if v1helper.IsExtendedResourceName(rName) {
    3.         // If this resource is one of the extended resources that should be
    4.         // ignored, we will skip checking it.
    5.         if ignoredExtendedResources.Has(string(rName)) {
    6.             continue
    7.         }
    8.     }
    9.     if allocatable.ScalarResources[rName] < rQuant+nodeInfo.RequestedResource().ScalarResources[rName] {
    10.         predicateFails = append(predicateFails, NewInsufficientResourceError(rName, podRequest.ScalarResources[rName], nodeInfo.RequestedResource().ScalarResources[rName], allocatable.ScalarResources[rName]))
    11.     }
    12. }

    7. 总结

    findNodesThatFit基于给定的预选函数过滤node，每个node传入到预选函数中来确实该节点是否符合要求。

    findNodesThatFit的入参是被调度的pod和当前的节点列表，返回预选节点列表和错误。

    findNodesThatFit基本流程如下：

    1. 设置可行节点的总数，作为预选节点数组的容量，避免总节点过多导致需要筛选的节点过多，效率低。
    2. 通过NodeTree不断获取下一个节点来判断该节点是否满足pod的调度条件。
    3. 通过之前注册的各种预选函数来判断当前节点是否符合pod的调度条件。
    4. 最后返回满足调度条件的node列表，供下一步的优选操作。

    7.1. checkNode

    checkNode是一个校验node是否符合要求的函数，其中实际调用到的核心函数是podFitsOnNode。再通过workqueue并发执行checkNode操作。

    checkNode主要流程如下：

    1. 通过cache中的nodeTree不断获取下一个node。
    2. 将当前node和pod传入podFitsOnNode判断当前node是否符合要求。
    3. 如果当前node符合要求就将当前node加入预选节点的数组中filtered。
    4. 如果当前node不满足要求，则加入到失败的数组中，并记录原因。
    5. 通过workqueue.ParallelizeUntil并发执行checkNode函数，一旦找到配置的可行节点数，就停止搜索更多节点。

    7.2. podFitsOnNode

    其中会调用到核心函数podFitsOnNode。

    podFitsOnNode主要内容如下：

    • podFitsOnNode会检查给定的某个Node是否满足预选的函数。

    • 对于给定的pod，podFitsOnNode会检查是否有相同的pod存在，尽量复用缓存过的预选结果。

    podFitsOnNode主要在Schedule（调度）和Preempt（抢占）的时候被调用。

    当在Schedule中被调用的时候，主要判断是否可以被调度到当前节点，依据为当前节点上所有已存在的pod及被提名要运行到该节点的具有相等或更高优先级的pod。

    当在Preempt中被调用的时候，即发生抢占的时候，通过SelectVictimsOnNode函数选出需要被移除的pod，移除后然后将预调度的pod调度到该节点上。

    podFitsOnNode基本流程如下：

    1. 遍历之前注册好的预选策略predicates.Ordering，并获取预选策略的执行函数。
    2. 遍历执行每个预选函数，并返回是否合适，预选失败的原因和错误。
    3. 如果预选函数执行的结果不合适，则加入预选失败的数组中。
    4. 最后返回预选失败的个数是否为0，和预选失败的原因。

    7.3. PodFitsResources

    本文只示例分析了其中一个重要的预选函数：PodFitsResources

    PodFitsResources用来检查一个节点是否有足够的资源来运行当前的pod，包括CPU、内存、GPU等。

    PodFitsResources基本流程如下：

    1. 判断当前节点上pod总数加上预调度pod个数是否大于node的可分配pod总数，若是则不允许调度。
    2. 判断pod的request值是否都为0，若是则允许调度。
    3. 判断pod的request值加上当前node上所有pod的request值总和是否大于node的可分配资源，若是则不允许调度。
    4. 判断pod的拓展资源request值加上当前node上所有pod对应的request值总和是否大于node对应的可分配资源，若是则不允许调度。

    参考：

    • 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/predicates/predicates.go