k8s client-go 的 Reflector 源码分析

type Reflector struct {
  // 反射器名称，默认为 文件:行数
  name string
  
  // 期望放到 Store 中的类型名称
  expectedTypeName string
  
  // 放到 Store 中的对象类型
  expectedType reflect.Type
  
  // 期望的 GVK（GroupVersionKind）
  expectedGVK *schema.GroupVersionKind
  
  // 存储目标，用于同步数据
  store Store
  
  // 最重要的接口，用于执行 List 和 Watch 操作
  listerWatcher ListerWatcher
  
  // 处理退避逻辑
  backoffManager wait.BackoffManager
  
  // 重新同步周期
  resyncPeriod time.Duration
  
  // 用于确定是否需要重新同步
  ShouldResync func() bool
  
  // 其他字段...
  lastSyncResourceVersion string
  isLastSyncResourceVersionGone bool
  // ...
}

type ListerWatcher interface {
  // List 返回资源对象列表
  List(options metav1.ListOptions) (runtime.Object, error)
  
  // Watch 从指定版本开始监视资源变化
  Watch(options metav1.ListOptions) (watch.Interface, error)
}

// Run 函数：不断地使用 ListAndWatch 方法获取所有对象和它们的变化
// 当 stopCh 关闭时，Run 函数才会退出
func (r *Reflector) Run(stopCh <-chan struct{}) {
	klog.V(3).Infof("Starting reflector %s (%s) from %s", r.typeDescription, r.resyncPeriod, r.name)
     // 开始工作
	wait.BackoffUntil(func() {
        // 调用 ListAndWatch 方法获取数据
		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.typeDescription, r.resyncPeriod, r.name)
}

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
    // 记录开始监视的资源类型和来源
    klog.V(3).Infof("Listing and watching %v from %s", r.typeDescription, r.name)
    var err error
    var w watch.Interface
    
    // 判断是否使用 watchList 功能（这是一个新特性）
    useWatchList := ptr.Deref(r.UseWatchList, false)
    fallbackToList := !useWatchList

    // 尝试使用 watchList 功能
    if useWatchList {
        w, err = r.watchList(stopCh)
        // 处理各种错误情况，如果watchList失败回退到标准的 LIST/WATCH 模式
        if err != nil {
            fallbackToList = true
            w = nil
        }
    }

    // 如果需要使用标准的 LIST 方式
    if fallbackToList {
        err = r.list(stopCh)  // 执行列表操作
        if err != nil {
            return err
        }
    }

    // 缓存填充完成的日志
    klog.V(2).Infof("Caches populated for %v from %s", r.typeDescription, r.name)

    // 设置 resync 通道和取消通道
    resyncerrc := make(chan error, 1)
    cancelCh := make(chan struct{})
    defer close(cancelCh)
    
    // 启动重新同步协程
    go r.startResync(stopCh, cancelCh, resyncerrc)
    
    // 开始 watch 操作
    return r.watch(w, stopCh, resyncerrc)
}

// watchList establishes a stream to get a consistent snapshot of data
// from the server as described in https://github.com/kubernetes/enhancements/tree/master/keps/sig-api-machinery/3157-watch-list#proposal
//
// case 1: start at Most Recent (RV="", ResourceVersionMatch=ResourceVersionMatchNotOlderThan)
// Establishes a consistent stream with the server.
// That means the returned data is consistent, as if, served directly from etcd via a quorum read.
// It begins with synthetic "Added" events of all resources up to the most recent ResourceVersion.
// It ends with a synthetic "Bookmark" event containing the most recent ResourceVersion.
// After receiving a "Bookmark" event the reflector is considered to be synchronized.
// It replaces its internal store with the collected items and
// reuses the current watch requests for getting further events.
//
// case 2: start at Exact (RV>"0", ResourceVersionMatch=ResourceVersionMatchNotOlderThan)
// Establishes a stream with the server at the provided resource version.
// To establish the initial state the server begins with synthetic "Added" events.
// It ends with a synthetic "Bookmark" event containing the provided or newer resource version.
// After receiving a "Bookmark" event the reflector is considered to be synchronized.
// It replaces its internal store with the collected items and
// reuses the current watch requests for getting further events.

func watchHandler(
    start time.Time,                            // 开始时间，用于计算持续时间
    w watch.Interface,                          // watch 接口，事件来源
    store Store,                                // 存储接口，保存事件对象
    expectedType reflect.Type,                  // 期望的对象类型
    expectedGVK *schema.GroupVersionKind,       // 期望的 GVK
    name string,                                // 反射器名称
    expectedTypeName string,                    // 期望类型名称
    setLastSyncResourceVersion func(string),    // 设置最后同步资源版本的函数
    exitOnInitialEventsEndBookmark *bool,       // WatchList 模式特有参数
    clock clock.Clock,                          // 时钟接口
    errc chan error,                            // 错误通道
    stopCh <-chan struct{},                     // 停止信号通道
)

   eventCount := 0
   if exitOnInitialEventsEndBookmark != nil {
       *exitOnInitialEventsEndBookmark = false
   }

   loop:
   for {
       select {
       case <-stopCh:
           return errorStopRequested
       case err := <-errc:
           return err
       case event, ok := <-w.ResultChan():
           // 处理事件...
       }
   }

   if expectedType != nil {
       if e, a := expectedType, reflect.TypeOf(event.Object); e != a {
           // 处理类型不匹配
           utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", name, e, a))
           continue
       }
   }

   switch event.Type {
   case watch.Added:
       err := store.Add(event.Object)
       // ...
   case watch.Modified:
       err := store.Update(event.Object)
       // ...
   case watch.Deleted:
       err := store.Delete(event.Object)
       // ...
   case watch.Bookmark:
       // 处理 Bookmark 事件
       // ...
   }

   if meta.GetAnnotations()["k8s.io/initial-events-end"] == "true" {
       if exitOnInitialEventsEndBookmark != nil {
           *exitOnInitialEventsEndBookmark = true
       }
   }

   setLastSyncResourceVersion(resourceVersion)
   if rvu, ok := store.(ResourceVersionUpdater); ok {
       rvu.UpdateResourceVersion(resourceVersion)
   }

   if exitOnInitialEventsEndBookmark != nil && *exitOnInitialEventsEndBookmark {
       watchDuration := clock.Since(start)
       klog.V(4).Infof("exiting %v Watch because received the bookmark...", name)
       return nil
   }

   watchDuration := clock.Since(start)
   if watchDuration < 1*time.Second && eventCount == 0 {
       return fmt.Errorf("very short watch: %s: Unexpected watch close...", name)
   }