Kubernetes Informer基本原理(3)

Kubernetes

首先通过 f.defaultInformer 方法生成 informer，然后通过 f.factory.InformerFor 方法，将 informer 注册到 sharedInformerFactory

3.3、Register event handler

这个过程展示如何注册一个回调函数，以及如何触发这个回调函数

### podInformer.AddEventHandler：
func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) {
  s.AddEventHandlerWithResyncPeriod(handler, s.defaultEventHandlerResyncPeriod)
}
func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) {
  ...
  listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(),  initialBufferSize)
  if !s.started {
    s.processor.addListener(listener)
    return
  }
  ...
}
### s.processor.addListener(listener)：
func (p *sharedProcessor) addListener(listener *processorListener) {
  p.addListenerLocked(listener)
  if p.listenersStarted {
    p.wg.Start(listener.run)
    p.wg.Start(listener.pop)
  }
}
### listener.run：
func (p *processorListener) run() {
  // this call blocks until the channel is closed.  When a panic happens during the notification
  // we will catch it, **the offending item will be skipped!**, and after a short delay (one second)
  // the next notification will be attempted.  This is usually better than the alternative of never
  // delivering again.
  stopCh := make(chan struct{})
  wait.Until(func() {
    for next := range p.nextCh {
      switch notification := next.(type) {        // 通过next结构体本身的类型来判断事件类型
      case updateNotification:
        p.handler.OnUpdate(notification.oldObj, notification.newObj)
      case addNotification:
        p.handler.OnAdd(notification.newObj)
      case deleteNotification:
        p.handler.OnDelete(notification.oldObj)
      default:
        utilruntime.HandleError(fmt.Errorf("unrecognized notification: %T", next))
      }
    }
    // the only way to get here is if the p.nextCh is empty and closed
    close(stopCh)
  }, 1*time.Second, stopCh)
}
### listener.pop：
func (p *processorListener) pop() {
  var nextCh chan<- interface{}
  var notification interface{}
  for {
    select {
    case nextCh <- notification:
      // Notification dispatched
      var ok bool
      notification, ok = p.pendingNotifications.ReadOne()
      if !ok { // Nothing to pop
        nextCh = nil // Disable this select case
      }
    case notificationToAdd, ok := <-p.addCh:
      if !ok {
        return
      }
      if notification == nil { // No notification to pop (and pendingNotifications is empty)
        // Optimize the case - skip adding to pendingNotifications
        notification = notificationToAdd
        nextCh = p.nextCh
      } else { // There is already a notification waiting to be dispatched
        p.pendingNotifications.WriteOne(notificationToAdd)
      }
    }
  }
}

这个过程总结就是：

（1）AddEventHandler 到 sharedProcessor，注册事件回调函数到 sharedProcessor