一、什么是 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)
}
三、运转逻辑 —— 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 有压力，什么时候需要分页，什么时候不需要呢？
?
