Kubernetes Informer基本原理(2)

Kubernetes

上面就是一个简单的 informer 的使用例子，整个过程如上述几个步骤,着重说一下（2）、（3）、（4）、（5）四个步骤

3、流程分析

3.1、New a sharedInformers factory

sharedInformers := informers.NewSharedInformerFactory(clientset, defaultResync)
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),
}

这个过程就是创建一个 informer 的工厂 sharedInformerFactory，sharedInformerFactory 中有一个 informers 对象，里面是一个 informer 的 map，sharedInformerFactory 是为了防止过多的重复 informer 监听 apiserver，导致 apiserver 压力过大，在同一个服务中，不同的 controller 使用同一个 informer

3.2、Register a informer

这个过程主要是生成和注册 informer 到 sharedInformerFactory

podInformer := sharedInformers.Core().V1().Pods().Informer()
func (f *podInformer) Informer() cache.SharedIndexInformer {
  return f.factory.InformerFor(&corev1.Pod{}, f.defaultInformer)
}
### f.factory.InformerFor:
### 注册 informer 
func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
  ...
  informer = newFunc(f.client, resyncPeriod)
  f.informers[informerType] = informer
  return informer
}
### f.defaultInformer:
### 生成 informer
func (f *podInformer) defaultInformer(client k.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
  return NewFilteredPodInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
}
func NewFilteredPodInformer(client k.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,
  )
}
### cache.NewSharedIndexInformer:
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
}