Kubernetes Informer基本原理(5)

Kubernetes

（2）消费—c.processLoop：

消费逻辑就是从 DeltaFifo pop 出对象，然后做两件事情：（1）触发前面注册的 eventhandler （2）更新本地索引缓存 indexer，保持数据和 etcd 一致

func (c *controller) processLoop() {
  for {
    obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
  }
}
### Queue.Pop：
## Queue.Pop是一个带有处理函数的pod方法，首先先看Pod逻辑，即为deltaFifo的pop方法：
func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
  for {                       // 无限循环
    for len(f.queue) == 0 {
      f.cond.Wait()       // 阻塞直到生产端broadcast方法通知
    }
    id := f.queue[0]
    item, ok := f.items[id]
    delete(f.items, id)
    err := process(item)        // 执行处理方法
    if e, ok := err.(ErrRequeue); ok {
      f.addIfNotPresent(id, item)     // 如果处理失败的重新加入到fifo中重新处理
      err = e.Err
    }
    return item, err
  }
}
### c.config.Process：
## c.config.Process是在初始化controller的时候赋值的，即为前面的s.HandleDeltas
### s.HandleDeltas：
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)
        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:
            // Sync events are only propagated to listeners that requested resync
            isSync = true
          case d.Type == 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)
        } 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
}

可以看到上面主要执行两部分逻辑：

s.processor.distribute

#### s.processor.distribute：
### 例如新增通知：s.processor.distribute(addNotification{newObj: d.Object}, false)
### 其中addNotification就是add类型的通知，后面会通过notification结构体的类型来执行不同的eventHandler
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 *processorListener) add(notification interface{}) {
  p.addCh <- notification     // 新增notification到addCh
}