Kubernetes informer机制深入剖析

// staging/src/k8s.io/client-go/informers/factory.go
// 这里使用了工厂模式
// 对于k8s内置的对象进行了所有对象的定义, 实现都在informers子库中
// 比如Core.V1.Pod的实现路径
// - staging/src/k8s.io/client-go/informers/core/interface.go
//   - staging/src/k8s.io/client-go/informers/core/v1/interface.go
//     - staging/src/k8s.io/client-go/informers/core/v1/pod.go


// sharedInformerFactory Core().V1().Pods() 接口的实现：
// core.New(f, f.namespace, f.tweakListOptions)
//   return &group{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
//     return v1.New(g.factory, g.namespace, g.tweakListOptions)
//       return &version{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
//         return &podInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
//           // 提供 根据schema定义的list wather接口


type SharedInformerFactory interface { 
  internalinterfaces.SharedInformerFactory
  ForResource(resource schema.GroupVersionResource) (GenericInformer, error)
  WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool


  Admissionregistration() admissionregistration.Interface
  Apps() apps.Interface
  Auditregistration() auditregistration.Interface
  Autoscaling() autoscaling.Interface
  Batch() batch.Interface
  Certificates() certificates.Interface
  Coordination() coordination.Interface
  Core() core.Interface
  Discovery() discovery.Interface
  Events() events.Interface
  Extensions() extensions.Interface
  Flowcontrol() flowcontrol.Interface
  Networking() networking.Interface
  Node() node.Interface
  Policy() policy.Interface
  Rbac() rbac.Interface
  Scheduling() scheduling.Interface
  Settings() settings.Interface
  Storage() storage.Interface
}

// `*sharedIndexInformer` implements SharedIndexInformer and has three
// main components.  One is an indexed local cache, `indexer Indexer`.
// The second main component is a Controller that pulls
// objects/notifications using the ListerWatcher and pushes them into
// a DeltaFIFO --- whose knownObjects is the informer's local cache
// --- while concurrently Popping Deltas values from that fifo and
// processing them with `sharedIndexInformer::HandleDeltas`.  Each
// invocation of HandleDeltas, which is done with the fifo's lock
// held, processes each Delta in turn.  For each Delta this both
// updates the local cache and stuffs the relevant notification into
// the sharedProcessor.  The third main component is that
// sharedProcessor, which is responsible for relaying those
// notifications to each of the informer's clients.
type sharedIndexInformer struct {
  indexer    Indexer
  controller Controller


  processor             *sharedProcessor
  cacheMutationDetector MutationDetector


  listerWatcher ListerWatcher


  // objectType is an example object of the type this informer is
  // expected to handle.  Only the type needs to be right, except
  // that when that is `unstructured.Unstructured` the object's
  // `"apiVersion"` and `"kind"` must also be right.
  objectType runtime.Object


  // resyncCheckPeriod is how often we want the reflector's resync timer to fire so it can call
  // shouldResync to check if any of our listeners need a resync.
  resyncCheckPeriod time.Duration
  // defaultEventHandlerResyncPeriod is the default resync period for any handlers added via
  // AddEventHandler (i.e. they don't specify one and just want to use the shared informer's default
  // value).
  defaultEventHandlerResyncPeriod time.Duration
  // clock allows for testability
  clock clock.Clock


  started, stopped bool
  startedLock      sync.Mutex


  // blockDeltas gives a way to stop all event distribution so that a late event handler
  // can safely join the shared informer.
  blockDeltas sync.Mutex
}

// staging/src/k8s.io/client-go/tools/cache/controller.go


// Run begins processing items, and will continue until a value is sent down stopCh or it is closed.
// It's an error to call Run more than once.
// Run blocks; call via go.
func (c *controller) Run(stopCh <-chan struct{}) {
  defer utilruntime.HandleCrash()
  go func() {
    <-stopCh
    c.config.Queue.Close()
  }()
  r := NewReflector(              // controller的config是shareIndexInformer初始化时设置的
    c.config.ListerWatcher,     // object本身实现的ListerWatcher
    c.config.ObjectType,        // 对象类型
    c.config.Queue,             // 实质是deltaFifo
    c.config.FullResyncPeriod,  // reysync的周期
  )
  r.ShouldResync = c.config.ShouldResync // 具体实现时遍历了所有的listner，只要有一个到期就会进行resync
  r.clock = c.clock


  c.reflectorMutex.Lock()
  c.reflector = r
  c.reflectorMutex.Unlock()


  var wg wait.Group
  defer wg.Wait()


  wg.StartWithChannel(stopCh, r.Run)  // 启动reflector


  wait.Until(c.processLoop, time.Second, stopCh) // processLoop会消费deltaFifo
}


// relector 入口
// Run repeatedly uses the reflector's ListAndWatch to fetch all the
// objects and subsequent deltas.
// Run will exit when stopCh is closed.
func (r *Reflector) Run(stopCh <-chan struct{}) {
  klog.V(2).Infof("Starting reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
  wait.BackoffUntil(func() {
    if err := r.ListAndWatch(stopCh); err != nil {
      utilruntime.HandleError(err)
    }
  }, r.backoffManager, true, stopCh)  // 一旦watch有错， 重新backoff进行listWatch。主要逻辑：1. list（list结束之后会调用syncWith使用Relplace接口把所有的item以replace事件塞给deltaFifo） 2. 启动resync routine 3. 启动watch routine
  klog.V(2).Infof("Stopping reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
}


// deltaFifo消费端
// processLoop drains the work queue.
// TODO: Consider doing the processing in parallel. This will require a little thought
// to make sure that we don't end up processing the same object multiple times
// concurrently.
//
// TODO: Plumb through the stopCh here (and down to the queue) so that this can
// actually exit when the controller is stopped. Or just give up on this stuff
// ever being stoppable. Converting this whole package to use Context would
// also be helpful.
func (c *controller) processLoop() {
  for {
    obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process)) // c.config.Process也是在shareIndexInformer初始化时设置的，是shareIndexInformer.HandleDeltas
    if err != nil {
      if err == ErrFIFOClosed {
        return
      }
      if c.config.RetryOnError {
        // This is the safe way to re-enqueue.
        c.config.Queue.AddIfNotPresent(obj) // 这个obj是某一个对象的所有event的list，所以对一批events看上去得像一个原子操作。
      }
    }
  }
}


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) // 默认情况下这个mutationDetector没有启动，是的dummyDetector，什么也没做。 defaultCacheMutationDetector也只是调用了obj的DeepCopy接口，存了一份原始值和一份深拷贝的值，然后周期性的对这两份值做relect.DeepEqual, 有问题就打error log
            // 先从indexer中查找否有该对象，有则update，否则add。 都先更新本地cache，然后通知client
      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: // resycn是使用Sync类型
          // Sync events are only propagated to listeners that requested resync
          isSync = true
        case d.Type == Replaced: // 初始化时也会使用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) // 通知processor的listeners
      } 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
}


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 *sharedProcessor) addListener(listener *processorListener) {
  p.listenersLock.Lock()
  defer p.listenersLock.Unlock()


  p.addListenerLocked(listener)
  if p.listenersStarted {
    p.wg.Start(listener.run)
    p.wg.Start(listener.pop)
  }
}
// 发现这两个listener已经同流合污，就是同一个实例
func (p *sharedProcessor) addListenerLocked(listener *processorListener) {
  p.listeners = append(p.listeners, listener)
  p.syncingListeners = append(p.syncingListeners, listener)
}

// ResourceEventHandler can handle notifications for events that happen to a
// resource. The events are informational only, so you can't return an
// error.
//  * OnAdd is called when an object is added.
//  * OnUpdate is called when an object is modified. Note that oldObj is the
//      last known state of the object-- it is possible that several changes
//      were combined together, so you can't use this to see every single
//      change. OnUpdate is also called when a re-list happens, and it will
//      get called even if nothing changed. This is useful for periodically
//      evaluating or syncing something.
//  * OnDelete will get the final state of the item if it is known, otherwise
//      it will get an object of type DeletedFinalStateUnknown. This can
//      happen if the watch is closed and misses the delete event and we don't
//      notice the deletion until the subsequent re-list.
type ResourceEventHandler interface {
  OnAdd(obj interface{})
  OnUpdate(oldObj, newObj interface{})
  OnDelete(obj interface{})
}


func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) {
  s.AddEventHandlerWithResyncPeriod(handler, s.defaultEventHandlerResyncPeriod)
}


func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) {
  s.startedLock.Lock()
  defer s.startedLock.Unlock()


  if s.stopped {
    klog.V(2).Infof("Handler %v was not added to shared informer because it has stopped already", handler)
    return
  }


    // 一个informer会share同一个resyncPeriod，以小值为准
  if resyncPeriod > 0 {
    if resyncPeriod < minimumResyncPeriod {
      klog.Warningf("resyncPeriod %d is too small. Changing it to the minimum allowed value of %d", resyncPeriod, minimumResyncPeriod)
      resyncPeriod = minimumResyncPeriod
    }


    if resyncPeriod < s.resyncCheckPeriod {
      if s.started {
        klog.Warningf("resyncPeriod %d is smaller than resyncCheckPeriod %d and the informer has already started. Changing it to %d", resyncPeriod, s.resyncCheckPeriod, s.resyncCheckPeriod)
        resyncPeriod = s.resyncCheckPeriod
      } else {
        // if the event handler's resyncPeriod is smaller than the current resyncCheckPeriod, update
        // resyncCheckPeriod to match resyncPeriod and adjust the resync periods of all the listeners
        // accordingly
        s.resyncCheckPeriod = resyncPeriod
        s.processor.resyncCheckPeriodChanged(resyncPeriod)
      }
    }
  }


    // 新建listener
  listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize)


    // 如果informer还没有启动，则之间添加listener
  if !s.started {
    s.processor.addListener(listener)
    return
  }


    // 如果informer已经启动，那么会暂停deltaFifo的消费，保证时序的一致性
    // 本地cache中的所有item都以add方式先推送给新的listener，然后再重新恢复deltaFifo的消费
  // in order to safely join, we have to
  // 1. stop sending add/update/delete notifications
  // 2. do a list against the store
  // 3. send synthetic "Add" events to the new handler
  // 4. unblock
  s.blockDeltas.Lock()
  defer s.blockDeltas.Unlock()


  s.processor.addListener(listener)
  for _, item := range s.indexer.List() {
    listener.add(addNotification{newObj: item})
  }
}


func (p *sharedProcessor) run(stopCh <-chan struct{}) {
  func() {
    p.listenersLock.RLock()
    defer p.listenersLock.RUnlock()
    for _, listener := range p.listeners {
      p.wg.Start(listener.run)  // 启动一个routine 消费 listener的nextCh
      p.wg.Start(listener.pop)  // 启动一个routine 消费 listener的addCh(通过informer的调用processor.distribute再调用每一个listener.add进行生产。通过addCh 接受消息，使用 pendingNotifications（大小为initialBufferSize，1024）bufferRing向nextCh注入消息。)
    }
    p.listenersStarted = true // 对于后加入的listener会根据这个flag决定是都要启动上面两个routinue（run，pop）
  }()
  <-stopCh
  p.listenersLock.RLock()
  defer p.listenersLock.RUnlock()
  for _, listener := range p.listeners {
    close(listener.addCh) // Tell .pop() to stop. .pop() will tell .run() to stop
  }
  p.wg.Wait() // Wait for all .pop() and .run() to stop
}

//cmd/kube-controller-manager/app/controllermanager.go


func CreateControllerContext(s *config.CompletedConfig, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) (ControllerContext, error) {
......
informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(s)())  // controller-manager初始化sharedInformer入口
......
}


StartControllers(controllerContext, saTokenControllerInitFunc, NewControllerInitializers(controllerContext.LoopMode), unsecuredMux) // NewControllerInitializers会初始化所有的controller， 此时会调用Informer()接口进行真正的实例化（Tips：如果不使用controller，而只是简单的使用lister，由于lister的indexer也是从informer中获取，因此此时也会实例化informer。）。
controllerContext.InformerFactory.Start(controllerContext.Stop) // 启动入口， 所有的informer的Run接口会被调用，此时会初始化这些informer，如果此时没有informer注册（调用），那么啥也不干。直到有client调用 类似sharedInformer.Core().V1().Pods().Informer() 进行实例化时，在调用Informer()接口时才会进行真正的初始化informer。 （Tips：对一个factory新增加了informer，需要重新调用factory的Start方法来启动对应的informer）






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


// 初始化sharedIndexInformer 3件套以及fifo
// 启动reflector processor等等

//staging/src/k8s.io/client-go/tools/cache/index.go


 // Index maps the indexed value to a set of keys in the store that match on that value
type Index map[string]sets.String // 索引值-> value e.g. idxVal(kube-system) -> sets of matched objs keys


// Indexers maps a name to a IndexFunc
type Indexers map[string]IndexFunc // 类型->获取该类型的索引值的func e.g. "namespace"-> idxVal=getNameSpace(obj), e.g. idxVal = kube-system


// Indices maps a name to an Index
type Indices map[string]Index // 类型->该类型的索引 e.g. "namespace" -> Index of  "namespace"


//staging/src/k8s.io/client-go/tools/cache/thread_safe_store.go
// threadSafeMap implements ThreadSafeStore
type threadSafeMap struct {
  lock  sync.RWMutex
  items map[string]interface{}  // 以key为索引的hash表


  // indexers maps a name to an IndexFunc
  indexers Indexers
  // indices maps a name to an Index
  indices Indices
}




//staging/src/k8s.io/client-go/tools/cache/delta_fifo.go


// DeltaFIFO is like FIFO, but differs in two ways.  One is that the
// accumulator associated with a given object's key is not that object
// but rather a Deltas, which is a slice of Delta values for that
// object.  Applying an object to a Deltas means to append a Delta
// except when the potentially appended Delta is a Deleted and the
// Deltas already ends with a Deleted.  In that case the Deltas does
// not grow, although the terminal Deleted will be replaced by the new
// Deleted if the older Deleted's object is a
// DeletedFinalStateUnknown.
//
// The other difference is that DeltaFIFO has an additional way that
// an object can be applied to an accumulator, called Sync.
//
// DeltaFIFO is a producer-consumer queue, where a Reflector is
// intended to be the producer, and the consumer is whatever calls
// the Pop() method.
//
// DeltaFIFO solves this use case:
//  * You want to process every object change (delta) at most once.
//  * When you process an object, you want to see everything
//    that's happened to it since you last processed it.
//  * You want to process the deletion of some of the objects.
//  * You might want to periodically reprocess objects.
//
// DeltaFIFO's Pop(), Get(), and GetByKey() methods return
// interface{} to satisfy the Store/Queue interfaces, but they
// will always return an object of type Deltas.
//
// A DeltaFIFO's knownObjects KeyListerGetter provides the abilities
// to list Store keys and to get objects by Store key.  The objects in
// question are called "known objects" and this set of objects
// modifies the behavior of the Delete, Replace, and Resync methods
// (each in a different way).
//
// A note on threading: If you call Pop() in parallel from multiple
// threads, you could end up with multiple threads processing slightly
// different versions of the same object.
type DeltaFIFO struct {
  // lock/cond protects access to 'items' and 'queue'.
  lock sync.RWMutex
  cond sync.Cond  // 用来通知进行Pop


  // We depend on the property that items in the set are in
  // the queue and vice versa, and that all Deltas in this
  // map have at least one Delta.
  items map[string]Deltas // 根据key(即不同的对象,以pod为例,就是pod的namespace/podname)把所有的event作为Deltas（slice列表）
  queue []string  // 根据先进先出原则，存放key


  // populated is true if the first batch of items inserted by Replace() has been populated
  // or Delete/Add/Update was called first.
  populated bool
  // initialPopulationCount is the number of items inserted by the first call of Replace()
  initialPopulationCount int


  // keyFunc is used to make the key used for queued item
  // insertion and retrieval, and should be deterministic.
  keyFunc KeyFunc


  // knownObjects list keys that are "known" --- affecting Delete(),
  // Replace(), and Resync()
  knownObjects KeyListerGetter // 实际会指向shareIndexInformer的indexer


  // Indication the queue is closed.
  // Used to indicate a queue is closed so a control loop can exit when a queue is empty.
  // Currently, not used to gate any of CRED operations.
  closed     bool
  closedLock sync.Mutex


  // emitDeltaTypeReplaced is whether to emit the Replaced or Sync
  // DeltaType when Replace() is called (to preserve backwards compat).
  emitDeltaTypeReplaced bool // 目前shareIndexInformer使用是初始化为true
}