sharedInformer

​ sharedInformer作为informer的另一种实现，相比普通的informer来说，sharedInformer缓解了多组件采用informer监听同一个资源对象带来的性能问题，采用共享的reflector节省带宽，采用共享的deltaFIFO队列暂存资源变更事件，采用共享的indexer缓存索引，只有controller是独占的，避免了同一对象落在不同缓存的问题。

1.源码分析

1.1.sharedInformer初始化

• sharedInformer实例化时，会通过NewSharedInformerFactory工厂进行创建，NewSharedInformerFactory方法最终会调用到NewSharedInformerFactoryWithOptions初始化一个sharedInformerFactory，sharedInformerFactory初始化时会创建一个informers，用于存储不同类型的informer。

• func NewSharedInformerFactory(client kubernetes.Interface, defaultResync time.Duration) SharedInformerFactory {
  	return NewSharedInformerFactoryWithOptions(client, defaultResync)
  }
  
  func NewSharedInformerFactoryWithOptions(client kubernetes.Interface, defaultResync time.Duration, options ...SharedInformerOption) SharedInformerFactory {
  	factory := &sharedInformerFactory{
  		client:           client,
  		namespace:        v1.NamespaceAll,
  		defaultResync:    defaultResync,
  		informers:        make(map[reflect.Type]cache.SharedIndexInformer),
  		startedInformers: make(map[reflect.Type]bool),
  		customResync:     make(map[reflect.Type]time.Duration),
  	}
  	  
  	// Apply all options
  	for _, opt := range options {
  		factory = opt(factory)
  	}
  	  
  	return factory
  }
  

1.2.informer初始化

• informer初始化时会调用sharedInformerFactory的方法进行初始化，并且可以调用不同资源的informer。不同的资源有各自的informer实现，不同资源调用informer()方法会根据类型创建各自的informer，同一类资源会共享同一个informer，避免同一资源对象落在不同informer造成的时序问题。

• 	podInformer := sharedInformers.Core().V1().Pods().Informer()
  	nodeInformer := sharedInformers.Node().V1beta1().RuntimeClasses().Informer()
  

• Informer()方法进行初始化时，会调用sharedInformerFactory的informerFor()方法，调用时会传入defaultInformer()方法用于真正创建informer，informerFor()方法内首先根据传入的类型从sharedInformerFactory的缓存中查找对应的informer，如果存在会直接返回(对应共享informer的特点)，否则根据defaultInformer()方法创建并设置到缓存中。

• func (f *podInformer) defaultInformer(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
      // 创建informer
  	return NewFilteredPodInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
  }
  
  func (f *podInformer) Informer() cache.SharedIndexInformer {
      // 根据类型和回调函数创建informer
  	return f.factory.InformerFor(&corev1.Pod{}, f.defaultInformer)
  }
  
  func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
  	f.lock.Lock()
  	defer f.lock.Unlock()
  
      // 获取informer类型
  	informerType := reflect.TypeOf(obj)
      // 查找map缓存，如果存在直接返回
  	informer, exists := f.informers[informerType]
  	if exists {
  		return informer
  	}
  	// 根据类型查找resync周期
  	resyncPeriod, exists := f.customResync[informerType]
  	if !exists {
  		resyncPeriod = f.defaultResync
  	}
  	// 调用defaultInformer创建informer
  	informer = newFunc(f.client, resyncPeriod)
  	f.informers[informerType] = informer
  
  	return informer
  }
  

• NewFilteredPodInformer中会真正创建informer，注册List&Watch的回调函数。

• func NewFilteredPodInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
  	return cache.NewSharedIndexInformer(
  		&cache.ListWatch{
  			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
  				if tweakListOptions != nil {
  					tweakListOptions(&options)
  				}
                  // 调用apiserver获取pod列表
  				return client.CoreV1().Pods(namespace).List(context.TODO(), options)
  			},
  			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
  				if tweakListOptions != nil {
  					tweakListOptions(&options)
  				}
                  // 调用apiserver监控pod列表
  				return client.CoreV1().Pods(namespace).Watch(context.TODO(), options)
  			},
  		},
  		&corev1.Pod{},
  		resyncPeriod,
  		indexers,
  	)
  }
  

• 这里的List回调函数的api其实本质上还是拿到restClient.Request请求apiserver

• func (c *pods) List(ctx context.Context, opts metav1.ListOptions) (result *v1.PodList, err error) {
  	var timeout time.Duration
  	if opts.TimeoutSeconds != nil {
  		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
  	}
  	result = &v1.PodList{}
  	err = c.client.Get().
  		Namespace(c.ns).
  		Resource("pods").
  		VersionedParams(&opts, scheme.ParameterCodec).
  		Timeout(timeout).
  		Do(ctx).
  		Into(result)
  	return
  }
  

• sharedInformer的构造通过NewSharedIndexInformer完成，这是sharedInformer的实例化方法。sharedIndexInformer实例化时会创建sharedProcessor，设置List&Watch的回调函数，创建Indexer。

• 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
  }
  

• NewIndexer方法会创建cache，其内部包装了threadSafeMap用于存储资源对象并自带索引功能，它的keyFunc是DeletionHandlingMetaNamespaceKeyFunc，即接受一个object，生成它的namespace/name字符串。

• func NewIndexer(keyFunc KeyFunc, indexers Indexers) Indexer {
  	return &cache{
  		cacheStorage: NewThreadSafeStore(indexers, Indices{}),
  		keyFunc:      keyFunc,
  	}
  }
  

1.3.注册EventHandler

• eventHandler事件的注册是通过informer的AddEventHandler方法进行的，调用AddEventHandler方法的时候，传入一个cache.ResourceEventHandlerFuncs结构体。AddEventHandler方法会调用到AddEventHandlerWithResyncPeriod方法中，然后调用newProcessListener方法初始化listener，接着会校验informer是否启动，没有启动会直接将listener添加到processor监听器列表中；否则会加锁将listener添加到processor监听器列表中，然后将indexer中缓存的数据写入到listener中。

• func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) (ResourceEventHandlerRegistration, error) {
  	return s.AddEventHandlerWithResyncPeriod(handler, s.defaultEventHandlerResyncPeriod)
  }
  
  func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) (ResourceEventHandlerRegistration, error) {
  	s.startedLock.Lock()
  	defer s.startedLock.Unlock()
  	...
      
  	// 初始化监听器
  	listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize)
  	
      // 如果informer还没启动，直接将监听器加入到processor的监听器列表
  	if !s.started {
  		return s.processor.addListener(listener), nil
  	}
  
  	s.blockDeltas.Lock()
  	defer s.blockDeltas.Unlock()
  
  	handle := s.processor.addListener(listener)
      // 将indexer中的缓存数据写道listener，即处理deltaFIFO同步到ThreadSafeMap中的数据
  	for _, item := range s.indexer.List() {
  		listener.add(addNotification{newObj: item})
  	}
  	return handle, nil
  }
  

• listener.add方法其实会将缓存数据写道addCh管道，等待回调事件处理

• func (p *processorListener) add(notification interface{}) {
  	p.addCh <- notification
  }
  

• 

1.3.启动sharedIndexInformer

• sharedInformer启动时，会调用NewDeltaFIFOWithOptions方法初始化deltaFIFO队列，初始化config结构体作为创建controller的参数，异步创建controller，然后会调用run启动processor和controller。

• func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
  	...
      // 实例化deltaFIFO
  	fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
  		KnownObjects:          s.indexer,
  		EmitDeltaTypeReplaced: true,
  		Transformer:           s.transform,
  	})
  
      // 初始化其他组件配置
  	cfg := &Config{
  		Queue:            fifo,
          // reflector内部依赖的List/Watch
  		ListerWatcher:    s.listerWatcher,
  		// 消费deltaFIFO的回调
  		Process:           s.HandleDeltas,
  	}
  
  	// 缓存变更检测,它会周期性执行对象比对更新缓存
  	wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
      
      // 启动sharedProcessor
  	wg.StartWithChannel(processorStopCh, s.processor.run)
  
  	// 启动controller
  	s.controller.Run(stopCh)
  }
  

• 

1.3.启动sharedProcessor

• 启动sharedProcessor处理器，遍历所有的listeners监听器，每个listener启动两个协程处理run和pop方法，两个方法合在一起完成事件回调。

• func (p *sharedProcessor) run(stopCh <-chan struct{}) {
  	func() {
  		p.listenersLock.RLock()
  		defer p.listenersLock.RUnlock()
          // 遍历所有listeners，运行run和pop方法
  		for listener := range p.listeners {
  			p.wg.Start(listener.run)
  			p.wg.Start(listener.pop)
  		}
  		p.listenersStarted = true
  	}()
  	<-stopCh
  
  	p.listenersLock.Lock()
  	defer p.listenersLock.Unlock()
      // 退出收尾操作
  	for listener := range p.listeners {
  		close(listener.addCh) // Tell .pop() to stop. .pop() will tell .run() to stop
  	}
  
  	p.listeners = nil
  
  	// Reset to false since no listeners are running
  	p.listenersStarted = false
  
      // 等待所有关联协程退出
  	p.wg.Wait() // Wait for all .pop() and .run() to stop
  }
  

• pop方法利用select语句获取addCh(注册listener时初始化了一个通知)管道中的数据，第一次循环时notification=nil，但p.addCh已经放入了一个addNotification{newObj: item}通知，此时会走case notificationToAdd, ok := <-p.addCh初始化notication，并设置nextCh=p.nextCh；第二次循环时会不停向pendingNotifications暂存区写数据，并不停读取放到nextCh(此时实际上是p.nextCh)中，供listener.run处理。其实从执行顺序上来说，s.processor更像先启动系统内部的listener进行监听处理，而kubebuilder controller注册的listener则通过AddEventHandler注册后执行，主要负责监听controller感兴趣的资源对象。

1.4.添加eventHandler

• sharedIndexInformer支持动态添加ResourceEventHandler事件方法，根据传入的handler对象构建listener监听器，然后把监听器加到listeners数组里，并启动run和pop两个协程。

• func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) (ResourceEventHandlerRegistration, error) {
  	return s.AddEventHandlerWithResyncPeriod(handler, s.defaultEventHandlerResyncPeriod)
  }
  
  const minimumResyncPeriod = 1 * time.Second
  
  func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) (ResourceEventHandlerRegistration, error) {
  	...
  
  	if resyncPeriod > 0 {
          // 至少1s
  		if resyncPeriod < minimumResyncPeriod {
  			klog.Warningf("resyncPeriod %v is too small. Changing it to the minimum allowed value of %v", resyncPeriod, minimumResyncPeriod)
  			resyncPeriod = minimumResyncPeriod
  		}
  		...
  	}
  
      // 基于注册的ResourceEventHandler实例化一个listener监听对象
  	listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize)
  
      // 把构建的listener放到processor的listeners数组里，并启动两个协程处理run和pop
  	if !s.started {
  		return s.processor.addListener(listener), nil
  	}
  	
  	s.blockDeltas.Lock()
  	defer s.blockDeltas.Unlock()
  
      // 同上注册listener
  	handle := s.processor.addListener(listener)
      // 增加资源对象的通知
  	for _, item := range s.indexer.List() {
  		listener.add(addNotification{newObj: item})
  	}
  	return handle, nil
  }
  

• 注

  1.添加eventHandler初始化Listener后，需要启动Listener，因为这种动态添加的Listener没有在sharedProcessor.run时启动

  2.启动添加的Listener后，需要把threadSafeMap中存储的资源对象信息单独通知给Listener一份，因为存量资源对象的事件通知发生在写入ThreadSafeMap时，之后新增的Listener缺失已经存储的资源对象通知信息

  3.Listener启动后会监听p.addCh管道，新进来的事件经过HandleDeltas处理时放入该管道，触发Listener通知

1.5.HandleDeltas

• HandleDeltas用来处理DeltaFIFO拿到的deltas事件列表，然后通知给所有Listener去处理

• func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
  	s.blockDeltas.Lock()
  	defer s.blockDeltas.Unlock()
  
  	if deltas, ok := obj.(Deltas); ok {
  		return processDeltas(s, s.indexer, deltas)
  	}
  	return errors.New("object given as Process argument is not Deltas")
  }
  

• 当delta事件类型位Sync,Replaced,Added,Updated时，判断在store中是否存储该对象，存在则执行store.update操作和OnUpdate回调，反之则执行store.add和OnAdd回调

• 当delta事件类型为deleted时，则执行store.Delete操作和OnDelete回调

• sharedIndexInformer设计里，store是Indexer存储索引，其内部利用threadSafeMap存储资源对象，handler则是sharedIndexInformer自身实现的ResourceEventHandler接口

• func processDeltas(
  	// Object which receives event notifications from the given deltas
  	handler ResourceEventHandler,
  	clientState Store,
  	deltas Deltas,
  ) error {
  	// from oldest to newest
  	for _, d := range deltas {
  		obj := d.Object
  
  		switch d.Type {
  		case Sync, Replaced, Added, Updated:
  			if old, exists, err := clientState.Get(obj); err == nil && exists {
  				if err := clientState.Update(obj); err != nil {
  					return err
  				}
  				handler.OnUpdate(old, obj)
  			} else {
  				if err := clientState.Add(obj); err != nil {
  					return err
  				}
  				handler.OnAdd(obj)
  			}
  		case Deleted:
  			if err := clientState.Delete(obj); err != nil {
  				return err
  			}
  			handler.OnDelete(obj)
  		}
  	}
  	return nil
  }
  

• sharedIndexInformer内的eventHandler其实就是调用processor.distribute，将通知事件放到addCh中触发Listener处理回调。

• // Conforms to ResourceEventHandler
  func (s *sharedIndexInformer) OnAdd(obj interface{}) {
  	s.cacheMutationDetector.AddObject(obj)
      // 添加，无需同步
  	s.processor.distribute(addNotification{newObj: obj}, false)
  }
  
  // Conforms to ResourceEventHandler
  func (s *sharedIndexInformer) OnUpdate(old, new interface{}) {
  	isSync := false
      // 如果新旧的resourceVersion相等时，标记isSync，不需要通知Listener更新事件
  	if accessor, err := meta.Accessor(new); err == nil {
  		if oldAccessor, err := meta.Accessor(old); err == nil {
  			isSync = accessor.GetResourceVersion() == oldAccessor.GetResourceVersion()
  		}
  	}
  
  	s.cacheMutationDetector.AddObject(new)
      // 更新事件通知，如果新旧资源没有变化则不需要通知
  	s.processor.distribute(updateNotification{oldObj: old, newObj: new}, isSync)
  }
  
  // Conforms to ResourceEventHandler
  func (s *sharedIndexInformer) OnDelete(old interface{}) {
      // 删除事件通知，无需同步
  	s.processor.distribute(deleteNotification{oldObj: old}, false)
  }
  

• distribute收到变更事件后，遍历通知给所有Listener监听器，这里的通知就是把事件写到Listener的addCh管道中。

• func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
  	p.listenersLock.RLock()
  	defer p.listenersLock.RUnlock()
  
  	for listener, isSyncing := range p.listeners {
  		switch {
  		case !sync:
  			// non-sync messages are delivered to every listener
  			listener.add(obj)
  		case isSyncing:
  			// sync messages are delivered to every syncing listener
  			listener.add(obj)
  		default:
  			// skipping a sync obj for a non-syncing listener
  		}
  	}
  }
  
  func (p *processorListener) add(notification interface{}) {
      // 把 obj 写到 listener 对应的 addCh 管道里.
      p.addCh <- notification
  }
  

• processorListener启动时会开启两个协程执行run()和pop()，pop()负责监听addCh队列把notification对象扔到nextCh管道；run()则会监听nextCh管道，然后根据不同类型调用不同的ResourceEventHandler方法。pop()进行通知事件转运时，会经历addCh->pendingNotifications->nextCh进行事件投递，保证处理过程的并发及非阻塞。这里的addCh和nextCh都是无缓冲的管道，仅作为通知暂存区使用，未设置大小。虽然默认的无缓冲的不定长管道处理能力有限，但K8S中组件采用ringbuffer环形队列缓解了这个问题，例如apiserver的watch缓冲区，kubelet的内部缓冲区等。

• func (p *processorListener) pop() {
      var nextCh chan<- interface{}
      var notification interface{}
  
      for {
          select {
          // 把从 addCh 获取的对象扔到 nextCh 里
          case nextCh <- notification:
              var ok bool
              notification, ok = p.pendingNotifications.ReadOne()
  
              // ok = false, ringbuffer 没有需要处理的
              if !ok {
                  // 既然没有事情要做, 那么就设 nextCh 为 nil.
                  // 当 nextCh 为 nil 时, select 忽略该 case.
                  nextCh = nil
              }
  
          // 从 addCh 获取对象, 如果上一次的 noti 还未扔到 nextCh 里, 那么之后的对象扔到 buffer 里
          case notificationToAdd, ok := <-p.addCh:
              ...
  
              // 当 notification 为空时, 给 nextCh 一个能用的 channel.
              if notification == nil {
                  notification = notificationToAdd
                  nextCh = p.nextCh
              } else {
                  // 如果不为空, 则扔到 ringbuffer 缓冲里
                  p.pendingNotifications.WriteOne(notificationToAdd)
              }
          }
      }
  }   
  
  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 {  
  				notification = notificationToAdd
  				nextCh = p.nextCh
  			} else {  
  				p.pendingNotifications.WriteOne(notificationToAdd)
  			}
  		}
  	}
  }
  

1.6.WaitForCacheSync

• WaitForCacheSync会同步等待informer完成数据同步，直到同步完成才退出，不然一直调用informer.HasSynced判断是否完成同步。

• func WaitForCacheSync(stopCh <-chan struct{}, cacheSyncs ...InformerSynced) bool {
  	err := wait.PollImmediateUntil(syncedPollPeriod,
  		func() (bool, error) {
  			for _, syncFunc := range cacheSyncs {
  				if !syncFunc() {
  					return false, nil
  				}
  			}
  			return true, nil
  		},
  		stopCh)
  	if err != nil {
  		klog.V(2).Infof("stop requested")
  		return false
  	}
  
  	klog.V(4).Infof("caches populated")
  	return true
  }
  

1.7.informer实现原理

• 

2.sharedInformerFactory

2.1.informer获取

• 针对某一资源，可以通过InformerFor获取资源对象的informer，这里的informer指的就是sharedIndexInformer。这里以pod资源举例，通过informers.Core().V1().Pods().Informer()可以获取pods资源的informer。其内部会调用informerFor来寻找各个资源类型的cache.SharedIndexInformer，存在会直接返回，否则会调用NewFilteredPodInformer创建SharedIndexInformer共享informer。

• // informers/factory.go
  func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
  	f.lock.Lock()
  	defer f.lock.Unlock()
  
      // 反射获取资源类型
  	informerType := reflect.TypeOf(obj)
      // 根据类型获取informer
  	informer, exists := f.informers[informerType]
  	if exists {
  		return informer
  	}
  
      // 配置resyncPeriod 时长
  	resyncPeriod, exists := f.customResync[informerType]
  	if !exists {
  		resyncPeriod = f.defaultResync
  	}
  
      // 创建一个 sharedIndexInformer 对象，其实就是调用 NewFilteredXXXInformer 方法
  	informer = newFunc(f.client, resyncPeriod)
      // 赋值记录
  	f.informers[informerType] = informer
  
  	return informer
  }
  

• pod资源对象实现了自己的informerFor，它会注册自己的sharedIndexInformer创建函数，通过defaultInformer实现传递给informerFor创建pod资源的informer。

• func (f *podInformer) defaultInformer(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
  	return NewFilteredPodInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
  }
  
  func (f *podInformer) Informer() cache.SharedIndexInformer {
  	return f.factory.InformerFor(&corev1.Pod{}, f.defaultInformer)
  }
  
  func NewFilteredPodInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
  	return cache.NewSharedIndexInformer(
  		&cache.ListWatch{
  			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
  				if tweakListOptions != nil {
  					tweakListOptions(&options)
  				}
  				return client.CoreV1().Pods(namespace).List(context.TODO(), options)
  			},
  			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
  				if tweakListOptions != nil {
  					tweakListOptions(&options)
  				}
  				return client.CoreV1().Pods(namespace).Watch(context.TODO(), options)
  			},
  		},
  		&corev1.Pod{},
  		resyncPeriod,
  		indexers,
  	)
  }
  

2.2.sharedInformerFactory启动

• // 启动当前informers里的所有informer
  func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {
  	f.lock.Lock()
  	defer f.lock.Unlock()
  
  	if f.shuttingDown {
  		return
  	}
  
  	for informerType, informer := range f.informers {
  		if !f.startedInformers[informerType] {
  			f.wg.Add(1)
  			// We need a new variable in each loop iteration,
  			// otherwise the goroutine would use the loop variable
  			// and that keeps changing.
  			informer := informer
  			go func() {
  				defer f.wg.Done()
  				informer.Run(stopCh)
  			}()
  			f.startedInformers[informerType] = true
  		}
  	}
  }
  

2.3.等待缓存同步

• 遍历当前的informers集合，依次调用WaitForCacheSync等待缓存同步完成

• func (f *sharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool {
      // 加锁获取informers map集合，由于等待缓存同步是阻塞的，所以不能长期持有锁，这里会先收集
  	informers := func() map[reflect.Type]cache.SharedIndexInformer {
  		f.lock.Lock()
  		defer f.lock.Unlock()
  
  		informers := map[reflect.Type]cache.SharedIndexInformer{}
  		for informerType, informer := range f.informers {
  			if f.startedInformers[informerType] {
  				informers[informerType] = informer
  			}
  		}
  		return informers
  	}()
  
  	res := map[reflect.Type]bool{}
  	for informType, informer := range informers {
          // 等待各个资源类型完成同步本地缓存
  		res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced)
  	}
  	return res
  }
  

2.4.informerLister

• sharedInformerFactory不仅可以拿到共享的informer实例，也可以拿到Listener实例，其实就是个根据资源对象自己实现的informer获取到对应的Lister，通过Lister实例Pods()获取某个namespace下所有的pods对象，通过List(labels.Selector)可以拿到符合label条件的pods集合。

• func main() {
      // 实例化 informers 集合对象
      informers := informers.NewSharedInformerFactory(client, 0)
  
      // 获取 pod informer
      podInformer := informers.Core().V1().Pods().Informer()
  
      // 获取 pod lister 实例
      podLister := informers.Core().V1().Pods().Lister()
  
      // 从缓存中获取 pods
      podLister.List(labels.Everything())
  }
  

2.5.podLister创建

• 传参informer indexer存储创建podLister对象，这里的indexer底层就是threadSafeMap

• func main() {
      // 实例化 informers 集合对象
      informers := informers.NewSharedInformerFactory(client, 0)
  
      // 获取 pod informer
      podInformer := informers.Core().V1().Pods().Informer()
  
      // 获取 pod lister 实例
      podLister := informers.Core().V1().Pods().Lister()
  
      // 从缓存中获取 pods
      podLister.List(labels.Everything())
  }
  

2.6.遍历查询

• Liter会使用cache.ListAll遍历indexer缓冲的所有对象，然后挑出符合labels条件的pods对象。虽然Lister是遍历查找的过程，在本地会产生一点计算压力，但节省了apiserver端的开销，也减少了因网络访问带来的时延latency。如果自定义的k8s operator有频繁的labels条件查询，可以增加自定义的索引方法indexFunc构建倒排索引，或者通过自定义的ResourceEventHandler自定义更好的倒排索引，毕竟store.Indexer模式的索引键是string，复杂条件查询下性能会受到影响。

• type podLister struct {
      // 内置 indexer store 对象
      indexer cache.Indexer
  }
  
  func (s *podLister) List(selector labels.Selector) (ret []*v1.Pod, err error) {
      // 传递 indexer, selector, 回调方法
      err = cache.ListAll(s.indexer, selector, func(m interface{}) {
          // 把符合条件的 pod 放到 ret 里
          ret = append(ret, m.(*v1.Pod))
      })
      return ret, err
  }
  
  func (s *podLister) Pods(namespace string) PodNamespaceLister {
      // 先从 store.index 里获取获取相关的索引对应的 names,
      // 再从 threadSafeMap 的 items 缓存里通过 names 获取对象集合.
      return podNamespaceLister{indexer: s.indexer, namespace: namespace}
  }
  

• ListAll实现匹配时会遍历每个对象，通过传入的label进行筛选。

• type AppendFunc func(interface{})
  
  func ListAll(store Store, selector labels.Selector, appendFn AppendFunc) error {
      selectAll := selector.Empty()
      // store.List 的内部实现是加锁, 然后把所有的对象放到 slice 里返回.
      for _, m := range store.List() {
          // 空的 selector
          if selectAll {
              appendFn(m)
              continue
          }
  
          metadata, err := meta.Accessor(m)
          // 满足匹配条件则回调添加
          if selector.Matches(labels.Set(metadata.GetLabels())) {
              appendFn(m)
          }
      }
      return nil
  }
  

• store.List()主要是从threadSafeMap中拷贝数据，作为for循环的输入。

• func (c *threadSafeMap) List() []interface{} {
      c.lock.RLock()
      defer c.lock.RUnlock()
  
      list := make([]interface{}, 0, len(c.items))
      for _, item := range c.items {
          list = append(list, item)
      }
      return list
  }