如何進行kubernetes scheduler基于map/reduce模式實現

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

優選階段通過分map/reduce模式來實現多個node和多種算法的并行計算，并且通過基于二級索引來設計最終的存儲結果，從而達到整個計算過程中的無鎖設計，同時為了保證分配的隨機性，針對同等優先級的采用了隨機的方式來進行最終節點的分配，如果大家后續有類似的需求，不妨可以借鑒借鑒

1. 設計基礎

1.1 兩階段: 單點與聚合

在進行優選的時候，除了最后一次計算，在進行針對單個算法的計算的時候，會分為兩個階段：單點和聚合

在單點階段，會根據當前算法針對單個node計算 在聚合階段，則會根據當前單點階段計算完成后，來進行聚合

1.2 并行: 節點與算法

單點和聚合兩階段在計算的時候，都是并行的，但是對象則不同，其中單點階段并行是針對單個node的計算，而聚合階段則是針對算法級別的計算，通過這種設計分離計算，從而避免多goroutine之間數據競爭，無鎖加速優選的計算

1.3 map與reduce

而map與reduce則是針對一個上面并行的兩種具體實現，其中map中負責單node打分，而reduce則是針對map階段的打分進行聚合后，根據匯總的結果進行二次打分計算

1.4 weight

map/reduce階段都是通過算法計算，如果我們要進行自定義的調整，針對單個算法，我們可以調整其在預選流程中的權重，從而進行定制自己的預選流程 

1.5 隨機分布

當進行優先級判斷的時候，肯定會出現多個node優先級相同的情況，在優選節點的時候，會進行隨機計算，從而決定是否用當前優先級相同的node替換之前的最合適的node

2. 源碼分析 

優選的核心流程主要是在PrioritizeNodes中，這里只介紹其關鍵的核心數據結構設計

2.1 無鎖計算結果保存

無鎖計算結果的保存主要是通過下面的二維數組實現， 如果要存儲一個算法針對某個node的結果，其實只需要通過兩個索引即可：算法索引和節點索引，同理如果我吧針對單個node的索引分配給一個goroutine，則其去其他的goroutine則就可以并行計算

// 在計算的時候，會傳入nodes []*v1.Node的數組，存儲所有的節點，節點索引主要是指的該部分
results := make([]schedulerapi.HostPriorityList, len(priorityConfigs), len(priorityConfigs))

2.2 基于節點索引的Map計算

之前在預選階段介紹過ParallelizeUntil函數的實現，其根據傳入的數量來生成計算索引，放入chan中，后續多個goroutine從chan中取出數據直接進行計算即可

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

			var err error
                
                // 通過節點索引，來進行針對單個node的計算結果的保存
			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計算

基于算法的并行，則是為每個算法的計算都啟動一個goroutine,每個goroutine通過算法索引來進行該算法的所有map階段的結果的讀取，并進行計算，后續結果仍然存儲在對應的位置

	// 計算策略的分值
	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 優先級打分結果統計

根據之前的map/reduce階段，接下來就是將針對所有node的所有算法計算結果進行累加即可

	// 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 根據優先級隨機篩選host

這里的隨機篩選是指的當多個host優先級相同的時候，會有一定的概率用當前的node替換之前的優先級相等的node(到目前為止的優先級最高的node), 其主要通過cntOfMaxScore和rand.Intn(cntOfMaxScore)來進行實現

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. 設計總結

優選階段通過分map/reduce模式來實現多個node和多種算法的并行計算，并且通過基于二級索引來設計最終的存儲結果，從而達到整個計算過程中的無鎖設計，同時為了保證分配的隨機性，針對同等優先級的采用了隨機的方式來進行最終節點的分配，如果大家后續有類似的需求，不妨可以借鑒借鑒

看完上述內容是否對您有幫助呢？如果還想對相關知識有進一步的了解或閱讀更多相關文章，請關注億速云行業資訊頻道，感謝您對億速云的支持。