背景
kubebuilder可以一键帮我们生成一个operator工程。具体是如何实现的呢？kubebuilder主要借助了k8s的另一个开源项目controller-runtime。下图中可以大体看出operator、controller-runtime、informer、kube-apiserver的关系
image.png
operator工程代码分析
1. 我们从下往上看，controller-runtime如何创建一个informer, 可以看出是使用了client-go的cache包。
.\controller-runtime\pkg\cache\internal\informers.go
import
(
"k8s.io/client-go/tools/cache"
)
func NewInformers(config *rest.Config, options *InformersOpts) *Informers {
newInformer := cache.NewSharedIndexInformer
1.1 通过informers构造listwacher
.\controller-runtime\pkg\cache\internal\informers.go
// groupVersionKind to the Resource API we will use.
mapping, err := ip.mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
if err != nil {
return nil, err
}
...
default:
client, err := apiutil.RESTClientForGVK(gvk, false, ip.config, ip.codecs, ip.httpClient)
if err != nil {
return nil, err
}
listGVK := gvk.GroupVersion().WithKind(gvk.Kind + "List")
listObj, err := ip.scheme.New(listGVK)
if err != nil {
return nil, err
}
return &cache.ListWatch{
ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
// Build the request.
req := client.Get().Resource(mapping.Resource.Resource).VersionedParams(&opts, ip.paramCodec)
if namespace != "" {
req.Namespace(namespace)
}
// Create the resulting object, and execute the request.
res := listObj.DeepCopyObject()
if err := req.Do(ip.ctx).Into(res); err != nil {
return nil, err
}
return res, nil
},
// Setup the watch function
WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
// Build the request.
req := client.Get().Resource(mapping.Resource.Resource).VersionedParams(&opts, ip.paramCodec)
if namespace != "" {
req.Namespace(namespace)
}
// Call the watch.
return req.Watch(ip.ctx)
},
}, nil
2. 给informer绑定hander钩子函数，如下Kind对象Type参数即用户的crd，会在NewControllerManagedBy时传入。
.\controller-runtime\pkg\internal\source\kind.go
func (ks *Kind[object, request]) Start(){
...
_, err := i.AddEventHandler(NewEventHandler(ctx, queue, ks.Handler,
ks.Predicates).HandlerFuncs())
...
}
2.1 真正的钩子函数ks.Handler里, 即如下文件中的EventHandler，文件中可以看到有OnAdd、OnUpdate、OnDelete方法。
.\controller-runtime\pkg\internal\source\event_handler.go
2.2 从钩子函数的实现看在函数结束时会执行predicates对事件进行过滤，而predicates事件处理可以在ctrl.NewControllerManagedBy().WithEventFilter()进行配置
func (r *KrouterReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
Named("krouter-controller").
For(&kroutersalpha1.Krouter{}).
// Owns(&kroutersalpha1.Krouter{}).
WithEventFilter(predicate.Funcs{
UpdateFunc: func(e event.UpdateEvent) bool {
log.Log.Info("update event: resource version:" + e.ObjectNew.GetResourceVersion() + " old version:" + e.ObjectOld.GetResourceVersion() + "name:" + e.ObjectNew.GetName() + " namespace:" + e.ObjectNew.GetNamespace())
return true
},
}).
Complete(r)
}
2.3. EventHandler代码中可以看到消息经过处理后塞进了EventHandler的queue里
3. 对queue里的事件进行消费
调协函数Reconcile(ctx context.Context, req ctrl.Request) ，kubebuilder已经帮我们生成好了
.\krouters\internal\controller\krouter_controller.go
3.1 调协函数再哪里触发工作的呢，如下代码可以看出，在控制器Start里会开启协程持续对队列进行消费
.\controller-runtime\pkg\internal\controller\controller.go
func (c *Controller[request]) Start(ctx context.Context) error {
...
wg.Add(c.MaxConcurrentReconciles)
for i := 0; i < c.MaxConcurrentReconciles; i++ {
go func() {
defer wg.Done()
// Run a worker thread that just dequeues items, processes them, and marks them done.
// It enforces that the reconcileHandler is never invoked concurrently with the same object.
for c.processNextWorkItem(ctx) {
}
}()
}
...
func (c *Controller[request]) processNextWorkItem(ctx context.Context) bool {
obj, shutdown := c.Queue.Get()
if shutdown {
// Stop working
return false
}
// We call Done here so the workqueue knows we have finished
// processing this item. We also must remember to call Forget if we
// do not want this work item being re-queued. For example, we do
// not call Forget if a transient error occurs, instead the item is
// put back on the workqueue and attempted again after a back-off
// period.
defer c.Queue.Done(obj)
ctrlmetrics.ActiveWorkers.WithLabelValues(c.Name).Add(1)
defer ctrlmetrics.ActiveWorkers.WithLabelValues(c.Name).Add(-1)
c.reconcileHandler(ctx, obj)
return true
}
3.2 reconcileHandler函数里会对object进行消费或者重新入队, 实现中可以看出，除了重新入队还有做一些记录监控指标的操作，在用户返回的result.RequeueAfter>0时，会等待后再入到队列里。
.\controller-runtime\pkg\internal\controller\controller.go
result, err := c.Reconcile(ctx, req)
switch {
case err != nil:
if errors.Is(err, reconcile.TerminalError(nil)) {
ctrlmetrics.TerminalReconcileErrors.WithLabelValues(c.Name).Inc()
} else {
c.Queue.AddRateLimited(req)
}
ctrlmetrics.ReconcileErrors.WithLabelValues(c.Name).Inc()
ctrlmetrics.ReconcileTotal.WithLabelValues(c.Name, labelError).Inc()
if !result.IsZero() {
log.Info("Warning: Reconciler returned both a non-zero result and a non-nil error. The result will always be ignored if the error is non-nil and the non-nil error causes reqeueuing with exponential backoff. For more details, see: https://pkg.go.dev/sigs.k8s.io/controller-runtime/pkg/reconcile#Reconciler")
}
log.Error(err, "Reconciler error")
case result.RequeueAfter > 0:
log.V(5).Info(fmt.Sprintf("Reconcile done, requeueing after %s", result.RequeueAfter))
// The result.RequeueAfter request will be lost, if it is returned
// along with a non-nil error. But this is intended as
// We need to drive to stable reconcile loops before queuing due
// to result.RequestAfter
c.Queue.Forget(req)
c.Queue.AddAfter(req, result.RequeueAfter)
ctrlmetrics.ReconcileTotal.WithLabelValues(c.Name, labelRequeueAfter).Inc()
case result.Requeue:
log.V(5).Info("Reconcile done, requeueing")
c.Queue.AddRateLimited(req)
ctrlmetrics.ReconcileTotal.WithLabelValues(c.Name, labelRequeue).Inc()
default:
log.V(5).Info("Reconcile successful")
// Finally, if no error occurs we Forget this item so it does not
// get queued again until another change happens.
c.Queue.Forget(req)
ctrlmetrics.ReconcileTotal.WithLabelValues(c.Name, labelSuccess).Inc()
}
