几乎所有的Controller manager 和CRD Controller 都会使用Client-go 的Informer 函数，这样通过Watch 或者Get List 可以获取对应的Object，下面我们从源码分析角度来看一下Client go Informer 的机制。
1kubeClient, err := kubernetes.NewForConfig(cfg) 2if err != nil { 3 klog.Fatalf("Error building kubernetes clientset: %s", err.Error()) 4} 5 6kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30) 7 8controller := NewController(kubeClient, exampleClient, 9 kubeInformerFactory.Apps().V1().Deployments(), 10 exampleInformerFactory.Samplecontroller().V1alpha1().Foos()) 11 12// notice that there is no need to run Start methods in a separate goroutine. (i.e. go kubeInformerFactory.Start(stopCh) 13// Start method is non-blocking and runs all registered informers in a dedicated goroutine. 14kubeInformerFactory.Start(stopCh)
这里的例子是以https://github.com/kubernetes/sample-controller/blob/master/main.go节选，主要以 k8s 默认的Deployment Informer 为例子。可以看到直接使用Client-go Informer 还是非常简单的，先不管NewCOntroller函数里面执行了什么，顺着代码来看一下kubeInformerFactory.Start 都干了啥。
1// Start initializes all requested informers. 2func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) { 3 f.lock.Lock() 4 defer f.lock.Unlock() 5 6 for informerType, informer := range f.informers { 7 if !f.startedInformers[informerType] { 8 go informer.Run(stopCh) 9 f.startedInformers[informerType] = true 10 } 11 } 12}
可以看到这里遍历了f.informers，而informers 的定义我们来看一眼数据结构
1type sharedInformerFactory struct { 2 client kubernetes.Interface 3 namespace string 4 tweakListOptions internalinterfaces.TweakListOptionsFunc 5 lock sync.Mutex 6 defaultResync time.Duration 7 customResync map[reflect.Type]time.Duration 8 9 informers map[reflect.Type]cache.SharedIndexInformer 10 // startedInformers is used for tracking which informers have been started. 11 // This allows Start() to be called multiple times safely. 12 startedInformers map[reflect.Type]bool 13}
我们这里的例子，在运行的时候，f.informers里面含有的内容如下
type *v1.Deployment informer &{0xc000379fa0 <nil> 0xc00038ccb0 {} 0xc000379f80 0xc00033bb00 30000000000 30000000000 0x28e5ec8 false false {0 0} {0 0}}
也就是说，每一种k8s 类型都会有自己的Informer函数。下面我们来看一下这个函数是在哪里注册的，这里以Deployment Informer 为例。
首先回到刚开始初始化kubeClient 的代码，
1controller := NewController(kubeClient, exampleClient, 2 kubeInformerFactory.Apps().V1().Deployments(), 3 exampleInformerFactory.Samplecontroller().V1alpha1().Foos()) 4 5 6deploymentInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ 7 AddFunc: controller.handleObject, 8 UpdateFunc: func(old, new interface{}) { 9 newDepl := new.(*appsv1.Deployment) 10 oldDepl := old.(*appsv1.Deployment) 11 if newDepl.ResourceVersion == oldDepl.ResourceVersion { 12 // Periodic resync will send update events for all known Deployments. 13 // Two different versions of the same Deployment will always have different RVs. 14 return 15 } 16 controller.handleObject(new) 17 }, 18 DeleteFunc: controller.handleObject, 19 })
注意这里的传参， kubeInformerFactory.Apps().V1().Deployments()， 这句话的意思就是指创建一个只关注Deployment 的Informer.
1controller := &Controller{ 2 kubeclientset: kubeclientset, 3 sampleclientset: sampleclientset, 4 deploymentsLister: deploymentInformer.Lister(), 5 deploymentsSynced: deploymentInformer.Informer().HasSynced, 6 foosLister: fooInformer.Lister(), 7 foosSynced: fooInformer.Informer().HasSynced, 8 workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Foos"), 9 recorder: recorder, 10 }
deploymentInformer.Lister() 这里就是初始化了一个Deployment Lister，下面来看一下Lister函数里面做了什么。
1// NewFilteredDeploymentInformer constructs a new informer for Deployment type. 2// Always prefer using an informer factory to get a shared informer instead of getting an independent 3// one. This reduces memory footprint and number of connections to the server. 4func NewFilteredDeploymentInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer { 5 return cache.NewSharedIndexInformer( 6 &cache.ListWatch{ 7 ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { 8 if tweakListOptions != nil { 9 tweakListOptions(&options) 10 } 11 return client.AppsV1().Deployments(namespace).List(options) 12 }, 13 WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { 14 if tweakListOptions != nil { 15 tweakListOptions(&options) 16 } 17 return client.AppsV1().Deployments(namespace).Watch(options) 18 }, 19 }, 20 &appsv1.Deployment{}, 21 resyncPeriod, 22 indexers, 23 ) 24} 25 26func (f *deploymentInformer) defaultInformer(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer { 27 return NewFilteredDeploymentInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions) 28} 29 30func (f *deploymentInformer) Informer() cache.SharedIndexInformer { 31 return f.factory.InformerFor(&appsv1.Deployment{}, f.defaultInformer) 32} 33 34func (f *deploymentInformer) Lister() v1.DeploymentLister { 35 return v1.NewDeploymentLister(f.Informer().GetIndexer()) 36}
注意这里的Lister 函数，它调用了Informer ，然后触发了f.factory.InformerFor ，
这就最终调用了sharedInformerFactory InformerFor函数，
1// InternalInformerFor returns the SharedIndexInformer for obj using an internal 2// client. 3func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer { 4 f.lock.Lock() 5 defer f.lock.Unlock() 6 7 informerType := reflect.TypeOf(obj) 8 informer, exists := f.informers[informerType] 9 if exists { 10 return informer 11 } 12 13 resyncPeriod, exists := f.customResync[informerType] 14 if !exists { 15 resyncPeriod = f.defaultResync 16 } 17 18 informer = newFunc(f.client, resyncPeriod) 19 f.informers[informerType] = informer 20 21 return informer 22}
这里可以看到，informer = newFunc(f.client, resyncPeriod)这句话最终完成了对于informer的创建，并且注册到了Struct object中，完成了前面我们的问题。
下面我们再回到informer start
1// Start initializes all requested informers. 2func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) { 3 f.lock.Lock() 4 defer f.lock.Unlock() 5 6 for informerType, informer := range f.informers { 7 if !f.startedInformers[informerType] { 8 go informer.Run(stopCh) 9 f.startedInformers[informerType] = true 10 } 11 } 12}
这里可以看到，它会遍历所有的informer，然后选择异步调用Informer 的RUN方法。我们来全局看一下Run方法
1func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) { 2 defer utilruntime.HandleCrash() 3 4 fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer) 5 6 cfg := &Config{ 7 Queue: fifo, 8 ListerWatcher: s.listerWatcher, 9 ObjectType: s.objectType, 10 FullResyncPeriod: s.resyncCheckPeriod, 11 RetryOnError: false, 12 ShouldResync: s.processor.shouldResync, 13 14 Process: s.HandleDeltas, 15 } 16 17 func() { 18 s.startedLock.Lock() 19 defer s.startedLock.Unlock() 20 21 s.controller = New(cfg) 22 s.controller.(*controller).clock = s.clock 23 s.started = true 24 }() 25 26 // Separate stop channel because Processor should be stopped strictly after controller 27 processorStopCh := make(chan struct{}) 28 var wg wait.Group 29 defer wg.Wait() // Wait for Processor to stop 30 defer close(processorStopCh) // Tell Processor to stop 31 wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run) 32 wg.StartWithChannel(processorStopCh, s.processor.run) 33 34 defer func() { 35 s.startedLock.Lock() 36 defer s.startedLock.Unlock() 37 s.stopped = true // Don't want any new listeners 38 }() 39 s.controller.Run(stopCh) 40}
首先它根据得到的 key 拆分函数和Store index 创建一个FIFO队列，这个队列是一个先进先出的队列，主要用来保存对象的各种事件。
1func NewDeltaFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter) *DeltaFIFO { 2 f := &DeltaFIFO{ 3 items: map[string]Deltas{}, 4 queue: []string{}, 5 keyFunc: keyFunc, 6 knownObjects: knownObjects, 7 } 8 f.cond.L = &f.lock 9 return f 10}
可以看到这个队列创建的比较简单，就是使用 Map 来存放数据，String 数组来存放队列的 Key。
后面根据client 创建的List 和Watch 函数，还有队列创建了一个 config，下面将根据这个config 来初始化controller. 这个controller是client-go 的Cache controller ，主要用来控制从 APIServer 获得的对象的 cache 以及更新对象。
下面主要关注这个函数调用
wg.StartWithChannel(processorStopCh, s.processor.run)
这里进行了真正的Listering 调用。
1func (p *sharedProcessor) run(stopCh <-chan struct{}) { 2 func() { 3 p.listenersLock.RLock() 4 defer p.listenersLock.RUnlock() 5 for _, listener := range p.listeners { 6 p.wg.Start(listener.run) 7 p.wg.Start(listener.pop) 8 } 9 p.listenersStarted = true 10 }() 11 <-stopCh 12 p.listenersLock.RLock() 13 defer p.listenersLock.RUnlock() 14 for _, listener := range p.listeners { 15 close(listener.addCh) // Tell .pop() to stop. .pop() will tell .run() to stop 16 } 17 p.wg.Wait() // Wait for all .pop() and .run() to stop 18}
主要看 run 方法，还记得前面已经把ADD UPDATE DELETE 注册了自定义的处理函数了吗。这里就实现了前面函数的触发
1func (p *processorListener) run() { 2 // this call blocks until the channel is closed. When a panic happens during the notification 3 // we will catch it, **the offending item will be skipped!**, and after a short delay (one second) 4 // the next notification will be attempted. This is usually better than the alternative of never 5 // delivering again. 6 stopCh := make(chan struct{}) 7 wait.Until(func() { 8 // this gives us a few quick retries before a long pause and then a few more quick retries 9 err := wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) { 10 for next := range p.nextCh { 11 switch notification := next.(type) { 12 case updateNotification: 13 p.handler.OnUpdate(notification.oldObj, notification.newObj) 14 case addNotification: 15 p.handler.OnAdd(notification.newObj) 16 case deleteNotification: 17 p.handler.OnDelete(notification.oldObj) 18 default: 19 utilruntime.HandleError(fmt.Errorf("unrecognized notification: %#v", next)) 20 } 21 } 22 // the only way to get here is if the p.nextCh is empty and closed 23 return true, nil 24 }) 25 26 // the only way to get here is if the p.nextCh is empty and closed 27 if err == nil { 28 close(stopCh) 29 } 30 }, 1*time.Minute, stopCh) 31}
可以看到当p.nexhCh channel 接收到一个对象进入的时候，就会根据通知类型的不同，选择对应的用户注册函数去调用。那么这个channel 谁来向其中传入参数呢
1func (p *processorListener) pop() { 2 defer utilruntime.HandleCrash() 3 defer close(p.nextCh) // Tell .run() to stop 4 5 var nextCh chan<- interface{} 6 var notification interface{} 7 for { 8 select { 9 case nextCh <- notification: 10 // Notification dispatched 11 var ok bool 12 notification, ok = p.pendingNotifications.ReadOne() 13 if !ok { // Nothing to pop 14 nextCh = nil // Disable this select case 15 } 16 case notificationToAdd, ok := <-p.addCh: 17 if !ok { 18 return 19 } 20 if notification == nil { // No notification to pop (and pendingNotifications is empty) 21 // Optimize the case - skip adding to pendingNotifications 22 notification = notificationToAdd 23 nextCh = p.nextCh 24 } else { // There is already a notification waiting to be dispatched 25 p.pendingNotifications.WriteOne(notificationToAdd) 26 } 27 } 28 } 29}
答案就是这个pop 函数，这里会从p.addCh中读取增加的通知，然后转给p.nexhCh 并且保证每个通知只会读取一次。
下面就是最终的Controller run 函数，我们来看看到底干了什么
1// Run begins processing items, and will continue until a value is sent down stopCh. 2// It's an error to call Run more than once. 3// Run blocks; call via go. 4func (c *controller) Run(stopCh <-chan struct{}) { 5 defer utilruntime.HandleCrash() 6 go func() { 7 <-stopCh 8 c.config.Queue.Close() 9 }() 10 r := NewReflector( 11 c.config.ListerWatcher, 12 c.config.ObjectType, 13 c.config.Queue, 14 c.config.FullResyncPeriod, 15 ) 16 r.ShouldResync = c.config.ShouldResync 17 r.clock = c.clock 18 19 c.reflectorMutex.Lock() 20 c.reflector = r 21 c.reflectorMutex.Unlock() 22 23 var wg wait.Group 24 defer wg.Wait() 25 26 wg.StartWithChannel(stopCh, r.Run) 27 28 wait.Until(c.processLoop, time.Second, stopCh) 29}
这里主要的就是wg.StartWithChannel(stopCh, r.Run)，
1// Run starts a watch and handles watch events. Will restart the watch if it is closed. 2// Run will exit when stopCh is closed. 3func (r *Reflector) Run(stopCh <-chan struct{}) { 4 klog.V(3).Infof("Starting reflector %v (%s) from %s", r.expectedType, r.resyncPeriod, r.name) 5 wait.Until(func() { 6 if err := r.ListAndWatch(stopCh); err != nil { 7 utilruntime.HandleError(err) 8 } 9 }, r.period, stopCh) 10}
这里就调用了r.ListAndWatch 方法，这个方法比较复杂，我们慢慢来看。
1// watchHandler watches w and keeps *resourceVersion up to date. 2func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error { 3 start := r.clock.Now() 4 eventCount := 0 5 6 // Stopping the watcher should be idempotent and if we return from this function there's no way 7 // we're coming back in with the same watch interface. 8 defer w.Stop() 9 // update metrics 10 defer func() { 11 r.metrics.numberOfItemsInWatch.Observe(float64(eventCount)) 12 r.metrics.watchDuration.Observe(time.Since(start).Seconds()) 13 }() 14 15loop: 16 for { 17 select { 18 case <-stopCh: 19 return errorStopRequested 20 case err := <-errc: 21 return err 22 case event, ok := <-w.ResultChan(): 23 if !ok { 24 break loop 25 } 26 if event.Type == watch.Error { 27 return apierrs.FromObject(event.Object) 28 } 29 if e, a := r.expectedType, reflect.TypeOf(event.Object); e != nil && e != a { 30 utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a)) 31 continue 32 } 33 meta, err := meta.Accessor(event.Object) 34 if err != nil { 35 utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event)) 36 continue 37 } 38 newResourceVersion := meta.GetResourceVersion() 39 switch event.Type { 40 case watch.Added: 41 err := r.store.Add(event.Object) 42 if err != nil { 43 utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", r.name, event.Object, err)) 44 } 45 case watch.Modified: 46 err := r.store.Update(event.Object) 47 if err != nil { 48 utilruntime.HandleError(fmt.Errorf("%s: unable to update watch event object (%#v) to store: %v", r.name, event.Object, err)) 49 } 50 case watch.Deleted: 51 // TODO: Will any consumers need access to the "last known 52 // state", which is passed in event.Object? If so, may need 53 // to change this. 54 err := r.store.Delete(event.Object) 55 if err != nil { 56 utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", r.name, event.Object, err)) 57 } 58 default: 59 utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event)) 60 } 61 *resourceVersion = newResourceVersion 62 r.setLastSyncResourceVersion(newResourceVersion) 63 eventCount++ 64 } 65 } 66 67 watchDuration := r.clock.Now().Sub(start) 68 if watchDuration < 1*time.Second && eventCount == 0 { 69 r.metrics.numberOfShortWatches.Inc() 70 return fmt.Errorf("very short watch: %s: Unexpected watch close - watch lasted less than a second and no items received", r.name) 71 } 72 klog.V(4).Infof("%s: Watch close - %v total %v items received", r.name, r.expectedType, eventCount) 73 return nil 74}
这里就是真正调用watch 方法，根据返回的watch 事件，将其放入到前面创建的 FIFO 队列中。
最终调用了controller 的POP 方法
1// processLoop drains the work queue. 2// TODO: Consider doing the processing in parallel. This will require a little thought 3// to make sure that we don't end up processing the same object multiple times 4// concurrently. 5// 6// TODO: Plumb through the stopCh here (and down to the queue) so that this can 7// actually exit when the controller is stopped. Or just give up on this stuff 8// ever being stoppable. Converting this whole package to use Context would 9// also be helpful. 10func (c *controller) processLoop() { 11 for { 12 obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process)) 13 if err != nil { 14 if err == FIFOClosedError { 15 return 16 } 17 if c.config.RetryOnError { 18 // This is the safe way to re-enqueue. 19 c.config.Queue.AddIfNotPresent(obj) 20 } 21 } 22 } 23}
前面是将 watch 到的对象加入到队列中，这里的goroutine 就是用来消费的。具体的消费函数就是前面创建的Process 函数
1func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error { 2 s.blockDeltas.Lock() 3 defer s.blockDeltas.Unlock() 4 5 // from oldest to newest 6 for _, d := range obj.(Deltas) { 7 switch d.Type { 8 case Sync, Added, Updated: 9 isSync := d.Type == Sync 10 s.cacheMutationDetector.AddObject(d.Object) 11 if old, exists, err := s.indexer.Get(d.Object); err == nil && exists { 12 if err := s.indexer.Update(d.Object); err != nil { 13 return err 14 } 15 s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync) 16 } else { 17 if err := s.indexer.Add(d.Object); err != nil { 18 return err 19 } 20 s.processor.distribute(addNotification{newObj: d.Object}, isSync) 21 } 22 case Deleted: 23 if err := s.indexer.Delete(d.Object); err != nil { 24 return err 25 } 26 s.processor.distribute(deleteNotification{oldObj: d.Object}, false) 27 } 28 } 29 return nil 30}
这个函数就是根据传进来的obj，先从自己的cache 中取一下，看是否存在，如果存在就代表是Update ，那么更新自己的队列后，调用用户注册的Update 函数，如果不存在，就调用用户的 Add 函数。
到此Client-go 的Informer 流程源码分析基本完毕。