如何進(jìn)行kubernetes scheduler基于map/reduce模式實(shí)現(xiàn)

如何進(jìn)行kubernetes scheduler基于map/reduce模式實(shí)現(xiàn)，很多新手對(duì)此不是很清楚，為了幫助大家解決這個(gè)難題，下面小編將為大家詳細(xì)講解，有這方面需求的人可以來學(xué)習(xí)下，希望你能有所收獲。

優(yōu)選階段通過分map/reduce模式來實(shí)現(xiàn)多個(gè)node和多種算法的并行計(jì)算，并且通過基于二級(jí)索引來設(shè)計(jì)最終的存儲(chǔ)結(jié)果，從而達(dá)到整個(gè)計(jì)算過程中的無鎖設(shè)計(jì)，同時(shí)為了保證分配的隨機(jī)性，針對(duì)同等優(yōu)先級(jí)的采用了隨機(jī)的方式來進(jìn)行最終節(jié)點(diǎn)的分配，如果大家后續(xù)有類似的需求，不妨可以借鑒借鑒

1. 設(shè)計(jì)基礎(chǔ)

1.1 兩階段: 單點(diǎn)與聚合

在進(jìn)行優(yōu)選的時(shí)候，除了最后一次計(jì)算，在進(jìn)行針對(duì)單個(gè)算法的計(jì)算的時(shí)候，會(huì)分為兩個(gè)階段：?jiǎn)吸c(diǎn)和聚合

在單點(diǎn)階段，會(huì)根據(jù)當(dāng)前算法針對(duì)單個(gè)node計(jì)算 在聚合階段，則會(huì)根據(jù)當(dāng)前單點(diǎn)階段計(jì)算完成后，來進(jìn)行聚合

1.2 并行: 節(jié)點(diǎn)與算法

單點(diǎn)和聚合兩階段在計(jì)算的時(shí)候，都是并行的，但是對(duì)象則不同，其中單點(diǎn)階段并行是針對(duì)單個(gè)node的計(jì)算，而聚合階段則是針對(duì)算法級(jí)別的計(jì)算，通過這種設(shè)計(jì)分離計(jì)算，從而避免多goroutine之間數(shù)據(jù)競(jìng)爭(zhēng)，無鎖加速優(yōu)選的計(jì)算

1.3 map與reduce

而map與reduce則是針對(duì)一個(gè)上面并行的兩種具體實(shí)現(xiàn)，其中map中負(fù)責(zé)單node打分，而reduce則是針對(duì)map階段的打分進(jìn)行聚合后，根據(jù)匯總的結(jié)果進(jìn)行二次打分計(jì)算

1.4 weight

map/reduce階段都是通過算法計(jì)算，如果我們要進(jìn)行自定義的調(diào)整，針對(duì)單個(gè)算法，我們可以調(diào)整其在預(yù)選流程中的權(quán)重，從而進(jìn)行定制自己的預(yù)選流程 

1.5 隨機(jī)分布

當(dāng)進(jìn)行優(yōu)先級(jí)判斷的時(shí)候，肯定會(huì)出現(xiàn)多個(gè)node優(yōu)先級(jí)相同的情況，在優(yōu)選節(jié)點(diǎn)的時(shí)候，會(huì)進(jìn)行隨機(jī)計(jì)算，從而決定是否用當(dāng)前優(yōu)先級(jí)相同的node替換之前的最合適的node

2. 源碼分析 

優(yōu)選的核心流程主要是在PrioritizeNodes中，這里只介紹其關(guān)鍵的核心數(shù)據(jù)結(jié)構(gòu)設(shè)計(jì)

2.1 無鎖計(jì)算結(jié)果保存

無鎖計(jì)算結(jié)果的保存主要是通過下面的二維數(shù)組實(shí)現(xiàn)， 如果要存儲(chǔ)一個(gè)算法針對(duì)某個(gè)node的結(jié)果，其實(shí)只需要通過兩個(gè)索引即可：算法索引和節(jié)點(diǎn)索引，同理如果我吧針對(duì)單個(gè)node的索引分配給一個(gè)goroutine，則其去其他的goroutine則就可以并行計(jì)算

// 在計(jì)算的時(shí)候，會(huì)傳入nodes []*v1.Node的數(shù)組，存儲(chǔ)所有的節(jié)點(diǎn)，節(jié)點(diǎn)索引主要是指的該部分
results := make([]schedulerapi.HostPriorityList, len(priorityConfigs), len(priorityConfigs))

2.2 基于節(jié)點(diǎn)索引的Map計(jì)算

之前在預(yù)選階段介紹過ParallelizeUntil函數(shù)的實(shí)現(xiàn)，其根據(jù)傳入的數(shù)量來生成計(jì)算索引，放入chan中，后續(xù)多個(gè)goroutine從chan中取出數(shù)據(jù)直接進(jìn)行計(jì)算即可

	workqueue.ParallelizeUntil(context.TODO(), 16, len(nodes), func(index int) {
		// 根據(jù)節(jié)點(diǎn)和配置的算法進(jìn)行計(jì)算
		nodeInfo := nodeNameToInfo[nodes[index].Name]
            // 獲取算法的索引
		for i := range priorityConfigs {
			if priorityConfigs[i].Function != nil {
				continue
			}

			var err error
                
                // 通過節(jié)點(diǎn)索引，來進(jìn)行針對(duì)單個(gè)node的計(jì)算結(jié)果的保存
			results[i][index], err = priorityConfigs[i].Map(pod, meta, nodeInfo)
			if err != nil {
				appendError(err)
				results[i][index].Host = nodes[index].Name
			}
		}
	})

2.3 基于算法索引的Reduce計(jì)算

基于算法的并行，則是為每個(gè)算法的計(jì)算都啟動(dòng)一個(gè)goroutine,每個(gè)goroutine通過算法索引來進(jìn)行該算法的所有map階段的結(jié)果的讀取，并進(jìn)行計(jì)算，后續(xù)結(jié)果仍然存儲(chǔ)在對(duì)應(yīng)的位置

	// 計(jì)算策略的分值
	for i := range priorityConfigs {
		if priorityConfigs[i].Reduce == nil {
			continue
		}
		wg.Add(1)
		go func(index int) {
			defer wg.Done()
			if err := priorityConfigs[index].Reduce(pod, meta, nodeNameToInfo, results[index]); err != nil {
				appendError(err)
			}
			if klog.V(10) {
				for _, hostPriority := range results[index] {
					klog.Infof("%v -> %v: %v, Score: (%d)", util.GetPodFullName(pod), hostPriority.Host, priorityConfigs[index].Name, hostPriority.Score)
				}
			}
		}(i)
	}
	// Wait for all computations to be finished.
	wg.Wait()

2.4 優(yōu)先級(jí)打分結(jié)果統(tǒng)計(jì)

根據(jù)之前的map/reduce階段，接下來就是將針對(duì)所有node的所有算法計(jì)算結(jié)果進(jìn)行累加即可

	// Summarize all scores.
	result := make(schedulerapi.HostPriorityList, 0, len(nodes))

	for i := range nodes {
		result = append(result, schedulerapi.HostPriority{Host: nodes[i].Name, Score: 0})
		// 便利所有的算法配置
		for j := range priorityConfigs {
			result[i].Score += results[j][i].Score * priorityConfigs[j].Weight
		}

		for j := range scoresMap {
			result[i].Score += scoresMap[j][i].Score
		}
	}

2.5 根據(jù)優(yōu)先級(jí)隨機(jī)篩選host

這里的隨機(jī)篩選是指的當(dāng)多個(gè)host優(yōu)先級(jí)相同的時(shí)候，會(huì)有一定的概率用當(dāng)前的node替換之前的優(yōu)先級(jí)相等的node(到目前為止的優(yōu)先級(jí)最高的node), 其主要通過cntOfMaxScore和rand.Intn(cntOfMaxScore)來進(jìn)行實(shí)現(xiàn)

func (g *genericScheduler) selectHost(priorityList schedulerapi.HostPriorityList) (string, error) {
	if len(priorityList) == 0 {
		return "", fmt.Errorf("empty priorityList")
	}
	maxScore := priorityList[0].Score
	selected := priorityList[0].Host
	cntOfMaxScore := 1
	for _, hp := range priorityList[1:] {
		if hp.Score > maxScore {
			maxScore = hp.Score
			selected = hp.Host
			cntOfMaxScore = 1
		} else if hp.Score == maxScore {
			cntOfMaxScore++
			if rand.Intn(cntOfMaxScore) == 0 {
				// Replace the candidate with probability of 1/cntOfMaxScore
				selected = hp.Host
			}
		}
	}
	return selected, nil
}

3. 設(shè)計(jì)總結(jié)

優(yōu)選階段通過分map/reduce模式來實(shí)現(xiàn)多個(gè)node和多種算法的并行計(jì)算，并且通過基于二級(jí)索引來設(shè)計(jì)最終的存儲(chǔ)結(jié)果，從而達(dá)到整個(gè)計(jì)算過程中的無鎖設(shè)計(jì)，同時(shí)為了保證分配的隨機(jī)性，針對(duì)同等優(yōu)先級(jí)的采用了隨機(jī)的方式來進(jìn)行最終節(jié)點(diǎn)的分配，如果大家后續(xù)有類似的需求，不妨可以借鑒借鑒

看完上述內(nèi)容是否對(duì)您有幫助呢？如果還想對(duì)相關(guān)知識(shí)有進(jìn)一步的了解或閱讀更多相關(guān)文章，請(qǐng)關(guān)注億速云行業(yè)資訊頻道，感謝您對(duì)億速云的支持。