Kubernetes 中 Informer 的使用
前面我們?cè)谑褂?Clientset 的時(shí)候了解到我們可以使用 Clientset 來(lái)獲取所有的原生資源對(duì)象，那么如果我們想要去一直獲取集群的資源對(duì)象數(shù)據(jù)呢？豈不是需要用一個(gè)輪詢?nèi)ゲ粩鄨?zhí)行 List() 操作？這顯然是不合理的，實(shí)際上除了常用的 CRUD 操作之外，我們還可以進(jìn)行 Watch 操作，可以監(jiān)聽(tīng)資源對(duì)象的增、刪、改、查操作，這樣我們就可以根據(jù)自己的業(yè)務(wù)邏輯去處理這些數(shù)據(jù)了。
Watch 通過(guò)一個(gè) event 接口監(jiān)聽(tīng)對(duì)象的所有變化（添加、刪除、更新）：
// staging/src/k8s.io/apimachinery/pkg/watch/watch.go// Interface 可以被任何知道如何 watch 和通知變化的對(duì)象實(shí)現(xiàn)type Interface interface {// Stops watching. Will close the channel returned by ResultChan(). Releases// any resources used by the watch.Stop()// Returns a chan which will receive all the events. If an error occurs// or Stop() is called, this channel will be closed, in which case the// watch should be completely cleaned up.ResultChan() <-chan Event}
watch 接口的 ResultChan 方法會(huì)返回如下幾種事件：
// staging/src/k8s.io/apimachinery/pkg/watch/watch.go// EventType 定義可能的事件類型type EventType stringconst (Added EventType = "ADDED"Modified EventType = "MODIFIED"Deleted EventType = "DELETED"Bookmark EventType = "BOOKMARK"Error EventType = "ERROR"DefaultChanSize int32 = 100)// Event represents a single event to a watched resource.// +k8s:deepcopy-gen=truetype Event struct {Type EventType// Object is:// * If Type is Added or Modified: the new state of the object.// * If Type is Deleted: the state of the object immediately before deletion.// * If Type is Bookmark: the object (instance of a type being watched) where// only ResourceVersion field is set. On successful restart of watch from a// bookmark resourceVersion, client is guaranteed to not get repeat event// nor miss any events.// * If Type is Error: *api.Status is recommended; other types may make sense// depending on context.Object runtime.Object}
這個(gè)接口雖然我們可以直接去使用，但是實(shí)際上并不建議這樣使用，因?yàn)橥捎诩褐械馁Y源較多，我們需要自己在客戶端去維護(hù)一套緩存，而這個(gè)維護(hù)成本也是非常大的，為此 client-go 也提供了自己的實(shí)現(xiàn)機(jī)制，那就是 Informers。Informers 是這個(gè)事件接口和帶索引查找功能的內(nèi)存緩存的組合，這樣也是目前最常用的用法。Informers 第一次被調(diào)用的時(shí)候會(huì)首先在客戶端調(diào)用 List 來(lái)獲取全量的對(duì)象集合，然后通過(guò) Watch 來(lái)獲取增量的對(duì)象更新緩存。
運(yùn)行原理
一個(gè)控制器每次需要獲取對(duì)象的時(shí)候都要訪問(wèn) APIServer，這會(huì)給系統(tǒng)帶來(lái)很高的負(fù)載，Informers 的內(nèi)存緩存就是來(lái)解決這個(gè)問(wèn)題的，此外 Informers 還可以幾乎實(shí)時(shí)的監(jiān)控對(duì)象的變化，而不需要輪詢請(qǐng)求，這樣就可以保證客戶端的緩存數(shù)據(jù)和服務(wù)端的數(shù)據(jù)一致，就可以大大降低 APIServer 的壓力了。
如上圖展示了 Informer 的基本處理流程：
以 events 事件的方式從 APIServer 獲取數(shù)據(jù)
提供一個(gè)類似客戶端的 Lister 接口，從內(nèi)存緩存中 get 和 list 對(duì)象
為添加、刪除、更新注冊(cè)事件處理程序
此外 Informers 也有錯(cuò)誤處理方式，當(dāng)長(zhǎng)期運(yùn)行的 watch 連接中斷時(shí)，它們會(huì)嘗試使用另一個(gè) watch 請(qǐng)求來(lái)恢復(fù)連接，在不丟失任何事件的情況下恢復(fù)事件流。如果中斷的時(shí)間較長(zhǎng)，而且 APIServer 丟失了事件（etcd 在新的 watch 請(qǐng)求成功之前從數(shù)據(jù)庫(kù)中清除了這些事件），那么 Informers 就會(huì)重新 List 全量數(shù)據(jù)。
而且在重新 List 全量操作的時(shí)候還可以配置一個(gè)重新同步的周期參數(shù)，用于協(xié)調(diào)內(nèi)存緩存數(shù)據(jù)和業(yè)務(wù)邏輯的數(shù)據(jù)一致性，每次過(guò)了該周期后，注冊(cè)的事件處理程序就將被所有的對(duì)象調(diào)用，通常這個(gè)周期參數(shù)以分為單位，比如10分鐘或者30分鐘。
Informers 的這些高級(jí)特性以及超強(qiáng)的魯棒性，都足以讓我們不去直接使用客戶端的 Watch() 方法來(lái)處理自己的業(yè)務(wù)邏輯，而且在 Kubernetes 中也有很多地方都有使用到 Informers。但是在使用 Informers 的時(shí)候，通常每個(gè) GroupVersionResource（GVR）只實(shí)例化一個(gè) Informers，但是有時(shí)候我們?cè)谝粋€(gè)應(yīng)用中往往有使用多種資源對(duì)象的需求，這個(gè)時(shí)候?yàn)榱朔奖愎蚕?Informers，我們可以通過(guò)使用共享 Informer 工廠來(lái)實(shí)例化一個(gè) Informer。
共享 Informer 工廠允許我們?cè)趹?yīng)用中為同一個(gè)資源共享 Informer，也就是說(shuō)不同的控制器循環(huán)可以使用相同的 watch 連接到后臺(tái)的 APIServer，例如，kube-controller-manager 中的控制器數(shù)據(jù)量就非常多，但是對(duì)于每個(gè)資源（比如 Pod），在這個(gè)進(jìn)程中只有一個(gè) Informer。
示例
首先我們創(chuàng)建一個(gè) Clientset 對(duì)象，然后使用 Clientset 來(lái)創(chuàng)建一個(gè)共享的 Informer 工廠，Informer 是通過(guò) informer-gen 這個(gè)代碼生成器工具自動(dòng)生成的，位于 k8s.io/client-go/informers 中。
這里我們來(lái)創(chuàng)建一個(gè)用于獲取 Deployment 的共享 Informer，代碼如下所示：
package mainimport ("flag""fmt""path/filepath""time"v1 "k8s.io/api/apps/v1""k8s.io/apimachinery/pkg/labels""k8s.io/client-go/informers""k8s.io/client-go/kubernetes""k8s.io/client-go/rest""k8s.io/client-go/tools/cache""k8s.io/client-go/tools/clientcmd""k8s.io/client-go/util/homedir")func main() {var err errorvar config *rest.Configvar kubeconfig *stringif home := homedir.HomeDir(); home != "" {kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "[可選] kubeconfig 絕對(duì)路徑")} else {kubeconfig = flag.String("kubeconfig", "", "kubeconfig 絕對(duì)路徑")}// 初始化 rest.Config 對(duì)象if config, err = rest.InClusterConfig(); err != nil {if config, err = clientcmd.BuildConfigFromFlags("", *kubeconfig); err != nil {panic(err.Error())}}// 創(chuàng)建 Clientset 對(duì)象clientset, err := kubernetes.NewForConfig(config)if err != nil {panic(err.Error())}// 初始化 informer factory（為了測(cè)試方便這里設(shè)置每30s重新 List 一次）informerFactory := informers.NewSharedInformerFactory(clientset, time.Second*30)// 對(duì) Deployment 監(jiān)聽(tīng)deployInformer := informerFactory.Apps().V1().Deployments()// 創(chuàng)建 Informer（相當(dāng)于注冊(cè)到工廠中去，這樣下面啟動(dòng)的時(shí)候就會(huì)去 List & Watch 對(duì)應(yīng)的資源）informer := deployInformer.Informer()// 創(chuàng)建 ListerdeployLister := deployInformer.Lister()// 注冊(cè)事件處理程序informer.AddEventHandler(cache.ResourceEventHandlerFuncs{AddFunc: onAdd,UpdateFunc: onUpdate,DeleteFunc: onDelete,})stopper := make(chan struct{})defer close(stopper)// 啟動(dòng) informer，List & WatchinformerFactory.Start(stopper)// 等待所有啟動(dòng)的 Informer 的緩存被同步informerFactory.WaitForCacheSync(stopper)// 從本地緩存中獲取 default 中的所有 deployment 列表deployments, err := deployLister.Deployments("default").List(labels.Everything())if err != nil {panic(err)}for idx, deploy := range deployments {fmt.Printf("%d -> %s\n", idx+1, deploy.Name)}<-stopper}func onAdd(obj interface{}) {deploy := obj.(*v1.Deployment)fmt.Println("add a deployment:", deploy.Name)}func onUpdate(old, new interface{}) {oldDeploy := old.(*v1.Deployment)newDeploy := new.(*v1.Deployment)fmt.Println("update deployment:", oldDeploy.Name, newDeploy.Name)}func onDelete(obj interface{}) {deploy := obj.(*v1.Deployment)fmt.Println("delete a deployment:", deploy.Name)}
上面的代碼運(yùn)行可以獲得 default 命名空間之下的所有 Deployment 信息以及整個(gè)集群的 Deployment 數(shù)據(jù)：
go run main.goadd a deployment: dingtalk-hookadd a deployment: spin-orcaadd a deployment: argocd-serveradd a deployment: istio-egressgatewayadd a deployment: vault-agent-injectoradd a deployment: rook-ceph-osd-0add a deployment: rook-ceph-osd-2add a deployment: code-server......1 -> nginx2 -> helloworld-v13 -> productpage-v14 -> details-v1......
這是因?yàn)槲覀兪紫韧ㄟ^(guò) Informer 注冊(cè)了事件處理程序，這樣當(dāng)我們啟動(dòng) Informer 的時(shí)候首先會(huì)將集群的全量 Deployment 數(shù)據(jù)同步到本地的緩存中，會(huì)觸發(fā) AddFunc 這個(gè)回調(diào)函數(shù)，然后我們又在下面使用 Lister() 來(lái)獲取 default 命名空間下面的所有 Deployment 數(shù)據(jù)，這個(gè)時(shí)候的數(shù)據(jù)是從本地的緩存中獲取的，所以就看到了上面的結(jié)果，由于我們還配置了每30s重新全量 List 一次，所以正常每30s我們也可以看到所有的 Deployment 數(shù)據(jù)出現(xiàn)在 UpdateFunc 回調(diào)函數(shù)下面，我們也可以嘗試去刪除一個(gè) Deployment，同樣也會(huì)出現(xiàn)對(duì)應(yīng)的 DeleteFunc 下面的事件。
Informers 是 client-go 中非常重要的概念，接下來(lái)我們將仔細(xì)分析 Informers 的實(shí)現(xiàn)原理，由于 Informers 實(shí)現(xiàn)非常復(fù)雜，我們將按照 Informers 的幾個(gè)核心知識(shí)點(diǎn)分別進(jìn)行講解。
K8S進(jìn)階訓(xùn)練營(yíng)，點(diǎn)擊下方圖片了解詳情
