1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

type sharedIndexInformer struct {
    indexer       Indexer                 // 本地缓存，底层是 threadSafeMap
    controller    Controller              // 内部 Reflector + DeltaFIFO 的驱动器
    processor     *sharedProcessor        // 多消费者事件分发
    listerWatcher ListerWatcher           // List/Watch 的入口
    objectType    runtime.Object          // 关注的资源类型

    resyncCheckPeriod          time.Duration // 检查是否需要 resync 的周期
    defaultEventHandlerResyncPeriod time.Duration

    clock    clock.Clock
    started  bool
    stopped  bool
}

 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

                  +-----------------+
   List+Watch     |                 |
----------------> |   Reflector     |
                  |                 |
                  +-------+---------+
                          | Deltas (Added/Updated/Deleted/Sync)
                          v
                  +-----------------+
                  |   DeltaFIFO     |   有序、按 key 去重
                  |                 |
                  +-------+---------+
                          | Pop()
                          v
                  +-----------------+
                  |  HandleDeltas   |   Informer 的处理循环
                  |  (Controller)   |
                  +---+----------+--+
                      |          |
       AddOrUpdate    |          | Delete
                      v          v
              +-------------+   +-------------+
              |   Indexer   |   |   Indexer   |
              | (本地缓存)  |   | .Delete()   |
              +------+------+   +-------------+
                     |
                     v
              +-----------------+
              | sharedProcessor |  分发给所有注册的 handler
              +-----------------+
                     |
        +------------+------------+
        v            v            v
   OnAdd/OnUpdate  OnDelete   (各 handler 顺序调用)

 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

func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
    fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer)
    cfg := &Config{
        Queue:            fifo,
        ListerWatcher:    s.listerWatcher,
        ObjectType:       s.objectType,
        Process:          s.HandleDeltas,
        FullResyncPeriod: s.resyncCheckPeriod,
    }
    s.controller = New(cfg)
    s.processor.Run(stopCh)
    s.controller.Run(stopCh)
}

func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
    for _, d := range obj.(Deltas) {
        switch d.Type {
        case Sync, Added, Updated:
            if old, exists, _ := s.indexer.Get(d.Object); exists {
                s.indexer.Update(d.Object)
                s.processor.OnUpdate(old, d.Object)
            } else {
                s.indexer.Add(d.Object)
                s.processor.OnAdd(d.Object)
            }
        case Deleted:
            s.indexer.Delete(d.Object)
            s.processor.OnDelete(d.Object)
        }
    }
    return nil
}

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

type Reflector struct {
    name                string                 // 用于日志和指标
    expectedType        reflect.Type           // 期望的对象类型，用于类型断言
    expectedGVK         *schema.GroupVersionKind
    store               Store                  // 通常是 DeltaFIFO
    listerWatcher       ListerWatcher          // List/Watch 的入口
    resyncPeriod        time.Duration          // resync 周期
    ShouldResync        func() bool            // 自定义 resync 判定
    clock               clock.Clock
    lastSyncResourceVersion string             // 上次同步到的 RV
    isLastSyncResourceVersionUnavailable bool  // RV 是否已经不可用（410）
    paginatedResult     bool                   // List 是否走分页
    WatchListPageSize   int64                  // List 分页大小
    nextResync          time.Time              // 下次 resync 时间
}

 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

func (r *Reflector) Run(stopCh <-chan struct{}) {
    wait.Until(func() {
        if err := r.ListAndWatch(stopCh); err != nil {
            r.watchErrorHandler(r, err)
        }
    }, r.period, stopCh)
}

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
    // 1. List：分页或一次性拉取
    list, listRV, err := r.list(stopCh)
    r.setLastSyncResourceVersion(listRV)
    // 2. 把 List 结果作为 Sync 事件同步到 store
    r.syncWith(list, listRV)
    // 3. 进入 Watch 循环
    for {
        select {
        case <-stopCh:
            return nil
        default:
        }
        // 4. Watch：阻塞读事件，直到出错
        w, err := r.watch(listRV, stopCh)
        if err := r.watchHandler(w, r.store, r.expectedType, ...); err != nil {
            if apierrors.IsResourceExpired(err) || apierrors.IsGone(err) {
                // 410 Gone：重新 List
                r.setLastSyncResourceVersionUnavailable(true)
                return nil
            }
            return err
        }
    }
}

 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

type DeltaFIFO struct {
    lock sync.RWMutex
    cond sync.Cond

    // 实际存储：key -> Deltas
    items map[string]Deltas

    // 按 FIFO 顺序记录的 key 列表
    queue []string

    // 已经在队列里的 key 集合（用于去重，新版用 keys map[string]struct{}）
    keyedMutexs keyedLock

    // 是否已经完成首次 List（Replace）
    populated bool
    // 第一次 Replace 进来的对象数
    initialPopulationCount int

    // 用来从对象算 key 的函数
    keyFunc KeyFunc

    // 已知对象集合，通常是 Indexer
    knownObjects KeyListerGetter

    // 控制是否把 Replaced 事件作为 Replaced 而非 Sync 暴露
    emitDeltaTypeReplaced bool
}

type DeltaType string

const (
    Added    DeltaType = "Added"
    Updated  DeltaType = "Updated"
    Deleted  DeltaType = "Deleted"
    Sync     DeltaType = "Sync"
    Replaced DeltaType = "Replaced"
)

type Delta struct {
    Type   DeltaType
    Object interface{}
}

type Deltas []Delta

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

func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
    f.lock.Lock()
    defer f.lock.Unlock()
    for {
        for len(f.queue) == 0 {
            f.cond.Wait()
        }
        id := f.queue[0]
        f.queue = f.queue[1:]
        item, exists := f.items[id]
        if !exists {
            continue
        }
        delete(f.items, id)
        if err := process(item); err != nil {
            // 重新入队
            f.AddIfNotPresent(item)
            return nil, err
        }
        return item, nil
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
    id := f.keyOf(obj)
    list := f.items[id]
    list = append(list, Delta{actionType, obj})
    // 去重相邻相同类型的 Delta，节省空间（DeletedFinalStateUnknown 特殊处理）
    list = dedupDeltas(list)
    f.items[id] = list
    if !f.keyInQueue(id) {
        f.queue = append(f.queue, id)
        f.cond.Signal()
    }
    return nil
}

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

type Indexer interface {
    Store
    Index(indexName string, obj interface{}) ([]interface{}, error)
    IndexKeys(indexName, indexKey string) ([]string, error)
    ListIndexFuncValues(indexName string) []string
    ByIndex(indexName, indexKey string) ([]interface{}, error)
    GetIndexers() Indexers
    AddIndexers(newIndexers Indexers) error
}

type Store interface {
    Add(obj interface{}) error
    Update(obj interface{}) error
    Delete(obj interface{}) error
    List() []interface{}
    ListKeys() []string
    Get(obj interface{}) (item interface{}, exists bool, err error)
    GetByKey(key string) (item interface{}, exists bool, err error)
    Resync() error
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

type threadSafeMap struct {
    lock  sync.RWMutex
    items map[string]interface{}     // key -> object

    indexers Indexers                 // 索引名 -> 索引函数
    indices  Indices                  // 索引名 -> {索引值 -> set(key)}
}

type Indexers map[string]IndexFunc
type Indices map[string]Index
type Index   map[string]sets.String

// IndexFunc 把对象映射为若干索引值
type IndexFunc func(obj interface{}) ([]string, error)

 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

func (c *threadSafeMap) Add(key string, obj interface{}) {
    c.lock.Lock()
    defer c.lock.Unlock()
    oldObj := c.items[key]
    c.items[key] = obj
    c.updateIndices(oldObj, obj, key)
}

func (c *threadSafeMap) updateIndices(oldObj interface{}, newObj interface{}, key string) {
    for name, indexFunc := range c.indexers {
        // 老对象的索引值集合，从 indices 里删掉 key
        if oldObj != nil {
            oldValues, _ := indexFunc(oldObj)
            for _, v := range oldValues {
                if set, ok := c.indices[name][v]; ok {
                    set.Delete(key)
                }
            }
        }
        // 新对象的索引值集合，写入 indices
        newValues, _ := indexFunc(newObj)
        for _, v := range newValues {
            if c.indices[name] == nil {
                c.indices[name] = Index{}
            }
            if c.indices[name][v] == nil {
                c.indices[name][v] = sets.NewString()
            }
            c.indices[name][v].Insert(key)
        }
    }
}

func (c *threadSafeMap) ByIndex(indexName, indexKey string) ([]interface{}, error) {
    c.lock.RLock()
    defer c.lock.RUnlock()
    set := c.indices[indexName][indexKey]
    var ret []interface{}
    for key := range set {
        ret = append(ret, c.items[key])
    }
    return ret, nil
}

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

type sharedProcessor struct {
    listenersStarted bool
    listenersLock    sync.RWMutex
    listeners        []*processorListener     // 所有已注册的 listener
    syncingListeners []*processorListener     // 还在首次同步的 listener 子集
    clock            clock.Clock
    wg               wait.Group
}

type processorListener struct {
    nextCh chan interface{}      // 给用户回调的 goroutine 读
    addCh  chan interface{}      // distribute 写入

    handler ResourceEventHandler // 用户注册的 OnAdd/OnUpdate/OnDelete

    pendingNotifications buffer.RingGrowing  // 缓冲队列

    requestedResyncPeriod time.Duration
    resyncPeriod          time.Duration
    nextResync            time.Time
    resyncLock            sync.Mutex
}

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

HandleDeltas
   |
   v
sharedProcessor.distribute(obj, sync)
   |
   | (持 listeners 锁)
   v
processorListener.addCh <- obj    // 非阻塞写，缓冲满则走 ring
   |
   v
processorListener.pop()           // 内部 goroutine 把 addCh 和 ring 合并
   |
   v
processorListener.run()           // 内部 goroutine 从 nextCh 读，调 handler
   |
   v
ResourceEventHandler.OnAdd/OnUpdate/OnDelete

 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

package main

import (
    "context"
    "time"

    "k8s.io/client-go/informers"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/cache"
)

func main() {
    var client kubernetes.Interface // 假设已初始化
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    factory := informers.NewSharedInformerFactory(client, time.Minute)
    podInformer := factory.Core().V1().Pods()
    podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) { /* 入队 */ },
        UpdateFunc: func(old, cur interface{}) { /* 入队 */ },
        DeleteFunc: func(obj interface{}) { /* 入队 */ },
    })
    factory.Start(ctx.Done())
    cache.WaitForCacheSync(ctx.Done(), podInformer.Informer().HasSynced)
}