client-go 提供了 Informer 机制，client 能够指定监听的资源对象类型，而后 Informer：
- 首先，全量 list 所有的该类型的对象；
- 而后，监听该类型对象的变动 (Add/Update/Delete)，并调用用户注册的 Handler；
一. 实现机制
由 sharedIndexInformer 类型实现，整体实现机制如下图所示：
- Reflector 负责向 apiserver list&watch 资源对象；
- Reflector list&watch 到的对象及其变动，被放入 DeltaFIFO 队列；
-
Informer 负责 pop DeltaFIFO 队列的变动事件，而后：
- 更新对象到 Indexer；
- 调用用户注册的 ResourceEventHandler；
二. 源码框架
整体对外接口：sharedInformerFactory，通过 map 让各种 resourceType 的 informer 能够共享；
客户端调用创立进去的 informer 对象：sharedIndexInformer，其中包含：
- indexer Indexer: 索引 & 存储；
-
controller Controller:
- 调用 Reflector&FIFO&indexer 实现事件的散发、LocalStore 的保护；
- 将 Reflector&FIFO 的逻辑串联起来；
- processor *sharedProcessor: 利用散发的事件，调用 client 的 ResourceEventHandler；
- listwatcher ListerWatcher: 被 reflector.Run() 调用，进行资源对象的 List&Watch；
- objectType runtime.Object: 监听的对象类型；
三. 源码：sharedInformerFactory 如何 share informer
client 创立 Informer 的过程：
- 先创立 sharedInformerFactory；
- 再用 factory 创立 informer；
factory := informers.NewSharedInformerFactoryWithOptions(clientset, 10*time.Second, informers.WithNamespace(""))
factory.Core().V1().Pods().Informer()sharedInformerFactory 的构造：
- informers：保留了所有已创立的 sharedIndexInformer；
- startedInformers: 保留了所有已启动的 sharedIndexInformer：
// staging/src/k8s.io/client-go/informers/factory.go
type sharedInformerFactory struct {
...
informers map[reflect.Type]cache.SharedIndexInformer
startedInformers map[reflect.Type]bool
...
}底层 client-go 创立 podInformer:
// staging/src/k8s.io/client-go/informers/core/v1/interface.go
// Pods returns a PodInformer.
func (v *version) Pods() PodInformer {return &podInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
}
// staging/src/k8s.io/client-go/informers/core/v1/pod.go
func (f *podInformer) Informer() cache.SharedIndexInformer {return f.factory.InformerFor(&corev1.Pod{}, f.defaultInformer)
}InformerFor() 执行创立，并将 podInformer 增加到 factory:
- 若 objectType 的 informer 曾经存在，则间接返回；
- 否则，创立 informer，并增加到 factory.informers；
// staging/src/k8s.io/client-go/informer/factory.go
func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
.......
informerType := reflect.TypeOf(obj)
informer, exists := f.informers[informerType]
//obj 的 informer 曾经存在
if exists {return informer}
.....
// 创立 informer 并增加到 informers
informer = newFunc(f.client, resyncPeriod)
f.informers[informerType] = informer
return informer
}三. 源码：sharedIndexInformer.Run() 流程
Run() 做初始化并调用其它组件 run:
- 创立 fifo，它对应 delta 存储；
- 创立 controller；
- 调用 process.run(): 执行 client 的 ResourceEventHandler；
- 调用 controller.Run(): 执行 list&watch–> 更新 fifo–>[ 更新 indexer，散发 notification]；
// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
...
fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{ // 创立 fifo
KnownObjects: s.indexer,
EmitDeltaTypeReplaced: true,
})
cfg := &Config{
Queue: fifo,
ListerWatcher: s.listerWatcher,
ObjectType: s.objectType,
FullResyncPeriod: s.resyncCheckPeriod,
RetryOnError: false,
ShouldResync: s.processor.shouldResync,
Process: s.HandleDeltas,
}
func() {
// 创立 controller
s.controller = New(cfg)
s.controller.(*controller).clock = s.clock
s.started = true
}()
wg.StartWithChannel(processorStopCh, s.processor.run) // process.run()，执行 ResourceEventHandler
s.controller.Run(stopCh) // controller.Run(): 主逻辑}四. 源码：controller.Run() 流程
controller 整合了 reflector/fifo/indexer 等各大组件：
- 应用 reflector.Run() 进行 list&watcher，将变更存入 fifo；
-
应用 processLoop() 进行 fifo.Pop()，解决变更：
- 应用 indexer 进行对象的更新；
- 应用 processor 进行变更告诉的散发；
// staging/src/k8s.io/client-go/tools/cache/controller.go
func (c *controller) Run(stopCh <-chan struct{}) {
....
r := NewReflector(
c.config.ListerWatcher,
c.config.ObjectType,
c.config.Queue,
c.config.FullResyncPeriod,
)
c.reflector = r
var wg wait.Group
defer wg.Wait()
// 执行 reflector.Run()：进行 List&watch
wg.StartWithChannel(stopCh, r.Run)
// 执行 processLoop(): 更新 indexer + 散发事件告诉
wait.Until(c.processLoop, time.Second, stopCh)
}1. reflector.Run()
调用 ListAndWatch() 监听 apiserver 而后进行 fifo 的更新；
- 首先，进行全量的 List()，而后将后果全量更新至 fifo(fifo.Replace());
-
而后，启动 resync 的定时器，定期进行 resync，行将 LocalStore 的数据再放入 DeltaFIFO 进行解决；
- resync 的起因：deltaFIFO 中的事件回调，可能存在解决失败的状况，定时的 resync 机制让失败的事件有了从新 onUpdate 解决的机会；
- 最初，进行 watch 监听，而后通过 watchHandler() 解决 watch 到的事件 ( 即进行 fifo.Add/Update/Delete)；
// staging/src/k8s.io/client-go/tools/cache/reflector.go
func (r *Reflector) Run(stopCh <-chan struct{}) {wait.BackoffUntil(func() {if err := r.ListAndWatch(stopCh); err != nil {utilruntime.HandleError(err)
}
}, r.backoffManager, true, stopCh)
}func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
...
// 1)list 所有的
if err := func() error {
...
var list runtime.Object
listCh := make(chan struct{}, 1)
go func() {
......
// 按页 List
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)
}()
select {
// 期待 list 实现
case <-listCh:
}
// 提取 list 中的 items
listMetaInterface, err := meta.ListAccessor(list)
resourceVersion = listMetaInterface.GetResourceVersion()
items, err := meta.ExtractList(list)
// 进行全量的 sync，最终调用 -->fifo.Replace() 全量更新
if err := r.syncWith(items, resourceVersion); err != nil {return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
}
r.setLastSyncResourceVersion(resourceVersion)
return nil
}(); err != nil {return err}
// 2) 定期进行 resync
resyncerrc := make(chan error, 1)
go func() {resyncCh, cleanup := r.resyncChan()
for {
select {case <-resyncCh: // 定时器到}
if r.ShouldResync == nil || r.ShouldResync() {
// 执行 resync
if err := r.store.Resync(); err != nil {
resyncerrc <- err
return
}
}
cleanup()
resyncCh, cleanup = r.resyncChan()}
}()
// 3) 监听 watch
for {timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
options = metav1.ListOptions{
ResourceVersion: resourceVersion,
TimeoutSeconds: &timeoutSeconds,
}
// 执行 watch
w, err := r.listerWatcher.Watch(options)
// 执行 watch 对象的 handler
if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh); err != nil {....}
}
}重点看一下 watcherHandler(): 监听到事件后，进行 store 的增删改；
// staging/src/k8s.io/client-go/tools/cache/reflector.go
// 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
loop:
for {
select {
// 监听到事件
case event, ok := <-w.ResultChan():
meta, err := meta.Accessor(event.Object)
newResourceVersion := meta.GetResourceVersion()
switch event.Type {case watch.Added: //fifo.Add()
err := r.store.Add(event.Object)
case watch.Modified: //fifo.Update()
err := r.store.Update(event.Object)
case watch.Deleted: //fifo.Delete()
err := r.store.Delete(event.Object)
}
*resourceVersion = newResourceVersion
r.setLastSyncResourceVersion(newResourceVersion)
eventCount++
}
}
...
return nil
}2. controller.processLoop()
processLoop() 的工作：
- Pop fifo 中的事件；
- 事件被 c.config.Process 函数解决；
- 处理结果呈现 RetryOnError 时，将事件从新入队；
// staging/src/k8s.io/client-go/tools/cache/controller.go
func (c *controller) processLoop() {
for {obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process)) // pop 进去的事件被 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)
}
}
}
}事件处理函数：c.config.Process = informer.HandleDeleta()
-
解决从 fifo pop 进去的变更：
- 一是更新 indexer；
- 二是向 processor 散发 notification；
// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
// from oldest to newest
for _, d := range obj.(Deltas) {
switch d.Type {
// 事件类型 =Sync/Replaced/Add/Updated
case Sync, Replaced, Added, Updated:
if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
// 若 LocalStore 中已存在，则更新；if err := s.indexer.Update(d.Object); err != nil {return err}
isSync := false
switch {
case d.Type == Sync:
// Sync events are only propagated to listeners that requested resync
isSync = true
}
// 向 processor 散发 UPDATE notification
s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
} else {
// 若 LocalStore 中不存在，则增加；if err := s.indexer.Add(d.Object); err != nil {return err}
// 向 processor 散发 ADD notification
s.processor.distribute(addNotification{newObj: d.Object}, false)
}
// 事件类型 =Deleted
case Deleted:
// 删除 LocalStore 中的对象
if err := s.indexer.Delete(d.Object); err != nil {return err}
// 向 processor 散发 delete notification
s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
}
}
return nil
}五. 源码：processor.run() 流程
processor 的类型是 sharedProcessor;
processor.run() 次要工作：
- 接管从 fifo.Pop()–>informer.HandleDelta() 散发进去的 notification；
- 调用 ResourceEventHandler 进行 notification 的解决；
sharedProcessor 中蕴含 N 个 processorListner，而 processorListener 封装了 ResourceEventHandler；
也就是说，sharedProcessor 中包含 N 个 ResourceEventHandler 的解决；
// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
type sharedProcessor struct {
...
listeners []*processorListener
syncingListeners []*processorListener}
type processorListener struct {nextCh chan interface{}
addCh chan interface{}
handler ResourceEventHandler
...
}process.run() 调用每个 processorListener.run()/pop()：
- 由 processorListener.pop() 进行 notification 的接管；
- 由 processorListener.run() 调用 ResourceEventHandler 进行 notification 的解决；
// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (p *sharedProcessor) run(stopCh <-chan struct{}) {func() {p.listenersLock.RLock()
defer p.listenersLock.RUnlock()
for _, listener := range p.listeners {p.wg.Start(listener.run) // 调用 ResourceEventHandler 进行 notification 的解决
p.wg.Start(listener.pop) // 进行 notification 的接管
}
p.listenersStarted = true
}()
<-stopCh
...
}processorListener.run()
- 生产 nextCh 中的 notification，调用 ResourceEventHandler 实现事件的解决；
// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (p *processorListener) run() {stopCh := make(chan struct{})
wait.Until(func() {
for next := range p.nextCh { // 生产 notification
switch notification := next.(type) {
case updateNotification:
p.handler.OnUpdate(notification.oldObj, notification.newObj)
case addNotification:
p.handler.OnAdd(notification.newObj)
case deleteNotification:
p.handler.OnDelete(notification.oldObj)
default:
utilruntime.HandleError(fmt.Errorf("unrecognized notification: %T", next))
}
}
// the only way to get here is if the p.nextCh is empty and closed
close(stopCh)
}, 1*time.Second, stopCh)
}