k8s调度器优化之步长+有限个数node的调度

发表于

当k8s的node数量变大的时候，k8s调度器中的2个调度环节会变的很长，可能会造成总体超过秒级别，这显然是不可接受的，所以我们会对调度的环节进行优化。
比如我们对findNodesThatFit进行优化
这个函数真正干活的一行代码workqueue.Parallelize(16, len(nodes), checkNode)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// Parallelize is a very simple framework that allow for parallelizing
// N independent pieces of work.
//以下代码就是实现了启动workers个协程并发运行，然后从toProcess里面取数字，丢给doWorkPiece运行
func Parallelize(workers, pieces int, doWorkPiece DoWorkPieceFunc) {
	toProcess := make(chan int, pieces)
	for i := 0; i < pieces; i++ {
		toProcess <- i
	}
	close(toProcess)

	if pieces < workers {
		workers = pieces
	}

	wg := sync.WaitGroup{}
	wg.Add(workers)
	for i := 0; i < workers; i++ {
		go func() {
			defer utilruntime.HandleCrash()
			defer wg.Done()
			for piece := range toProcess {
				doWorkPiece(piece)
			}
		}()
	}
	wg.Wait()
}

那么其实我们就可以有个最大想法就是我们其实没用必要遍历所有的节点，而可以设计成遍历部分节点，并且能找到我们需要的个数个的节点停止就可以了，对这个函数进行改造

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

// ParallelizeUntil is a framework that allows for parallelizing N
// independent pieces of work until done or the context is canceled.
//当ctx被取消的时候我们就停止，然后可以在doWorkPiece中实现当找到对的node的数量到我们设定的值的时候就取消
func ParallelizeUntil(ctx context.Context, workers, pieces int, doWorkPiece DoWorkPieceFunc) {
	var stop <-chan struct{}
	if ctx != nil {
		stop = ctx.Done()
	}

	toProcess := make(chan int, pieces)
	for i := 0; i < pieces; i++ {
		toProcess <- i
	}
	close(toProcess)

	if pieces < workers {
		workers = pieces
	}

	wg := sync.WaitGroup{}
	wg.Add(workers)
	for i := 0; i < workers; i++ {
		go func() {
			defer utilruntime.HandleCrash()
			defer wg.Done()
			for piece := range toProcess {
				select {
				case <-stop: //注意这里
					return
				default:
					doWorkPiece(piece)
				}
			}
		}()
	}
	wg.Wait()
}

并且修改checkNode中

1
2
3
4
5
6
7
length := atomic.AddInt32(&filteredLen, 1)
				if length > numNodesToFind {
					cancel() //这里cancel掉
					atomic.AddInt32(&filteredLen, -1)
				} else {
					filtered[length-1] = g.cachedNodeInfoMap[nodeName].Node()
				}

以上功能，其实在官方实现之前就实现了，同事们没有提交PR，看到官方也有这个功能了，那么可以对外写写这个东西。
至于我的标题步长的是什么意思呢，其实关注了下社区的PR，发现之前也有类似的实现过程中的问题，我们的实现是先找一定数量的机器，找到满足数量个的之后就退出，否则继续找下个步长个的机器。官方其实也经历了类似的实现过程，不过最终改成了上面的代码，巧妙的利用了context，去掉了要分步长的逻辑，值得学习.

Gitalking ...