kubebuilder是创建operator的常用工具，学会在kubebuilder的框架中使用并发可以降低开发难度，同时提高operator的效率。
控制器会监听cr（custom resource）状态的变化，并交给 Reconcile函数处理。当某个cr状态产生的变化，recoucile函数处理时间过长的话，那其余cr状态的变化将被放在队列中，排队等待被reconcile函数处理。因此自然出现了需求：希望有多个reconcile函数并发的处理cr状态的变化。
kubebuilder的worker并发参数
// Options are the arguments for creating a new Controller.
type Options struct {
// MaxConcurrentReconciles is the maximum number of concurrent Reconciles which can be run. Defaults to 1.
MaxConcurrentReconciles int
...
}
Option参数定义了最大的Reconcile并发数量，默认值为1: 即不会引入并发处理
使用默认参数启动operator，会看到 "worker count": 1
2023-06-13T17:19:32+08:00 INFO Starting workers {"controller": "guestbook", "controllerGroup": "webapp.my.domain", "controllerKind": "Guestbook", "worker count": 1}
修改kubebuilder的worker并发参数
可以通过修改option的参数来修改MaxConcurrentReconciles
import "sigs.k8s.io/controller-runtime/pkg/controller"
// SetupWithManager sets up the controller with the Manager.
func (r *GuestbookReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&webappv1.Guestbook{}).
WithOptions(controller.Options{
MaxConcurrentReconciles: 3,
}).Complete(r)
}
2023-06-13T17:24:07+08:00 INFO Starting workers {"controller": "guestbook", "controllerGroup": "webapp.my.domain", "controllerKind": "Guestbook", "worker count": 3}
可以根据计算机的cpu数量来选择传入的MaxConcurrentReconciles的数值
import (
goruntime "runtime"
"sigs.k8s.io/controller-runtime/pkg/controller"
)
// SetupWithManager sets up the controller with the Manager.
func (r *GuestbookReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&webappv1.Guestbook{}).
WithOptions(controller.Options{
MaxConcurrentReconciles: goruntime.NumCPU(),
}).Complete(r)
}
2023-06-13T17:29:33+08:00 INFO Starting workers {"controller": "guestbook", "controllerGroup": "webapp.my.domain", "controllerKind": "Guestbook", "worker count": 8}
引入并发的问题
答案是否定的，这是因为kubebuilder使用了client-go中的工作队列，该队列限制了同一对象的不同请求不会被同时处理
client-go队列原理分析
// Type is a work queue (see the package comment).
type Type struct {
// queue defines the order in which we will work on items. Every
// element of queue should be in the dirty set and not in the
// processing set.
queue []t
// dirty defines all of the items that need to be processed.
dirty set
// Things that are currently being processed are in the processing set.
// These things may be simultaneously in the dirty set. When we finish
// processing something and remove it from this set, we'll check if
// it's in the dirty set, and if so, add it to the queue.
processing set
cond *sync.Cond
shuttingDown bool
drain bool
metrics queueMetrics
unfinishedWorkUpdatePeriod time.Duration
clock clock.WithTicker
}
由client-go中队列的源码可知：work queue主要维护了三个数据结构：
// 存储所有要被worker处理的对象
queue []t
// 存储所有要处理的请求对象（丢弃重复的）
dirty set
// 存储所有正在被处理的的请求对象
processing set
下图描述了client-go队列的工作流程
由第2步知：cr的请求会先被加入dirty set，这一步确保不会去处理相同的请求。
由第4步知：如果一个cr的请求正在被处理，则该cr的其它请求不会再被加入队列，这一步确保了，即使有多个worker空闲可以并发的处理请求。相同cr的不同请求也不会被同时处理。
同时，这意味着如果一个的cr的请求事件过长，那么即使有多个worker（MaxConcurrentReconciles参数大于一），该cr的其它请求仍然会在Dirty set中等待。
