kubernetes 事件反射器

我就是我 2022-04-24 16:40 37阅读 0赞

# kubernetes 事件反射器 #

*  kubernetes 事件反射器
    
     *  ListWatcher 接口
        
         *  实现 ListWatcher 接口的 ListWatch
         *  使用 ListWatch 的 Informer
     *  Reflector
        
         *  Reflector 类型定义
         *  创建 Reflector 对象的函数
         *  Run() 方法
         *  ListAndWatch() 方法
     *  使用 Reflector 的 controller
        
         *  Run() 方法
         *  processLoop() 方法
     *  使用 controller 的 Informer
        
         *  HasSynced() 方法
         *  自定义 Controller 使用 HasSynced() 方法的场景

`Reflector` 是 Kubernetes 的事件反射器，`controller`、`Informer` 使用它 List 和 Watch apiserver 中特定类型资源对象变化，并更新到 DeltaFIFO 和对象缓存中，这样客户端后续可以从本地缓存获取对象，而不需要每次都和 apiserver 交互。

在介绍 Relfector 前，先介绍 Reflector 使用的 `ListerWatcher` 接口。

## ListWatcher 接口 ##

ListWatcher 接口定义了 List 和 Watch 特定资源类型的方法：

1.  List(): 从 apiserver 获取一批**特定类型**的对象；
2.  Watch(): 从上面 List() 获取的 ResourceVersion 开始，通过 apiserver Watch etcd 中对象的变化；

ListWatcher 需要通过 RESTClient 的 Get 方法和 apiserver 通信，Getter 接口定义了该方法。

// 来源于 k8s.io/client-go/tools/cache/listwatch.go`
    type ListerWatcher interface { 
    	// List() 返回对象列表；从这些对象中提取 ResourceVersion，用户后续的 Watch() 方法参数；
    	List(options metav1.ListOptions) (runtime.Object, error)
    	// Watch() 方法从指定的版本(options 中指定)开始 Watch。
    	Watch(options metav1.ListOptions) (watch.Interface, error)
    }
    
    // ListFunc knows how to list resources
    type ListFunc func(options metav1.ListOptions) (runtime.Object, error)
    
    // WatchFunc knows how to watch resources
    type WatchFunc func(options metav1.ListOptions) (watch.Interface, error)
    
    // Getter interface knows how to access Get method from RESTClient.
    type Getter interface { 
    	Get() *restclient.Request
    }

### 实现 ListWatcher 接口的 ListWatch ###

ListWatch 类型实现了 `ListWatcher` 接口，被 `NewReflector()` 函数和各 K8S 内置资源对象的 `Informer` 创建函数（如 `NewFilteredDeploymentInformer()`)使用：

// 来源于 k8s.io/client-go/tools/cache/listwatch.go
    type ListWatch struct { 
    	ListFunc  ListFunc
    	WatchFunc WatchFunc
    	DisableChunking bool
    }

函数 `NewListWatchFromClient()` 和 `NewFilteredListWatchFromClient()` 返回 ListWatch 对象。

传入的 Getter 参数是**已经配置**的 K8S **特定 API Group/Version 的 RESTClient**，这里的“已经配置” 指创建 RESTClient 的 rest.Config 已经配置了 GroupVersion、APIPath、NegotiatedSerializer 等参数：

对于 K8S 内置对象，已经配置的 RESTClient 位于 `k8s.io/client-go/kubernetes/<group>/<version>/<group>_client.go` 文件中，如  
ExtensionsV1beta1Client

对于自定义类型对象，已经配置的 RESTClient 位于 `pkg/client/clientset/versioned/typed/<group>/<version>/<group>_client.go` 文件中。

这个已配置的 RESTClient 适用于特定 API Group/Version 下的**所有**资源类型，在发送 Get() 请求时，还需要给`Resource()` 方法传入具体的**资源类型名称**，如 `Resource("deployments")`，这样才能唯一确定资源类型：`/<APIPath>/<group>/<version>/namespaces/<resource>/<name>`，如 `/apis/apps/v1beta1/namespaces/default/deployment/my-nginx-111`。

// 来源于 k8s.io/client-go/tools/cache/listwatch.go
    func NewListWatchFromClient(c Getter, resource string, namespace string, fieldSelector fields.Selector) *ListWatch { 
    	optionsModifier := func(options *metav1.ListOptions) { 
    		options.FieldSelector = fieldSelector.String()
    	}
    	return NewFilteredListWatchFromClient(c, resource, namespace, optionsModifier)
    }
    
    // 调用 RESTClient 的 Get() 方法 List 和 Watch 对应的资源类型对象
    func NewFilteredListWatchFromClient(c Getter, resource string, namespace string, optionsModifier func(options *metav1.ListOptions)) *ListWatch { 
    	listFunc := func(options metav1.ListOptions) (runtime.Object, error) { 
    		optionsModifier(&options)
    		// c 是某一个 API Group/Version 的 RESTClient
    		return c.Get().
    			Namespace(namespace).
    			Resource(resource). // 指定某一个资源类型名称，如 "deployments"
    			VersionedParams(&options, metav1.ParameterCodec).
    			Do().
    			Get()
    	}
    	// 实际调用 WatchFunc 函数时，options 中包含上次 ListFunc 放回的对象列表的 ResourceVersion 值
    	watchFunc := func(options metav1.ListOptions) (watch.Interface, error) { 
    		options.Watch = true
    		optionsModifier(&options)
    		return c.Get().
    			Namespace(namespace).
    			Resource(resource).
    			VersionedParams(&options, metav1.ParameterCodec).
    			Watch()
    	}
    	return &ListWatch{ ListFunc: listFunc, WatchFunc: watchFunc}
    }

ListWatch 的 `List()` 和 `Watch()` 方法是直接调用内部的 `ListFunc()` 或 `WatchFunc()` 函数来实现的，比较直接和简单，故不再分析。

### 使用 ListWatch 的 Informer ###

后文会介绍，各资源类型都有自己特定的 Informer（codegen 工具自动生成），如 DeploymentInformer，它们使用自己资源类型的 ClientSet 来初始化 ListWatch，只返回对应类型的对象：

// 来源于 k8s.io/client-go/informers/extensions/v1beta1/deployment.go
    func NewFilteredDeploymentInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer { 
    	return cache.NewSharedIndexInformer(
    		// 使用特定资源类型的 RESTClient 创建 ListWatch
    		&cache.ListWatch{ 
    			ListFunc: func(options v1.ListOptions) (runtime.Object, error) { 
    				if tweakListOptions != nil { 
    					tweakListOptions(&options)
    				}
    				return client.ExtensionsV1beta1().Deployments(namespace).List(options)
    			},
    			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) { 
    				if tweakListOptions != nil { 
    					tweakListOptions(&options)
    				}
    				return client.ExtensionsV1beta1().Deployments(namespace).Watch(options)
    			},
    		},
    		&extensionsv1beta1.Deployment{ },
    		resyncPeriod,
    		indexers,
    	)
    }

## Reflector ##

在解析完 ListWatch 垫后，我们终于可以开始解析 Reflector 的实现了！

### Reflector 类型定义 ###

Reflector 使用 ListerWatcher 从 apiserver 同步 expectedType 类型的对象及其事件，缓存到 store（DeltaFIFO 类型）中。

// 来源于：k8s.io/client-go/tools/cache/reflector.go
    type Reflector struct { 
    	// Reflector 的名称，缺省值是 file:line
    	name string
    	
    	// metrics tracks basic metric information about the reflector
    	metrics *reflectorMetrics
    
    	// 一个 Reflector 指定监控一种对象类型，该 field 指定对象类型
    	expectedType reflect.Type
    	
    	// store 用于缓存对象变化事件，一般是 DeltaFIFO，可以参考 NewInformer/NewIndexerInformer 函数
    	store Store
    
    	// listerWatcher 用于 List 和 Watch 资源对象
    	listerWatcher ListerWatcher
    
    	// 当 ListAndWatch() 方法出错(超时)返回后，等待 period 时间后再次执行 ListAndWatch() 方法
    	period       time.Duration
    
    	// Reflector 周期调用 store 的 Resync() 方法，
    	// 对于 DeltaFIFO 而言，是将 knownObjects 对象缓存中的所有对象同步到 DeltaFIFO 中
    	resyncPeriod time.Duration
    	
    	// 在执行 resync 时额外的判断函数
    	ShouldResync func() bool
    
    	// clock allows tests to manipulate time
    	clock clock.Clock
    	
    	// lastSyncResourceVersion is the resource version token last
    	// observed when doing a sync with the underlying store
    	// it is thread safe, but not synchronized with the underlying store
    	lastSyncResourceVersion string
    	
    	// lastSyncResourceVersionMutex guards read/write access to lastSyncResourceVersion
    	lastSyncResourceVersionMutex sync.RWMutex
    }

### 创建 Reflector 对象的函数 ###

函数 `NewNamespaceKeyedIndexerAndReflector()`、`NewReflector()`、`NewNamedReflector()` 返回 Reflector 对象，其中 `NewNamedReflector()` 是其它两个方法的基础：

// 来源于：k8s.io/client-go/tools/cache/reflector.go
    var internalPackages = []string{ "client-go/tools/cache/"}
    
    func NewNamespaceKeyedIndexerAndReflector(lw ListerWatcher, expectedType interface{ }, resyncPeriod time.Duration) (indexer Indexer, reflector *Reflector) { 
    	// 使用 Namespace 作为 IndexFunc 和 KeyFunc
    	indexer = NewIndexer(MetaNamespaceKeyFunc, Indexers{ "namespace": MetaNamespaceIndexFunc})
    	reflector = NewReflector(lw, expectedType, indexer, resyncPeriod)
    	return indexer, reflector
    }
    
    func NewReflector(lw ListerWatcher, expectedType interface{ }, store Store, resyncPeriod time.Duration) *Reflector { 
    	// 使用 Reflector 所在的目录文件创建 Reflector 名称，格式为 file:linenum
    	return NewNamedReflector(naming.GetNameFromCallsite(internalPackages...), lw, expectedType, store, resyncPeriod)
    }
    
    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, // Run() 方法 wait.Until() 的重新执行时间间隔
    		resyncPeriod:  resyncPeriod, // 周期调用 store 的 Resync() 方法的时间间隔
    		clock:         &clock.RealClock{ },
    	}
    	return r
    }

Relectore 对象一般是 `Informer` 的 `controller` 创建的，例如（具体参考：4.controller-informer.md）：

// 来源于：k8s.io/client-go/tools/cache/controller.go
    func (c *controller) Run(stopCh <-chan struct{ }) { 
    	...
    	// 使用 controller 的 Config 参数创建 Reflector
    	r := NewReflector(
    		c.config.ListerWatcher,
    		c.config.ObjectType,
    		c.config.Queue, // DeltaFIFO
    		c.config.FullResyncPeriod,
    	)
    	r.ShouldResync = c.config.ShouldResync
    	r.clock = c.clock
    	...
    	// 运行 Reflector 的 Run 方法
    	wg.StartWithChannel(stopCh, r.Run)
    	...
    }

### Run() 方法 ###

Run() 方法一直运行 `ListAndWatch()` 方法，如果出错，则等待 r.period 时间后再次执行 `ListAndWatch()` 方法方法，所以 **Run() 方法是不会返回的**，直到 stopCh 被关闭。

// 来源于：k8s.io/client-go/tools/cache/reflector.go
    func (r *Reflector) Run(stopCh <-chan struct{ }) { 
    	klog.V(3).Infof("Starting reflector %v (%s) from %s", r.expectedType, r.resyncPeriod, r.name)
    	wait.Until(func() { 
    		if err := r.ListAndWatch(stopCh); err != nil { 
    			utilruntime.HandleError(err)
    		}
    	}, r.period, stopCh)
    }

### ListAndWatch() 方法 ###

该方法是 Refelector 的核心方法，实现了：

1.  从 apiserver List 资源类型的所有对象（ResourceVersion 为 0）；
2.  从对象列表中获取该类型对象的 resourceVersion；
3.  调用内部 Store（DeltaFIFO）的 Replace() 方法，将 List 的对象更新到内部的 DeltaFIFO 中（生成 Sync 事件，或 DeletedFinalStateUnknown 类型的 Deleted 事件）；
4.  周期(resyncPeriod)调用 DeltaFIFO 的 Resycn() 方法，将 knownObjects 缓存中的对象同步到 DeltaFIFO 中（SYNC 事件），从而实现**周期处理所有**资源类型对象的功能；
5.  从 List 获取的 resourceVersion 开始阻塞式 Watch apiserver，根据收到的事件类型更新 DeltaFIFO；

// 来源于：k8s.io/client-go/tools/cache/reflector.go
    
    // Watch 的最小 timeout 时间，实际是 [minWatchTimeout, 2*minWatchTimeout] 之间的随机值
    var minWatchTimeout = 5 * time.Minute
    
    func (r *Reflector) ListAndWatch(stopCh <-chan struct{ }) error { 
    	...
    	// ResourceVersion: "0" 表示 etcd 中对象的当前值，这里列出 etcd 中当前该类型的所有对象
    	options := metav1.ListOptions{ ResourceVersion: "0"}
    	...
    	list, err := r.listerWatcher.List(options)
    	...
    	listMetaInterface, err := meta.ListAccessor(list)
    	...
    	// 从对象列表中获取该类型对象的 resourceVersion，后续 Watch 的时候使用
    	resourceVersion = listMetaInterface.GetResourceVersion()
    	...
    	items, err := meta.ExtractList(list)
    	...
    	// 将 items 同步到 r.store 即 DeltaFIFO 中，使用的是 DeltaFIFO 的 Replace() 方法
    	if err := r.syncWith(items, resourceVersion); err != nil { 
    		return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
    	}
    	// 缓存 resourceVersion
    	r.setLastSyncResourceVersion(resourceVersion)
    	...
    	
        go func() { 
    		resyncCh, cleanup := r.resyncChan()
    		defer func() { 
    			cleanup() // Call the last one written into cleanup
    		}()
    		// 周期调用 r.store 的 Resync() 方法，实现将 knownObjects 对象缓存中的对象同步到 DeltaFIFO 中
    		for { 
    			select { 
    			case <-resyncCh:
    			case <-stopCh:
    				return
    			case <-cancelCh:
    				return
    			}
    			if r.ShouldResync == nil || r.ShouldResync() { 
    				klog.V(4).Infof("%s: forcing resync", r.name)
    				if err := r.store.Resync(); err != nil { 
    					resyncerrc <- err
    					return
    				}
    			}
    			cleanup()
    			resyncCh, cleanup = r.resyncChan()
    		}
    	}()
    
        for { 
            ...
            // Watch 会在 timeoutSeconds 后超时，这时 r.watchHandler() 出错，ListAndWatch() 方法出错返回
            // Reflecter 会等待 r.period 事件后重新执行 ListAndWatch() 方法；
            timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
    		options = metav1.ListOptions{ 
    			ResourceVersion: resourceVersion,
    			TimeoutSeconds: &timeoutSeconds,
    		}
            w, err := r.listerWatcher.Watch(options)
    		...
    		// 阻塞处理 Watch 事件
            if err := r.watchHandler(w, &resourceVersion, resyncerrc, stopCh); err != nil { 
    			if err != errorStopRequested { 
    				klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedType, err)
    			}
    			return nil
    		}
        }
    }
    
    // 用 items 中的对象替换 r.store
    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 到的事件类型，调用 r.store 的方法，将对象更新到 r.store 中
    func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{ }) error { 
    ...
    loop:
    	for { 
    		select { 
            ...
    		case event, ok := <-w.ResultChan():
                ...
    			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:
    				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))
    				}
    			default:
    				utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
    			}
    			*resourceVersion = newResourceVersion
    			r.setLastSyncResourceVersion(newResourceVersion)
    			eventCount++
    		}
    	}
        ...
    }

## 使用 Reflector 的 controller ##

Controller 封装了 Reflector，Reflector 使用 ListerWatcher 从 apiserver 获取对象列表和事件，存到 DeltaFIFO 中，Controller 持续弹出 Reflector 的 DeltaFIFO，用弹出的 Delta 更新它的 ClientState 缓存，并调用 Informer 设置的 OnAdd/OnUpdate/OnDelete 回调函数。

// 来源于：k8s.io/client-go/tools/cache/controller.go
    type Config struct { 
        // 缓存 ObjectType 类型对象的队列，同时也被 Reflector 使用。
        // 后续 NewInformer、NewIndexerInformer 创建该配置时，实际创建的是 DeltaFIFO 类型的 Queue。
    	Queue
    
        // Controller 为 ObjectType 创建 Reflector 使用的 ListerWatcher;
    	ListerWatcher
    
        // 对于对象的每个事件，调用该处理函数
    	Process ProcessFunc
    
        // 该 Controller 关注和管理的对象类型
    	ObjectType runtime.Object
    
    	// 周期调用 Queue 的 Resync() 方法的周期
    	FullResyncPeriod time.Duration
    
    	// 而外判断是否需要 Resync() 的函数（一般是 nil）
    	ShouldResync ShouldResyncFunc
    
    	// 如果 Process 处理弹出的对象失败，是否将对象重新加到 Queue 中（一般是 false）
    	RetryOnError bool
    }
    
    // controller 是 Controller 的实现，但是没有创建它的 New 方法
    // 所以 controller 实际是由 NewInformer、NewIndexerInformer 来创建和使用的
    type controller struct { 
    	config         Config
    	reflector      *Reflector
    	reflectorMutex sync.RWMutex
    	clock          clock.Clock
    }

### Run() 方法 ###

func (c *controller) Run(stopCh <-chan struct{ }) { 
    	defer utilruntime.HandleCrash()
    	go func() { 
    		<-stopCh
            // controller 运行结束时关闭队列
    		c.config.Queue.Close()
    	}()
    
        // 根据 Controller 的配置，初始化监控 ObjectType 类型对象的 Reflector
        // 初始化 Refector 时，传入了 Controller 的 Queue(DeltaFIFO)，所以 Refector 会同步更新该 Queue；
    	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()
    
    	var wg wait.Group
    	defer wg.Wait()
    
        // 在另一个 goroutine 里启动 Reflector
    	wg.StartWithChannel(stopCh, r.Run)
    
    	// 阻塞执行 c.processLoop，该 Loop 是个死循环，只是在出错时才返回值，然后等待 1s 后重新执行
    	wait.Until(c.processLoop, time.Second, stopCh)
    }

### processLoop() 方法 ###

从 DeltaFIFO 中弹出 Deltas 事件，然后调用配置的 PopProcessFunc 函数，该函数：

1.  使用弹出的对象更新 DeltaFIFO 使用的 knownObjests 对象缓存（controller 创建的 clientState）；
2.  调用用户注册的回调函数；

DelteFIFO 的 Pop() 方法是在**加锁的情况下**执行 PopProcessFunc 的，所以即使多个 goroutine 并发调用 Pop() 方法，它们也是**串行执行**的，所以**不会出现多个 goroutine 同时处理一个资源对象的情况**。

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

c.config.RetryOnError 一般为 false（参考后续的 NewInformer() 函数），所以当 PopProcessFunc 执行出错时并不会将对象添加到 DeltaFIFO 中。

## 使用 controller 的 Informer ##

NewInformer()、NewIndexInformer() 函数使用 controller 来 List/Watch 特定资源类型的对象，缓存到本地，并调用用户提供的回调函数（保存在 ResourceEventHandler 中）。

// 来源于：k8s.io/client-go/tools/cache/controller.go
    func NewInformer(
    	lw ListerWatcher,
    	objType runtime.Object,
    	resyncPeriod time.Duration,
    	h ResourceEventHandler,
    ) (Store, Controller) { 
        // Store 是一个存储对象的内存数据库，它使用 KeyFunc 函数获取对象的唯一访问 key。
        // NewStore 实际返回的是一个 ThreadSafe 类型
    	clientState := NewStore(DeletionHandlingMetaNamespaceKeyFunc)
    	fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, clientState)
    
    	cfg := &Config{ 
    		Queue:            fifo,
    		ListerWatcher:    lw,
    		ObjectType:       objType,
    		FullResyncPeriod: resyncPeriod,
    		// 不将执行失败 Process 失败的对象放回 Queue
    		RetryOnError:     false,
    
    		// DeltaFIFO 的 Pop() 方法是在内部加锁的情况下执行 Process 函数，所以对 clientState 的更新和调用 OnUpdate/OnAdd/OnDelted 处理函数是串行的。
    		Process: func(obj interface{ }) error { 
    			// from oldest to newest
    			for _, d := range obj.(Deltas) { 
    				switch d.Type { 
    				case Sync, Added, Updated:
    					// 根据 clientState 中是否有该对象决定调用 OnUpdate() 还是 OnAdd() 处理函数
    					// 所以，Controller 刚启动时，由于 clientState 为空，会为所有 list 到的对象调用 OnAdd() 处理函数；
    					if old, exists, err := clientState.Get(d.Object); err == nil && exists { 
    						if err := clientState.Update(d.Object); err != nil { 
    							return err
    						}
    						h.OnUpdate(old, d.Object)
    					} else { 
    						if err := clientState.Add(d.Object); err != nil { 
    							return err
    						}
    						h.OnAdd(d.Object)
    					}
    				case Deleted:
    					// 删除的时候，是先将对象从 clientState 中删除，再调用用户处理函数
    					if err := clientState.Delete(d.Object); err != nil { 
    						return err
    					}
    					// d.Object 可能是原生资源类型对象，也可能是 DeletedFinalStateUnknown 类型对象，所以 OnDeleted()函数需要能区分和处理
    					h.OnDelete(d.Object)
    				}
    			}
    			return nil
    		},
    	}
    	return clientState, New(cfg)
    }

### HasSynced() 方法 ###

调用 DeltaFIFO 的 HasSynced() 方法。

当 processLoop() 弹出 DeltaFIFO 中第一批 Reflector List 到的对象并处理结束后，该方法返回 true，而且后续一直返回 true；

func (c *controller) HasSynced() bool { 
    	return c.config.Queue.HasSynced()
    }

sharedInformer/sharedIndexInformer 的 HasSynced() 方法实际是调用 controller 的 HasSynced() 方法，而且该方法的签名与 InformerSynced 函数类型一致：

// 来源：k8s.io/client-go/tools/cache/shared_informer.go
    func (s *sharedIndexInformer) HasSynced() bool { 
    	s.startedLock.Lock()
    	defer s.startedLock.Unlock()
    
    	if s.controller == nil { 
    		return false
    	}
    	return s.controller.HasSynced()
    }
    
    type InformerSynced func() bool

### 自定义 Controller 使用 HasSynced() 方法的场景 ###

在开发 K8S Controller 的时候，一个惯例是调用 cache.WaitForCacheSync，等待所有 Informer Cache 都同步后才启动消费 workqueue 的 worker：

// 来源于：https://github.com/kubernetes/sample-controller/blob/master/controller.go
    
    // 自定义的 Controller
    type Controller struct { 
    	...
    	deploymentsLister appslisters.DeploymentLister
    	deploymentsSynced cache.InformerSynced  // InformerSynced 函数类型
    	...
    }
    
    func NewController(
    	kubeclientset kubernetes.Interface,
    	sampleclientset clientset.Interface,
    	deploymentInformer appsinformers.DeploymentInformer,
    	fooInformer informers.FooInformer) *Controller { 
    		...
    		controller := &Controller{ 
    			kubeclientset:     kubeclientset,
    			sampleclientset:   sampleclientset,
    			deploymentsLister: deploymentInformer.Lister(),
    			deploymentsSynced: deploymentInformer.Informer().HasSynced, // Infomer 的 HasSynced() 方法
    		}
    		...
    }
    
    // 等待所有类型的 Informer 的 HasSynced() 方法返回为 true 时再启动 workers
    func (c *Controller) Run(threadiness int, stopCh <-chan struct{ }) error { 
    	...
    	// Wait for the caches to be synced before starting workers
    	klog.Info("Waiting for informer caches to sync")
    	if ok := cache.WaitForCacheSync(stopCh, c.deploymentsSynced, c.foosSynced); !ok { 
    		return fmt.Errorf("failed to wait for caches to sync")
    	}
    
    	klog.Info("Starting workers")
    	// Launch two workers to process Foo resources
    	for i := 0; i < threadiness; i++ { 
    		go wait.Until(c.runWorker, time.Second, stopCh)
    	}
    	...
    }

为何要等各 informer 的 HasSynced() 返回为 true 时才开始启动 worker 呢？

因为 HasSynced() 返回 true 时表明 Reflecter List 的第一批对象都从 DeltaFIFO 弹出，并由 controller **更新到 clientState 缓存中，这样 worker 才能通过通过对象名称（Key）从 Lister Get 到对象**。否则对象可能还在 DeltaFIFO 中且没有同步到 clientState 缓存中，这样 worker 通过对象名称从 Lister 中 Get 不到对象。