client-go SharedInformer

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

func main() {
    var err error
    var config *rest.Config
    var kubeconfig *string

    if home := homedir.HomeDir(); home != "" {
        kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "[可选] kubeconfig 绝对路径")
    } else {
        kubeconfig = flag.String("kubeconfig", "", "kubeconfig 绝对路径")
    }
    flag.Parse()

    // 初始化 *rest.Config 对象
    if config, err = rest.InClusterConfig(); err != nil {
        if config, err = clientcmd.BuildConfigFromFlags("", *kubeconfig); err != nil {
            panic(err.Error())
        }
    }

    // 创建 *ClientSet 对象
    clientSet, err := kubernetes.NewForConfig(config)
    if err != nil {
        panic(err.Error())
    }

    // 初始化 SharedInformerFactory，10s 做一次 resync：Reflector 定时遍历 Indexer 缓存，把全部对象以 Sync 事件重新入队（不访问 API Server）
    sharedInformerFactory := informers.NewSharedInformerFactory(clientSet, time.Second*10)

    // 每种kubernetes内置资源都实现了Informer
    // 如创建podInformer：podInformer := sharedInformerFactory.Core().V1().Pods()
    // 下面创建deploymentInformer
    deployInformer := sharedInformerFactory.Apps().V1().Deployments()

    // 实际上调用了 InformerFor(&appsv1.Deployment{}, f.defaultInformer) 构造 deployInformer 中的 SharedIndexInformer
    informer := deployInformer.Informer()

    // 创建 DeploymentLister，拥有list方法
    //type deploymentLister struct {
    //    indexer cache.Indexer
    //}
    deployLister := deployInformer.Lister()

    // 注册事件处理程序
    informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    onAdd,
        UpdateFunc: onUpdate,
        DeleteFunc: onDelete,
    })

    // 程序进程退出前，通知 informer 退出
    stopper := make(chan struct{})
	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	go func() { <-quit; close(stopper) }()

    // 启动所有由 factory 创建出来的 informer
    // 这里会触发 SharedIndexInformer 的 Run，进而启动 Reflector/DeltaFIFO/controller
    sharedInformerFactory.Start(stopper)
    // 等待所有启动的 informer 的缓存同步完成
    sharedInformerFactory.WaitForCacheSync(stopper)

    // 从本地缓存中获取 default 中的所有 deployment 列表
    deployments, err := deployLister.Deployments("default").List(labels.Everything())
    if err != nil {
        panic(err)
    }
    for idx, deploy := range deployments {
        fmt.Printf("%d -> %s\n", idx+1, deploy.Name)
    }
    <-stopper
}

// 注意这些回调不能阻塞，handler 在 processorListener 的独立协程中执行，一旦阻塞，事件堆积，容易导致整个Informer事件消费卡住
func onAdd(obj interface{}) {
    deploy := obj.(*v1.Deployment)
    fmt.Println("add a deployment:", deploy.Name)
}

func onUpdate(old, new interface{}) {
    oldDeploy := old.(*v1.Deployment)
    newDeploy := new.(*v1.Deployment)
    fmt.Println("update deployment:", oldDeploy.Name, newDeploy.Name)
}

func onDelete(obj interface{}) {
    deploy := obj.(*v1.Deployment)
    fmt.Println("delete a deployment:", deploy.Name)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

func NewSharedInformerFactory(client kubernetes.Interface, defaultResync time.Duration) SharedInformerFactory

factory := &sharedInformerFactory{
    client:           client,
    namespace:        v1.NamespaceAll,
    defaultResync:    defaultResync, // 示例中我们传入的是10s
    informers:        make(map[reflect.Type]cache.SharedIndexInformer),
    startedInformers: make(map[reflect.Type]bool),
    customResync:     make(map[reflect.Type]time.Duration),
}

1
2
3
4
5
6

type deploymentInformer struct {
    factory          internalinterfaces.SharedInformerFactory
    // TweakListOptionsFunc is a function that transforms a v1.ListOptions
    tweakListOptions internalinterfaces.TweakListOptionsFunc
    namespace        string
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

// 给InformerFor传入两个参数：
// 1. Deployment对象空结构体
// 2. Deployment的默认Informer构造函数（它在调用后返回SharedIndexInformer）
// 返回Deployment对应的SharedIndexInformer
func (f *deploymentInformer) Informer() cache.SharedIndexInformer {
    return f.factory.InformerFor(&appsv1.Deployment{}, f.defaultInformer)
}

// InformerFor的参数defaultInformer方法的实现，返回了构造出的SharedIndexInformer
func (f *deploymentInformer) defaultInformer(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
    // 传入了参数cache.Indexers
    // Indexers的键是索引分类名，为常量namespace，值为IndexFunc，取出对象的namespace值
    return NewFilteredDeploymentInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
}

func NewFilteredDeploymentInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
    // 传入了ListerWatcher对象和Indexer参数来构造SharedIndexInformer
    return cache.NewSharedIndexInformer(
        &cache.ListWatch{
            // ListFunc/WatchFunc函数就是来list和watch APIServer资源的动作，具体实现在clientset里面
            ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                if tweakListOptions != nil {
                    tweakListOptions(&options)
                }
                return client.AppsV1().Deployments(namespace).List(context.TODO(), options)
            },
            WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                if tweakListOptions != nil {
                    tweakListOptions(&options)
                }
                return client.AppsV1().Deployments(namespace).Watch(context.TODO(), options)
            },
        },
        &appsv1.Deployment{},
        resyncPeriod, // NewSharedInformerFactory时传入的10s
        indexers,
    )
}

func NewSharedIndexInformer(lw ListerWatcher, exampleObject runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer {
    realClock := &clock.RealClock{}
    // 最终返回的informer是这个sharedIndexInformer
    sharedIndexInformer := &sharedIndexInformer{
        processor:                       &sharedProcessor{clock: realClock},
        indexer:                         NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers),
        listerWatcher:                   lw,
        objectType:                      exampleObject,
        resyncCheckPeriod:               defaultEventHandlerResyncPeriod, // 10s
        defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod, // 10s
        cacheMutationDetector:           NewCacheMutationDetector(fmt.Sprintf("%T", exampleObject)),
        clock:                           realClock,
    }
    return sharedIndexInformer
}

// 入参：
// 1. 通用对象Object，此处即Deployment{}
// 2. deployment资源类型对应的sharedIndexInformer的构造函数
func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
    f.lock.Lock()
    defer f.lock.Unlock()

    // 反射取得对象类型为Deployment
    informerType := reflect.TypeOf(obj)
    informer, exists := f.informers[informerType]
    if exists {
        return informer
    }

    resyncPeriod, exists := f.customResync[informerType]
    if !exists {
        resyncPeriod = f.defaultResync // 默认是零值，所以赋予我们传入的10s
    }

    // 如果没有此类型的informer的话就新建一个，此例是deployment类型的informer
    informer = newFunc(f.client, resyncPeriod)
    // 填充了sharedInformerFactory的informers字段
    // informers map[reflect.Type]cache.SharedIndexInformer
    f.informers[informerType] = informer
    // 返回sharedIndexInformer实例
    return informer
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

func (f *deploymentInformer) Lister() v1.DeploymentLister {
    // 传入一个参数为cache.Indexer，即前面func (f *deploymentInformer) defaultInformer方法中
    // 初始化的cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}
    return v1.NewDeploymentLister(f.Informer().GetIndexer())
}

func (s *sharedIndexInformer) GetIndexer() Indexer {
    return s.indexer
}

func NewDeploymentLister(indexer cache.Indexer) DeploymentLister {
    return &deploymentLister{indexer: indexer}
}

// deploymentLister implements the DeploymentLister interface.
type deploymentLister struct {
    indexer cache.Indexer
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

// 添加默认的ResyncPeriod事件处理器
func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) {
    s.AddEventHandlerWithResyncPeriod(handler, s.defaultEventHandlerResyncPeriod)
}

type ResourceEventHandlerFuncs struct {
    AddFunc    func(obj interface{})
    UpdateFunc func(oldObj, newObj interface{})
    DeleteFunc func(obj interface{})
}

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
    }

    // 这里示例中是传入的10s
    if resyncPeriod > 0 {
        // 同步周期不能小于1s
        // const minimumResyncPeriod = 1 * time.Second
        if resyncPeriod < minimumResyncPeriod {
            klog.Warningf("resyncPeriod %v is too small. Changing it to the minimum allowed value of %v", resyncPeriod, minimumResyncPeriod)
            resyncPeriod = minimumResyncPeriod
        }

        // 若通过AddEventHandler添加的任何handler的默认resync周期 小于 Reflector定期检查是否需要resync的同步周期
        if resyncPeriod < s.resyncCheckPeriod {
            // 已启动，那么同步周期是resyncCheckPeriod
            if s.started {
                klog.Warningf("resyncPeriod %v is smaller than resyncCheckPeriod %v and the informer has already started. Changing it to %v", resyncPeriod, s.resyncCheckPeriod, s.resyncCheckPeriod)
                resyncPeriod = s.resyncCheckPeriod
            } else {
                // 如果没有启动，则更新resyncCheckPeriod为resyncPeriod，并调整所有监听器的同步周期
                s.resyncCheckPeriod = resyncPeriod
                s.processor.resyncCheckPeriodChanged(resyncPeriod)
            }
        }
    }

    // 用ResourceEventHandler创建processorListener，里面有ringbuffer和两个通道
    listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize)

    // 如果sharedIndexInformer未启动，则把processorListener添加到sharedProcessor里面去。addListener方法在下文有说明
    if !s.started {
        s.processor.addListener(listener)
        return
    }

    // 下面是listener已启动的情况下再添加processorListener
    // 为了安全的加入，按照以下顺序：
    // 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})
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

// 添加processorListener到sharedProcessor里面去，并开启两个协程运行processorListener.run/pop
func (p *sharedProcessor) addListener(listener *processorListener) {
    p.listenersLock.Lock()
    defer p.listenersLock.Unlock()

    p.addListenerLocked(listener)
    // 若已启动，则新添加的listener开启俩协程分别执行run和pop方法
    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 *processorListener) run() {
    // 开启一个协程周期性调用一个匿名函数，从nextCh里面接收事件并根据事件类型交给对应的handler处理
    stopCh := make(chan struct{})
    wait.Until(func() {
        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
        close(stopCh)
    }, 1*time.Second, stopCh)
}

// 
func (p *processorListener) pop() {
    defer utilruntime.HandleCrash()
    defer close(p.nextCh) // Tell .run() to stop

    var nextCh chan <- interface{}
    var notification interface{}
    // 死循环不停的接收notification并发送给handler
    for {
        select {
        // 未初始化的通道进行读写会阻塞的
        // nextCh通道有数据了，这个case段会立马执行，run 的 for range 持续阻塞读取 nextCh，消费跟不上时事件才暂存进环形缓冲区
        // 如果环状buffer里面没数据，则下次有notification了意味着可以直接放到nextCh
        // 如果环状buffer里面有数据，那么后来的notification应该放到buffer里面去
        // 假设第一个notification由run正在处理中的1s内，这时又来了两个notification
        // 第一个notification由于未被run消费，使得nextCh阻塞，第二个notification就会被存入到环状buffer
        case nextCh <- notification:
            // Notification dispatched
            var ok bool
            // 环形缓冲区顺序读取
            notification, ok = p.pendingNotifications.ReadOne()
            if !ok { // Nothing to pop
                nextCh = nil // Disable nextCh通道
            }
        // addCh通道有数据了的情况
        case notificationToAdd, ok := <- p.addCh:
            if !ok {
                // 通道已关闭则直接退出
                return
            }
            // 如果notification为空，说明还没有发送事件给handler
            if notification == nil { // No notification to pop (and pendingNotifications is empty)
                // 把从addCh获取到的notification写入到p.nextCh通道，run函数里面会消费此通道
                notification = notificationToAdd
                nextCh = p.nextCh
            } else { // There is already a notification waiting to be dispatched
                // 到这里说明还有未被消费的notification，先存入到buffer
                p.pendingNotifications.WriteOne(notificationToAdd)
            }
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

// Start initializes all requested informers.
func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {
    f.lock.Lock()
    defer f.lock.Unlock()

    for informerType, informer := range f.informers {
        if !f.startedInformers[informerType] {
            go informer.Run(stopCh)
            f.startedInformers[informerType] = true
        }
    }
}

func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
    defer utilruntime.HandleCrash()

    // 构造一个DeltaFIFO对象
    fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
        KnownObjects:          s.indexer,
        EmitDeltaTypeReplaced: true,
    })

    // 构造Config对象，包含需要的所有配置
    cfg := &Config{
        Queue:            fifo, // DeltaFIFO
        ListerWatcher:    s.listerWatcher, // 在创建资源对应的informer时定义的，每种资源都有其对应的ListerWatcher
        ObjectType:       s.objectType, // 关注的对象类型
        FullResyncPeriod: s.resyncCheckPeriod,
        RetryOnError:     false,
        ShouldResync:     s.processor.shouldResync,

        Process:           s.HandleDeltas, // sharedIndexInformer的方法
        WatchErrorHandler: s.watchErrorHandler,
    }

    func() {
        s.startedLock.Lock()
        defer s.startedLock.Unlock()

        // 构造一个controller对象，
        s.controller = New(cfg)
        s.controller.(*controller).clock = s.clock
        s.started = true
    }()

    // Separate stop channel because Processor should be stopped strictly after controller
    processorStopCh := make(chan struct{})
    var wg wait.Group
    defer wg.Wait()              // Wait for Processor to stop
    defer close(processorStopCh) // Tell Processor to stop
    wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
    wg.StartWithChannel(processorStopCh, s.processor.run) // 启动listener

    defer func() {
        s.startedLock.Lock()
        defer s.startedLock.Unlock()
        s.stopped = true // Don't want any new listeners
    }()
    s.controller.Run(stopCh)
}

func (c *controller) Run(stopCh <-chan struct{}) {
    defer utilruntime.HandleCrash()
    go func() {
        <-stopCh // 阻塞以等待有情况关闭队列
        c.config.Queue.Close()
    }()
    r := NewReflector(
        c.config.ListerWatcher,
        c.config.ObjectType,
        c.config.Queue,
        c.config.FullResyncPeriod,
    )
    r.ShouldResync = c.config.ShouldResync
    r.WatchListPageSize = c.config.WatchListPageSize
    r.clock = c.clock
    if c.config.WatchErrorHandler != nil {
        r.watchErrorHandler = c.config.WatchErrorHandler
    }

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

    var wg wait.Group

    // Reflector开始运行ListAndWatch
    wg.StartWithChannel(stopCh, r.Run)

    // 也是周期性地执行，需要从DeltaFIFO中弹出对象供后续处理
    wait.Until(c.processLoop, time.Second, stopCh)
    wg.Wait()
}

// 调用ListAndWatch获取对象和对应的事件类型
func (r *Reflector) Run(stopCh <-chan struct{}) {
    klog.V(3).Infof("Starting reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
    wait.BackoffUntil(func() {
        if err := r.ListAndWatch(stopCh); err != nil {
            r.watchErrorHandler(r, err)
        }
    }, r.backoffManager, true, stopCh)
    klog.V(3).Infof("Stopping reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
}

// 周期性的执行
func (c *controller) processLoop() {
    for {
        // 从队列中弹出一个对象然后处理它
        // 此处的c.config.Process就是func (s *sharedIndexInformer) HandleDeltas
        obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
        if err != nil {
            if err == ErrFIFOClosed {
                return
            }
            if c.config.RetryOnError {
                // 重新入队
                c.config.Queue.AddIfNotPresent(obj)
            }
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

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
}

// 这儿就把从DeltaFIFO中pop出来的对象，给到processorListener的addCh通道中，中间可以经过ringbuffer放入到nextCh通道，由事件处理器消化它
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
}