带你读《云原生应用开发：Operator原理与实践》——2.2.6　Client-go Informer 解析

type Reflector struct {
 name string // 这个 Reflector 的名称，默认为文件 : 行数
 metrics *reflectorMetrics // 用于保存 Reflector 的一些监控指标
 expectedType reflect.Type // 期望放到 Store 中的类型名称
 store Store // 与 Watch 源同步的目标 Store
 listerWatcher ListerWatcher //ListerWatcher 接口，用于指定 List-Watch 方法
 period time.Duration //Watch 周期
 resyncPeriod time.Duration // 重新同步周期
 ShouldResync func() bool
 // clock allows tests to manipulate time
 clock clock.Clock
  lastSyncResourceVersion string // 最后同步的资源的版本号
 lastSyncResourceVersionMutex sync.RWMutex //lastSyncResourceVersion 的读写锁
}

// NewNamedReflector same as NewReflector, but with a specified name for logging
func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, 
store Store, resyncPeriod time.Duration) *Reflector {
 reflectorSuffix := atomic.AddInt64(&reflectorDisambiguator, 1)
 r := &Reflector{
 name: name,
 // we need this to be unique per process (some names are still the same) 
but obvious who it belongs to
 metrics: newReflectorMetrics(makeValidPrometheusMetricLabel(fmt.
Sprintf("reflector_"+name+"_%d", reflectorSuffix))),
 listerWatcher: lw,
 store: store,
 expectedType: reflect.TypeOf(expectedType),
 period: time.Second,
 resyncPeriod: resyncPeriod,
 clock: &clock.RealClock{},
 }
 return r
}

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
 glog.V(3).Infof("Listing and watching %v from %s", r.expectedType, r.name)
 var resourceVersion string
 options := metav1.ListOptions{ResourceVersion: "0"}
 r.metrics.numberOfLists.Inc()
 start := r.clock.Now()
 list, err := r.listerWatcher.List(options)
 if err != nil {
 return fmt.Errorf("%s: Failed to list %v: %v", r.name, r.expectedType, 
err)
 }
 r.metrics.listDuration.Observe(time.Since(start).Seconds())
 listMetaInterface, err := meta.ListAccessor(list)
 if err != nil {
 return fmt.Errorf("%s: Unable to understand list result %#v: %v", 
r.name, list, err)
 }
 resourceVersion = listMetaInterface.GetResourceVersion()
 items, err := meta.ExtractList(list)
 if err != nil {
 return fmt.Errorf("%s: Unable to understand list result %#v (%v)", 
r.name, list, err)
 }
 r.metrics.numberOfItemsInList.Observe(float64(len(items)))
 if err := r.syncWith(items, resourceVersion); err != nil {
 return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
 }
 r.setLastSyncResourceVersion(resourceVersion)
 resyncerrc := make(chan error, 1)
 cancelCh := make(chan struct{})
 defer close(cancelCh)
 go func() {
 resyncCh, cleanup := r.resyncChan()
 defer func() {
 cleanup() // Call the last one written into cleanup
 }()
 for {
 select {
 case <-resyncCh:
 case <-stopCh:
 return
 case <-cancelCh:
 return
 }
 if r.ShouldResync == nil || r.ShouldResync() {
 glog.V(4).Infof("%s: forcing resync", r.name)
 if err := r.store.Resync(); err != nil {
 resyncerrc <- err
 return
 }
 }
 cleanup()
 resyncCh, cleanup = r.resyncChan()
 }
 }()
 for {
 // give the stopCh a chance to stop the loop, even in case of continue 
statements further down on errors
 select {
 case <-stopCh:
 return nil
 default:
 }
 timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 
1.0))
 options = metav1.ListOptions{
 ResourceVersion: resourceVersion,
 TimeoutSeconds: &timeoutSeconds,
 }
 r.metrics.numberOfWatches.Inc()
 w, err := r.listerWatcher.Watch(options)
if err != nil {
 switch err {
 case io.EOF:
 // watch closed normally
 case io.ErrUnexpectedEOF:
 glog.V(1).Infof("%s: Watch for %v closed with unexpected 
EOF: %v", r.name, r.expectedType, err)
 default:
 utilruntime.HandleError(fmt.Errorf("%s: Failed to watch 
%v: %v", r.name, r.expectedType, err))
 }
 if urlError, ok := err.(*url.Error); ok {
 if opError, ok := urlError.Err.(*net.OpError); ok {
 if errno, ok := opError.Err.(syscall.Errno); ok && 
errno == syscall.ECONNREFUSED {
 time.Sleep(time.Second)
 continue
 }
 }
 }
 return nil
 }
 if err := r.watchHandler(w, &resourceVersion, resyncerrc, stopCh); 
err != nil {
 if err != errorStopRequested {
 glog.Warningf("%s: watch of %v ended with: %v", r.name, 
r.expectedType, err)
 }
 return nil
 }
 }
} 

type DeltaFIFO struct {
 lock sync.RWMutex
 cond sync.Cond // 条件变量，唤醒等待的协程
 items map[string]Deltas //Delta 存储桶
 queue []string // 队列存储对象键实际就是和 items 一起形成了一个有序 Map
 // true 通过 Replace() 第一批元素被插入队列或者 Delete/Add/Update 首次被调用
 populated bool
 // Replace() 被首次调用时插入的元素数目
 initialPopulationCount int
 // 函数计算元素 Key 值
 keyFunc KeyFunc
 // 列出已知的对象
 knownObjects KeyListerGetter
 // 队列是否被关闭，关闭互斥锁
 closed bool
 closedLock sync.Mutex
}

type Delta struct {
 Type DeltaType // Delta 类型，比如增、减，后面有详细说明
 Object interface{} // 对象，Delta 的粒度是一个对象
}
type DeltaType string // Delta 的类型用字符串表达
const ( 
 Added DeltaType = "Added" // 增加
 Updated DeltaType = "Updated" // 更新
 Deleted DeltaType = "Deleted" // 删除
 Sync DeltaType = "Sync" // 同步
)
type Deltas []Delta // Delta 数组

type Queue interface {
 Store
 Pop(PopProcessFunc) (interface{}, error)
 AddIfNotPresent(interface{}) error
 HasSynced() bool
 Close()
}

func (f *FIFO) Add(obj interface{}) error {
 id, err := f.keyFunc(obj)
 if err != nil {
 return KeyError{obj, err}
 }
 f.lock.Lock()
 defer f.lock.Unlock() 
 f.populated = true
 if _, exists := f.items[id]; !exists {
 f.queue = append(f.queue, id)
 }
 f.items[id] = obj
 f.cond.Broadcast()
 return nil
}

func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
 f.lock.Lock()
 defer f.lock.Unlock()
 for {
 for len(f.queue) == 0 {
 // When the queue is empty, invocation of Pop() is blocked 
until new item is enqueued.
 // When Close() is called, the f.closed is set and the 
condition is broadcasted.
 // Which causes this loop to continue and return from the 
Pop().
 if f.IsClosed() {
 return nil, FIFOClosedError
 }
 f.cond.Wait()
 }
 id := f.queue[0]
 f.queue = f.queue[1:]
 item, ok := f.items[id]
 if f.initialPopulationCount > 0 {
 f.initialPopulationCount--
 }
 if !ok {
 // Item may have been deleted subsequently.
 continue
 }
 delete(f.items, id)
 err := process(item)
 if e, ok := err.(ErrRequeue); ok {
 f.addIfNotPresent(id, item)
 err = e.Err
 }
 // Don't need to copyDeltas here, because we're transferring
 // ownership to the caller.
 return item, err
 }
}
func (f *DeltaFIFO) KeyOf(obj interface{}) (string, error) {
 if d, ok := obj.(Deltas); ok {
 if len(d) == 0 {
 return "", KeyError{obj, ErrZeroLengthDeltasObject}
 }
 obj = d.Newest().Object
 }
 if d, ok := obj.(DeletedFinalStateUnknown); ok {
 return d.Key, nil
 }
 return f.keyFunc(obj)
}

func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
 ...
 cfg := &Config{
 Queue: fifo,
 ListerWatcher: s.listerWatcher,
 ObjectType: s.objectType,
 FullResyncPeriod: s.resyncCheckPeriod,
 RetryOnError: false,
 ShouldResync: s.processor.shouldResync,
 Process: s.HandleDeltas,
 WatchErrorHandler: s.watchErrorHandler,
 }
 func() {
 s.startedLock.Lock()
 defer s.startedLock.Unlock()
 s.controller = New(cfg)
 s.controller.(*controller).clock = s.clock
 s.started = true
 }()
 ...
 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.clock = c.clock
 c.reflectorMutex.Lock()
 c.reflector = r
 c.reflectorMutex.Unlock()
 ...
 wait.Until(c.processLoop, time.Second, stopCh)
}
func (c *controller) processLoop() {
 for {
 obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
 if err != nil {
 if err == FIFOClosedError {
 return
 }
 if c.config.RetryOnError {
 // This is the safe way to re-enqueue.
 c.config.Queue.AddIfNotPresent(obj)
 }
 }
 }
}

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
}

type Indexer interface {
 Store
 // indexName 索引类，obj 是对象，计算 obj 在 indexName 索引类中的索引键，通过索引键
获取所有的对象
 // 基本就是获取符合 obj 特征的所有对象，所谓的特征就是对象在索引类中的索引键
 Index(indexName string, obj interface{}) ([]interface{}, error)
 // indexKey 是 indexName 索引类中的一个索引键，函数返回 indexKey 指定的所有对象键
 IndexKeys(indexName, indexedValue string) ([]string, error)
 // 获取 indexName 索引类中的所有索引键
 ListIndexFuncValues(indexName string) []string
 // 这个函数和 Index 类似，只是返回值不是对象键，而是所有对象
 ByIndex(indexName, indexedValue string) ([]interface{}, error)
 // 返回 Indexers
 GetIndexers() Indexers
 // 添加 Indexers，就是增加更多的索引分类
 AddIndexers(newIndexers Indexers) error
}

func (c *threadSafeMap) ByIndex(indexName, indexKey string) ([]interface{}, error) {
 c.lock.RLock()
 defer c.lock.RUnlock()
 indexFunc := c.indexers[indexName]
 if indexFunc == nil {
 return nil, fmt.Errorf("Index with name %s does not exist", indexName)
 }
 index := c.indices[indexName]
 set := index[indexKey]
 list := make([]interface{}, 0, set.Len())
 for _, key := range set.List() {
 list = append(list, c.items[key])
 }
 return list, nil
}