k8s-Informer之Reflector的解析

Reflect 概述

Reflector 从 kube-apiserver 中 list&watch 资源对象，用于监听指定资源的 Kubernetes 。当资源对象发生变化时（如：添加和删除等事件），Reflector 会将其这些资源对象的变化包装成Delta并将其丢到DeltaFIFO中。其实就是将 Etcd 的对象及其变化反射到DeltaFIFO中，实时更新本地缓存，确保本地数据和 ETCD 数据一致。

源码位置：k8s.io/client-go/tools/cache/reflector.go
（1）Reflector 它的数据结构如下：

 type Reflector struct {name stringexpectedTypeName stringexpectedType reflect.Type // 放到Store中（即DeltaFIFO中）的对象类型expectedGVK *schema.GroupVersionKindstore Store // 与 Watch 源同步的⽬标，会赋值为 DeltaFIFOlisterWatcher ListerWatcher // ListerWatcher是个interface（含list和watch）backoffManager wait.BackoffManagerinitConnBackoffManager wait.BackoffManagerMaxInternalErrorRetryDuration time.DurationresyncPeriod time.Duration // 重新同步周期ShouldResync func() boolclock clock.ClockpaginatedResult boollastSyncResourceVersion stringisLastSyncResourceVersionUnavailable boollastSyncResourceVersionMutex sync.RWMutexWatchListPageSize int64watchErrorHandler WatchErrorHandler
}

（2）Reflector 初始化

通过 NewReflector 实例化 Reflector 对象，在实例中需要传入的 ListerWatcher 数据接口对象，这个包含核心 List 和 Watch 方法，主要是负责 List 和 Watch 指定的 Kubernetes APIServer 资源。


func NewReflector(lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {  return NewNamedReflector(naming.GetNameFromCallsite(internalPackages...), lw, expectedType, store, resyncPeriod)  
}// 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 {  realClock := &clock.RealClock{}  r := &Reflector{  name:          name,  listerWatcher: lw,  store:         store,  initConnBackoffManager: wait.NewExponentialBackoffManager(800*time.Millisecond, 30*time.Second, 2*time.Minute, 2.0, 1.0, realClock),  resyncPeriod:           resyncPeriod,  clock:                  realClock,  watchErrorHandler:      WatchErrorHandler(DefaultWatchErrorHandler),  }  r.setExpectedType(expectedType)  return r  
}

（3）ListerWatcher interface

type Lister interface {  List(options metav1.ListOptions) (runtime.Object, error)  
}  type Watcher interface {  
Watch(options metav1.ListOptions) (watch.Interface, error)  
}  type ListerWatcher interface {  Lister  Watcher
}

（4）ListWatch struct

type ListFunc func(options metav1.ListOptions) (runtime.Object, error)type WatchFunc func(options metav1.ListOptions) (watch.Interface, error)type ListWatch struct {  ListFunc  ListFunc  WatchFunc WatchFunc  DisableChunking bool  
}

（5）Reflector 启动

创建 Reflector 对象后， Run 方法启动监听并处理事件，通过 wait.BackoffUntil 不断调用 ListAndWatch 方法，如果该方法 return 了，那么就会发生re-list，watch过程则被嵌套在for循环中。 Run() 中最核心的就是 List-Watch 方法。

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

ListAndWatch 核心代码：

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {  // ...// list 获取资源下的所有对象的数据  err := r.list(stopCh)  if err != nil {  return err  }  // go 部分go func() {  // 返回重新同步的定时通道  resyncCh, cleanup := r.resyncChan()  // ...for {  select {  case <-resyncCh:  case <-stopCh:  return  case <-cancelCh:  return  }  // 判断是否需要执行Resync操作，即重新同步  if r.ShouldResync == nil || r.ShouldResync() {  klog.V(4).Infof("%s: forcing resync", r.name)  // Resync 机制会将本地存储(LocalStore)的资源对象同步到 DeltaFIFO 中  if err := r.store.Resync(); err != nil {  resyncerrc <- err  return  }  }  cleanup()  // 重新启⽤定时器定时触发  resyncCh, cleanup = r.resyncChan()  }  }()  // for 部分 for {  // 1、stopCh处理，判断是否需要退出循环 select {  case <-stopCh:  return nil  default:  }  // 2、将resourceVersion为最新的resourceVersion，即从list回来的最新resourceVersion开始执行watch操作timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))  options := metav1.ListOptions{  ResourceVersion: r.LastSyncResourceVersion(),TimeoutSeconds: &timeoutSeconds,AllowWatchBookmarks: true}   // 3、 开始监听start := r.clock.Now()  w, err := r.listerWatcher.Watch(options)// 4、Reflctor 组件的功能: 事件处理函数  // 事件处理函数,当触发增删改时，将对应的资源对象更新到本地缓存 DeltaFIFO，并设置 ResouceVersion 最新  err = watchHandler(start, w, r.store, r.expectedType, r.expectedGVK, r.name, r.expectedTypeName, r.setLastSyncResourceVersion, r.clock, resyncerrc, stopCh)  // ... }  
}

r.watchHandler() 函数：

watchHandler()函数是 reflector组件的一个重要函数，它负责监听Kubernetes API server中的对象变更事件。如当触发增删改时，将对应的资源对象更新到本地缓存 DeltaFIFO，并设置 ResouceVersion 最新。

主要逻辑：
（1）从watch操作返回来的结果中获取event事件；
（2）接收到每个事件后，watchHandler()函数会判断该事件是否为错误事件及根据事件类型作出处理；
（3）获得当前watch到资源的ResourceVersion；
（4）判断不同类型的event事件作出相应处理；
（5）调用r.setLastSyncResourceVersion，更新Reflector对象中存储的最新的资源版本号。

循环操作，直至event事件处理完毕。


func watchHandler(start time.Time,  w watch.Interface,  store Store,  expectedType reflect.Type,  expectedGVK *schema.GroupVersionKind,  name string,  expectedTypeName string,  setLastSyncResourceVersion func(string),  clock clock.Clock,  errc chan error,  stopCh <-chan struct{},  
) error {  eventCount := 0  // Stopping the watcher should be idempotent and if we return from this function there's no way  // we're coming back in with the same watch interface.   defer w.Stop()  loop:  for {  select {  case <-stopCh:  return errorStopRequested  case err := <-errc:  return err  case event, ok := <-w.ResultChan():  if !ok {  // 错误事件，可能与客户端的连接已断开，则重试机制下尝试重新连接break loop  }  if event.Type == watch.Error {  return apierrors.FromObject(event.Object)  }  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  }  }  if expectedGVK != nil {  if e, a := *expectedGVK, event.Object.GetObjectKind().GroupVersionKind(); e != a {  utilruntime.HandleError(fmt.Errorf("%s: expected gvk %v, but watch event object had gvk %v", name, e, a))  continue  }  }  meta, err := meta.Accessor(event.Object)  if err != nil {  utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", name, event))  continue  }  // 获得当前watch到资源的ResourceVersionresourceVersion := meta.GetResourceVersion()  switch event.Type {  // 不同类型的event事件，调用不同函数处理。如事件为Added则调用 store.Add 处理case watch.Added:  err := store.Add(event.Object)  if err != nil {  utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", name, event.Object, err))  }  case watch.Modified:  err := store.Update(event.Object)  if err != nil {  utilruntime.HandleError(fmt.Errorf("%s: unable to update watch event object (%#v) to store: %v", name, event.Object, err))  }  case watch.Deleted:  // TODO: Will any consumers need access to the "last known  // state", which is passed in event.Object? If so, may need  // to change this.            err := store.Delete(event.Object)  if err != nil {  utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", name, event.Object, err))  }  case watch.Bookmark:  // A `Bookmark` means watch has synced here, just update the resourceVersion  default:  utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", name, event))  }  // 记录Reflector对象已经处理过的最新的资源版本号，以便在下次请求资源数据时能够从该版本号开始监听资源变更。setLastSyncResourceVersion(resourceVersion)  if rvu, ok := store.(ResourceVersionUpdater); ok {  rvu.UpdateResourceVersion(resourceVersion)  }  eventCount++  }  }  watchDuration := clock.Since(start)  if watchDuration < 1*time.Second && eventCount == 0 {  return fmt.Errorf("very short watch: %s: Unexpected watch close - watch lasted less than a second and no items received", name)  }  klog.V(4).Infof("%s: Watch close - %v total %v items received", name, expectedTypeName, eventCount)  return nil  
}