圖解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和多種演算法的平行計算，並且通過基於二級索引來設計最終的儲存結果，從而達到整個計算過程中的無鎖設計，同時為了保證分配的隨機性，針對同等優先順序的採用了隨機的方式來進行最終節點的分配，如果大家後續有類似的需求，不妨可以借鑑借鑑
本系列純屬個人臆測僅供參考，如果有看出錯誤的大佬歡迎指正
微訊號：baxiaoshi2020
關注公告號閱讀更多原始碼分析文章
更多文章關注 www.sreguide.com
本文由部落格一文多發平臺 OpenWrite 釋出