自定义 Controller

拼搏现实的明天。 2022-04-24 16:50 188阅读 0赞

# 自定义 Controller #

*  自定义 Controller
    
     *  使用 Informer 的自定义 Controller
     *  参考

一般自定义 controller 的模式是：

1.  创建一个 SharedIndexInformer 和 workerqueue。
2.  在 SharedIndexInformer 中注册 OnAdd/OnUpdate/OnDelete 的处理函数是 enqueue，它向队列压入对象的 Key。
3.  运行几个 worker，分别从 workerqueue 中 Pop 对象 Key，从 Key 提取 namespace 和 objname；
4.  再调用从 SharedIndexInformer 获取的对象 Lister，根据 namespace 和 objname 获取对象；

workqueue 可以通过通过内部的机制可以保证，同时只能有一个 worker 处理某个对象，处理结束后（调用队列的 Done() 或 Forget() 方法），才能向队列中 Add/Get 对象。

由于 SharedIndexInformer 内部会有循环队列缓存 Controller Pop 出的对象事件，所以如果快速的 Add 再 Delete 对象， worker 用 Add 事件的对象 key 查找缓存，可能出现**找不到**的情况。

对于 CRD 对象的删除事件，一般是**不需**要定义处理函数（如 enqueue 或自定义函数），因为删除意味着该对象已经不在 K8S 中了。但是如果需要清理该 CRD 创建的 K8S 资源对象，则可能需要为 CRD 对象的 OnDelete 事件绑定处理函数，或者使用 finalized 机制。

如何快速查找 CRD 创建的 K8S 资源对象呢？简单的办法是给它创建的对象打上**包含 CRD 对象名的标签**。另外 CRD 对象名最好是随机的，否则删除和创建同一个 CRD 对象的资源时可能出现**竞争**关系。

对于 CRD 创建的类型对象，获取到后，需要看它是否是 CRD 创建的，如果不是则需要忽略。另外，对这些对象的删除事件，需要捕获，一般需要再次创建该对象。

## 使用 Informer 的自定义 Controller ##

在实现自定义 Controller 时， SharedIndexInformer 一般和 workqueue 联合使用：

1.  OnAdd/OnUpdate 的时候，都将对象的 Key 保存到 workqueue 中，后续弹出后的处理逻辑是一致的：
    
    1.  从 Key 提取 namespace 和 name；
    2.  从 Lister 获取 namespace 和 name 对应的资源对象（它的最新版本）；
    3.  通过 Lister 获取受控资源对象（一般是通过主资源对象的 name、spec 等来查找它的受控资源对象）；
    4.  如果受控资源对象不存在，则创建它；
    5.  比较主资源对象和受控资源对象，如果不一致，则更新受控资源对象；
    6.  用受控资源对象 Status 更新主资源对象；
2.  对于 OnDelete：
    
     *  如果是主资源对象，则需要特殊的 Delete Handler 直接处理；
     *  对于受控资源类型对象，则查找它的主资源对象，然后将主资源对象的 Key 存入 workqueue。对受控资源类型对象自身，一般除不需要处理；
3.  实现真正并发：可以为 workqueue 定义多个 workers；
4.  通过 workqueue 可以做到限流和重试；

对于 OnDelete() 处理函数，Informer 传入的 Obj 可能是 `cache.DeletedFinalStateUnknown` 类型而非特定的资源类型如 aolV1alpha1.AolDeployment，处理函数应该能判断和处理这种情况：

// 来源于：https://github.com/kubernetes/sample-controller/blob/master/controller.go
    func (c *Controller) handleObject(obj interface{ }) { 
    	var object metav1.Object
    	var ok bool
    	if object, ok = obj.(metav1.Object); !ok { 
    		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
    		if !ok { 
    			utilruntime.HandleError(fmt.Errorf("error decoding object, invalid type"))
    			return
    		}
    		object, ok = tombstone.Obj.(metav1.Object)
    		if !ok { 
    			utilruntime.HandleError(fmt.Errorf("error decoding object tombstone, invalid type"))
    			return
    		}
    		klog.V(4).Infof("Recovered deleted object '%s' from tombstone", object.GetName())
    	}
    	klog.V(4).Infof("Processing object: %s", object.GetName())
    	...
    }

同时需要等 informer 的 HasSynced() 返回为 true 时才开始启动 worker：

// 来源于：https://github.com/kubernetes/sample-controller/blob/master/controller.go
    func (c *Controller) Run(threadiness int, stopCh <-chan struct{ }) error { 
    	defer utilruntime.HandleCrash()
    	defer c.workqueue.ShutDown()
    
    	// Start the informer factories to begin populating the informer caches
    	klog.Info("Starting Foo controller")
    
    	// Wait for the caches to be synced before starting workers
    	klog.Info("Waiting for informer caches to sync")
    	if ok := cache.WaitForCacheSync(stopCh, c.deploymentsSynced, c.foosSynced); !ok { 
    		return fmt.Errorf("failed to wait for caches to sync")
    	}
    	...
    }

1.  HasSynced() 返回 false 时，表明部分对象还处于 DeltaFIFO 中，没有被 Informer 内部的 controller 更新到 clientState 缓存；这时如果执行 worker，则 worker 使用对象名称（Key）查找 clientState 缓存(通过 Informer 的 Lister)时会找不到对象；
2.  HasSynced() 返回 true 时表明 Reflecter List 的第一批对象都从 DeltaFIFO 弹出，并由 controller **更新到 clientState 缓存中，这样 worker 才能通过通过对象名称（Key）从 Lister Get 到对象**。

考虑一种情况，有多个 worker 处理 workqueue，如果用户先 Update 再 Delete 资源对象：

1.  worker1 弹出 Update 事件；
2.  worker1 还在执行过程中，worker2 弹出和执行 Delete 事件；
3.  worker1 执行 Update 失败，将 key 放回 workqueue；
4.  下一次 work3 弹出 key，查找 Lister，发现对象不存在。**这时除了打印日志外，啥都做不了**；

因为多个 worker 并发处理 workqueue，有可能出现两个 worker 同时处理一个对象的不同版本情况么？  
答案是否定的。

## 参考 ##

*  [cache.NewIndexerInformer 的用法示例][cache.NewIndexerInformer]
 *  [sample-controller][]

[cache.NewIndexerInformer]: https://github.com/kubernetes/client-go/blob/master/examples/workqueue/main.go#L164-L195
[sample-controller]: https://github.com/kubernetes/sample-controller