9.深入k8s：排程器及其原始碼分析
阿新 • • 發佈：2020-09-05
> 轉載請宣告出處哦~，本篇文章釋出於luozhiyun的部落格：https://www.luozhiyun.com
>
> 原始碼版本是[1.19](https://github.com/kubernetes/kubernetes/tree/release-1.19)
這次講解的是k8s的排程器部分的程式碼，相對來說比較複雜，慢慢的梳理清楚邏輯花費了不少的時間，不過在梳理過程中也對k8s有了一個更深刻的理解。
## 排程的邏輯介紹
排程器的主要職責，就是為一個新創建出來的 Pod，尋找一個最合適的節點（Node）。[kube-scheduler](https://kubernetes.io/docs/reference/command-line-tools-reference/kube-scheduler/) 就是 Kubernetes 叢集的預設排程器。
預設排程器會首先呼叫一組Filter過濾器，也就是使用相應的Predicates的排程演算法來進行過濾。然後，再呼叫一組叫作 Priority 的排程演算法，來給上一步得到的結果裡的每個 Node 打分，然後根據打分來對Node進行排序，找出最優節點，如果多個節點都有最高的優先順序分數，那麼則迴圈分配，確保平均分配給pod。
排程演算法執行完成後，排程器就需要將 Pod 物件的 nodeName 欄位的值，修改為上述 Node 的名字。
Filter過濾器的作用主要是從當前叢集的所有節點中，“過濾”出一系列符合條件的節點，有如下幾種排程策略：
1. GeneralPredicates
這一組過濾規則，負責的是最基礎的排程策略。比如，計算宿主機的 CPU 和記憶體資源等是否夠用； ，等等。
2. Volume過濾規則
這一組過濾規則，負責的是跟容器持久化 Volume 相關的排程策略。如：檢查多個 Pod 宣告掛載的持久化 Volume 是否有衝突；檢查一個節點上某種型別的持久化 Volume 是不是已經超過了一定數目；檢查Pod 對應的 PV 的 nodeAffinity 欄位，是否跟某個節點的標籤相匹配等等。
3. 檢查排程 Pod 是否滿足 Node 本身的某些條件
如PodToleratesNodeTaints負責檢查的就是我們前面經常用到的 Node 的“汙點”機制。NodeMemoryPressurePredicate，檢查的是當前節點的記憶體是不是已經不夠充足。
4. 檢查親密與反親密關係
檢查待排程 Pod 與 Node 上的已有 Pod 之間的親密（affinity）和反親密（anti-affinity）關係。
在呼叫Filter過濾器的時候需要關注整個叢集的資訊，Kubernetes 排程器會在為每個待排程 Pod 執行該排程演算法之前，先將演算法需要的叢集資訊初步計算一遍，然後快取起來。這樣也可以加快執行速度。
而Priorities裡的打分規則包含如：空閒資源（CPU 和 Memory）多的宿主機可以得高權重；CPU和Memory使用都比較均衡則可以得高權重；為了避免這個演算法引發排程堆疊如果大映象分佈的節點數目很少，那麼這些節點的權重就會被調低等。
整個的流程圖如下：
## 原始碼分析
整個排程過程如流程圖：
### 例項化Scheduler物件
程式碼路徑：pkg/scheduler/scheduler.go
Scheduler物件是執行kube-scheduler元件的主物件，所以kube-scheduler會在執行的時候建立一個scheduler物件：
```go
sched, err := scheduler.New(...)
```
呼叫的scheduler的New方法，這個方法會例項化scheduler物件並返回。
在建立scheduler例項的時候會根據Schedule rAlgorithm Source來例項化排程演算法函式：
程式碼路徑：pkg/scheduler/apis/config/types.go
```go
type SchedulerAlgorithmSource struct {
Policy *SchedulerPolicySource
Provider *string
}
```
Policy是通過引數--policy-config-file引數指定排程策略檔案來定義策略。
Providre是通用排程器，是kube-scheduler預設排程方式。
然後會根據設定的策略來建立不同的scheduler：
```go
func New(...) (*Scheduler, error) {
...
case source.Provider != nil:
sc, err := configurator.createFromProvider(*source.Provider)
...
case source.Policy != nil:
...
sc, err := configurator.createFromConfig(*policy)
...
}
```
createFromProvider方法裡面設定好Filter和Score，也就是過濾策略和打分策略：
程式碼路徑：pkg/scheduler/algorithmprovider/registry.go
```go
func getDefaultConfig() *schedulerapi.Plugins {
return &schedulerapi.Plugins{
...
Filter: &schedulerapi.PluginSet{
Enabled: []schedulerapi.Plugin{
{Name: nodeunschedulable.Name},
{Name: noderesources.FitName},
{Name: nodename.Name},
{Name: nodeports.Name},
{Name: nodeaffinity.Name},
{Name: volumerestrictions.Name},
{Name: tainttoleration.Name},
{Name: nodevolumelimits.EBSName},
{Name: nodevolumelimits.GCEPDName},
{Name: nodevolumelimits.CSIName},
{Name: nodevolumelimits.AzureDiskName},
{Name: volumebinding.Name},
{Name: volumezone.Name},
{Name: podtopologyspread.Name},
{Name: interpodaffinity.Name},
},
},
...
Score: &schedulerapi.PluginSet{
Enabled: []schedulerapi.Plugin{
{Name: noderesources.BalancedAllocationName, Weight: 1},
{Name: imagelocality.Name, Weight: 1},
{Name: interpodaffinity.Name, Weight: 1},
{Name: noderesources.LeastAllocatedName, Weight: 1},
{Name: nodeaffinity.Name, Weight: 1},
{Name: nodepreferavoidpods.Name, Weight: 10000},
// Weight is doubled because:
// - This is a score coming from user preference.
// - It makes its signal comparable to NodeResourcesLeastAllocated.
{Name: podtopologyspread.Name, Weight: 2},
{Name: tainttoleration.Name, Weight: 1},
},
},
...
}
}
```
最後kube-scheduler處理完一系列的邏輯，最後會呼叫到Scheduler的run方法：
```go
func (sched *Scheduler) Run(ctx context.Context) {
if !cache.WaitForCacheSync(ctx.Done(), sched.scheduledPodsHasSynced) {
return
}
sched.SchedulingQueue.Run()
wait.UntilWithContext(ctx, sched.scheduleOne, 0)
sched.SchedulingQueue.Close()
}
```
### 排程主邏輯
sched.scheduleOne會被wait.UntilWithContext定時呼叫，直到ctx.Done()返回true為止。sched.scheduleOne是核心實現，主要做了以下幾件事：
1. 通過sched.NextPod()函式從優先佇列中獲取一個優先順序最高的待排程Pod資源物件，如果沒有獲取到，那麼該方法會阻塞住；
2. 通過sched.Algorithm.Schedule排程函式執行Predicates的排程演算法與Priorities演算法，挑選出一個合適的節點；
3. 當沒有找到合適的節點時，排程器會嘗試呼叫prof.RunPostFilterPlugins搶佔低優先順序的Pod資源物件的節點；
4. 當排程器為Pod資源物件選擇了一個合適的節點時，通過sched.bind函式將合適的節點與Pod資源物件繫結在一起；
下面我們直接看一下sched.Algorithm.Schedule方法的實現：
程式碼路徑：pkg/scheduler/core/generic_scheduler.go
```go
//將pod排程到某一node上，如果成功則返回node的名稱，如果成功則返回失敗資訊
func (g *genericScheduler) Schedule(ctx context.Context, prof *profile.Profile, 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)
//檢查pod上宣告的pvc，包括pvc是否存在，是否已被刪除等
if err := podPassesBasicChecks(pod, g.pvcLister); err != nil {
return result, err
}
trace.Step("Basic checks done")
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
}
startPredicateEvalTime := time.Now()
//這裡是Predicates部分的邏輯，負責選出一系列符合條件的節點
feasibleNodes, filteredNodesStatuses, err := g.findNodesThatFitPod(ctx, prof, state, pod)
if err != nil {
return result, err
}
trace.Step("Computing predicates done")
//表示沒有 找到合適的節點
if len(feasibleNodes) == 0 {
return result, &FitError{
Pod: pod,
NumAllNodes: g.nodeInfoSnapshot.NumNodes(),
FilteredNodesStatuses: filteredNodesStatuses,
}
}
metrics.DeprecatedSchedulingAlgorithmPredicateEvaluationSecondsDuration.Observe(metrics.SinceInSeconds(startPredicateEvalTime))
metrics.DeprecatedSchedulingDuration.WithLabelValues(metrics.PredicateEvaluation).Observe(metrics.SinceInSeconds(startPredicateEvalTime))
startPriorityEvalTime := time.Now()
// When only one node after predicate, just use it.
//找到唯一的node節點，並返回
if len(feasibleNodes) == 1 {
metrics.DeprecatedSchedulingAlgorithmPriorityEvaluationSecondsDuration.Observe(metrics.SinceInSeconds(startPriorityEvalTime))
return ScheduleResult{
SuggestedHost: feasibleNodes[0].Name,
EvaluatedNodes: 1 + len(filteredNodesStatuses),
FeasibleNodes: 1,
}, nil
}
//如果節點不是唯一，那麼需要進行打分排序
priorityList, err := g.prioritizeNodes(ctx, prof, state, pod, feasibleNodes)
if err != nil {
return result, err
}
metrics.DeprecatedSchedulingAlgorithmPriorityEvaluationSecondsDuration.Observe(metrics.SinceInSeconds(startPriorityEvalTime))
metrics.DeprecatedSchedulingDuration.WithLabelValues(metrics.PriorityEvaluation).Observe(metrics.SinceInSeconds(startPriorityEvalTime))
//選擇最佳的節點
host, err := g.selectHost(priorityList)
trace.Step("Prioritizing done")
return ScheduleResult{
SuggestedHost: host,
EvaluatedNodes: len(feasibleNodes) + len(filteredNodesStatuses),
FeasibleNodes: len(feasibleNodes),
}, err
}
```
這個方法邏輯還是比較清晰的，總共分為如下幾部分：
1. 對pod進行校驗，檢查是否聲明瞭pvc，以及對應的pvc是否已經被刪除等；
2. 呼叫findNodesThatFitPod方法，負責選出一系列符合條件的節點；
3. 如果沒有找到節點或唯一節點，那麼直接返回；
4. 如果找到的節點數超過1，那麼需要呼叫prioritizeNodes方法，進行打分排序；
5. 最後呼叫selectHost選出合適的唯一節點，並返回。
### Filter過濾篩選節點
下面我們看看findNodesThatFitPod時如何實現篩選過濾的。
程式碼位置：pkg/scheduler/core/generic_scheduler.go
```go
func (g *genericScheduler) findNodesThatFitPod(ctx context.Context, prof *profile.Profile, state *framework.CycleState, pod *v1.Pod) ([]*v1.Node, framework.NodeToStatusMap, error) {
filteredNodesStatuses := make(framework.NodeToStatusMap)
//前置過濾外掛用於預處理 Pod 的相關資訊，或者檢查叢集或 Pod 必須滿足的某些條件。
//如果 PreFilter 外掛返回錯誤，則排程週期將終止
s := prof.RunPreFilterPlugins(ctx, state, pod)
if !s.IsSuccess() {
if !s.IsUnschedulable() {
return nil, nil, s.AsError()
}
// All nodes will have the same status. Some non trivial refactoring is
// needed to avoid this copy.
allNodes, err := g.nodeInfoSnapshot.NodeInfos().List()
if err != nil {
return nil, nil, err
}
for _, n := range allNodes {
filteredNodesStatuses[n.Node().Name] = s
}
return nil, filteredNodesStatuses, nil
}
//過濾掉不符合條件的node
feasibleNodes, err := g.findNodesThatPassFilters(ctx, prof, state, pod, filteredNodesStatuses)
if err != nil {
return nil, nil, err
}
//SchdulerExtender是kubernets外部擴充套件方式，使用者可以根據需求獨立構建排程服務
feasibleNodes, err = g.findNodesThatPassExtenders(pod, feasibleNodes, filteredNodesStatuses)
if err != nil {
return nil, nil, err
}
return feasibleNodes, filteredNodesStatuses, nil
}
```
這個方法首先會通過前置過濾器來校驗pod是否符合條件，然後呼叫findNodesThatPassFilters方法過濾掉不符合條件的node。findNodesThatPassExtenders是kubernets留給使用者的外部擴充套件方式，暫且不表。
下面我們接著看findNodesThatPassFilters方法：
```go
func (g *genericScheduler) findNodesThatPassFilters(ctx context.Context, prof *profile.Profile, state *framework.CycleState, pod *v1.Pod, statuses framework.NodeToStatusMap) ([]*v1.Node, error) {
allNodes, err := g.nodeInfoSnapshot.NodeInfos().List()
if err != nil {
return nil, err
}
//根據叢集節點數量選擇參與排程的節點的數量
numNodesToFind := g.numFeasibleNodesToFind(int32(len(allNodes)))
//初始化一個大小和numNodesToFind一樣的陣列，用來存放node節點
feasibleNodes := make([]*v1.Node, numNodesToFind)
...
checkNode := func(i int) {
//我們從上一個排程週期中離開的節點開始檢查節點，以確保所有節點在Pod中被檢查的機會相同。
nodeInfo := allNodes[(g.nextStartNodeIndex+i)%len(allNodes)]
fits, status, err := PodPassesFiltersOnNode(ctx, prof.PreemptHandle(), state, pod, nodeInfo)
if err != nil {
errCh.SendErrorWithCancel(err, cancel)
return
}
//如果該節點合適，那麼放入到feasibleNodes列表中
if fits {
length := atomic.AddInt32(&feasibleNodesLen, 1)
if length > numNodesToFind {
cancel()
atomic.AddInt32(&feasibleNodesLen, -1)
} else {
feasibleNodes[length-1] = nodeInfo.Node()
}
} else {
statusesLock.Lock()
if !status.IsSuccess() {
statuses[nodeInfo.Node().Name] = status
}
statusesLock.Unlock()
}
}
...
//開啟16個執行緒尋找符合條件的node節點，數量等於feasibleNodes
parallelize.Until(ctx, len(allNodes), checkNode)
processedNodes := int(feasibleNodesLen) + len(statuses)
//設定下次開始尋找node的位置
g.nextStartNodeIndex = (g.nextStartNodeIndex + processedNodes) % len(allNodes)
feasibleNodes = feasibleNodes[:feasibleNodesLen]
if err := errCh.ReceiveError(); err != nil {
statusCode = framework.Error
return nil, err
}
return feasibleNodes, nil
}
```
在這個方法中首先會根據numFeasibleNodesToFind方法選擇參與排程的節點的數量，然後呼叫parallelize.Until方法開啟16個執行緒來呼叫checkNode方法尋找合適的節點。
對於numFeasibleNodesToFind的邏輯如下：
```go
func (g *genericScheduler) numFeasibleNodesToFind(numAllNodes int32) (numNodes int32) {
//對於一個小於100的節點，全部節點參與排程
//percentageOfNodesToScore引數值是一個叢集中所有節點的百分比，範圍是1和100之間，0表示不啟用
if numAllNodes < minFeasibleNodesToFind || g.percentageOfNodesToScore >= 100 {
return numAllNodes
}
adaptivePercentage := g.percentageOfNodesToScore
//當numAllNodes大於100時，如果沒有設定percentageOfNodesToScore，那麼這裡需要計算出一個值
if adaptivePercentage <= 0 {
basePercentageOfNodesToScore := int32(50)
adaptivePercentage = basePercentageOfNodesToScore - numAllNodes/125
if adaptivePercentage < minFeasibleNodesPercentageToFind {
adaptivePercentage = minFeasibleNodesPercentageToFind
}
}
numNodes = numAllNodes * adaptivePercentage / 100
if numNodes < minFeasibleNodesToFind {
return minFeasibleNodesToFind
}
return numNodes
}
```
找出能夠進行排程的節點，如果節點小於100，那麼全部節點參與排程。
percentageOfNodesToScore引數值是一個叢集中所有節點的百分比，範圍是1和100之間，0表示不啟用。如果叢集節點數大於100，那麼就會根據這個值來計算讓合適的節點數參與排程。
如果一個5000個節點的叢集，percentageOfNodesToScore會預設設定為10%，也就是500個節點參與排程。
因為如果一個5000節點的叢集來進行排程的話，不進行控制時，每個pod排程都需要嘗試5000次的節點預選過程時非常消耗資源的。
然後我們回到findNodesThatPassFilters方法中，我們看一下PodPassesFiltersOnNode是如何篩選出合適的節點的：
```go
func PodPassesFiltersOnNode(
ctx context.Context,
ph framework.PreemptHandle,
state *framework.CycleState,
pod *v1.Pod,
info *framework.NodeInfo,
) (bool, *framework.Status, error) {
var status *framework.Status
podsAdded := false
//待檢查的 Node 是一個即將被搶佔的節點，排程器就會對這個 Node ，將同樣的 Predicates 演算法執行兩遍。
for i := 0; i < 2; i++ {
stateToUse := state
nodeInfoToUse := info
//處理優先順序pod的邏輯
if i == 0 {
var err error
//查詢是否有優先順序大於或等於當前pod的NominatedPods，然後加入到nodeInfoToUse中
podsAdded, stateToUse, nodeInfoToUse, err = addNominatedPods(ctx, ph, pod, state, info)
if err != nil {
return false, nil, err
}
} else if !podsAdded || !status.IsSuccess() {
break
}
//執行過濾器檢查該pod是否能執行在該節點上
statusMap := ph.RunFilterPlugins(ctx, stateToUse, pod, nodeInfoToUse)
status = statusMap.Merge()
if !status.IsSuccess() && !status.IsUnschedulable() {
return false, status, status.AsError()
}
}
return status.IsSuccess(), status, nil
}
```
這個方法用來檢測node是否能通過過濾器，此方法會在排程Schedule和搶佔Preempt的時被呼叫，如果在Schedule時被呼叫，那麼會測試nod，能否可以讓所有存在的pod以及更高優先順序的pod在該node上執行。如果在搶佔時被呼叫，那麼我們首先要移除搶佔失敗的pod，新增將要搶佔的pod。
然後RunFilterPlugins會呼叫runFilterPlugin方法來執行我們上面講的getDefaultConfig中設定的過濾器：
```go
func (f *frameworkImpl) runFilterPlugin(ctx context.Context, pl framework.FilterPlugin, state *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
if !state.ShouldRecordPluginMetrics() {
return pl.Filter(ctx, state, pod, nodeInfo)
}
startTime := time.Now()
status := pl.Filter(ctx, state, pod, nodeInfo)
f.metricsRecorder.observePluginDurationAsync(Filter, pl.Name(), status, metrics.SinceInSeconds(startTime))
return status
}
```
過濾器總共有這些：nodeunschedulable,noderesources,nodename,nodeports,nodeaffinity,volumerestrictions,tainttoleration,nodevolumelimits,nodevolumelimits,nodevolumelimits,nodevolumelimits,volumebinding,volumezone,podtopologyspread,interpodaffinity
過濾器太多就不一一看了，裡面的邏輯還是很清晰的，感興趣的自己可以看看具體實現。
### prioritize為節點打分
下面我們繼續回到Schedule方法，執行完findNodesThatFitPod後會找到一系列符合條件的node節點，然後會呼叫prioritizeNodes進行打分排序：
```go
func (g *genericScheduler) prioritizeNodes(
ctx context.Context,
prof *profile.Profile,
state *framework.CycleState,
pod *v1.Pod,
nodes []*v1.Node,
) (framework.NodeScoreList, error) {
...
scoresMap, scoreStatus := prof.RunScorePlugins(ctx, state, pod, nodes)
if !scoreStatus.IsSuccess() {
return nil, scoreStatus.AsError()
}
// Summarize all scores.
result := make(framework.NodeScoreList, 0, len(nodes))
//將分數按照node維度進行彙總
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
}
}
...
return result, nil
}
```
prioritizeNodes裡面會呼叫RunScorePlugins方法，裡面會遍歷一系列的外掛的方式為node打分。然後遍歷scoresMap將結果按照node維度進行聚合。
```go
func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodes []*v1.Node) (ps framework.PluginToNodeScores, status *framework.Status) {
...
//開啟16個執行緒為node進行打分
parallelize.Until(ctx, len(nodes), func(index int) {
for _, pl := range f.scorePlugins {
nodeName := nodes[index].Name
s, status := f.runScorePlugin(ctx, pl, state, pod, nodeName)
if !status.IsSuccess() {
errCh.SendErrorWithCancel(fmt.Errorf(status.Message()), cancel)
return
}
pluginToNodeScores[pl.Name()][index] = framework.NodeScore{
Name: nodeName,
Score: int64(s),
}
}
})
if err := errCh.ReceiveError(); err != nil {
msg := fmt.Sprintf("error while running score plugin for pod %q: %v", pod.Name, err)
klog.Error(msg)
return nil, framework.NewStatus(framework.Error, msg)
}
//用於在排程程式計算節點的最終排名之前修改分數,保證 Score 外掛的輸出必須是 [MinNodeScore，MaxNodeScore]（[0-100]） 範圍內的整數
parallelize.Until(ctx, len(f.scorePlugins), func(index int) {
pl := f.scorePlugins[index]
nodeScoreList := pluginToNodeScores[pl.Name()]
if pl.ScoreExtensions() == nil {
return
}
status := f.runScoreExtension(ctx, pl, state, pod, nodeScoreList)
if !status.IsSuccess() {
err := fmt.Errorf("normalize score plugin %q failed with error %v", pl.Name(), status.Message())
errCh.SendErrorWithCancel(err, cancel)
return
}
})
if err := errCh.ReceiveError(); err != nil {
msg := fmt.Sprintf("error while running normalize score plugin for pod %q: %v", pod.Name, err)
klog.Error(msg)
return nil, framework.NewStatus(framework.Error, msg)
}
// 為每個節點的分數乘上一個權重
parallelize.Until(ctx, len(f.scorePlugins), func(index int) {
pl := f.scorePlugins[index]
// Score plugins' weight has been checked when they are initialized.
weight := f.pluginNameToWeightMap[pl.Name()]
nodeScoreList := pluginToNodeScores[pl.Name()]
for i, nodeScore := range nodeScoreList {
// return error if score plugin returns invalid score.
if nodeScore.Score >