圖解kubernetes調度器SchedulingQueue核心源碼實現

時間  2020-01-26
標籤 圖解 kubernetes 調度 schedulingqueue 核心 源碼 實現 简体版
原文   原文鏈接

SchedulingQueue是kubernetes scheduler中負責進行等待調度pod存儲的對，Scheduler經過SchedulingQueue來獲取當前系統中等待調度的Pod，本文主要討論SchedulingQueue的設計與實現的各類實現, 瞭解探究其內部實現與底層源碼,本系列代碼基於kubernets1.1.6分析而來,圖解主要位於第二部分node

SchedulingQueue設計

隊列與優先級

隊列與場景

類型 描述 一般實現
隊列 普通隊列是一個FIFO的數據結構，根據元素入隊的次序依次出隊 數組或者鏈表
優先級隊列 優先級隊列一般是指根據某些優先級策略，高優先級會優先被獲取 數組或者樹

其實在大多數的調度場景中，大多都是採用優先級隊列來實現，優先知足優先級比較高的任務或者需求，從而減小後續高優先級對低優先級的搶佔，scheduler中也是如此算法

優先級的選擇

k8s中調度的單元是Pod，scheduler中根據pod的優先級的高低來進行優先級隊列的構建, 這個實際上是在kubernets的adminission准入插件中，會爲用戶建立的pod根據用戶的設置，進行優先級字段的計算編程

三級隊列

活動隊列

活動隊列存儲當前系統中全部正在等待調度的隊列數組

不可調度隊列

當pod的資源在當前集羣中不能被知足時，則會被加入到一個不可調度隊列中，而後等待稍後再進行嘗試緩存

backoff隊列

backoff機制是併發編程中常見的一種機制，即若是任務反覆執行依舊失敗，則會按次增加等待調度時間，下降重試效率，從而避免反覆失敗浪費調度資源安全

針對調度失敗的pod會優先存儲在backoff隊列中，等待後續重試微信

阻塞與搶佔

阻塞設計

當隊列中不存在等待調度的pod的時候，會阻塞scheduler等待有須要調度的pod的時候再喚醒調度器，獲取pod進行調度數據結構

搶佔相關

nominatedPods存儲pod被提議運行的node,主要用於搶佔調度流程中使用，本節先不分析併發

源碼分析

數據結構

kubernetes中默認的schedulingQueue實現是PriorityQueue，本章就以該數據結構來分析app

type PriorityQueue struct {
	stop  <-chan struct{}
	clock util.Clock
    // 存儲backoff的pod計時器
	podBackoff *PodBackoffMap

	lock sync.RWMutex
    // 用於協調通知由於獲取不到調度pod而阻塞的cond
	cond sync.Cond

    // 活動隊列
	activeQ *util.Heap
	
    // backoff隊列
	podBackoffQ *util.Heap
	
    // 不可調度隊列
	unschedulableQ *UnschedulablePodsMap
	// 存儲pod和被提名的node, 實際上就是存儲pod和建議的node節點
	nominatedPods *nominatedPodMap
	// schedulingCycle是一個調度週期的遞增序號，當pod pop的時候會遞增
	schedulingCycle int64
	// moveRequestCycle緩存schedulingCycle, 當未調度的pod從新被添加到activeQueue中
	// 會保存schedulingCycle到moveRequestCycle中
	moveRequestCycle int64
	closed bool
}

PriorityQueue做爲實現SchedulingQueue的實現，其核心數據結構主要包含三個隊列:activeQ、podBackoffQ、unscheduleQ內部經過cond來實現Pop操做的阻塞與通知，接下來先分析核心的調度流程，最後再分析util.Heap裏面的具體實現

activeQ

存儲全部等待調度的Pod的隊列，默認是基於堆來實現，其中元素的優先級則經過對比pod的建立時間和pod的優先級來進行排序

// activeQ is heap structure that scheduler actively looks at to find pods to
	// schedule. Head of heap is the highest priority pod.
	activeQ *util.Heap

優先級比較函數

// activeQComp is the function used by the activeQ heap algorithm to sort pods.
// It sorts pods based on their priority. When priorities are equal, it uses
// PodInfo.timestamp.
func activeQComp(podInfo1, podInfo2 interface{}) bool {
	pInfo1 := podInfo1.(*framework.PodInfo)
	pInfo2 := podInfo2.(*framework.PodInfo)
	prio1 := util.GetPodPriority(pInfo1.Pod)
	prio2 := util.GetPodPriority(pInfo2.Pod)
    // 首先根據優先級的高低進行比較，而後根據pod的建立時間，越高優先級的Pod越被優先調度
    // 越早建立的pod越優先
	return (prio1 > prio2) || (prio1 == prio2 && pInfo1.Timestamp.Before(pInfo2.Timestamp))
}

podbackOffQ

podBackOffQ主要存儲那些在多個schedulingCycle中依舊調度失敗的狀況下，則會經過以前說的backOff機制，延遲等待調度的時間

// podBackoffQ is a heap ordered by backoff expiry. Pods which have completed backoff
	// are popped from this heap before the scheduler looks at activeQ
	podBackoffQ *util.Heap

podBackOff

上面提到podBackOffQ隊列中並無存儲pod的backOff的具體信息，好比backoff的計數器，最後一次更新的時間等，podBackOff則相似一個記分板，記錄這些信息，供podBackOffQ使用

// podBackoff tracks backoff for pods attempting to be rescheduled
	podBackoff *PodBackoffMap

// PodBackoffMap is a structure that stores backoff related information for pods
type PodBackoffMap struct {
	// lock for performing actions on this PodBackoffMap
	lock sync.RWMutex
	// initial backoff duration
	initialDuration time.Duration // 當前值是1秒
	// maximal backoff duration
	maxDuration time.Duration // 當前值是1分鐘
	// map for pod -> number of attempts for this pod
	podAttempts map[ktypes.NamespacedName]int
	// map for pod -> lastUpdateTime pod of this pod
	podLastUpdateTime map[ktypes.NamespacedName]time.Time
}

unschedulableQ

存儲已經嘗試調度可是當前集羣資源不知足的pod的隊列

moveRequestCycle

當由於集羣資源發生變化會嘗試進行unschedulableQ中的pod轉移到activeQ,moveRequestCycle就是存儲資源變動時的schedulingCycle

func (p *PriorityQueue) MoveAllToActiveQueue() {
    // 省略其餘代碼
    p.moveRequestCycle = p.schedulingCycle
}

schedulingCycle

schedulingCycle是一個遞增的序列每次從activeQ中pop出一個pod都會遞增

func (p *PriorityQueue) Pop() (*v1.Pod, error) {
    //省略其餘
    	p.schedulingCycle++
}

併發活動隊列

併發從活動隊列中獲取pod

image.png SchedulingQueue提供了一個Pop接口用於從獲取當前集羣中等待調度的pod,其內部實現主要經過上面cond與activeQ來實現

當前隊列中沒有可調度的pod的時候，則經過cond.Wait來進行阻塞，而後在忘activeQ中添加pod的時候經過cond.Broadcast來實現通知

func (p *PriorityQueue) Pop() (*v1.Pod, error) {
	p.lock.Lock()
	defer p.lock.Unlock()
	for p.activeQ.Len() == 0 {
		if p.closed {
			return nil, fmt.Errorf(queueClosed)
		}
        // 
		p.cond.Wait()
	}
	obj, err := p.activeQ.Pop()
	if err != nil {
		return nil, err
	}
	pInfo := obj.(*framework.PodInfo)
	p.schedulingCycle++
	return pInfo.Pod, err
}

加入調度pod到活動隊列

image.png 當pod加入活動隊列中，除了加入activeQ的優先級隊列中，還須要從podBackoffQ和unschedulableQ中移除當前的pod,最後進行廣播通知阻塞在Pop操做的scheudler進行最新pod的獲取

func (p *PriorityQueue) Add(pod *v1.Pod) error {
	p.lock.Lock()
	defer p.lock.Unlock()
	pInfo := p.newPodInfo(pod)
	// 加入activeQ
	if err := p.activeQ.Add(pInfo); err != nil {
		klog.Errorf("Error adding pod %v/%v to the scheduling queue: %v", pod.Namespace, pod.Name, err)
		return err
	}
	// 從unschedulableQ刪除
	if p.unschedulableQ.get(pod) != nil {
		klog.Errorf("Error: pod %v/%v is already in the unschedulable queue.", pod.Namespace, pod.Name)
		p.unschedulableQ.delete(pod)
	}
	// Delete pod from backoffQ if it is backing off
	// 從podBackoffQ刪除
	if err := p.podBackoffQ.Delete(pInfo); err == nil {
		klog.Errorf("Error: pod %v/%v is already in the podBackoff queue.", pod.Namespace, pod.Name)
	}
	// 存儲pod和被提名的node
	p.nominatedPods.add(pod, "")
	p.cond.Broadcast()

	return nil
}

schedulingCycle與moveRequestCycle

未調度的隊列的及時重試

image.png 致使調度週期schedulingCyclye變動主要因素以下: 1.當集羣資源發生變化的時候:好比新添加pv、node等資源，那以前在unschedulableQ中由於資源不知足需求的pod就能夠進行放入activeQ中或者podBackoffQ中，及時進行調度 2.pod被成功調度: 以前因爲親和性不知足被放入到unschedulableQ中的pod，此時也能夠進行嘗試，而沒必要等到超時以後，再加入

這兩種狀況下會分別觸發MoveAllToActiveQueue和movePodsToActiveQueue變動moveRequestCycle使其等於schedulingCycle

對重試機制的影響

當前一個pod失敗的時候，有兩種選擇一是加入podBackoffQ中，二是加入unschedulableQ中，那麼針對一個失敗的pod如何選擇該進入那個隊列中呢 image.png 結合上面的moveRequestCycle變動時機，何時moveRequestCycle會大於等於podSchedulingCycle呢？答案就是當前集羣中進行過集羣資源的變動或者pod被成功分配，那這個時候咱們若是重試一個失敗的調度則可能會成功，由於集羣資源變動了可能有新的資源加入

if p.moveRequestCycle >= podSchedulingCycle {
		if err := p.podBackoffQ.Add(pInfo); err != nil {
			return fmt.Errorf("error adding pod %v to the backoff queue: %v", pod.Name, err)
		}
	} else {
		p.unschedulableQ.addOrUpdate(pInfo)
	}

失敗處理邏輯的注入

注入調度失敗邏輯處理

在建立scheduler Config的時候會經過MakeDefaultErrorFunc注入一個失敗處理函數, 在scheduler調度的時候會進行調用 kubernetes/pkg/scheduler/factory/factory.go: MakeDefaultErrorFunc會將沒有調度到任何一個node的pod從新放回到優先級隊列中

podSchedulingCycle := podQueue.SchedulingCycle()
	// 省略非核心代碼
	if len(pod.Spec.NodeName) == 0 {
        //從新放回隊列
        if err := podQueue.AddUnschedulableIfNotPresent(pod, podSchedulingCycle); err != nil {
            klog.Error(err)
        }
    }

失敗處理的回調

當調度pod的失敗的時候, scheduler會同時調用sched.Error就是上面注入的失敗處理邏輯，來將調度失敗未分配node的pod節點從新加入到隊裏鍾 kubernetes/pkg/scheduler/scheduler.go

func (sched *Scheduler) recordSchedulingFailure(pod *v1.Pod, err error, reason string, message string) {
	// 錯誤回調
    sched.Error(pod, err)
	sched.Recorder.Eventf(pod, nil, v1.EventTypeWarning, "FailedScheduling", "Scheduling", message)
	if err := sched.PodConditionUpdater.Update(pod, &v1.PodCondition{
		Type:    v1.PodScheduled,
		Status:  v1.ConditionFalse,
		Reason:  reason,
		Message: err.Error(),
	}); err != nil {
		klog.Errorf("Error updating the condition of the pod %s/%s: %v", pod.Namespace, pod.Name, err)
	}
}

PodBackoffMap

PodBackoffMap主要用於存儲pod的最後一次失敗的更新時間與實現次數，從而根據這些數據來進行pod的backoffTime的計算

數據結構設計

type PodBackoffMap struct {
	// lock for performing actions on this PodBackoffMap
	lock sync.RWMutex
	// 初始化 backoff duration
	initialDuration time.Duration // 當前值是1秒
	// 最大 backoff duration
	maxDuration time.Duration // 當前值是1分鐘
	// 記錄pod重試的次數
	podAttempts map[ktypes.NamespacedName]int
	// 記錄pod的最後一次的更新時間
	podLastUpdateTime map[ktypes.NamespacedName]time.Time
}

backoffTime計算算法

初始化的時候回設定initialDuration和maxDuration，在當前版本中分別是1s和10s,也就是backoffQ中的pod最長10s就會從新加入activeQ中(須要等待定時任務進行輔助)

在每次失敗回調的時候，都會進行BackoffPod方法來進行計數更新，在後續獲取pod的backoffTime的時候，只須要獲取次數而後結合initialDuration進行算法計算，結合pod最後一次的更新時間,就會獲取pod的backoffTime的終止時間 image.png

backoffDuration計算

其實最終的計算很簡單就是2的N次冪

func (pbm *PodBackoffMap) calculateBackoffDuration(nsPod ktypes.NamespacedName) time.Duration {
	// initialDuration是1s
	backoffDuration := pbm.initialDuration
	if _, found := pbm.podAttempts[nsPod]; found {
		// podAttempts裏面包含pod的嘗試失敗的次數
		for i := 1; i < pbm.podAttempts[nsPod]; i++ {
			backoffDuration = backoffDuration * 2
			// 最大10s
			if backoffDuration > pbm.maxDuration {
				return pbm.maxDuration
			}
		}
	}
	return backoffDuration
}

podBackoffQ

image.png

優先級函數

podBackoffQ實際上會根據pod的backoffTime來進行優先級排序，因此podBackoffQ的隊列頭部，就是最近一個要過時的pod

func (p *PriorityQueue) podsCompareBackoffCompleted(podInfo1, podInfo2 interface{}) bool {
	pInfo1 := podInfo1.(*framework.PodInfo)
	pInfo2 := podInfo2.(*framework.PodInfo)
	bo1, _ := p.podBackoff.GetBackoffTime(nsNameForPod(pInfo1.Pod))
	bo2, _ := p.podBackoff.GetBackoffTime(nsNameForPod(pInfo2.Pod))
	return bo1.Before(bo2)
}

調度失敗加入到podBackoffQ

若是調度失敗，而且moveRequestCycle=podSchedulingCycle的時候就加入podBackfoffQ中

func (p *PriorityQueue) AddUnschedulableIfNotPresent(pod *v1.Pod, podSchedulingCycle int64) error {
	// 省略檢查性代碼
	// 更新pod的backoff 信息
	p.backoffPod(pod)

	// moveRequestCycle將pod從unscheduledQ大於pod的調度週期添加到 若是pod的調度週期小於當前的調度週期
	if p.moveRequestCycle >= podSchedulingCycle {
		if err := p.podBackoffQ.Add(pInfo); err != nil {
			return fmt.Errorf("error adding pod %v to the backoff queue: %v", pod.Name, err)
		}
	} else {
		p.unschedulableQ.addOrUpdate(pInfo)
	}

	p.nominatedPods.add(pod, "")
	return nil

}

從unschedulableQ遷移

在前面介紹的當集羣資源發生變動的時候，會觸發嘗試unschedulabelQ中的pod進行轉移，若是發現當前pod還未到達backoffTime,就加入到podBackoffQ中

if p.isPodBackingOff(pod) {
			if err := p.podBackoffQ.Add(pInfo); err != nil {
				klog.Errorf("Error adding pod %v to the backoff queue: %v", pod.Name, err)
				addErrorPods = append(addErrorPods, pInfo)
			}
		} else {
			if err := p.activeQ.Add(pInfo); err != nil {
				klog.Errorf("Error adding pod %v to the scheduling queue: %v", pod.Name, err)
				addErrorPods = append(addErrorPods, pInfo)
			}
		}

podBackoffQ定時轉移

在建立PriorityQueue的時候，會建立兩個定時任務其中一個就是講backoffQ中的pod到期後的轉移，每秒鐘嘗試一次

func (p *PriorityQueue) run() {
	go wait.Until(p.flushBackoffQCompleted, 1.0*time.Second, p.stop)
	go wait.Until(p.flushUnschedulableQLeftover, 30*time.Second, p.stop)
}

由於是一個堆結果，因此只須要獲取堆頂的元素，而後肯定是否到期，若是到期後則進行pop處來，加入到activeQ中

func (p *PriorityQueue) flushBackoffQCompleted() {
	p.lock.Lock()
	defer p.lock.Unlock()

	for {
        // 獲取堆頂元素
		rawPodInfo := p.podBackoffQ.Peek()
		if rawPodInfo == nil {
			return
		}
		pod := rawPodInfo.(*framework.PodInfo).Pod
        // 獲取到期時間
		boTime, found := p.podBackoff.GetBackoffTime(nsNameForPod(pod))
		if !found {
			// 若是當前已經不在podBackoff中，則就pop出來而後放入到activeQ
			klog.Errorf("Unable to find backoff value for pod %v in backoffQ", nsNameForPod(pod))
			p.podBackoffQ.Pop()
			p.activeQ.Add(rawPodInfo)
			defer p.cond.Broadcast()
			continue
		}

		// 未超時
		if boTime.After(p.clock.Now()) {
			return
		}
		// 超時就pop出來
		_, err := p.podBackoffQ.Pop()
		if err != nil {
			klog.Errorf("Unable to pop pod %v from backoffQ despite backoff completion.", nsNameForPod(pod))
			return
		}
        // 加入到activeQ中
		p.activeQ.Add(rawPodInfo)
		defer p.cond.Broadcast()
	}
}

unschedulableQ

image.png

調度失敗

調度失敗後，若是當前集羣資源沒有發生變動，就加入到unschedulable，緣由上面說過

func (p *PriorityQueue) AddUnschedulableIfNotPresent(pod *v1.Pod, podSchedulingCycle int64) error {
	// 省略檢查性代碼
	// 更新pod的backoff 信息
	p.backoffPod(pod)

	// moveRequestCycle將pod從unscheduledQ大於pod的調度週期添加到 若是pod的調度週期小於當前的調度週期
	if p.moveRequestCycle >= podSchedulingCycle {
		if err := p.podBackoffQ.Add(pInfo); err != nil {
			return fmt.Errorf("error adding pod %v to the backoff queue: %v", pod.Name, err)
		}
	} else {
		p.unschedulableQ.addOrUpdate(pInfo)
	}

	p.nominatedPods.add(pod, "")
	return nil

}

定時轉移任務

定時任務每30秒執行一次

func (p *PriorityQueue) run() {
	go wait.Until(p.flushUnschedulableQLeftover, 30*time.Second, p.stop)
}

邏輯其實就很是簡單若是當前時間-pod的最後調度時間大於60s,就從新調度，轉移到podBackoffQ或者activeQ中

func (p *PriorityQueue) flushUnschedulableQLeftover() {
	p.lock.Lock()
	defer p.lock.Unlock()

	var podsToMove []*framework.PodInfo
	currentTime := p.clock.Now()
	for _, pInfo := range p.unschedulableQ.podInfoMap {
		lastScheduleTime := pInfo.Timestamp
		// 若是該pod1分鐘內沒有被調度就加入到podsToMove
		if currentTime.Sub(lastScheduleTime) > unschedulableQTimeInterval {
			podsToMove = append(podsToMove, pInfo)
		}
	}

	if len(podsToMove) > 0 {
		// podsToMove將這些pod移動到activeQ
		p.movePodsToActiveQueue(podsToMove)
	}
}

調度隊列總結

數據流設計總結

image.png

3.1.1 三隊列與後臺定時任務

從設計上三隊列分別存儲：活動隊列、bakcoff隊列、不可調度隊列，其中backoff中會根據任務的失敗來逐步遞增重試時間(最長10s)、unschedulableQ隊列則延遲60s

經過後臺定時任務分別將backoffQ隊列、unschedulableQ隊列來進行重試，加入到activeQ中，從而加快完成pod的失敗重試調度

cycle與優先調度

schedulingCycle、moveRequestCycle兩個cycle其實本質上也是爲了加快失敗任務的重試調度，當集羣資源發生變化的時候，進行當即重試，那些失敗的優先級比較高、親和性問題的pod均可能會被優先調度

鎖與cond實現線程安全pop

內部經過lock保證線程安全，並經過cond來實現阻塞等待，從而實現阻塞scheduler worker的通知

今天就分析到這裏，其實參考這個實現，咱們也能夠從中抽象出一些設計思想，實現本身的一個具備優先級、快速重試、高可用的任務隊列，先分析到這，下一個分析的組件是SchedulerCache, 感興趣能夠加我微信一塊兒交流學習,畢竟三個臭皮匠算計不過諸葛亮

> 微信號：baxiaoshi2020 > 關注公告號閱讀更多源碼分析文章 21天大棚 > 更多文章關注 www.sreguide.com > 本文由博客一文多發平臺 OpenWrite 發佈

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程