kubernetes源码-kube-scheduler 原理和源码分析（三）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

  // Tell the cache to assume that a pod now is running on a given node, even though it hasn't been bound yet.
  // This allows us to keep scheduling without waiting on binding to occur.
  assumedPodInfo := podInfo.DeepCopy()
  assumedPod := assumedPodInfo.Pod
  // assume modifies `assumedPod` by setting NodeName=scheduleResult.SuggestedHost
  err = sched.assume(assumedPod, scheduleResult.SuggestedHost)
  if err != nil {
    metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
    // This is most probably result of a BUG in retrying logic.
    // We report an error here so that pod scheduling can be retried.
    // This relies on the fact that Error will check if the pod has been bound
    // to a node and if so will not add it back to the unscheduled pods queue
    // (otherwise this would cause an infinite loop).
    sched.recordSchedulingFailure(fwk, assumedPodInfo, err, SchedulerError, "")
    return
  }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

  // Run the Reserve method of reserve plugins.
  if sts := fwk.RunReservePluginsReserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
    metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
    // trigger un-reserve to clean up state associated with the reserved Pod
    fwk.RunReservePluginsUnreserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
    if forgetErr := sched.SchedulerCache.ForgetPod(assumedPod); forgetErr != nil {
      klog.ErrorS(forgetErr, "scheduler cache ForgetPod failed")
    }
    sched.recordSchedulingFailure(fwk, assumedPodInfo, sts.AsError(), SchedulerError, "")
    return
  }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

  // Run "permit" plugins.
  runPermitStatus := fwk.RunPermitPlugins(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
  if runPermitStatus.Code() != framework.Wait && !runPermitStatus.IsSuccess() {
    var reason string
    if runPermitStatus.IsUnschedulable() {
      metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
      reason = v1.PodReasonUnschedulable
    } else {
      metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
      reason = SchedulerError
    }
    // One of the plugins returned status different than success or wait.
    fwk.RunReservePluginsUnreserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
    if forgetErr := sched.SchedulerCache.ForgetPod(assumedPod); forgetErr != nil {
      klog.ErrorS(forgetErr, "scheduler cache ForgetPod failed")
    }
    sched.recordSchedulingFailure(fwk, assumedPodInfo, runPermitStatus.AsError(), reason, "")
    return
  }

  // At the end of a successful scheduling cycle, pop and move up Pods if needed.
  if len(podsToActivate.Map) != 0 {
    sched.SchedulingQueue.Activate(podsToActivate.Map)
    // Clear the entries after activation.
    podsToActivate.Map = make(map[string]*v1.Pod)
  }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

  // bind the pod to its host asynchronously (we can do this b/c of the assumption step above).
  go func() {
    bindingCycleCtx, cancel := context.WithCancel(ctx)
    defer cancel()
    metrics.SchedulerGoroutines.WithLabelValues(metrics.Binding).Inc()
    defer metrics.SchedulerGoroutines.WithLabelValues(metrics.Binding).Dec()

    waitOnPermitStatus := fwk.WaitOnPermit(bindingCycleCtx, assumedPod)
    if !waitOnPermitStatus.IsSuccess() {
      var reason string
      if waitOnPermitStatus.IsUnschedulable() {
        metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
        reason = v1.PodReasonUnschedulable
      } else {
        metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
        reason = SchedulerError
      }
      // trigger un-reserve plugins to clean up state associated with the reserved Pod
      fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
      if forgetErr := sched.SchedulerCache.ForgetPod(assumedPod); forgetErr != nil {
        klog.ErrorS(forgetErr, "scheduler cache ForgetPod failed")
      }
      sched.recordSchedulingFailure(fwk, assumedPodInfo, waitOnPermitStatus.AsError(), reason, "")
      return
    }

    // Run "prebind" plugins.
    preBindStatus := fwk.RunPreBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
    if !preBindStatus.IsSuccess() {
      metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
      // trigger un-reserve plugins to clean up state associated with the reserved Pod
      fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
      if forgetErr := sched.SchedulerCache.ForgetPod(assumedPod); forgetErr != nil {
        klog.ErrorS(forgetErr, "scheduler cache ForgetPod failed")
      }
      sched.recordSchedulingFailure(fwk, assumedPodInfo, preBindStatus.AsError(), SchedulerError, "")
      return
    }

    err := sched.bind(bindingCycleCtx, fwk, assumedPod, scheduleResult.SuggestedHost, state)
    if err != nil {
      metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
      // trigger un-reserve plugins to clean up state associated with the reserved Pod
      fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
      if err := sched.SchedulerCache.ForgetPod(assumedPod); err != nil {
        klog.ErrorS(err, "scheduler cache ForgetPod failed")
      }
      sched.recordSchedulingFailure(fwk, assumedPodInfo, fmt.Errorf("binding rejected: %w", err), SchedulerError, "")
    } else {
      // Calculating nodeResourceString can be heavy. Avoid it if klog verbosity is below 2.
      if klog.V(2).Enabled() {
        klog.InfoS("Successfully bound pod to node", "pod", klog.KObj(pod), "node", scheduleResult.SuggestedHost, "evaluatedNodes", scheduleResult.EvaluatedNodes, "feasibleNodes", scheduleResult.FeasibleNodes)
      }
      metrics.PodScheduled(fwk.ProfileName(), metrics.SinceInSeconds(start))
      metrics.PodSchedulingAttempts.Observe(float64(podInfo.Attempts))
      metrics.PodSchedulingDuration.WithLabelValues(getAttemptsLabel(podInfo)).Observe(metrics.SinceInSeconds(podInfo.InitialAttemptTimestamp))

      // Run "postbind" plugins.
      fwk.RunPostBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)

      // At the end of a successful binding cycle, move up Pods if needed.
      if len(podsToActivate.Map) != 0 {
        sched.SchedulingQueue.Activate(podsToActivate.Map)
        // Unlike the logic in scheduling cycle, we don't bother deleting the entries
        // as `podsToActivate.Map` is no longer consumed.
      }
    }
  }()