kube-scheduler 源码分析（3）- 抢占调度分析

apiVersion: kubescheduler.config.k8s.io/v1alpha1kind: KubeSchedulerConfiguration...disablePreemption: true

// pkg/scheduler/scheduler.gofunc (sched *Scheduler) scheduleOne(ctx context.Context) {    ...    // 调度pod    scheduleResult, err := sched.Algorithm.Schedule(schedulingCycleCtx, state, pod)  if err != nil {    ...    if fitError, ok := err.(*core.FitError); ok {        // 判断是否关闭了抢占调度功能      if sched.DisablePreemption {        klog.V(3).Infof("Pod priority feature is not enabled or preemption is disabled by scheduler configuration." +          " No preemption is performed.")      } else {      // 抢占调度逻辑        preemptionStartTime := time.Now()        sched.preempt(schedulingCycleCtx, state, fwk, pod, fitError)        ...      }      ...  }  ...

// pkg/scheduler/scheduler.gofunc (sched *Scheduler) preempt(ctx context.Context, state *framework.CycleState, fwk framework.Framework, preemptor *v1.Pod, scheduleErr error) (string, error) {  ...    // （1）模拟pod抢占调度过程，返回pod可以抢占的node节点、被抢占的pod列表、需要去除nominateName属性的pod列表  node, victims, nominatedPodsToClear, err := sched.Algorithm.Preempt(ctx, state, preemptor, scheduleErr)  if err != nil {    klog.Errorf("Error preempting victims to make room for %v/%v: %v", preemptor.Namespace, preemptor.Name, err)    return "", err  }  var nodeName = ""  if node != nil {    nodeName = node.Name        sched.SchedulingQueue.UpdateNominatedPodForNode(preemptor, nodeName)
    // （2）请求apiserver，将可以抢占的node节点名称设置到pod的nominatedNode属性值中，然后该pod会重新进入待调度pod队列，等待再一次调度    err = sched.podPreemptor.setNominatedNodeName(preemptor, nodeName)    if err != nil {      klog.Errorf("Error in preemption process. Cannot set 'NominatedPod' on pod %v/%v: %v", preemptor.Namespace, preemptor.Name, err)      sched.SchedulingQueue.DeleteNominatedPodIfExists(preemptor)      return "", err    }                // （3）遍历被抢占的pod列表，请求apiserver，删除pod    for _, victim := range victims {      if err := sched.podPreemptor.deletePod(victim); err != nil {        klog.Errorf("Error preempting pod %v/%v: %v", victim.Namespace, victim.Name, err)        return "", err      }      // If the victim is a WaitingPod, send a reject message to the PermitPlugin      if waitingPod := fwk.GetWaitingPod(victim.UID); waitingPod != nil {        waitingPod.Reject("preempted")      }      sched.Recorder.Eventf(victim, preemptor, v1.EventTypeNormal, "Preempted", "Preempting", "Preempted by %v/%v on node %v", preemptor.Namespace, preemptor.Name, nodeName)
    }    metrics.PreemptionVictims.Observe(float64(len(victims)))  }  // （4）遍历需要去除nominateName属性的pod列表，请求apiserver，更新pod，去除pod的nominateName属性值  for _, p := range nominatedPodsToClear {    rErr := sched.podPreemptor.removeNominatedNodeName(p)    if rErr != nil {      klog.Errorf("Cannot remove 'NominatedPod' field of pod: %v", rErr)      // We do not return as this error is not critical.    }  }  return nodeName, err}

// pkg/scheduler/core/generic_scheduler.gofunc (g *genericScheduler) Preempt(ctx context.Context, state *framework.CycleState, pod *v1.Pod, scheduleErr error) (*v1.Node, []*v1.Pod, []*v1.Pod, error) {  // Scheduler may return various types of errors. Consider preemption only if  // the error is of type FitError.  fitError, ok := scheduleErr.(*FitError)  if !ok || fitError == nil {    return nil, nil, nil, nil  }  if !podEligibleToPreemptOthers(pod, g.nodeInfoSnapshot.NodeInfoMap, g.enableNonPreempting) {    klog.V(5).Infof("Pod %v/%v is not eligible for more preemption.", pod.Namespace, pod.Name)    return nil, nil, nil, nil  }  if len(g.nodeInfoSnapshot.NodeInfoMap) == 0 {    return nil, nil, nil, ErrNoNodesAvailable  }  // （1）获取预选失败且移除部分pod之后可能可以满足调度条件的节点；  potentialNodes := nodesWherePreemptionMightHelp(g.nodeInfoSnapshot.NodeInfoMap, fitError)  if len(potentialNodes) == 0 {    klog.V(3).Infof("Preemption will not help schedule pod %v/%v on any node.", pod.Namespace, pod.Name)    // In this case, we should clean-up any existing nominated node name of the pod.    return nil, nil, []*v1.Pod{pod}, nil  }  var (    pdbs []*policy.PodDisruptionBudget    err  error  )  // （2）获取PodDisruptionBudget对象，用于后续筛选可以被抢占的node节点列表（关于PodDisruptionBudget的用法，可自行搜索资料查看）；  if g.pdbLister != nil {    pdbs, err = g.pdbLister.List(labels.Everything())    if err != nil {      return nil, nil, nil, err    }  }  // （3）获取可以被抢占的node节点列表；    nodeToVictims, err := g.selectNodesForPreemption(ctx, state, pod, potentialNodes, pdbs)  if err != nil {    return nil, nil, nil, err  }      // （4）遍历scheduler-extender（kube-scheduler的一种webhook扩展机制），执行extender的抢占处理逻辑，根据处理逻辑过滤可以被抢占的node节点列表；   // We will only check nodeToVictims with extenders that support preemption.  // Extenders which do not support preemption may later prevent preemptor from being scheduled on the nominated  // node. In that case, scheduler will find a different host for the preemptor in subsequent scheduling cycles.  nodeToVictims, err = g.processPreemptionWithExtenders(pod, nodeToVictims)  if err != nil {    return nil, nil, nil, err  }        // （5）从可被抢占的node节点列表中挑选出一个node节点；    candidateNode := pickOneNodeForPreemption(nodeToVictims)  if candidateNode == nil {    return nil, nil, nil, nil  }        // （6）获取被抢占node节点上nominateName属性不为空且优先级比抢占pod低的pod列表；    // Lower priority pods nominated to run on this node, may no longer fit on  // this node. So, we should remove their nomination. Removing their  // nomination updates these pods and moves them to the active queue. It  // lets scheduler find another place for them.  nominatedPods := g.getLowerPriorityNominatedPods(pod, candidateNode.Name)  if nodeInfo, ok := g.nodeInfoSnapshot.NodeInfoMap[candidateNode.Name]; ok {    return nodeInfo.Node(), nodeToVictims[candidateNode].Pods, nominatedPods, nil  }
  return nil, nil, nil, fmt.Errorf(    "preemption failed: the target node %s has been deleted from scheduler cache",    candidateNode.Name)}

// pkg/scheduler/core/generic_scheduler.gofunc nodesWherePreemptionMightHelp(nodeNameToInfo map[string]*schedulernodeinfo.NodeInfo, fitErr *FitError) []*v1.Node {  potentialNodes := []*v1.Node{}  for name, node := range nodeNameToInfo {    if fitErr.FilteredNodesStatuses[name].Code() == framework.UnschedulableAndUnresolvable {      continue    }    failedPredicates := fitErr.FailedPredicates[name]
    // If we assume that scheduler looks at all nodes and populates the failedPredicateMap    // (which is the case today), the !found case should never happen, but we'd prefer    // to rely less on such assumptions in the code when checking does not impose    // significant overhead.    // Also, we currently assume all failures returned by extender as resolvable.    if predicates.UnresolvablePredicateExists(failedPredicates) == nil {      klog.V(3).Infof("Node %v is a potential node for preemption.", name)      potentialNodes = append(potentialNodes, node.Node())    }  }  return potentialNodes}

// pkg/scheduler/algorithm/predicates/error.govar unresolvablePredicateFailureErrors = map[PredicateFailureReason]struct{}{  ErrNodeSelectorNotMatch:      {},  ErrPodAffinityRulesNotMatch:  {},  ErrPodNotMatchHostName:       {},  ErrTaintsTolerationsNotMatch: {},  ErrNodeLabelPresenceViolated: {},  // Node conditions won't change when scheduler simulates removal of preemption victims.  // So, it is pointless to try nodes that have not been able to host the pod due to node  // conditions. These include ErrNodeNotReady, ErrNodeUnderPIDPressure, ErrNodeUnderMemoryPressure, ....  ErrNodeNotReady:            {},  ErrNodeNetworkUnavailable:  {},  ErrNodeUnderDiskPressure:   {},  ErrNodeUnderPIDPressure:    {},  ErrNodeUnderMemoryPressure: {},  ErrNodeUnschedulable:       {},  ErrNodeUnknownCondition:    {},  ErrVolumeZoneConflict:      {},  ErrVolumeNodeConflict:      {},  ErrVolumeBindConflict:      {},}
// UnresolvablePredicateExists checks if there is at least one unresolvable predicate failure reason, if true// returns the first one in the list.func UnresolvablePredicateExists(reasons []PredicateFailureReason) PredicateFailureReason {  for _, r := range reasons {    if _, ok := unresolvablePredicateFailureErrors[r]; ok {      return r    }  }  return nil}

// pkg/scheduler/core/generic_scheduler.go// selectNodesForPreemption finds all the nodes with possible victims for// preemption in parallel.func (g *genericScheduler) selectNodesForPreemption(  ctx context.Context,  state *framework.CycleState,  pod *v1.Pod,  potentialNodes []*v1.Node,  pdbs []*policy.PodDisruptionBudget,) (map[*v1.Node]*extenderv1.Victims, error) {  nodeToVictims := map[*v1.Node]*extenderv1.Victims{}  var resultLock sync.Mutex        // （1）定义checkNode函数  // We can use the same metadata producer for all nodes.  meta := g.predicateMetaProducer(pod, g.nodeInfoSnapshot)  checkNode := func(i int) {    nodeName := potentialNodes[i].Name    if g.nodeInfoSnapshot.NodeInfoMap[nodeName] == nil {      return    }    nodeInfoCopy := g.nodeInfoSnapshot.NodeInfoMap[nodeName].Clone()    var metaCopy predicates.Metadata    if meta != nil {      metaCopy = meta.ShallowCopy()    }    stateCopy := state.Clone()    stateCopy.Write(migration.PredicatesStateKey, &migration.PredicatesStateData{Reference: metaCopy})    // 调用g.selectVictimsOnNode方法，方法返回某node是否适合被抢占，并返回该node节点上被抢占的pod的最小集合、与PDB冲突的pod数量；     pods, numPDBViolations, fits := g.selectVictimsOnNode(ctx, stateCopy, pod, metaCopy, nodeInfoCopy, pdbs)    if fits {      resultLock.Lock()      victims := extenderv1.Victims{        Pods:             pods,        NumPDBViolations: int64(numPDBViolations),      }      nodeToVictims[potentialNodes[i]] = &victims      resultLock.Unlock()    }  }  // （2）拉起16个goroutine，并发调用checkNode函数，对预选失败的node节点列表进行是否适合被抢占的检查；  workqueue.ParallelizeUntil(context.TODO(), 16, len(potentialNodes), checkNode)  return nodeToVictims, nil}

// pkg/scheduler/core/generic_scheduler.gofunc (g *genericScheduler) selectVictimsOnNode(  ctx context.Context,  state *framework.CycleState,  pod *v1.Pod,  meta predicates.Metadata,  nodeInfo *schedulernodeinfo.NodeInfo,  pdbs []*policy.PodDisruptionBudget,) ([]*v1.Pod, int, bool) {  var potentialVictims []*v1.Pod
  removePod := func(rp *v1.Pod) error {    if err := nodeInfo.RemovePod(rp); err != nil {      return err    }    if meta != nil {      if err := meta.RemovePod(rp, nodeInfo.Node()); err != nil {        return err      }    }    status := g.framework.RunPreFilterExtensionRemovePod(ctx, state, pod, rp, nodeInfo)    if !status.IsSuccess() {      return status.AsError()    }    return nil  }  addPod := func(ap *v1.Pod) error {    nodeInfo.AddPod(ap)    if meta != nil {      if err := meta.AddPod(ap, nodeInfo.Node()); err != nil {        return err      }    }    status := g.framework.RunPreFilterExtensionAddPod(ctx, state, pod, ap, nodeInfo)    if !status.IsSuccess() {      return status.AsError()    }    return nil  }  // （1）首先，假设把该node节点上比抢占pod优先级低的所有pod都删除掉，然后调用预选算法，看pod在该node上是否满足调度条件，假如还是不符合调度条件，则该node节点不适合被抢占，直接return  // As the first step, remove all the lower priority pods from the node and  // check if the given pod can be scheduled.  podPriority := podutil.GetPodPriority(pod)  for _, p := range nodeInfo.Pods() {    if podutil.GetPodPriority(p) < podPriority {      potentialVictims = append(potentialVictims, p)      if err := removePod(p); err != nil {        return nil, 0, false      }    }  }  // If the new pod does not fit after removing all the lower priority pods,  // we are almost done and this node is not suitable for preemption. The only  // condition that we could check is if the "pod" is failing to schedule due to  // inter-pod affinity to one or more victims, but we have decided not to  // support this case for performance reasons. Having affinity to lower  // priority pods is not a recommended configuration anyway.  if fits, _, _, err := g.podFitsOnNode(ctx, state, pod, meta, nodeInfo, false); !fits {    if err != nil {      klog.Warningf("Encountered error while selecting victims on node %v: %v", nodeInfo.Node().Name, err)    }
    return nil, 0, false  }  var victims []*v1.Pod  numViolatingVictim := 0  // （2）将所有比抢占pod优先级低的pod按优先级高低进行排序，优先级最低的排在最前面；    sort.Slice(potentialVictims, func(i, j int) bool { return util.MoreImportantPod(potentialVictims[i], potentialVictims[j]) })  // Try to reprieve as many pods as possible. We first try to reprieve the PDB  // violating victims and then other non-violating ones. In both cases, we start  // from the highest priority victims.  // （3）将排好序的pod列表按是否定义了PDB分成两个pod列表；   violatingVictims, nonViolatingVictims := filterPodsWithPDBViolation(potentialVictims, pdbs)  reprievePod := func(p *v1.Pod) (bool, error) {    if err := addPod(p); err != nil {      return false, err    }    fits, _, _, _ := g.podFitsOnNode(ctx, state, pod, meta, nodeInfo, false)    if !fits {      if err := removePod(p); err != nil {        return false, err      }      victims = append(victims, p)      klog.V(5).Infof("Pod %v/%v is a potential preemption victim on node %v.", p.Namespace, p.Name, nodeInfo.Node().Name)    }    return fits, nil  }  // （4）先遍历定义了PDB的pod列表，逐一删除pod（被删除的pod称为被抢占pod），每删除一个pod，调度预选算法，看pod在该node上是否满足调度条件，如满足则直接返回该node适合被抢占、被抢占的pod列表、被抢占pod中定义了PDB的pod数量；    for _, p := range violatingVictims {    if fits, err := reprievePod(p); err != nil {      klog.Warningf("Failed to reprieve pod %q: %v", p.Name, err)      return nil, 0, false    } else if !fits {      numViolatingVictim++    }  }  // （5）假如遍历完定义了PDB的pod列表后，抢占pod在该node上任然不满足调度条件，则继续遍历没有定义PDB的pod列表，逐一删除pod，每删除一个pod，调度预选算法，看pod在该node上是否满足调度条件，如满足则直接返回该node适合被抢占、被抢占的pod列表、被抢占pod中定义了PDB的pod数量；    // Now we try to reprieve non-violating victims.  for _, p := range nonViolatingVictims {    if _, err := reprievePod(p); err != nil {      klog.Warningf("Failed to reprieve pod %q: %v", p.Name, err)      return nil, 0, false    }  }  // （6）如果上述两个pod列表里的pod都被删除后，抢占pod在该node上任然不满足调度条件，则该node不适合被抢占，return。   return victims, numViolatingVictim, true}

// pkg/scheduler/core/generic_scheduler.gofunc pickOneNodeForPreemption(nodesToVictims map[*v1.Node]*extenderv1.Victims) *v1.Node {  if len(nodesToVictims) == 0 {    return nil  }  minNumPDBViolatingPods := int64(math.MaxInt32)  var minNodes1 []*v1.Node  lenNodes1 := 0  for node, victims := range nodesToVictims {      // （1）node节点没有被抢占pod的，优先选择    if len(victims.Pods) == 0 {      // We found a node that doesn't need any preemption. Return it!      // This should happen rarely when one or more pods are terminated between      // the time that scheduler tries to schedule the pod and the time that      // preemption logic tries to find nodes for preemption.      return node    }    //（2）与PDB冲突的pod数量最少的节点    numPDBViolatingPods := victims.NumPDBViolations    if numPDBViolatingPods < minNumPDBViolatingPods {      minNumPDBViolatingPods = numPDBViolatingPods      minNodes1 = nil      lenNodes1 = 0    }    if numPDBViolatingPods == minNumPDBViolatingPods {      minNodes1 = append(minNodes1, node)      lenNodes1++    }  }  if lenNodes1 == 1 {    return minNodes1[0]  }        // （3）高优先级pod数量最少的节点  // There are more than one node with minimum number PDB violating pods. Find  // the one with minimum highest priority victim.  minHighestPriority := int32(math.MaxInt32)  var minNodes2 = make([]*v1.Node, lenNodes1)  lenNodes2 := 0  for i := 0; i < lenNodes1; i++ {    node := minNodes1[i]    victims := nodesToVictims[node]    // highestPodPriority is the highest priority among the victims on this node.    highestPodPriority := podutil.GetPodPriority(victims.Pods[0])    if highestPodPriority < minHighestPriority {      minHighestPriority = highestPodPriority      lenNodes2 = 0    }    if highestPodPriority == minHighestPriority {      minNodes2[lenNodes2] = node      lenNodes2++    }  }  if lenNodes2 == 1 {    return minNodes2[0]  }        // （4）对node节点上所有被抢占pod的优先级进行相加，选取其值最小的节点  // There are a few nodes with minimum highest priority victim. Find the  // smallest sum of priorities.  minSumPriorities := int64(math.MaxInt64)  lenNodes1 = 0  for i := 0; i < lenNodes2; i++ {    var sumPriorities int64    node := minNodes2[i]    for _, pod := range nodesToVictims[node].Pods {      // We add MaxInt32+1 to all priorities to make all of them >= 0. This is      // needed so that a node with a few pods with negative priority is not      // picked over a node with a smaller number of pods with the same negative      // priority (and similar scenarios).      sumPriorities += int64(podutil.GetPodPriority(pod)) + int64(math.MaxInt32+1)    }    if sumPriorities < minSumPriorities {      minSumPriorities = sumPriorities      lenNodes1 = 0    }    if sumPriorities == minSumPriorities {      minNodes1[lenNodes1] = node      lenNodes1++    }  }  if lenNodes1 == 1 {    return minNodes1[0]  }        // （5）选择被抢占pod数量最少的node节点；   // There are a few nodes with minimum highest priority victim and sum of priorities.  // Find one with the minimum number of pods.  minNumPods := math.MaxInt32  lenNodes2 = 0  for i := 0; i < lenNodes1; i++ {    node := minNodes1[i]    numPods := len(nodesToVictims[node].Pods)    if numPods < minNumPods {      minNumPods = numPods      lenNodes2 = 0    }    if numPods == minNumPods {      minNodes2[lenNodes2] = node      lenNodes2++    }  }  if lenNodes2 == 1 {    return minNodes2[0]  }        // （6）选择被抢占pod中运行时间最短的pod所在node节点；   // There are a few nodes with same number of pods.  // Find the node that satisfies latest(earliestStartTime(all highest-priority pods on node))  latestStartTime := util.GetEarliestPodStartTime(nodesToVictims[minNodes2[0]])  if latestStartTime == nil {    // If the earliest start time of all pods on the 1st node is nil, just return it,    // which is not expected to happen.    klog.Errorf("earliestStartTime is nil for node %s. Should not reach here.", minNodes2[0])    return minNodes2[0]  }  nodeToReturn := minNodes2[0]  for i := 1; i < lenNodes2; i++ {    node := minNodes2[i]    // Get earliest start time of all pods on the current node.    earliestStartTimeOnNode := util.GetEarliestPodStartTime(nodesToVictims[node])    if earliestStartTimeOnNode == nil {      klog.Errorf("earliestStartTime is nil for node %s. Should not reach here.", node)      continue    }    if earliestStartTimeOnNode.After(latestStartTime.Time) {      latestStartTime = earliestStartTimeOnNode      nodeToReturn = node    }  }        // （7）返回符合上述条件的最后一个node节点  return nodeToReturn}