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

1
2
3
4
5
6
7
8
9

// Run begins watching and scheduling. It starts scheduling and blocked until the context is done.
func (sched *Scheduler) Run(ctx context.Context) {
  // 队列相关
  sched.SchedulingQueue.Run()
  // 真正的逻辑
  wait.UntilWithContext(ctx, sched.scheduleOne, 0)
  // 关闭队列
  sched.SchedulingQueue.Close()
}

 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

// scheduleOne does the entire scheduling workflow for a single pod. It is serialized on the scheduling algorithm's host fitting.
func (sched *Scheduler) scheduleOne(ctx context.Context) {
  podInfo := sched.NextPod()
  // pod could be nil when schedulerQueue is closed
  if podInfo == nil || podInfo.Pod == nil {
    return
  }
  pod := podInfo.Pod
  fwk, err := sched.frameworkForPod(pod)
  if err != nil {
    // This shouldn't happen, because we only accept for scheduling the pods
    // which specify a scheduler name that matches one of the profiles.
    klog.ErrorS(err, "Error occurred")
    return
  }
  if sched.skipPodSchedule(fwk, pod) {
    return
  }

  klog.V(3).InfoS("Attempting to schedule pod", "pod", klog.KObj(pod))

  // Synchronously attempt to find a fit for the pod.
  start := time.Now()
  state := framework.NewCycleState()
  state.SetRecordPluginMetrics(rand.Intn(100) < pluginMetricsSamplePercent)
  // Initialize an empty podsToActivate struct, which will be filled up by plugins or stay empty.
  podsToActivate := framework.NewPodsToActivate()
  state.Write(framework.PodsToActivateKey, podsToActivate)
  ...
  schedulingCycleCtx, cancel := context.WithCancel(ctx)
  defer cancel()
  scheduleResult, err := sched.Algorithm.Schedule(schedulingCycleCtx, sched.Extenders, fwk, state, 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

func (g *genericScheduler) Schedule(ctx context.Context, extenders []framework.Extender, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) (result ScheduleResult, err error) {
  trace := utiltrace.New("Scheduling", utiltrace.Field{Key: "namespace", Value: pod.Namespace}, utiltrace.Field{Key: "name", Value: pod.Name})
  defer trace.LogIfLong(100 * time.Millisecond)

  if err := g.snapshot(); err != nil {
    return result, err
  }
  trace.Step("Snapshotting scheduler cache and node infos done")

  if g.nodeInfoSnapshot.NumNodes() == 0 {
    return result, ErrNoNodesAvailable
  }

  feasibleNodes, diagnosis, err := g.findNodesThatFitPod(ctx, extenders, fwk, state, pod)
  if err != nil {
    return result, err
  }
  trace.Step("Computing predicates done")

  if len(feasibleNodes) == 0 {
    return result, &framework.FitError{
      Pod:         pod,
      NumAllNodes: g.nodeInfoSnapshot.NumNodes(),
      Diagnosis:   diagnosis,
    }
  }

  // When only one node after predicate, just use it.
  if len(feasibleNodes) == 1 {
    return ScheduleResult{
      SuggestedHost:  feasibleNodes[0].Name,
      EvaluatedNodes: 1 + len(diagnosis.NodeToStatusMap),
      FeasibleNodes:  1,
    }, nil
  }

  priorityList, err := prioritizeNodes(ctx, extenders, fwk, state, pod, feasibleNodes)
  if err != nil {
    return result, err
  }

  host, err := g.selectHost(priorityList)
  trace.Step("Prioritizing done")

  return ScheduleResult{
    SuggestedHost:  host,
    EvaluatedNodes: len(feasibleNodes) + len(diagnosis.NodeToStatusMap),
    FeasibleNodes:  len(feasibleNodes),
  }, err
}

  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
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112

func (g *genericScheduler) findNodesThatFitPod(ctx context.Context, extenders []framework.Extender, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) ([]*v1.Node, framework.Diagnosis, error) {
  diagnosis := framework.Diagnosis{
    NodeToStatusMap:      make(framework.NodeToStatusMap),
    UnschedulablePlugins: sets.NewString(),
  }

  // Run "prefilter" plugins.
  s := fwk.RunPreFilterPlugins(ctx, state, pod)
  allNodes, err := g.nodeInfoSnapshot.NodeInfos().List()
  if err != nil {
    return nil, diagnosis, err
  }
  ...
  // "NominatedNodeName" can potentially be set in a previous scheduling cycle as a result of preemption.
  // This node is likely the only candidate that will fit the pod, and hence we try it first before iterating over all nodes.
  if len(pod.Status.NominatedNodeName) > 0 && feature.DefaultFeatureGate.Enabled(features.PreferNominatedNode) {
    feasibleNodes, err := g.evaluateNominatedNode(ctx, extenders, pod, fwk, state, diagnosis)
    if err != nil {
      klog.ErrorS(err, "Evaluation failed on nominated node", "pod", klog.KObj(pod), "node", pod.Status.NominatedNodeName)
    }
    // Nominated node passes all the filters, scheduler is good to assign this node to the pod.
    if len(feasibleNodes) != 0 {
      return feasibleNodes, diagnosis, nil
    }
  }
  ...
  feasibleNodes, err := g.findNodesThatPassFilters(ctx, fwk, state, pod, diagnosis, allNodes)
  if err != nil {
    return nil, diagnosis, err
  }

  feasibleNodes, err = findNodesThatPassExtenders(extenders, pod, feasibleNodes, diagnosis.NodeToStatusMap)
  if err != nil {
    return nil, diagnosis, err
  }
  return feasibleNodes, diagnosis, nil
}

// findNodesThatPassFilters finds the nodes that fit the filter plugins.
func (g *genericScheduler) findNodesThatPassFilters(
  ctx context.Context,
  fwk framework.Framework,
  state *framework.CycleState,
  pod *v1.Pod,
  diagnosis framework.Diagnosis,
  nodes []*framework.NodeInfo) ([]*v1.Node, error) {
  numNodesToFind := g.numFeasibleNodesToFind(int32(len(nodes)))

  // Create feasible list with enough space to avoid growing it
  // and allow assigning.
  feasibleNodes := make([]*v1.Node, numNodesToFind)

  if !fwk.HasFilterPlugins() {
    length := len(nodes)
    for i := range feasibleNodes {
      feasibleNodes[i] = nodes[(g.nextStartNodeIndex+i)%length].Node()
    }
    g.nextStartNodeIndex = (g.nextStartNodeIndex + len(feasibleNodes)) % length
    return feasibleNodes, nil
  }

  errCh := parallelize.NewErrorChannel()
  var statusesLock sync.Mutex
  var feasibleNodesLen int32
  ctx, cancel := context.WithCancel(ctx)
  checkNode := func(i int) {
    // We check the nodes starting from where we left off in the previous scheduling cycle,
    // this is to make sure all nodes have the same chance of being examined across pods.
    nodeInfo := nodes[(g.nextStartNodeIndex+i)%len(nodes)]
    status := fwk.RunFilterPluginsWithNominatedPods(ctx, state, pod, nodeInfo)
    if status.Code() == framework.Error {
      errCh.SendErrorWithCancel(status.AsError(), cancel)
      return
    }
    if status.IsSuccess() {
      length := atomic.AddInt32(&feasibleNodesLen, 1)
      if length > numNodesToFind {
        cancel()
        atomic.AddInt32(&feasibleNodesLen, -1)
      } else {
        feasibleNodes[length-1] = nodeInfo.Node()
      }
    } else {
      statusesLock.Lock()
      diagnosis.NodeToStatusMap[nodeInfo.Node().Name] = status
      diagnosis.UnschedulablePlugins.Insert(status.FailedPlugin())
      statusesLock.Unlock()
    }
  }

  beginCheckNode := time.Now()
  statusCode := framework.Success
  defer func() {
    // We record Filter extension point latency here instead of in framework.go because framework.RunFilterPlugins
    // function is called for each node, whereas we want to have an overall latency for all nodes per scheduling cycle.
    // Note that this latency also includes latency for `addNominatedPods`, which calls framework.RunPreFilterAddPod.
    metrics.FrameworkExtensionPointDuration.WithLabelValues(runtime.Filter, statusCode.String(), fwk.ProfileName()).Observe(metrics.SinceInSeconds(beginCheckNode))
  }()

  // Stops searching for more nodes once the configured number of feasible nodes
  // are found.
  fwk.Parallelizer().Until(ctx, len(nodes), checkNode)
  processedNodes := int(feasibleNodesLen) + len(diagnosis.NodeToStatusMap)
  g.nextStartNodeIndex = (g.nextStartNodeIndex + processedNodes) % len(nodes)

  feasibleNodes = feasibleNodes[:feasibleNodesLen]
  if err := errCh.ReceiveError(); err != nil {
    statusCode = framework.Error
    return nil, err
  }
  return feasibleNodes, nil
}

 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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

func prioritizeNodes(
  ctx context.Context,
  extenders []framework.Extender,
  fwk framework.Framework,
  state *framework.CycleState,
  pod *v1.Pod,
  nodes []*v1.Node,
) (framework.NodeScoreList, error) {
  // If no priority configs are provided, then all nodes will have a score of one.
  // This is required to generate the priority list in the required format
  if len(extenders) == 0 && !fwk.HasScorePlugins() {
    result := make(framework.NodeScoreList, 0, len(nodes))
    for i := range nodes {
      result = append(result, framework.NodeScore{
        Name:  nodes[i].Name,
        Score: 1,
      })
    }
    return result, nil
  }

  // Run PreScore plugins.
  preScoreStatus := fwk.RunPreScorePlugins(ctx, state, pod, nodes)
  if !preScoreStatus.IsSuccess() {
    return nil, preScoreStatus.AsError()
  }

  // Run the Score plugins.
  scoresMap, scoreStatus := fwk.RunScorePlugins(ctx, state, pod, nodes)
  if !scoreStatus.IsSuccess() {
    return nil, scoreStatus.AsError()
  }

  if klog.V(10).Enabled() {
    for plugin, nodeScoreList := range scoresMap {
      for _, nodeScore := range nodeScoreList {
        klog.InfoS("Plugin scored node for pod", "pod", klog.KObj(pod), "plugin", plugin, "node", nodeScore.Name, "score", nodeScore.Score)
      }
    }
  }

  // Summarize all scores.
  result := make(framework.NodeScoreList, 0, len(nodes))

  for i := range nodes {
    result = append(result, framework.NodeScore{Name: nodes[i].Name, Score: 0})
    for j := range scoresMap {
      result[i].Score += scoresMap[j][i].Score
    }
  }

  if len(extenders) != 0 && nodes != nil {
    var mu sync.Mutex
    var wg sync.WaitGroup
    combinedScores := make(map[string]int64, len(nodes))
    for i := range extenders {
      if !extenders[i].IsInterested(pod) {
        continue
      }
      wg.Add(1)
      go func(extIndex int) {
        metrics.SchedulerGoroutines.WithLabelValues(metrics.PrioritizingExtender).Inc()
        defer func() {
          metrics.SchedulerGoroutines.WithLabelValues(metrics.PrioritizingExtender).Dec()
          wg.Done()
        }()
        prioritizedList, weight, err := extenders[extIndex].Prioritize(pod, nodes)
        if err != nil {
          // Prioritization errors from extender can be ignored, let k8s/other extenders determine the priorities
          return
        }
        mu.Lock()
        for i := range *prioritizedList {
          host, score := (*prioritizedList)[i].Host, (*prioritizedList)[i].Score
          if klog.V(10).Enabled() {
            klog.InfoS("Extender scored node for pod", "pod", klog.KObj(pod), "extender", extenders[extIndex].Name(), "node", host, "score", score)
          }
          combinedScores[host] += score * weight
        }
        mu.Unlock()
      }(i)
    }
    // wait for all go routines to finish
    wg.Wait()
    for i := range result {
      // MaxExtenderPriority may diverge from the max priority used in the scheduler and defined by MaxNodeScore,
      // therefore we need to scale the score returned by extenders to the score range used by the scheduler.
      result[i].Score += combinedScores[result[i].Name] * (framework.MaxNodeScore / extenderv1.MaxExtenderPriority)
    }
  }

  if klog.V(10).Enabled() {
    for i := range result {
      klog.InfoS("Calculated node's final score for pod", "pod", klog.KObj(pod), "node", result[i].Name, "score", result[i].Score)
    }
  }
  return result, nil
}