Inside Kubernetes controller-02

二 低層架構-client-go

2.1 Informer

Informer watches 對象事件（added，updated，deleted）

當控制器每次向api服務器查詢對象狀態時，監視對象更改，api服務器高負載。

informer watch api，通過controller設置關聯cache。

func main() { ... clientset, err := kubernetes.NewForConfig(config) // Create InformerFactory informerFactory := informers.NewSharedInformerFactory(clientset, time.Second*30) // Create pod informer by informerFactory podInformer := informerFactory.Core().V1().Pods() // Add EventHandler to informer podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(new interface{}) { log.Println("Added") }, UpdateFunc: func(old, new interface{}) { log.Println("Updated") }, DeleteFunc: func(old interface{}) { log.Println("Deleted") }, }) // Start Go routines informerFactory.Start(wait.NeverStop) // Wait until finish caching with List API informerFactory.WaitForCacheSync(wait.NeverStop) // Create Pod Lister podLister := podInformer.Lister() // Get List of pods _, err = podLister.List(labels.Nothing()) … }

當我們使用Informer，我們不直接使用informer自己，而是使用共享informer。

共享informer 共享相同的資源在一個單獨的二進制文件中。

Reflector list & watch api-server

Reflector將對象添加到Delta FIFO 隊列中

informer 從Delta FIFO中彈出對象

將對象添加到Indexer

Indexer 存儲對象到安全線程存儲中

informer 設置事件reference

用戶自定義的Controller，事件Handlers 將對象添加至workqueue，

進程獲取key，process item

獲取對象的key從indexer reference中。

Informer 整體架構

reflector 開始listAndWatch api對象中的用戶創建的心資源pod對象。

informer 從DeltaFIFO 隊列中Pop初事件。

HandleDelta

indexer.Add 增加事件到內存存儲中

開始循環彈出事件并進行處理

后期get or list 請求都從indexer中獲取。

Informer 和其組件

Informer：watch 一個對象事件并且存儲在內存緩存中

Reflector：listAndWatch api-server

DeltaFIFO：FIFO 隊列存儲對象臨時

Indexer：設置或者獲取對象

Store：存儲對象

Lister：獲取對象在內存通過index

2.2 WorkQueue

WorkQueue不同于DeltaFIFO隊列，

WorkQueue是使用存儲Contrl Loop 的item。

調諧將執行WorkQWueue中的內容。

感覺的控制存儲enqueues item 到工作隊列中，當事件發生時。

func main() { ... clientset, err := kubernetes.NewForConfig(config) // Create InformerFactory informerFactory := informers.NewSharedInformerFactory(clientset, time.Second*30) // Create pod informer by informerFactory podInformer := informerFactory.Core().V1().Pods() // Create RateLimitQueue queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()) // shutdown when process ends defer queue.ShutDown() // Add EventHandler to informer podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(old interface{}) { var key string var err error if key, err = cache.MetaNamespaceKeyFunc(old); err != nil { runtime.HandleError(err) return } queue.Add(key) log.Println("Added: " + key) }, UpdateFunc: func(old, new interface{}) { … }, DeleteFunc: func(old interface{}) { … }, }) … }

整體示意圖

Reflector watch k8s-api 將其中資源發送到DeltaFIFO中

Informer EventHandler 從DeltaFIFO中pop出數據。

Informer EventHandler 調用資源事件函數，例如：AddFunc、UpdateFunc、DeleteFunc來將事件對象放進WorkQueue中。

注意：再次添加的key為資源的唯一標示，后期使用的時候從index中之間獲取，從而減輕api-server 的壓力。

workqueue.Add 進入WorkQueue中，例如namespace/name、default/nginx

循環3-4，處理事件放入隊列中。

用戶自定義的controller 邏輯從WorkQueue中獲取對象進行處理。

Informer 重新同步周期

重新同步周期對于Informer是可選的。

Informer監視api服務器上的對象事件。

再同步期過后，無論發生什么事件，UpdateFunc被回調。結果，調諧將再次執行。

2.3 Controller Cycle Main Logic

完整示意圖：

假設用戶創建了兩個pod，生成了兩個update 事件

informer 執行UpdateFunc

Informer 調用workQueue.Add 函數 增加事件到WorkQueue中。

程序處理調用workqueue.Get 獲取事件

調諧處理具體更新業務邏輯。

當調諧成功后，處理程序進行workQueue.Forget/workQueue.Done，這個item從workQueue中移除。

當調諧是錯誤結束后，workQueue 增加速率限制，并且控制器重新入隊item，并且開始新一輪調諧。

每次事件發生時，項都會繼續存儲在工作隊列中,控制器處理工作隊列中的項并執行協調。該循環持續不斷，直到控制器停止。

從內存中讀取，寫向api-server。

但是，如果我們直接更新緩存中的對象，很難保證其一致性。

所以，我們在更新對象時使用DeepCopy（獲取克隆數據）。

例如代碼：kubernetes/pkg/controller/replicaset/replica_set.go

rs = rs.DeepCopy() newStatus := calculateStatus(rs, filteredPods, manageReplicasErr) // Always updates status as pods come up or die. updatedRS, err := updateReplicaSetStatus(rsc.kubeClient.AppsV1(). ? ReplicaSets(rs.Namespace), rs, newStatus)

worker：無限循環 processNextWorkItem。

processNet WorkItem：操作WorkQueue(Get, Add)并且調用調諧邏輯。

syncHandler：這就是調諧具體邏輯。

基于K8s v1.16

worker：https://github.com/kubernetes/kubernetes/blob/release-1.16/pkg/controller/replicaset/replica_set.go#L432

processNextWorkItem：https://github.com/kubernetes/kubernetes/blob/release-1.16/pkg/controller/replicaset/replica_set.go#L437

syncReplicaSet：https://github.com/kubernetes/kubernetes/blob/release-1.16/pkg/controller/replicaset/replica_set.go#L562

其他

Informer 從etcd 同步對象數據到內存中。

再次需要考慮內存數據是否和etcd數據不相同呢。

這是沒有問題的，對象有resourceVersion，如果etcd的resourceVersion 于內存緩存中的resourceVersion不同，Controller在更新對象狀態時出錯，控制器請求調諧，直到遷移完成。

review

Informer：通過eventandler將Control Loop的項添加到WorkQueue中

Lister：通過Indexer從內存緩存中獲取對象數據

WorkQueue：存儲控制循環項的隊列。該項是Reconcile Logic的目標。如果在調解結束時發生錯誤，控制器將項請求到工作隊列。控制器再次執行Reconcile。

Kubernetes

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。