K8S百万资源预list加载数据方案

联邦控制器资源已经超过百万，每次重启时，需要一次性list所有资源，需要10分钟，给服务启动带来很多风险。

现通过主从预list加载数据，避免升级或者leader切换过程中长时间拉取数据。

// IndexField registers Indexer functions to controller manager.
func IndexField(mgr controllerruntime.Manager) error {
    rbIndexerFunc := func(obj client.Object) []string {
       rb, ok := obj.(*v1.RB)
       if !ok {
          return nil
       }
       return util.GetBindingClusterNames(&rb.Spec)
    }

    return utilerrors.NewAggregate([]error{
       mgr.GetFieldIndexer().IndexField(context.TODO(), &v1.RB{}, rbClusterKeyIndex, rbIndexerFunc),
    })
}

这里面通过IndexFiled方法去构建v1.RB资源的Informer （会NewSharedIndexInformer)，并加入到InformerCache全局缓存里面

// IndexField adds an indexer to the underlying cache, using extraction function to get
// value(s) from the given field.  This index can then be used by passing a field selector
// to List. For one-to-one compatibility with "normal" field selectors, only return one value.
// The values may be anything.  They will automatically be prefixed with the namespace of the
// given object, if present.  The objects passed are guaranteed to be objects of the correct type.
func (ip *informerCache) IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error {
    informer, err := ip.GetInformer(ctx, obj)
    if err != nil {
       return err
    }
    return indexByField(informer, field, extractValue)
}

---构建Informer

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
}

---构建Indexer索引器

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

---构建数据Cache

func NewThreadSafeStore(indexers Indexers, indices Indices) ThreadSafeStore {
    return &threadSafeMap{
       items: map[string]interface{}{},
       index: &storeIndex{
          indexers: indexers,
          indices:  indices,
       },
    }
}

--- 加入informerCache的时候调用NewSharedInformer

func (ip *specificInformersMap) addInformerToMap(gvk schema.GroupVersionKind, obj runtime.Object) (*MapEntry, bool, error) {
    ip.mu.Lock()
    defer ip.mu.Unlock()

    // Check the cache to see if we already have an Informer.  If we do, return the Informer.
    // This is for the case where 2 routines tried to get the informer when it wasn't in the map
    // so neither returned early, but the first one created it.
    if i, ok := ip.informersByGVK[gvk]; ok {
       return i, ip.started, nil
    }

    // Create a NewSharedIndexInformer and add it to the map.
    var lw *cache.ListWatch
    lw, err := ip.createListWatcher(gvk, ip)
    if err != nil {
       return nil, false, err
    }
    ni := cache.NewSharedIndexInformer(lw, obj, resyncPeriod(ip.resync)(), cache.Indexers{
       cache.NamespaceIndex: cache.MetaNamespaceIndexFunc,
    })
...

// Start the Informer if need by
// TODO(seans): write thorough tests and document what happens here - can you add indexers?
// can you add eventhandlers?
if ip.started {
    go i.Informer.Run(ip.stop) //Run informer时，创建FIFO和reflector
}

}

--- Run informer时 构建DeltaFIFO

func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {

fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
    KnownObjects:          s.indexer,
    EmitDeltaTypeReplaced: true,
})

cfg := &Config{
    Queue:            fifo,
    ListerWatcher:    s.listerWatcher,
    ObjectType:       s.objectType,
    FullResyncPeriod: s.resyncCheckPeriod,
    RetryOnError:     false,
    ShouldResync:     s.processor.shouldResync,

    Process:           s.HandleDeltas,
    WatchErrorHandler: s.watchErrorHandler,
}
。。。

s.controller.Run(stopCh)
}

--- Controller Run的时候构建v1.RB这个资源对应的Reflector（一种gvk，一个Reflector)

func (c *controller) Run(stopCh <-chan struct{}) {
    defer utilruntime.HandleCrash()
    go func() {
       <-stopCh
       c.config.Queue.Close()
    }()
    r := NewReflector(
       c.config.ListerWatcher,
       c.config.ObjectType,
       c.config.Queue,
       c.config.FullResyncPeriod,
    )
    ...
}

获取Leader后，针对Controller watch的资源建立Informer时，会现从InformerCache里面根据gvk get一下，如果有，就不再New，同时把Informer里面的Indexer，Indexer里面的CacheStorage（里面有之前listAndWatch的数据，都存在这里）带过来。
然后将Informer跟Controller的EventHandler拼接在一起。
--- 切Leader Controller Start时，将Informer.AddEventHandler

_, err := i.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
if err != nil {
    ks.started <- err
    return
}
if !ks.cache.WaitForCacheSync(ctx) {
    // Would be great to return something more informative here
    ks.started <- errors.New("cache did not sync")
}

--- Reflector Run的时候 listAndWatch数据

func (r *Reflector) Run(stopCh <-chan struct{}) {
    klog.V(3).Infof("Starting reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
    wait.BackoffUntil(func() {
       if err := r.ListAndWatch(stopCh); err != nil {
          r.watchErrorHandler(r, err)
       }
    }, r.backoffManager, true, stopCh)
    klog.V(3).Infof("Stopping reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
}

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
    klog.V(3).Infof("Listing and watching %v from %s", r.expectedTypeName, r.name)
    if r.expectedTypeName == "*unstructured.Unstructured" {
       return nil
    }
    
    err := r.list(stopCh)   //LIST
    if err != nil {
       return err
    }
    ...

    w, err := r.listerWatcher.Watch(options)
}