Kubernetes Informer基本原理(4)

Kubernetes

（2）listener pop 方法里会监听 p.addCh，通过 nextCh = p.nextCh 将 addCh 将事件传递给 p.nextCh

（3）listener run 方法里会监听 p.nextCh，收到信号之后，判断是属于什么类型的方法，并且执行前面注册的 Handler

所以后面需要关注当资源对象发生变更时，是如何将变更信号给 p.addCh，进一步触发回调函数的

3.4、Start all informers

通过 sharedInformers.Start(stopCh)启动所有的 informer，代码如下：

// Start initializes all requested informers.
func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {
  for informerType, informer := range f.informers {
    if !f.startedInformers[informerType] {
      go informer.Run(stopCh)
      f.startedInformers[informerType] = true
    }
  }
}

我们的例子中其实就只启动了 PodInformer，接下来看到 podInformer 的 Run 方法做了什么

### go informer.Run(stopCh)：
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}){
  defer utilruntime.HandleCrash()
  fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{   // Deltafifo
    KnownObjects:          s.indexer,
    EmitDeltaTypeReplaced: true,
  })
  cfg := &Config{
    Queue:            fifo,         // Deltafifo
    ListerWatcher:    s.listerWatcher,  // listerWatcher
    ObjectType:       s.objectType,
    FullResyncPeriod: s.resyncCheckPeriod,
    RetryOnError:     false,
    ShouldResync:     s.processor.shouldResync,
    // HandleDeltas, added to process, and done in processloop
    Process:           s.HandleDeltas,
    WatchErrorHandler: s.watchErrorHandler,
  }
  func() {
    ...
    s.controller = New(cfg)
    ...
  }
  
  s.controller.Run(stopCh)
}
### s.controller.Run(stopCh)
func (c *controller) Run(stopCh <-chan struct{}) {
  r := NewReflector(
    c.config.ListerWatcher,
    c.config.ObjectType,
    c.config.Queue,
    c.config.FullResyncPeriod,
  )
  c.reflector = r
  // Run reflector
  wg.StartWithChannel(stopCh, r.Run)  
  // Run processLoop, pop from deltafifo and do ProcessFunc,
  // ProcessFunc is the s.HandleDeltas before
  wait.Until(c.processLoop, time.Second, stopCh)
}

可以看到上面的逻辑首先生成一个 DeltaFifo，然后接下来的逻辑分为两块，生产和消费：

（1）生产—r.Run:

主要的逻辑就是利用 list and watch 将资源对象包括操作类型压入队列 DeltaFifo

#### r.Run：
func (r *Reflector) Run(stopCh <-chan struct{}) {
// 执行listAndWatch
if err := r.ListAndWatch(stopCh);
}
// 执行ListAndWatch流程
func (r *Reflector)ListAndWatch(stopCh <-chan struct{}) error{
  // 1、list：
  // （1）、list pods, 实际调用的是podInformer里的ListFunc方法,
  // client.CoreV1().Pods(namespace).List(context.TODO(), options)
  
  r.listerWatcher.List(opts)
  // （2）、获取资源版本号，用于watch
  resourceVersion = listMetaInterface.GetResourceVersion()
  
  //  （3）、数据转换，转换成列表
  items, err := meta.ExtractList(list)
  
  // （4）、将资源列表中的资源对象和版本号存储到DeltaFifo中
  r.syncWith(items, resourceVersion);
  
  // 2、watch,无限循环去watch apiserver，当watch到事件的时候，执行watchHandler将event事件压入fifo
  for {
    // （1）、watch pods, 实际调用的是podInformer里的WatchFunc方法,
    // client.CoreV1().Pods(namespace).Watch(context.TODO(), options)
    w, err := r.listerWatcher.Watch(options)
    
    // （2）、watchHandler
    // watchHandler watches pod，更新DeltaFifo信息，并且更新resourceVersion
    if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh);
  }
}
### r.watchHandler
// watchHandler watches w and keeps *resourceVersion up to date.
func (r *Reflector) watchHandler(start time.Time, w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error {
    ...
loop:
  for {
    select {
    case event, ok := <-w.ResultChan():
      newResourceVersion := meta.GetResourceVersion()
      switch event.Type {
      case watch.Added:
        err := r.store.Add(event.Object)    // Add event to srore, store的具体方法在fifo中
        if err != nil {
            utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", r.name, event.Object, err))
        }
      ...
      }
      *resourceVersion = newResourceVersion
      r.setLastSyncResourceVersion(newResourceVersion)
      eventCount++
    }
  }
  ...
}
### r.store.Add：
## 即为deltaFifo的add方法：
func (f *DeltaFIFO) Add(obj interface{}) error {
  ...
  return f.queueActionLocked(Added, obj)
  ...
}
func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
  id, err := f.KeyOf(obj)
  if err != nil {
    return KeyError{obj, err}
  }
  newDeltas := append(f.items[id], Delta{actionType, obj})
  newDeltas = dedupDeltas(newDeltas)
  if len(newDeltas) > 0 {
    if _, exists := f.items[id]; !exists {
      f.queue = append(f.queue, id)
    }
    f.items[id] = newDeltas
    f.cond.Broadcast()          // 通知所有阻塞住的消费者
  }
  ...
  return nil
}