K8s 核心组件 Reflector 源码剖析 (基于 Informer 的实现)

一、什么是 Reflector

Reflector watches a specified resource and causes all changes to be reflected in the given store.

K8s 对于 Reflector 的描述十分干练：Reflector 是监听某项资源并将 “变更” “落库” 的组件。

除了 Reflector 本身，需要额外关注的便是 Store 这个接口，这里我们只简单看下 Store 的接口定义，了解一下这个东西大体是个什么东西：

// Store is a generic object storage and processing interface.  A
// Store holds a map from string keys to accumulators, and has
// operations to add, update, and delete a given object to/from the
// accumulator currently associated with a given key.  A Store also
// knows how to extract the key from a given object, so many operations
// are given only the object.
//
// In the simplest Store implementations each accumulator is simply
// the last given object, or empty after Delete, and thus the Store's
// behavior is simple storage.
//
// Reflector knows how to watch a server and update a Store.  This
// package provides a variety of implementations of Store.
type Store interface {

	// Add adds the given object to the accumulator associated with the given object's key
	Add(obj interface{}) error

	// Update updates the given object in the accumulator associated with the given object's key
	Update(obj interface{}) error

	// Delete deletes the given object from the accumulator associated with the given object's key
	Delete(obj interface{}) error

	// List returns a list of all the currently non-empty accumulators
	List() []interface{}

	// ListKeys returns a list of all the keys currently associated with non-empty accumulators
	ListKeys() []string

	// Get returns the accumulator associated with the given object's key
	Get(obj interface{}) (item interface{}, exists bool, err error)

	// GetByKey returns the accumulator associated with the given key
	GetByKey(key string) (item interface{}, exists bool, err error)

	// Replace will delete the contents of the store, using instead the
	// given list. Store takes ownership of the list, you should not reference
	// it after calling this function.
	Replace([]interface{}, string) error

	// Resync is meaningless in the terms appearing here but has
	// meaning in some implementations that have non-trivial
	// additional behavior (e.g., DeltaFIFO).
	Resync() error
}

通俗解释一下，就是 Store 是一个通用对象存储和处理的接口，持有 keys 到 accumulators 的映射，可以增删改查。

以最简单的一种 case 来说，这个 accumulator 可以是一个对象，存储的行为也可以是一个简单存储，比如 Indexer client-go/tools/cache/store#cache 的实现，就是维护了内置了一个 map（严格意义来说是 client-go/tools/cache/thread_safe_store#goThreadSafeStore，基于 map 实现了并发安全、索引）。

二、准备逻辑 —— List

那么回到正题，Reflector 最主要的运转逻辑就是 ListAndWatch，资源的 watch 和 store 的更新是如何实现的呢，先从启动入口开始，启动的入口如下：

// Run repeatedly uses the reflector's ListAndWatch to fetch all the
// objects and subsequent deltas.
// Run will exit when stopCh is closed.
func (r *Reflector) Run(stopCh <-chan struct{}) {
	klog.V(2).Infof("Starting reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
	wait.BackoffUntil(func() {
		if err := r.ListAndWatch(stopCh); err != nil {
			utilruntime.HandleError(err)
		}
	}, r.backoffManager, true, stopCh)
	klog.V(2).Infof("Stopping reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
}

这里的核心逻辑可以看到是 r.ListAndWatch()，外面套了一层 Backoff 的管理器，用于崩溃时退避与恢复，这里不作赘述。

ListAndWatch 的代码结构如下，从注释也很容易看出，这里先进行了一次全量的 list，然后根据资源版本去做 watch。可以看到一开始的全量 list，这里由 Pager 代理了 listerWatcher 的 list 逻辑，在拿到结果后，调用 r.syncWith，将结果存储到 Store 中：

// ListAndWatch first lists all items and get the resource version at the moment of call,
// and then use the resource version to watch.
// It returns error if ListAndWatch didn't even try to initialize watch.
func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
  klog.V(3).Infof("Listing and watching %v from %s", r.expectedTypeName, r.name)
	var resourceVersion string

	options := metav1.ListOptions{ResourceVersion: r.relistResourceVersion()}

	if err := func() error {
    ... // 有删减
    
    var list runtime.Object
    go func() {
      pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
				return r.listerWatcher.List(opts)
			}))
      list, paginatedResult, err = pager.List(context.Background(), options)
      close(listCh) 
    }
    
    listMetaInterface, err := meta.ListAccessor(list)
    resourceVersion = listMetaInterface.GetResourceVersion()
    items, err := meta.ExtractList(list)
    if err := r.syncWith(items, resourceVersion); err != nil {
			return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
		}
    
    r.setLastSyncResourceVersion(resourceVersion)
    ... // 有删减
  }(); err != nil {
		return err
	}

	resyncerrc := make(chan error, 1)
	cancelCh := make(chan struct{})
	defer close(cancelCh)
	go func() {...Block1...}()

	for {...Block2...}
}

r.syncWith 逻辑很简单，本质上就是调用了 store.Replace：

// syncWith replaces the store's items with the given list.
func (r *Reflector) syncWith(items []runtime.Object, resourceVersion string) error {
   found := make([]interface{}, 0, len(items))
   for _, item := range items {
      found = append(found, item)
   }
   return r.store.Replace(found, resourceVersion)
}

2.1 ListWatcher

一步步来，首先是最里层的执行者，r.listerWatcher.List(opts)，它有很多个实现，本质上就是如何去进行首次拉取，我们这里拿两个例子可以很直观的了解它做了什么：

第一个是测试用例中，如何声明 listerWatcher 的实现，ListFunc 就是很简单的返回一个列表对象，WatchFunc 则是简单的声明了一个基于 chan 的实现：

func TestReflectorResync(t *testing.T) {
	iteration := 0
	stopCh := make(chan struct{})
	rerr := errors.New("expected resync reached")
	s := &FakeCustomStore{
		ResyncFunc: func() error {
			iteration++
			if iteration == 2 {
				return rerr
			}
			return nil
		},
	}

	lw := &testLW{
		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
			fw := watch.NewFake()
			return fw, nil
		},
		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
			return &v1.PodList{ListMeta: metav1.ListMeta{ResourceVersion: "0"}}, nil
		},
	}
	resyncPeriod := 1 * time.Millisecond
	r := NewReflector(lw, &v1.Pod{}, s, resyncPeriod)
	if err := r.ListAndWatch(stopCh); err != nil {
		// error from Resync is not propaged up to here.
		t.Errorf("expected error %v", err)
	}
	if iteration != 2 {
		t.Errorf("exactly 2 iterations were expected, got: %v", iteration)
	}
}

//
// fw:
//

func (f *FakeWatcher) ResultChan() <-chan Event {
	return f.result
}

// Add sends an add event.
func (f *FakeWatcher) Add(obj runtime.Object) {
	f.result <- Event{Added, obj}
}

另外则是看一下通常的 NewDeploymentInformer 是如何声明的，可以看到，就是依靠基础的 client 来完成的：

// NewFilteredDeploymentInformer constructs a new informer for Deployment type.
// Always prefer using an informer factory to get a shared informer instead of getting an independent
// one. This reduces memory footprint and number of connections to the server.
func NewFilteredDeploymentInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
	return cache.NewSharedIndexInformer(
		&cache.ListWatch{
			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
				if tweakListOptions != nil {
					tweakListOptions(&options)
				}
				return client.ExtensionsV1beta1().Deployments(namespace).List(context.TODO(), options)
			},
			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
				if tweakListOptions != nil {
					tweakListOptions(&options)
				}
				return client.ExtensionsV1beta1().Deployments(namespace).Watch(context.TODO(), options)
			},
		},
		&extensionsv1beta1.Deployment{},
		resyncPeriod,
		indexers,
	)
}

2.2 Pager

了解了 listerWatcher 的职责，另一快则是 Pager。Pager 本质上是控制 List 这个动作是否分批执行，以及如何分批执行，很容易联想到一个 for 循环不断去捞取数据。

代码也确实如此，这个这里不做赘述。值得注意的是进行分页前的一些配置代码，如 page.PageSize 、options.ResourceVersion 和 Continue ，这些配置可以渗透到 pager 的设置中，在 Pager 启动前，可以看到：

			// Attempt to gather list in chunks, if supported by listerWatcher, if not, the first
			// list request will return the full response.
			pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
				return r.listerWatcher.List(opts)
			}))
			switch {
			case r.WatchListPageSize != 0:
				pager.PageSize = r.WatchListPageSize
			case r.paginatedResult:
				// We got a paginated result initially. Assume this resource and server honor
				// paging requests (i.e. watch cache is probably disabled) and leave the default
				// pager size set.
			case options.ResourceVersion != "" && options.ResourceVersion != "0":
				// User didn't explicitly request pagination.
				//
				// With ResourceVersion != "", we have a possibility to list from watch cache,
				// but we do that (for ResourceVersion != "0") only if Limit is unset.
				// To avoid thundering herd on etcd (e.g. on master upgrades), we explicitly
				// switch off pagination to force listing from watch cache (if enabled).
				// With the existing semantic of RV (result is at least as fresh as provided RV),
				// this is correct and doesn't lead to going back in time.
				//
				// We also don't turn off pagination for ResourceVersion="0", since watch cache
				// is ignoring Limit in that case anyway, and if watch cache is not enabled
				// we don't introduce regression.
				pager.PageSize = 0
			}

大体围绕着要不要进行分页，以及进行多大的分页来进行。

先看看分页的描述，这里提到了，使用分页应是一个谨慎的操作，它将直接由 etcd 提供，可能会导致性能问题，也就是第一个 case，我们应尽量避免。分页由 Continue 来进行提供，：

	// WatchListPageSize is the requested chunk size of initial and resync watch lists.
	// If unset, for consistent reads (RV="") or reads that opt-into arbitrarily old data
	// (RV="0") it will default to pager.PageSize, for the rest (RV != "" && RV != "0")
	// it will turn off pagination to allow serving them from watch cache.
	// NOTE: It should be used carefully as paginated lists are always served directly from
	// etcd, which is significantly less efficient and may lead to serious performance and
	// scalability problems.
	WatchListPageSize int64

再看下服务端（ApiServer）的代码，具体的代码在 apiserver/pkg/storage/cacher/cacher.go 中，可以看到，启用了 APIListChunking 的前提下，使用 Continue 或 不指定 RV 或（使用 Limit 且 RV > 0）都将导致压力传导到 etcd 中。

	pagingEnabled := utilfeature.DefaultFeatureGate.Enabled(features.APIListChunking)
	hasContinuation := pagingEnabled && len(pred.Continue) > 0
	hasLimit := pagingEnabled && pred.Limit > 0 && resourceVersion != "0"
	if resourceVersion == "" || hasContinuation || hasLimit {
		// If resourceVersion is not specified, serve it from underlying
		// storage (for backward compatibility). If a continuation is
		// requested, serve it from the underlying storage as well.
		// Limits are only sent to storage when resourceVersion is non-zero
		// since the watch cache isn't able to perform continuations, and
		// limits are ignored when resource version is zero.
		return c.storage.List(ctx, key, resourceVersion, pred, listObj)
	}

所以综合下来，通常情况的优解就是 RV >= 0，且不使用 Limit 这个 case 。我们看看官方文档中对 RV 的描述：

---

对于 get 和 list 而言，resourceVersion 的语义为：

get:

resourceVersion 未设置	resourceVersion="0"	resourceVersion="<非零值>"
最新版本	任何版本	不老于给定版本

list:

除非你对一致性有着非常强烈的需求，使用 resourceVersionMatch=NotOlderThan 同时为 resourceVersion 设定一个已知值是优选的交互方式，因为与不设置 resourceVersion 和 resourceVersionMatch 相比，这种配置可以取得更好的 集群性能和可扩缩性。后者需要提供带票选能力的读操作。

---

2.3 Store 与 RV

Pager 拿到数据之后，要做的就是三件事情：

• 第一个是将数据存起来
• 第二个是将数据最新的版本（RV）记录起来
• 第三个是拿着这个最新的数据版本去调用 watch

这里先看前两件事：

		listMetaInterface, err := meta.ListAccessor(list)
		resourceVersion = listMetaInterface.GetResourceVersion()

		items, err := meta.ExtractList(list)

		if err := r.syncWith(items, resourceVersion); err != nil {
			return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
		}

		r.setLastSyncResourceVersion(resourceVersion)

• 【Store】将数据存储起来这个事情很好理解，看起来也很 “简单”，这里调用了 Reflector 中的重要成员变量 Store，并将拉取到的数据存入，Reflector 中 Store 实际上比较复杂，值得长篇大论，这里可以先简单把它看作一个内存 KV mapping：

// syncWith replaces the store's items with the given list.
func (r *Reflector) syncWith(items []runtime.Object, resourceVersion string) error {
	found := make([]interface{}, 0, len(items))
	for _, item := range items {
		found = append(found, item)
	}
	return r.store.Replace(found, resourceVersion)
}

• 【RV】 本质上是 MVCC 中基础的设计，是一个逻辑时间戳，对应 Etcd 中的 revision。为每个操作附带一个递增的逻辑时间戳，来达到整个系统的串行化。当然它还有其他的功能，如隔离级别、一致性视图等等，K8s 中最常接触到的是使用 RV 来做并发控制。

  更新一个资源的时候，怎么知道其他人在这期间有没有修改过此资源？靠的就是 RV 的对比。

三、运转逻辑 —— Watch

3.1 Watch

拿到 list 的结果后，可以从中获取到 RV，Watch 实际上就是基于这个 RV 来进行增量 watch：

    ... // 有删减
    options = metav1.ListOptions{
			ResourceVersion: resourceVersion,
			// We want to avoid situations of hanging watchers. Stop any wachers that do not
			// receive any events within the timeout window.
			TimeoutSeconds: &timeoutSeconds,
			// To reduce load on kube-apiserver on watch restarts, you may enable watch bookmarks.
			// Reflector doesn't assume bookmarks are returned at all (if the server do not support
			// watch bookmarks, it will ignore this field).
			AllowWatchBookmarks: true,
		}

		// start the clock before sending the request, since some proxies won't flush headers until after the first watch event is sent
		start := r.clock.Now()
		w, err := r.listerWatcher.Watch(options)

    r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh);
    ... // 有删减

r.listerWatcher.Watch(options) 和前面提到的 list 是一样的，本质上都是借助基础的 client 来实现的。

watch 的实现这里不多讲，核心就是 Transfer-Encoding: chunked + 长连接。内容简单，往 body 不断塞入以下内容即可，长度开头，换行后为内容，再次换行来结束。灵活，简单，高效：

25
This is the data in the first chunk

1C
and this is the second one

3
con

8
sequence

类似的流式技术有很多，我也没想太明白为什么是这个，大概是因为简单吧。有了上面简单的认知后，就很好理解 client.watch 在做什么了，先发起一个 Http 长连接，根据 mediaType 拿到相应的解码器，使用 frameReader 来读取流：

func (r *Request) Watch(ctx context.Context) (watch.Interface, error)

  resp, err := client.Do(req)

	contentType := resp.Header.Get("Content-Type")

	mediaType, params, err := mime.ParseMediaType(contentType)

	objectDecoder, streamingSerializer, framer, err := r.c.content.Negotiator.StreamDecoder(mediaType, params)

	frameReader := framer.NewFrameReader(resp.Body)
	watchEventDecoder := streaming.NewDecoder(frameReader, streamingSerializer)

	return watch.NewStreamWatcher(
		restclientwatch.NewDecoder(watchEventDecoder, objectDecoder),
		// use 500 to indicate that the cause of the error is unknown - other error codes
		// are more specific to HTTP interactions, and set a reason
		errors.NewClientErrorReporter(http.StatusInternalServerError, r.verb, "ClientWatchDecoding"),
	), nil

这个 NewStreamWatcher 启了一个协程，不断从 Reader 中读取内容，我们可以看到内容由两部分构成，一个叫 action，另一个则是 runtime.Object。解码 + 反序列化后，将内容投入 channel 中：

// NewStreamWatcher creates a StreamWatcher from the given decoder.
func NewStreamWatcher(d Decoder, r Reporter) *StreamWatcher {
	sw := &StreamWatcher{
		source:   d,
		reporter: r,
		// It's easy for a consumer to add buffering via an extra
		// goroutine/channel, but impossible for them to remove it,
		// so nonbuffered is better.
		result: make(chan Event),
	}
	go sw.receive()
	return sw
}

const (
	Added    EventType = "ADDED"
	Modified EventType = "MODIFIED"
	Deleted  EventType = "DELETED"
	Bookmark EventType = "BOOKMARK"
	Error    EventType = "ERROR"

	DefaultChanSize int32 = 100
)

// receive reads result from the decoder in a loop and sends down the result channel.
func (sw *StreamWatcher) receive() {
	defer close(sw.result)
	defer sw.Stop()
	defer utilruntime.HandleCrash()
	for {
		action, obj, err := sw.source.Decode()
		if err != nil {
			// Ignore expected error.
			if sw.stopping() {
				return
			}
			switch err {
			case io.EOF:
				// watch closed normally
			case io.ErrUnexpectedEOF:
				klog.V(1).Infof("Unexpected EOF during watch stream event decoding: %v", err)
			default:
				if net.IsProbableEOF(err) || net.IsTimeout(err) {
					klog.V(5).Infof("Unable to decode an event from the watch stream: %v", err)
				} else {
					sw.result <- Event{
						Type:   Error,
						Object: sw.reporter.AsObject(fmt.Errorf("unable to decode an event from the watch stream: %v", err)),
					}
				}
			}
			return
		}
		sw.result <- Event{
			Type:   action,
			Object: obj,
		}
	}
}

3.2 WatchHandler

前面说到，Watcher 负责把 HTTP 长连接的内容进行解码、反序列化，拿到的内容投入 channel，那么消费 channel 的地方，在 Reflector 中叫做 WatchHandler ，整体的逻辑清晰简单，就是根据 Event，去调用不同的 store 实现，代码看起来多，本质上很简单：

// watchHandler watches w and keeps *resourceVersion up to date.
func (r *Reflector) watchHandler(start time.Time, w watch.Interface, resourceVersion *string, 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 r.expectedType != nil {
				if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a {
					utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a))
					continue
				}
			}
			if r.expectedGVK != nil {
				if e, a := *r.expectedGVK, event.Object.GetObjectKind().GroupVersionKind(); e != a {
					utilruntime.HandleError(fmt.Errorf("%s: expected gvk %v, but watch event object had gvk %v", r.name, e, a))
					continue
				}
			}
			meta, err := meta.Accessor(event.Object)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
				continue
			}
			newResourceVersion := meta.GetResourceVersion()
			switch event.Type {
			case watch.Added:
				err := r.store.Add(event.Object)
				if err != nil {
					utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", r.name, event.Object, err))
				}
			case watch.Modified:
				err := r.store.Update(event.Object)
				if err != nil {
					utilruntime.HandleError(fmt.Errorf("%s: unable to update watch event object (%#v) to store: %v", r.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 := r.store.Delete(event.Object)
				if err != nil {
					utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", r.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", r.name, event))
			}
			*resourceVersion = newResourceVersion
			r.setLastSyncResourceVersion(newResourceVersion)
			eventCount++
		}
	}

	watchDuration := r.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", r.name)
	}
	klog.V(4).Infof("%s: Watch close - %v total %v items received", r.name, r.expectedTypeName, eventCount)
	return nil
}

四、DeltaFIFO

经过前面的分析，我们已经了解到，Reflector 的核心逻辑是先进行了一次 List

• 通过 Pager 配合 listerWatcher#ListFunc 尽可能拿到所有数据
• 并将数据存入到 Store( Replace )

再进行 watch

• 通过 listerWatcher#WatchFunc 请求了 ApiServer，不断解析 Body，产出 Event
• WatchHandler 中通过事件的不同，进行处理，或者调用 Store 对应的增删改( Add、Delete、Update )

那么这里还剩下一个疑点没解，就是 Store 里面究竟做了什么？首先 Store 和 listWatcher 一样，都是很开放的实现，我们看看在 informer 的实现中，Store 是如何工作的。

代码很多，但是只需要关注 Store 这块内容即可，在初始化 informer 的过程中，new 了一个 DeltaFifo，注入到 Config 中，然后在运行 informer 的时候，将这个 DeltaFifo 当作 Store 传入了 Reflector 的构造函数。

// newInformer returns a controller for populating the store while also
// providing event notifications.
//
// Parameters
//  * lw is list and watch functions for the source of the resource you want to
//    be informed of.
//  * objType is an object of the type that you expect to receive.
//  * resyncPeriod: if non-zero, will re-list this often (you will get OnUpdate
//    calls, even if nothing changed). Otherwise, re-list will be delayed as
//    long as possible (until the upstream source closes the watch or times out,
//    or you stop the controller).
//  * h is the object you want notifications sent to.
//  * clientState is the store you want to populate
//
func newInformer(
	lw ListerWatcher,
	objType runtime.Object,
	resyncPeriod time.Duration,
	h ResourceEventHandler,
	clientState Store,
	transformer TransformFunc,
) Controller {
	// This will hold incoming changes. Note how we pass clientState in as a
	// KeyLister, that way resync operations will result in the correct set
	// of update/delete deltas.
	fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
		KnownObjects:          clientState,
		EmitDeltaTypeReplaced: true,
	})

	cfg := &Config{
		Queue:            fifo,
		ListerWatcher:    lw,
		ObjectType:       objType,
		FullResyncPeriod: resyncPeriod,
		RetryOnError:     false,

		Process: func(obj interface{}) error {
			if deltas, ok := obj.(Deltas); ok {
				return processDeltas(h, clientState, transformer, deltas)
			}
			return errors.New("object given as Process argument is not Deltas")
		},
	}
	return New(cfg)
}

// New makes a new Controller from the given Config.
func New(c *Config) Controller {
	ctlr := &controller{
		config: *c,
		clock:  &clock.RealClock{},
	}
	return ctlr
}

// Run begins processing items, and will continue until a value is sent down stopCh or it is closed.
// It's an error to call Run more than once.
// Run blocks; call via go.
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

	wg.StartWithChannel(stopCh, r.Run)

	wait.Until(c.processLoop, time.Second, stopCh)
	wg.Wait()
}

4.1 DeltaFIFO

我们来关注一下 Informer 中的 Store，也就是 DeltaFIFO 承担了什么角色。先看看 DeltaFIFO 的设计。本质上 DeltaFIFO 像 FIFIO 一样，但是有两个不同：

• 其一是通过 Delta，可以认为是一个包装类（增量）来存储 K -> V 的 V
• 其二是有一个 Sync 机制

用于解决以下问题：

• 不错过任何一个操作历史的处理
• 处理某个 V 时，可以看到从上次处理后，到此次处理期间的操作
• 有感删除
• 可定期重新同步

实现的方式有很多种，来看看 K8s 是怎么实现这套逻辑的，主要声明的成员变量：

• KV 映射：items
• 队列：queue
• 从对象到 key 怎么转换：keyFunc
• 怎么通过 key 拿到对象、维护所有 key：knownObjects

其中 keyFunc 是最简单的，默认会取这个对象的 meta.GetNamespace() + "/" + meta.GetName()，或者 meta.GetName()，这里不细说，本质上就是怎么唯一去区分一个对象。

knownObjects 本质上是一个 Indexer（ infomer 背景下），在本篇文章的最开头，就介绍了这个 Indexer，DeltaFIFO 和 Indexer 都实现了 Store 接口。可以这么理解，Indexer 是 DeltaFIFO 中实际进行对象存储的成员，而 DeltaFIFO 则补充了期间的增量变化。这里也不多说，这个东西太简单了，有兴趣的可以去看下 client-go/tools/cache/store#cache。

// DeltaFIFO is like FIFO, but differs in two ways.  One is that the
// accumulator associated with a given object's key is not that object
// but rather a Deltas, which is a slice of Delta values for that
// object.  Applying an object to a Deltas means to append a Delta
// except when the potentially appended Delta is a Deleted and the
// Deltas already ends with a Deleted.  In that case the Deltas does
// not grow, although the terminal Deleted will be replaced by the new
// Deleted if the older Deleted's object is a
// DeletedFinalStateUnknown.
//
// The other difference is that DeltaFIFO has an additional way that
// an object can be applied to an accumulator, called Sync.
//
// DeltaFIFO is a producer-consumer queue, where a Reflector is
// intended to be the producer, and the consumer is whatever calls
// the Pop() method.
//
// DeltaFIFO solves this use case:
//  * You want to process every object change (delta) at most once.
//  * When you process an object, you want to see everything
//    that's happened to it since you last processed it.
//  * You want to process the deletion of some of the objects.
//  * You might want to periodically reprocess objects.
//
// DeltaFIFO's Pop(), Get(), and GetByKey() methods return
// interface{} to satisfy the Store/Queue interfaces, but they
// will always return an object of type Deltas.
//
// A DeltaFIFO's knownObjects KeyListerGetter provides the abilities
// to list Store keys and to get objects by Store key.  The objects in
// question are called "known objects" and this set of objects
// modifies the behavior of the Delete, Replace, and Resync methods
// (each in a different way).
//
// A note on threading: If you call Pop() in parallel from multiple
// threads, you could end up with multiple threads processing slightly
// different versions of the same object.
type DeltaFIFO struct {
	// lock/cond protects access to 'items' and 'queue'.
	lock sync.RWMutex
	cond sync.Cond

	// We depend on the property that items in the set are in
	// the queue and vice versa, and that all Deltas in this
	// map have at least one Delta.
	items map[string]Deltas
	queue []string

	// populated is true if the first batch of items inserted by Replace() has been populated
	// or Delete/Add/Update was called first.
	populated bool
	// initialPopulationCount is the number of items inserted by the first call of Replace()
	initialPopulationCount int

	// keyFunc is used to make the key used for queued item
	// insertion and retrieval, and should be deterministic.
	keyFunc KeyFunc

	// knownObjects list keys that are "known" --- affecting Delete(),
	// Replace(), and Resync()
	knownObjects KeyListerGetter

	// Indication the queue is closed.
	// Used to indicate a queue is closed so a control loop can exit when a queue is empty.
	// Currently, not used to gate any of CRED operations.
	closed     bool
	closedLock sync.Mutex

	// emitDeltaTypeReplaced is whether to emit the Replaced or Sync
	// DeltaType when Replace() is called (to preserve backwards compat).
	emitDeltaTypeReplaced bool
}

4.2 从增删改看 Deltas 工作

先纵览一下 增删改，发现它的核心干活逻辑都是 queueActionLocked，其中删除多了一点点逻辑，本质上只是做了下重入，在删除操作发生的时候，如果 Indexer 和 Queue 中都没有了，就不做任何处理了。其他的都是通过 queueActionLocked 来完成：

// Add inserts an item, and puts it in the queue. The item is only enqueued
// if it doesn't already exist in the set.
func (f *DeltaFIFO) Add(obj interface{}) error {
	f.lock.Lock()
	defer f.lock.Unlock()
	f.populated = true
	return f.queueActionLocked(Added, obj)
}

// Update is just like Add, but makes an Updated Delta.
func (f *DeltaFIFO) Update(obj interface{}) error {
	f.lock.Lock()
	defer f.lock.Unlock()
	f.populated = true
	return f.queueActionLocked(Updated, obj)
}

// Delete is just like Add, but makes a Deleted Delta. If the given
// object does not already exist, it will be ignored. (It may have
// already been deleted by a Replace (re-list), for example.)  In this
// method `f.knownObjects`, if not nil, provides (via GetByKey)
// _additional_ objects that are considered to already exist.
func (f *DeltaFIFO) Delete(obj interface{}) error {
	id, err := f.KeyOf(obj)
	if err != nil {
		return KeyError{obj, err}
	}
	f.lock.Lock()
	defer f.lock.Unlock()
	f.populated = true
	if f.knownObjects == nil {
		if _, exists := f.items[id]; !exists {
			// Presumably, this was deleted when a relist happened.
			// Don't provide a second report of the same deletion.
			return nil
		}
	} else {
		// We only want to skip the "deletion" action if the object doesn't
		// exist in knownObjects and it doesn't have corresponding item in items.
		// Note that even if there is a "deletion" action in items, we can ignore it,
		// because it will be deduped automatically in "queueActionLocked"
		_, exists, err := f.knownObjects.GetByKey(id)
		_, itemsExist := f.items[id]
		if err == nil && !exists && !itemsExist {
			// Presumably, this was deleted when a relist happened.
			// Don't provide a second report of the same deletion.
			return nil
		}
	}

	return f.queueActionLocked(Deleted, obj)
}

所以目光只需要专注于 queueActionLocked 即可，来看看它是如何完成的，先是根据唯一标识找到了 Delta 数组中的对象，将此次的 Object 与 ActionType，一起 append 到原有的 Delta 数组中。可以看到，不管是增、删还是改，都会以一条记录的形式，塞入到 Delta 数组中：

func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
	id, err := f.KeyOf(obj)
	if err != nil {
		return KeyError{obj, err}
	}

	newDeltas := append(f.items[id], Delta{actionType, obj})
	newDeltas = dedupDeltas(newDeltas)

	if len(newDeltas) > 0 {
		if _, exists := f.items[id]; !exists {
			f.queue = append(f.queue, id)
		}
		f.items[id] = newDeltas
		f.cond.Broadcast()
	} else {
		// This never happens, because dedupDeltas never returns an empty list
		// when given a non-empty list (as it is here).
		// But if somehow it ever does return an empty list, then
		// We need to remove this from our map (extra items in the queue are
		// ignored if they are not in the map).
		delete(f.items, id)
	}
	return nil
}

4.3 DeltaFIFO Replace

Reflector 有一个定期从 Api 重新拉取数据的 case，比较简单，这里不细说。值得注意的是，DeltaFIFO 的 Replace 操作，也就是说，触发了数据重新拉取之后，DeltaFIFO 将怎么处理。

重新同步拉取的列表，将全量变成新的 Event 塞入 DeltaFIFO，而同步过来时，本地有（Indexer），服务端却没有的数据，则会以占位符 DeletedFinalStateUnknown 的形式，塞入队列中。这样，在这种数据不一致场景出现的情况下，将模拟出一条删除事件供上层处理：

... // 有删减
func (f *DeltaFIFO) Replace(list []interface{}, _ string) error {
  
	// keep backwards compat for old clients
	action := Sync
	if f.emitDeltaTypeReplaced {
		action = Replaced
	}
  
	// Add Sync/Replaced action for each new item.
	for _, item := range list {
		key, err := f.KeyOf(item)
		if err != nil {
			return KeyError{item, err}
		}
		keys.Insert(key)
		if err := f.queueActionLocked(action, item); err != nil {
			return fmt.Errorf("couldn't enqueue object: %v", err)
		}
	}

	// Detect deletions not already in the queue.
	knownKeys := f.knownObjects.ListKeys()
	queuedDeletions := 0
	for _, k := range knownKeys {
		if keys.Has(k) {
			continue
		}

		deletedObj, exists, err := f.knownObjects.GetByKey(k)
		if err != nil {
			deletedObj = nil
			klog.Errorf("Unexpected error %v during lookup of key %v, placing DeleteFinalStateUnknown marker without object", err, k)
		} else if !exists {
			deletedObj = nil
			klog.Infof("Key %v does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", k)
		}
		queuedDeletions++
		if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil {
			return err
		}
	}

	return nil
}

看到了这里，还是有些没头没尾，这里似乎没有 Indexer 什么事，那么数据是什么时候落入 Indexer 的呢？DeltaFIFO 的消费者又在哪里？实际上 Reflector 并没有定义这些逻辑，它只负责生产消息，如何消费则非常开放。

五、消息消费者

在第四节中，我们提到了 Store 的定义者 informer（实体类叫 controller），负责管理 Reflector 的生命周期。它定义了 Reflector 中的组件，如 DeltaFIFO，它同时也定义了消息的消费逻辑：

// Multiplexes updates in the form of a list of Deltas into a Store, and informs
// a given handler of events OnUpdate, OnAdd, OnDelete
func processDeltas(
	// Object which receives event notifications from the given deltas
	handler ResourceEventHandler,
	clientState Store,
	transformer TransformFunc,
	deltas Deltas,
) error {
	// from oldest to newest
	for _, d := range deltas {
		obj := d.Object
		if transformer != nil {
			var err error
			obj, err = transformer(obj)
			if err != nil {
				return err
			}
		}

		switch d.Type {
		case Sync, Replaced, Added, Updated:
			if old, exists, err := clientState.Get(obj); err == nil && exists {
				if err := clientState.Update(obj); err != nil {
					return err
				}
				handler.OnUpdate(old, obj)
			} else {
				if err := clientState.Add(obj); err != nil {
					return err
				}
				handler.OnAdd(obj)
			}
		case Deleted:
			if err := clientState.Delete(obj); err != nil {
				return err
			}
			handler.OnDelete(obj)
		}
	}
	return nil
}

func newInformer(
	lw ListerWatcher,
	objType runtime.Object,
	resyncPeriod time.Duration,
	h ResourceEventHandler,
	clientState Store,
	transformer TransformFunc,
) Controller {
	// This will hold incoming changes. Note how we pass clientState in as a
	// KeyLister, that way resync operations will result in the correct set
	// of update/delete deltas.
	fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
		KnownObjects:          clientState,
		EmitDeltaTypeReplaced: true,
	})

	cfg := &Config{
		Queue:            fifo,
		ListerWatcher:    lw,
		ObjectType:       objType,
		FullResyncPeriod: resyncPeriod,
		RetryOnError:     false,

		Process: func(obj interface{}) error {
			if deltas, ok := obj.(Deltas); ok {
				return processDeltas(h, clientState, transformer, deltas)
			}
			return errors.New("object given as Process argument is not Deltas")
		},
	}
	return New(cfg)
}

// New makes a new Controller from the given Config.
func New(c *Config) Controller {
	ctlr := &controller{
		config: *c,
		clock:  &clock.RealClock{},
	}
	return ctlr
}

// Run begins processing items, and will continue until a value is sent down stopCh or it is closed.
// It's an error to call Run more than once.
// Run blocks; call via go.
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

	wg.StartWithChannel(stopCh, r.Run)

	wait.Until(c.processLoop, time.Second, stopCh)
	wg.Wait()
}

// processLoop drains the work queue.
// TODO: Consider doing the processing in parallel. This will require a little thought
// to make sure that we don't end up processing the same object multiple times
// concurrently.
//
// TODO: Plumb through the stopCh here (and down to the queue) so that this can
// actually exit when the controller is stopped. Or just give up on this stuff
// ever being stoppable. Converting this whole package to use Context would
// also be helpful.
func (c *controller) processLoop() {
	for {
		obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
		if err != nil {
			if err == ErrFIFOClosed {
				return
			}
			if c.config.RetryOnError {
				// This is the safe way to re-enqueue.
				c.config.Queue.AddIfNotPresent(obj)
			}
		}
	}
}

可以看到，informer 启动了一个协程，不断的从 DeltaFIFO 中 Pop 出 Event，并交给 PopProcessFunc，也就是消费逻辑中处理。

在 Informer 的消费逻辑中，我们看到了 Indexer（clientState） 的更新逻辑，每当有新的事件进来，process 逻辑中便会从旧到新，遍历 Event 中的事件与当时的对象，分别对本地存储进行操作，并调用对应注册进来的 handler。

六：Self-Reflective

这里算是阅读源码的一些思考，其实很早就读过这部分代码，只是没有细看，细看之后产生了一些疑问，同时感谢同事解答。有一些可能我目前也不是太清楚，也欢迎大佬解答。

保存状态变化有必要吗，申明式的资源，控制器不是只需要对 “终态” 负责吗？

这块自己也是先入为主、以偏概全了，其实不然，有不少的控制器是需要知道每次变更的，比如 Deployment 的 Revision，Rollback 等机制，后需要能清楚处理每一次的变更。

为何 processLoop 需要 Requeue 机制？

先入为主 +1，从实现（Indexer）去倒推了抽象，认为 Indexer 的操作不会报错。实际上 Store 可以是任意实现，如果是其他的实现，则无法保证落库一定成功。

已知分页会对 etcd 有压力，什么时候需要分页，什么时候不需要呢？

?