go语言K8S 的 informer机制浅析

Kubernetes的控制器模式是其非常重要的一个设计模式，整个Kubernetes定义的资源对象以及其状态都保存在etcd数据库中，通过apiserver对其进行增删查改，而各种各样的控制器需要从apiserver及时获取这些对象，然后将其应用到实际中，即将这些对象的实际状态调整为期望状态，让他们保持匹配。

不过这因为这样，各种控制器需要和apiserver进行频繁交互，需要能够及时获取对象状态的变化，而如果简单的通过暴力轮询的话，会给apiserver造成很大的压力，且效率很低，因此，Kubernetes设计了Informer这个机制，用来作为控制器跟apiserver交互的桥梁，它主要有两方面的作用：

依赖Etcd的List&Watch机制，在本地维护了一份目标对象的缓存。

Etcd的Watch机制能够使客户端及时获知这些对象的状态变化，然后通过List机制，更新本地缓存，这样就在客户端为这些API对象维护了一份和Etcd数据库中几乎一致的数据，然后控制器等客户端就可以直接访问缓存获取对象的信息，而不用去直接访问apiserver，这一方面显著提高了性能，另一方面则大大降低了对apiserver的访问压力；

依赖Etcd的Watch机制，触发控制器等客户端注册到Informer中的事件方法。

客户端可能会对某些对象的某些事件感兴趣，当这些事件发生时，希望能够执行某些操作，比如通过apiserver新建了一个pod，那么kube-scheduler中的控制器收到了这个事件，然后将这个pod加入到其队列中，等待进行调度。

Kubernetes的各个组件本身就内置了非常多的控制器，而自定义的控制器也需要通过Informer跟apiserver进行交互，因此，Informer在Kubernetes中应用非常广泛，本篇文章就重点分析下Informer的机制原理，以加深对其的理解。

使用方法

先来看看Informer是怎么用的，以Endpoint为例，来看下其使用Informer的相关代码：

创建Informer工厂

# client-go/informers/factory.go
sharedInformers := informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(s)())

首先创建了一个SharedInformerFactory，这个结构主要有两个作用：

• 一个是用来作为创建Informer的工厂，典型的工厂模式，在Kubernetes中这种设计模式也很常用；

• 一个是共享Informer，所谓共享，就是多个Controller可以共用同一个Informer，因为不同的Controller可能对同一种API对象感兴趣，这样相同的API对象，缓存就只有一份，通知机制也只有一套，大大提高了效率，减少了资源浪费。

创建对象Informer结构体

# client-go/informers/core/v1/endpoints.go
type EndpointsInformer interface {
 Informer() cache.SharedIndexInformer
 Lister() v1.EndpointsLister
}
endpointsInformer := kubeInformerFactory.Core().V1().Endpoints()

使用InformerFactory创建出对应版本的对象的Informer结构体，如Endpoints对象对应的就是EndpointsInformer结构体，该结构体实现了两个方法：Informer()和Lister()

• 前者用来构建出最终的Informer，即我们本篇文章的重点：SharedIndexInformer，

• 后者用来获取创建出来的Informer的缓存接口：Indexer，该接口可以用来查询缓存的数据，我准备下一篇文章单独介绍其底层如何实现缓存的。

注册事件方法

# Client-go/tools/cache/shared_informer.go
informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
   AddFunc:    onAdd,
   UpdateFunc: func(interface{}, interface{}) { fmt.Println("update not implemented") }, // 此处省略 workqueue 的使用
   DeleteFunc: func(interface{}) { fmt.Println("delete not implemented") },
 })
func onAdd(obj interface{}) {
 node := obj.(*corev1.Endpoint)
 fmt.Println("add a endpoint:", endpoint.Name)
}

这里，首先调用Infomer()创建出来SharedIndexInformer，然后向其中注册事件方法，这样当有对应的事件发生时，就会触发这里注册的方法去做相应的事情。其次调用Lister()获取到缓存接口，就可以通过它来查询Informer中缓存的数据了，而且Informer中缓存的数据，是可以有索引的，这样可以加快查询的速度。

启动Informer

# kubernetes/cmd/kube-controller-manager/app/controllermanager.go
controllerContext.InformerFactory.Start(controllerContext.Stop)

这里InformerFactory的启动，会遍历Factory中创建的所有Informer，依次将其启动。

机制解析

Informer的实现都是在client-go这个库中，通过上述的工厂方法，其实最终创建出来的是一个叫做SharedIndexInformer的结构体：

# k8s.io/client-go/tools/cache/shared_informer.go
type sharedIndexInformer struct {
   indexer    Indexer
   controller Controller
   processor             *sharedProcessor
   cacheMutationDetector MutationDetector
   listerWatcher ListerWatcher
   ......
}
func NewSharedIndexInformer(lw ListerWatcher, exampleObject runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer {
   realClock := &clock.RealClock{}
   sharedIndexInformer := &sharedIndexInformer{
       processor:                       &sharedProcessor{clock: realClock},
       indexer:                         NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers),
       listerWatcher:                   lw,
       objectType:                      exampleObject,
       resyncCheckPeriod:               defaultEventHandlerResyncPeriod,
       defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod,
       cacheMutationDetector:           NewCacheMutationDetector(fmt.Sprintf("％T", exampleObject)),
       clock:                           realClock,
   }
   return sharedIndexInformer
}

可以看到，在创建SharedIndexInformer时，就创建出了processor, indexer等结构，而在Informer启动时，还创建出了controller, fifo queue, reflector等结构。

Reflector

Reflector的作用，就是通过List&Watch的方式，从apiserver获取到感兴趣的对象以及其状态，然后将其放到一个称为”Delta”的先进先出队列中。

所谓的Delta FIFO Queue，就是队列中的元素除了对象本身外，还有针对该对象的事件类型：

type Delta struct {
   Type   DeltaType
   Object interface{}
}

目前有5种Type: Added, Updated, Deleted, Replaced, Resync，所以，针对同一个对象，可能有多个Delta元素在队列中，表示对该对象做了不同的操作，比如短时间内，多次对某一个对象进行了更新操作，那么就会有多个Updated类型的Delta放入到队列中。后续队列的消费者，可以根据这些Delta的类型，来回调注册到Informer中的事件方法。

而所谓的List&Watch，就是

• 先调用该API对象的List接口，获取到对象列表，将它们添加到队列中，Delta元素类型为Replaced，

• 然后再调用Watch接口，持续监听该API对象的状态变化事件，将这些事件按照不同的事件类型，组成对应的Delta类型，添加到队列中，Delta元素类型有Added, Updated, Deleted三种。

此外，Informer还会周期性的发送Resync类型的Delta元素到队列中，目的是为了周期性的触发注册到Informer中的事件方法UpdateFunc，保证对象的期望状态和实际状态一致，该周期是由一个叫做resyncPeriod的参数决定的，在向Informer中添加EventHandler时，可以指定该参数，若为0的话，则关闭该功能。需要注意的是，Resync类型的Delta元素中的对象，是通过Indexer从缓存中获取到的，而不是直接从apiserver中拿的，即这里resync的，其实是”缓存”的对象的期望状态和实际状态的一致性。

根据以上Reflector的机制，依赖Etcd的Watch机制，通过事件来获知对象变化状态，建立本地缓存。即使在Informer中，也没有周期性的调用对象的List接口，正常情况下，List&Watch只会执行一次，即先执行List把数据拉过来，放入队列中，后续就进入Watch阶段。

那什么时候才会再执行List呢？其实就是异常的时候，在List或者Watch的过程中，如果有异常，比如apiserver重启了，那么Reflector就开始周期性的执行List&Watch，直到再次正常进入Watch阶段。为了在异常时段，不给apiserver造成压力，这个周期是一个称为backoff的可变的时间间隔，默认是一个指数型的间隔，即越往后重试的间隔越长，到一定时间又会重置回一开始的频率。而且，为了让不同的apiserver能够均匀负载这些Watch请求，客户端会主动断开跟apiserver的连接，这个超时时间为60秒，然后重新发起Watch请求。此外，在控制器重启过程中，也会再次执行List，所以会观察到之前已经创建好的API对象，又重新触发了一遍AddFunc方法。

从以上这些点，可以看出来，Kubernetes在性能和稳定性的提升上，还是下了很多功夫的。

Controller

这里Controller的作用是通过轮询不断从队列中取出Delta元素，根据元素的类型，一方面通过Indexer更新本地的缓存，一方面调用Processor来触发注册到Informer的事件方法：

# k8s.io/client-go/tools/cache/controller.go
func (c *controller) processLoop() {
   for {
       obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
   }
}

这里的c.config.Process是定义在shared_informer.go中的HandleDeltas()方法：

# k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
   s.blockDeltas.Lock()
   defer s.blockDeltas.Unlock()
   // from oldest to newest
   for _, d := range obj.(Deltas) {
       switch d.Type {
       case Sync, Replaced, Added, Updated:
           s.cacheMutationDetector.AddObject(d.Object)
           if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
               if err := s.indexer.Update(d.Object); err != nil {
                   return err
               }
               isSync := false
               switch {
               case d.Type == Sync:
                   // Sync events are only propagated to listeners that requested resync
                   isSync = true
               case d.Type == Replaced:
                   if accessor, err := meta.Accessor(d.Object); err == nil {
                       if oldAccessor, err := meta.Accessor(old); err == nil {
                           // Replaced events that didn't change resourceVersion are treated as resync events
                           // and only propagated to listeners that requested resync
                           isSync = accessor.GetResourceVersion() == oldAccessor.GetResourceVersion()
                       }
                   }
               }
               s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
           } else {
               if err := s.indexer.Add(d.Object); err != nil {
                   return err
               }
               s.processor.distribute(addNotification{newObj: d.Object}, false)
           }
       case Deleted:
           if err := s.indexer.Delete(d.Object); err != nil {
               return err
           }
           s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
       }
   }
   return nil
}

Processer & Listener

Processer和Listener则是触发事件方法的机制，在创建Informer时，会创建一个Processer，而在向Informer中通过调用AddEventHandler()注册事件方法时，会为每一个Handler生成一个Listener，然后将该Lisener中添加到Processer中，每一个Listener中有两个channel：addCh和nextCh。Listener通过select监听在这两个channel上，当Controller从队列中取出新的元素时，会调用processer来给它的listener发送“通知”，这个“通知”就是向addCh中添加一个元素，即add()，然后一个goroutine就会将这个元素从addCh转移到nextCh，即pop()，从而触发另一个goroutine执行注册的事件方法，即run()。

# k8s.io/client-go/tools/cache/shared_informer.go
func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
   p.listenersLock.RLock()
   defer p.listenersLock.RUnlock()
   if sync {
       for _, listener := range p.syncingListeners {
           listener.add(obj)
       }
   } else {
       for _, listener := range p.listeners {
           listener.add(obj)
       }
   }
}
func (p *processorListener) add(notification interface{}) {
   p.addCh <- notification
}
func (p *processorListener) pop() {
   defer utilruntime.HandleCrash()
   defer close(p.nextCh) // Tell .run() to stop
   var nextCh chan<- interface{}
   var notification interface{}
   for {
       select {
       case nextCh <- notification:
           // Notification dispatched
           var ok bool
           notification, ok = p.pendingNotifications.ReadOne()
           if !ok { // Nothing to pop
               nextCh = nil // Disable this select case
           }
       case notificationToAdd, ok := <-p.addCh:
           if !ok {
               return
           }
           if notification == nil { // No notification to pop (and pendingNotifications is empty)
               // Optimize the case - skip adding to pendingNotifications
               notification = notificationToAdd
               nextCh = p.nextCh
           } else { // There is already a notification waiting to be dispatched
               p.pendingNotifications.WriteOne(notificationToAdd)
           }
       }
   }
}
func (p *processorListener) run() {
   // this call blocks until the channel is closed.  When a panic happens during the notification
   // we will catch it, **the offending item will be skipped!**, and after a short delay (one second)
   // the next notification will be attempted.  This is usually better than the alternative of never
   // delivering again.
   stopCh := make(chan struct{})
   wait.Until(func() {
       for next := range p.nextCh {
           switch notification := next.(type) {
           case updateNotification:
               p.handler.OnUpdate(notification.oldObj, notification.newObj)
           case addNotification:
               p.handler.OnAdd(notification.newObj)
           case deleteNotification:
               p.handler.OnDelete(notification.oldObj)
           default:
               utilruntime.HandleError(fmt.Errorf("unrecognized notification: ％T", next))
           }
       }
       // the only way to get here is if the p.nextCh is empty and closed
       close(stopCh)
   }, 1*time.Second, stopCh)
}

Indexer

Indexer是对缓存进行增删查改的接口，缓存本质上就是用map构建的key:value键值对，都存在items这个map中，key为/：

type threadSafeMap struct {
   lock  sync.RWMutex
   items map[string]interface{}
   // indexers maps a name to an IndexFunc
   indexers Indexers
   // indices maps a name to an Index
   indices Indices
}

而为了加速查询，还可以选择性的给这些缓存添加索引，索引存储在indecies中，所谓索引，就是在向缓存中添加记录时，就将其key添加到索引结构中，在查找时，可以根据索引条件，快速查找到指定的key记录，比如默认有个索引是按照namespace进行索引，可以根据快速找出属于某个namespace的某种对象，而不用去遍历所有的缓存。

Indexer对外提供了Replace(), Resync(), Add(), Update(), Delete(), List(), Get(), GetByKey(), ByIndex()等接口。

来源：https://juejin.cn/post/7158860202865852429