1. 前言

2. 接口与类

2.1 ResourceEventHandler

// client-go/tools/cache/controller.go
type ResourceEventHandler interface {
    OnAdd(obj interface{})
    OnUpdate(oldObj, newObj interface{})
    OnDelete(obj interface{})
}

2.2 processorListener

type processorListener struct {
    nextCh chan interface{}
    addCh  chan interface{}
    // 一个自定义处理数据的handler
    handler ResourceEventHandler
    // 一个环形的buffer 存着那些还没有被分发的notifications
    pendingNotifications buffer.RingGrowing
    // requestedResyncPeriod is how frequently the listener wants a full resync from the shared informer
    requestedResyncPeriod time.Duration
    // informer's overall resync check period.
    resyncPeriod time.Duration
    // 下次要resync的时候
    nextResync time.Time
    resyncLock sync.Mutex
}

func newProcessListener(handler ResourceEventHandler, requestedResyncPeriod, resyncPeriod time.Duration, now time.Time, bufferSize int) *processorListener {
    ret := &processorListener{
        nextCh:                make(chan interface{}),
        addCh:                 make(chan interface{}),
        handler:               handler,
        pendingNotifications:  *buffer.NewRingGrowing(bufferSize),
        requestedResyncPeriod: requestedResyncPeriod,
        resyncPeriod:          resyncPeriod,
    }
    ret.determineNextResync(now)
    return ret
}
func (p *processorListener) determineNextResync(now time.Time) {
    p.resyncLock.Lock()
    defer p.resyncLock.Unlock()
    // now加上该listener的resyncPeriod就是下次要resync的时间
    p.nextResync = now.Add(p.resyncPeriod)
}

add

func (p *processorListener) add(notification interface{}) {
    p.addCh <- notification
}

pop 和 run

func (p *processorListener) pop() {
    defer utilruntime.HandleCrash()
    defer close(p.nextCh) // Tell .run() to stop
    var nextCh chan<- interface{}
    var notification interface{}
    for {
        select {
        case nextCh <- notification:
            // Notification dispatched
            var ok bool
            // notification从缓冲区pendingNotifications中读 然后传递给nextCh
            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 { 
                // 如果notification还没有初始化 则进行初始化notification和nextCh
                notification = notificationToAdd
                nextCh = p.nextCh
            } else { // There is already a notification waiting to be dispatched
                // 直接往pendingNotifications中写
                p.pendingNotifications.WriteOne(notificationToAdd)
            }
        }
    }
}
func (p *processorListener) run() {
    stopCh := make(chan struct{})
    wait.Until(func() {
        // this gives us a few quick retries before a long pause and then a few more quick retries
        err := wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
            // 从nextCh读取并调用该listener的handler进行处理
            for next := range p.nextCh {
                switch notification := next.(type) {
                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
            return true, nil
        })

        // the only way to get here is if the p.nextCh is empty and closed
        if err == nil {
            close(stopCh)
        }
    }, 1*time.Minute, stopCh)
}

2.3 sharedProcessor

type sharedProcessor struct {
    // 判断listeners有没有启动
    listenersStarted bool
    listenersLock    sync.RWMutex
    // 所有的processorListener
    listeners        []*processorListener
    // 所有的需要sync的processorListener 动态变化
    syncingListeners []*processorListener
    clock            clock.Clock
    wg               wait.Group
}

resyncCheckPeriodChanged

func (p *sharedProcessor) resyncCheckPeriodChanged(resyncCheckPeriod time.Duration) {
    p.listenersLock.RLock()
    defer p.listenersLock.RUnlock()

    for _, listener := range p.listeners {
        // 根据listener自己要求的requestedResyncPeriod和resyncCheckPeriod来决定该listener真正的resyncPeriod
        resyncPeriod := determineResyncPeriod(listener.requestedResyncPeriod, resyncCheckPeriod)
        listener.setResyncPeriod(resyncPeriod)
    }
}
// 1. 如果desired或check其中一个是0 则返回0
// 2. 返回max(desired, check)
func determineResyncPeriod(desired, check time.Duration) time.Duration {
    if desired == 0 {
        return desired
    }
    if check == 0 {
        klog.Warningf("The specified resyncPeriod %v is invalid because this shared informer doesn't support resyncing", desired)
        return 0
    }
    if desired < check {
        klog.Warningf("The specified resyncPeriod %v is being increased to the minimum resyncCheckPeriod %v", desired, check)
        return check
    }
    return desired
}

shouldResync

func (p *sharedProcessor) shouldResync() bool {
    p.listenersLock.Lock()
    defer p.listenersLock.Unlock()
    p.syncingListeners = []*processorListener{}
    resyncNeeded := false
    now := p.clock.Now()
    for _, listener := range p.listeners {
        if listener.shouldResync(now) {
            resyncNeeded = true
            p.syncingListeners = append(p.syncingListeners, listener)
            listener.determineNextResync(now)
        }
    }
    return resyncNeeded
}

run

func (p *sharedProcessor) run(stopCh <-chan struct{}) {
    func() {
        p.listenersLock.RLock()
        defer p.listenersLock.RUnlock()
        // 以goroutine的方式启动所有的listeners监听
        for _, listener := range p.listeners {
            p.wg.Start(listener.run)
            p.wg.Start(listener.pop)
        }
        p.listenersStarted = true
    }()
    // 等待有信号告知退出
    <-stopCh
    p.listenersLock.RLock()
    defer p.listenersLock.RUnlock()
    // 关闭所有listener的addCh channel
    for _, listener := range p.listeners {
        // 通知pop()停止 pop()会告诉run()停止
        close(listener.addCh) // Tell .pop() to stop. .pop() will tell .run() to stop
    }
    // 等待所有的pop()和run()方法退出
    p.wg.Wait() // Wait for all .pop() and .run() to stop
}

其余方法

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)
    }
}

func (p *sharedProcessor) addListenerLocked(listener *processorListener) {
    p.listeners = append(p.listeners, listener)
    p.syncingListeners = append(p.syncingListeners, listener)
}

// 分发信息
func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
    p.listenersLock.RLock()
    defer p.listenersLock.RUnlock()

    if sync {
        // 如果是sync操作 只需要分发给那些resync时间到了的listener即可
        for _, listener := range p.syncingListeners {
            listener.add(obj)
        }
    } else {
        // 如果不是sync操作 则通知所有的listeners
        for _, listener := range p.listeners {
            listener.add(obj)
        }
    }
}

informer整体