k8s源码阅读：Informer源码解析

写在之前

Kubernetes的Informer机制是一种用于监控资源对象变化的机制。它提供了一种简化开发者编写控制器的方式，允许控制器能够及时感知并响应 Kubernetes 集群中资源对象的变化。Informer通过与Kubernetes API服务器进行交互，通过监听API服务器上资源对象的修改事件来实现实时的资源对象状态更新。当一个资源对象被创建、更新或删除时，Informer会收到相应的通知，并在内部维护一个缓存用于存储最新的资源对象状态。同时，Informer还能够为特定类型的资源对象设置过滤器以进行更精细的事件监听。使用Informer机制，开发者可以通过注册事件处理函数来定义对资源对象变化的响应逻辑，从而实现自定义的控制器逻辑。在运行时，Informer会定期向API服务器发起请求以获取最新的资源对象的状态，从而保证缓存中的数据与实际集群的状态保持同步。话不多说，直接进入到源码的解析环节。

1.源码阅读环境搭建

informer的源码是在client-go工程中的，需要从仓库clone完整的代码到本地。首先在GoPath路径下创建文件夹k8s.io，进入到该文件夹下，执行以下步骤。

git clone https://github.com/kubernetes/client-go.git

源码的阅读不依赖环境和操作系统，但是需要提供一个k8s集群的环境，提前将集群的kubeconfig保存在本地，以供后文中调试源码使用。

值得注意的是，笔者看的是master分支的代码，其中的go mod依赖的golang版本是1.21,建议大家选择适合自己的版本分支阅读，否则可能会出现编译报错的情况。

2. 一个informer的简单用例

将下面的代码放到kubernetes_test目录下，以方便调试源码

package kubernetesimport ("fmt""k8s.io/apimachinery/pkg/labels""k8s.io/client-go/informers""k8s.io/client-go/kubernetes""k8s.io/client-go/tools/cache""k8s.io/client-go/tools/clientcmd""log""testing""time"metav1 "k8s.io/api/apps/v1"
)func getKubernetesClient(kubeconfig string) (*kubernetes.Clientset, error) {// 配置文件路径，根据你的实际情况修改// 使用指定的 kubeconfig 文件创建一个 Config 对象config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)if err != nil {return nil, err}//config.Insecure = true// 创建客户端集clientset, err := kubernetes.NewForConfig(config)if err != nil {return nil, err}return clientset, nil
}// 展示informer的核心用法
func TestInformer(t *testing.T) {client, err := getKubernetesClient("XXX这里是集群kubeconfig")if err != nil {log.Fatal(err.Error())}// 初始化 informer factory（为了测试方便这里设置每30s重新 List 一次）informerFactory := informers.NewSharedInformerFactory(client, time.Second*30)// 对 Deployment 监听deployInformer := informerFactory.Apps().V1().Deployments()// 创建 Informer（相当于注册到工厂中去，这样下面启动的时候就会去 List & Watch 对应的资源）informer := deployInformer.Informer()// 创建 ListerdeployLister := deployInformer.Lister()// 注册事件处理程序informer.AddEventHandler(cache.ResourceEventHandlerFuncs{AddFunc:    onAdd,UpdateFunc: onUpdate,DeleteFunc: onDelete,})stopper := make(chan struct{})defer close(stopper)// 启动 informer，List & WatchinformerFactory.Start(stopper)// 等待所有启动的 Informer 的缓存被同步informerFactory.WaitForCacheSync(stopper)// 从本地缓存中获取 default 中的所有 deployment 列表deployments, err := deployLister.Deployments("default").List(labels.Everything())if err != nil {panic(err)}for idx, deploy := range deployments {fmt.Printf("%d -> %s\n", idx+1, deploy.Name)}<-stopper}func onAdd(obj interface{}) {deploy := obj.(*metav1.Deployment)fmt.Println("add a deployment:", deploy.Name)
}func onUpdate(old, new interface{}) {oldDeploy := old.(*metav1.Deployment)newDeploy := new.(*metav1.Deployment)fmt.Println("update deployment:", oldDeploy.Name, newDeploy.Name)
}func onDelete(obj interface{}) {deploy := obj.(*metav1.Deployment)fmt.Println("delete a deployment:", deploy.Name)
}

简单mark一下源码分析的链路，建议大家还是一步一步使用编辑器的debug模式来阅读。
在这里插入图片描述

3.2. informers.NewSharedInformerFactory(client, time.Second*30)

初始化一个SharedInformerFactory，传入了一个定时30进行数据同步的配置项。核心代码如下：

// NewSharedInformerFactoryWithOptions constructs a new instance of a SharedInformerFactory with additional options.
func NewSharedInformerFactoryWithOptions(client kubernetes.Interface, defaultResync time.Duration, options ...SharedInformerOption) SharedInformerFactory {factory := &sharedInformerFactory{client:           client,namespace:        v1.NamespaceAll,defaultResync:    defaultResync,informers:        make(map[reflect.Type]cache.SharedIndexInformer),startedInformers: make(map[reflect.Type]bool),customResync:     make(map[reflect.Type]time.Duration),}// Apply all optionsfor _, opt := range options {//这一步执行，在我们追踪的链路中，options为空factory = opt(factory)}return factory
}

4.deployInformer := informerFactory.Apps().V1().Deployments()

这一步是deployInformer的初始化方式。我们展开步骤2中的初始化的SharedInformerFactory结构体会发现，下面的属性字段core，policy，apps似乎是将k8s原生的资源分组进行了类别、版本的抽象。类如通过.Apps().V1().Deployments()，其实获取的就是deploymentInformer的一个实例对象：

// Deployments returns a DeploymentInformer.
func (v *version) Deployments() DeploymentInformer {return &deploymentInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
}

看一下deploymentInformer这个结构体的定义，这里的factory就是在informerFactory := informers.NewSharedInformerFactory(client, time.Second*30)初始化的factory，之所以关注这个factory的实例，是因为在后文中还有提及。

type deploymentInformer struct {factory          internalinterfaces.SharedInformerFactorytweakListOptions internalinterfaces.TweakListOptionsFuncnamespace        string
}

5.informer := deployInformer.Informer()

这里调用了informerfactory结构InformerFor的方法，我们上一节已经分析过了factory的实例是什么了，我们追踪到方法中看一下这个方法是在所做什么

func (f *deploymentInformer) Informer() cache.SharedIndexInformer {// 注册Deployment的informer到factory中return f.factory.InformerFor(&appsv1.Deployment{}, f.defaultInformer)
}func (f *deploymentInformer) defaultInformer(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {return NewFilteredDeploymentInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
}

现在继续追踪informerFor源码，这里看起来是将deployment的cache.SharedIndexInformer注册到factory里的缓存中去。

func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {//  f.factory.InformerFor(&appsv1.Deployment{}, f.defaultInformer)f.lock.Lock()defer f.lock.Unlock()informerType := reflect.TypeOf(obj)// 1.f.informers:  informers map[reflect.Type]cache.SharedIndexInformer 这里缓存不同k8s资源的informer对象informer, exists := f.informers[informerType]if exists {return informer}resyncPeriod, exists := f.customResync[informerType]if !exists {resyncPeriod = f.defaultResync}// 2.customResync     map[reflect.Type]time.Duration 缓存了资源的resync时间informer = newFunc(f.client, resyncPeriod)informer.SetTransform(f.transform)f.informers[informerType] = informerreturn informer
}

6.deployLister := deployInformer.Lister()

返回lister实例，这个listener主要是用来从内存中获取资源对象详细信息的客户端，类比于数据库查询客户端。
我们看一下f.Informer().GetIndexer()函数是什么，返回了一个Index结构体的实现类。

func (s *sharedIndexInformer) GetIndexer() Indexer {return s.indexer
}

// 这个类就是在集群中的类别是deployment的所有object缓存存放的结构体
type Indexer interface {Store// Index returns the stored objects whose set of indexed values// intersects the set of indexed values of the given object, for// the named indexIndex(indexName string, obj interface{}) ([]interface{}, error)// IndexKeys returns the storage keys of the stored objects whose// set of indexed values for the named index includes the given// indexed valueIndexKeys(indexName, indexedValue string) ([]string, error)// ListIndexFuncValues returns all the indexed values of the given indexListIndexFuncValues(indexName string) []string// ByIndex returns the stored objects whose set of indexed values// for the named index includes the given indexed valueByIndex(indexName, indexedValue string) ([]interface{}, error)// GetIndexers return the indexersGetIndexers() Indexers// AddIndexers adds more indexers to this store. This supports adding indexes after the store already has items.AddIndexers(newIndexers Indexers) error
}

7.informer.AddEventHandler

注册事件处理回调函数，下面是核心逻辑实现的具体细节代码。

func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) (ResourceEventHandlerRegistration, error) {.....//6.1这个是一个事件处理器的生成函数listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize, s.HasSynced)if !s.started {//6.2首次初始化的时候就started是false的，进入到这个逻辑里return s.processor.addListener(listener), nil}.....
}

7.1 newProcessListener

processListener实例化的方法,该实例将消息事件路由到不同的ResourceEventHandler方法中处理。它包含两个 goroutine、两个无缓冲通道和一个无界环形缓冲区区。 add(notification) 函数将给定的通知事件发送到 addCh。一个 goroutine 运行“pop()”，它使用环形缓冲区中的存储将通知从“addCh”传递到“nextCh”。另一个 goroutine 运行 run()，它接收来自 nextCh 的通知并同步调用适当的处理程序方法。

func newProcessListener(handler ResourceEventHandler, requestedResyncPeriod, resyncPeriod time.Duration, now time.Time, bufferSize int, hasSynced func() bool) *processorListener {ret := &processorListener{// 这里是事件处理channel通道nextCh:                make(chan interface{}),// 这里是watch到k8s资源变动的接收通道addCh:                 make(chan interface{}),// 这里就是传递的handler事件处理函数handler:               handler,// 这个结构体是一个线程安全的计数器syncTracker:           &synctrack.SingleFileTracker{UpstreamHasSynced: hasSynced},// 无界环，个人理解主要是用来在生产、消费者模型中速率不一致时进行缓冲的一个存储pendingNotifications:  *buffer.NewRingGrowing(bufferSize),requestedResyncPeriod: requestedResyncPeriod,resyncPeriod:          resyncPeriod,}// 生成下一次resync的时间并保存下来ret.determineNextResync(now)return ret
}

7.2 processor.addListener(listener)

这里的processor是sharedProcessor实例化的结果，实例化的过程放在
defaultInformer这个函数中，在前文informerFor中由提及，我们把这个函数展开，看一下这个processor是怎么实例化的。

func (f *deploymentInformer) defaultInformer(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {return NewFilteredDeploymentInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
}
func NewFilteredDeploymentInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {return cache.NewSharedIndexInformer(//初始化listwatch&cache.ListWatch{ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {if tweakListOptions != nil {tweakListOptions(&options)}return client.AppsV1().Deployments(namespace).List(context.TODO(), options)},WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {if tweakListOptions != nil {tweakListOptions(&options)}return client.AppsV1().Deployments(namespace).Watch(context.TODO(), options)},},&appsv1.Deployment{},resyncPeriod,indexers,)
}
//中间还有一跳，我们省略了，直接找到process的实例化代码
func NewSharedIndexInformerWithOptions(lw ListerWatcher, exampleObject runtime.Object, options SharedIndexInformerOptions) SharedIndexInformer {realClock := &clock.RealClock{}return &sharedIndexInformer{// indexre的初始化逻辑在这里indexer: NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, options.Indexers),//sharedProcessor初始的位置processor:                       &sharedProcessor{clock: realClock},listerWatcher:                   lw,objectType:                      exampleObject,objectDescription:               options.ObjectDescription,resyncCheckPeriod:               options.ResyncPeriod,defaultEventHandlerResyncPeriod: options.ResyncPeriod,clock:                           realClock,cacheMutationDetector:           NewCacheMutationDetector(fmt.Sprintf("%T", exampleObject)),}
}

addListener主要做了两件事情
1）listeners map[*processorListener]bool，是shardProcessor中的一个缓存，主要是保存定义的processorListener
2）如果p.listenersStarted 是true的话，这里会启动两个线程，一个线程运行listener.run，定时从nextCh这个channel中获取事件进行处理
一个线程运行listener.pop，定时将addCh这个channel中的事件转移到nextCh这个channel中去，两者之间是通过一个无界环状结构进行传递的
但是，按照前文的追踪逻辑这里的p.listenersStarted是false， p.wg.Start(listener.run和p.wg.Start(listener.pop)这两个典型的生产-消费者逻辑没有执行。

func (p *sharedProcessor) addListener(listener *processorListener) ResourceEventHandlerRegistration {p.listenersLock.Lock()defer p.listenersLock.Unlock()if p.listeners == nil {p.listeners = make(map[*processorListener]bool)}p.listeners[listener] = true//这里貌似没进来if p.listenersStarted {p.wg.Start(listener.run)p.wg.Start(listener.pop)}return listener
}

8.informerFactory.Start(stopper)

这一步是启动factory的核心逻辑，前文中主要是进行了一些结构体的定义及初始化，并没有任何执行逻辑存在。戏台子搭好了，这一个步骤应该是核心的执行逻辑了吧。核心代码在这里，主要是把所有注册的informer启动起来。

func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {f.lock.Lock()defer f.lock.Unlock()if f.shuttingDown {return}// 上文已经提到了，informer主要是保存k8s的资源的类runtimeObject和Informer之间的映射关系的缓存for informerType, informer := range f.informers {if !f.startedInformers[informerType] {f.wg.Add(1)// We need a new variable in each loop iteration,// otherwise the goroutine would use the loop variable// and that keeps changing.informer := informergo func() {defer f.wg.Done()//这里另起协程把factory中全部的informer启动起来//8.1 在这里启动indexInformerinformer.Run(stopCh)}()f.startedInformers[informerType] = true}}
}

8.1 informer.Run(stopCh)

func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {....// s.controller = New(cfg)，就是构建controller这个结构体，把shardInformer的start标识修改成truefunc() {s.startedLock.Lock()defer s.startedLock.Unlock()//7.1.1 搞了一个队列fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KnownObjects:          s.indexer,EmitDeltaTypeReplaced: true,Transformer: s.transform,})//1)构建了一个配置项cfg := &Config{Queue:             fifo,ListerWatcher:     s.listerWatcher,ObjectType:        s.objectType,ObjectDescription: s.objectDescription,FullResyncPeriod:  s.resyncCheckPeriod,RetryOnError:      false,ShouldResync:      s.processor.shouldResync,Process:           s.HandleDeltas,WatchErrorHandler: s.watchErrorHandler,}//2)创建controllers.controller = New(cfg)s.controller.(*controller).clock = s.clocks.started = true}()// Separate stop channel because Processor should be stopped strictly after controllerprocessorStopCh := make(chan struct{})var wg wait.Groupdefer wg.Wait()              // Wait for Processor to stopdefer close(processorStopCh) // Tell Processor to stop//这个不知道做了什么事情，用到的时候再说wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)//3)启动controller这里是真正启动listeners的地方wg.StartWithChannel(processorStopCh, s.processor.run)defer func() {s.startedLock.Lock()defer s.startedLock.Unlock()s.stopped = true // Don't want any new listeners}()//4) 启动controllers.controller.Run(stopCh)
}

8.1.1 wg.StartWithChannel(processorStopCh, s.processor.run)

这里启动了一个协程用来执行processor.run方法，我们进到这个函数中看一下这个函数做了些什么事情.

func (p *sharedProcessor) run(stopCh <-chan struct{}) {func() {p.listenersLock.RLock()defer p.listenersLock.RUnlock()// p.listener还记得是做什么用的吗，就是保存所有的listener的缓存for listener := range p.listeners {// 启动listener，逻辑在上文已经提及p.wg.Start(listener.run)p.wg.Start(listener.pop)}p.listenersStarted = true}()<-stopCh// 后续都是中止逻辑了...
}

8.1.2 s.controller.Run(stopCh),controller

1）构建reflector
2）启动reflector
3）启动循环处理逻辑

func (c *controller) Run(stopCh <-chan struct{}) {defer utilruntime.HandleCrash()go func() {<-stopChc.config.Queue.Close()}()// 1. 初始化reflect来同步服务器和store的objectr := NewReflectorWithOptions(c.config.ListerWatcher,c.config.ObjectType,c.config.Queue,ReflectorOptions{ResyncPeriod:    c.config.FullResyncPeriod,TypeDescription: c.config.ObjectDescription,Clock:           c.clock,},)r.ShouldResync = c.config.ShouldResyncr.WatchListPageSize = c.config.WatchListPageSizeif c.config.WatchErrorHandler != nil {r.watchErrorHandler = c.config.WatchErrorHandler}c.reflectorMutex.Lock()// 对reflector赋值c.reflector = rc.reflectorMutex.Unlock()var wg wait.Group// 2.启动reflectorwg.StartWithChannel(stopCh, r.Run)// 3.循环处理队列中的元素wait.Until(c.processLoop, time.Second, stopCh)wg.Wait()
}

我们来看一下reflector启动后在做些什么事情。记住一句话就好：首先从k8s的apiserver中利用list的方式获取指定资源的全部数据，并塞到deltaqueue中。后续通过watch的方式，监听资源的变动信息并同步到队列中国。


func (r *Reflector) Run(stopCh <-chan struct{}) {klog.V(3).Infof("Starting reflector %s (%s) from %s", r.typeDescription, r.resyncPeriod, r.name)wait.BackoffUntil(func() {// 核心逻辑入口if err := r.ListAndWatch(stopCh); err != nil {r.watchErrorHandler(r, err)}}, r.backoffManager, true, stopCh)klog.V(3).Infof("Stopping reflector %s (%s) from %s", r.typeDescription, r.resyncPeriod, r.name)
}func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {klog.V(3).Infof("Listing and watching %v from %s", r.typeDescription, r.name)var err errorvar w watch.InterfacefallbackToList := !r.UseWatchList//这里看起来不会落到这个执行逻辑中if r.UseWatchList {// 维持本地的资源缓存w, err = r.watchList(stopCh)if w == nil && err == nil {// stopCh was closedreturn nil}if err != nil {klog.Warningf("The watchlist request ended with an error, falling back to the standard LIST/WATCH semantics because making progress is better than deadlocking, err = %v", err)fallbackToList = true// ensure that we won't accidentally pass some garbage down the watch.w = nil}}if fallbackToList {// 直接借用listwatcher的list方法，从服务器全量同步资源数据err = r.list(stopCh)if err != nil {return err}}klog.V(2).Infof("Caches populated for %v from %s", r.typeDescription, r.name)resyncerrc := make(chan error, 1)cancelCh := make(chan struct{})defer close(cancelCh)//启动一个协程定期同步deltafifo这个结构体，这里埋个坑，有空写deltafifo的时候解释go r.startResync(stopCh, cancelCh, resyncerrc)return r.watch(w, stopCh, resyncerrc)
}

err = r.list(stopCh)这个方法我们看一下里面做了什么事情。

func (r *Reflector) list(stopCh <-chan struct{}) error {.....// 启动一个线程全量同步资源go func() {.....list, paginatedResult, err = pager.ListWithAlloc(context.Background(), options).....}()// 解析成指定的结构体items, err := meta.ExtractListWithAlloc(list) ....initTrace.Step("Objects extracted")// 将数据同步到indexer本地存储中if err := r.syncWith(items, resourceVersion); err != nil {return fmt.Errorf("unable to sync list result: %v", err)}...
}

还有一个函数没有分析，这里的watch函数主要是从listwatch的接口中获取数据变动的事件后，动态维护DeltaFiFO队列

func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc chan error) error {.....w, err = r.listerWatcher.Watch(options)....err = watchHandler(start, w, r.store, r.expectedType, r.expectedGVK, r.name, r.typeDescription, r.setLastSyncResourceVersion, nil, r.clock, resyncerrc, stopCh).....
}func watchHandler(start time.Time,w watch.Interface,store Store,expectedType reflect.Type,expectedGVK *schema.GroupVersionKind,name string,expectedTypeName string,setLastSyncResourceVersion func(string),exitOnInitialEventsEndBookmark *bool,clock clock.Clock,errc chan error,stopCh <-chan struct{},
) error {eventCount := 0if exitOnInitialEventsEndBookmark != nil {// set it to false just in case somebody// made it positive*exitOnInitialEventsEndBookmark = false}loop:for {select {case <-stopCh:return errorStopRequestedcase err := <-errc:return err//从watch接口中获取事件case event, ok := <-w.ResultChan():......case watch.Added:err := store.Add(event.Object)if err != nil {utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", name, event.Object, err))}case watch.Modified:err := store.Update(event.Object)if err != nil {utilruntime.HandleError(fmt.Errorf("%s: unable to update watch event object (%#v) to store: %v", name, event.Object, err))}case watch.Deleted:// TODO: Will any consumers need access to the "last known// state", which is passed in event.Object? If so, may need// to change this.err := store.Delete(event.Object)if err != nil {utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", name, event.Object, err))}case watch.Bookmark:// A `Bookmark` means watch has synced here, just update the resourceVersionif meta.GetAnnotations()["k8s.io/initial-events-end"] == "true" {if exitOnInitialEventsEndBookmark != nil {*exitOnInitialEventsEndBookmark = true}}}

我们看一下这个c.processLoop这个循环做了些什么事情。

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

围绕着queue，展开了腥风血雨，看起来对于queue中元素的处理落在在c.config.Process这个函数中，我们找一下这个函数在哪里被初始化的

func (s *sharedIndexInformer) HandleDeltas(obj interface{}, isInInitialList bool) error {s.blockDeltas.Lock()defer s.blockDeltas.Unlock()if deltas, ok := obj.(Deltas); ok {return processDeltas(s, s.indexer, deltas, isInInitialList)}return errors.New("object given as Process argument is not Deltas")
}func processDeltas(// Object which receives event notifications from the given deltashandler ResourceEventHandler,clientState Store,deltas Deltas,isInInitialList bool,
) error {// from oldest to newestfor _, d := range deltas {obj := d.Objectswitch 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, isInInitialList)}case Deleted:if err := clientState.Delete(obj); err != nil {return err}handler.OnDelete(obj)}}return nil
}

clientState.Update(obj)这个方法我们点击去看会发现，他更新的就是我们上文提及结构体的indexer内部缓存。这里的handler就是s *sharedIndexInformer，我们把这个结构体的OnAdd，OnUpdate，OnDelete方法截取出来。

// Conforms to ResourceEventHandler
func (s *sharedIndexInformer) OnAdd(obj interface{}, isInInitialList bool) {// Invocation of this function is locked under s.blockDeltas, so it is// save to distribute the notifications.cacheMutationDetector.AddObject(obj)s.processor.distribute(addNotification{newObj: obj, isInInitialList: isInInitialList}, false)
}// Conforms to ResourceEventHandler
func (s *sharedIndexInformer) OnUpdate(old, new interface{}) {isSync := false// If is a Sync event, isSync should be true// If is a Replaced event, isSync is true if resource version is unchanged.// If RV is unchanged: this is a Sync/Replaced event, so isSync is trueif accessor, err := meta.Accessor(new); err == nil {if oldAccessor, err := meta.Accessor(old); err == nil {// Events that didn't change resourceVersion are treated as resync events// and only propagated to listeners that requested resyncisSync = accessor.GetResourceVersion() == oldAccessor.GetResourceVersion()}}// Invocation of this function is locked under s.blockDeltas, so it is// save to distribute the notifications.cacheMutationDetector.AddObject(new)s.processor.distribute(updateNotification{oldObj: old, newObj: new}, isSync)
}// Conforms to ResourceEventHandler
func (s *sharedIndexInformer) OnDelete(old interface{}) {// Invocation of this function is locked under s.blockDeltas, so it is// save to distribute the notifications.processor.distribute(deleteNotification{oldObj: old}, false)
}

其中核心的方法大概就是s.processor.distribute(addNotification{newObj: obj, isInInitialList: isInInitialList}, false)，我们看一下这个方法是在做什么。

func (p *sharedProcessor) distribute(obj interface{}, sync bool) {p.listenersLock.RLock()defer p.listenersLock.RUnlock()for listener, isSyncing := range p.listeners {switch {case !sync:// non-sync messages are delivered to every listenerlistener.add(obj)case isSyncing:// sync messages are delivered to every syncing listenerlistener.add(obj)default:// skipping a sync obj for a non-syncing listener}}
}
// 还记得processorListener中的两个channel吗，这里就是往addCh中添加元素了
func (p *processorListener) add(notification interface{}) {if a, ok := notification.(addNotification); ok && a.isInInitialList {p.syncTracker.Start()}p.addCh <- notification
}

9.informerFactory.WaitForCacheSync(stopper)

等待所有启动的 Informer 的缓存被同步，真正的实现方法如下所示

func (f *sharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool {//这里是map[reflect.Type]cache.SharedIndexInformer 这个结构的一个深度拷贝，没有做额外的逻辑informers := func() map[reflect.Type]cache.SharedIndexInformer {f.lock.Lock()defer f.lock.Unlock()informers := map[reflect.Type]cache.SharedIndexInformer{}for informerType, informer := range f.informers {if f.startedInformers[informerType] {informers[informerType] = informer}}return informers}()//这里是判断所有的informer的数据是否被完全同步res := map[reflect.Type]bool{}for informType, informer := range informers {//这是一个同步的方法，只有当每一个informer的queue都已经被完全同步，保存执行结果。如果有一个informer没有同步，就一直卡在则立res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced)}return res
}

10.deployments, err := deployLister.Deployments(“default”).List(labels.Everything())

我们追踪一下这个List方法是从远程k8s apiserver获取的还是从本地的indexer存储中获取的。

func ListAllByNamespace(indexer Indexer, namespace string, selector labels.Selector, appendFn AppendFunc) error {.....if namespace == metav1.NamespaceAll {return ListAll(indexer, selector, appendFn)}// 结论在这里items, err := indexer.Index(NamespaceIndex, &metav1.ObjectMeta{Namespace: namespace}).....}

写在之后

絮絮叨叨终于理完了informer的全部源码逻辑，笔者记录了自己的阅读源码的过程，期望能在未来温习的时候能够有一个参考，可能之后会更新一些流程图去加深一下记忆吧，但是至少现在，洋洋洒洒的分析确实耗尽了我的耐心。接下来有时间的话，我会继续更新controller-runtime的源码解析。如果大家觉得有什么问题，还请在评论里不吝赐教。如果大家觉得有那么一丝帮助，还请大家帮忙点赞关注，再次拜谢。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/636544.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Idea 开发环境不断切换git代码分支导致冲掉别人代码

Idea 开发环境不断切换git代码分支导致冲掉别人代码

问题分析 使用git reflog查看执行命令&#xff0c;以下是发生事故的切换和提交动作 46f72622e1 HEAD{41}: commit: feat: 【Sales - 6.3】小程序端不登录也可以录入客户线索 c5e7d9f6e1 HEAD{42}: fetch origin feature/20240102_Sales6.3_xingang:feature/20240102_Sales6.3…
阅读更多...
零基础小白刚刚入门Python的注意点总结~

零基础小白刚刚入门Python的注意点总结~

文章目录 一、注意你的Python版本1.print()函数2.raw_input()与input()3.比较符号&#xff0c;使用!替换<>4.repr函数5.exec()函数 二、新手常遇到的问题1、如何写多行程序&#xff1f;2、如何执行.py文件&#xff1f;3、and&#xff0c;or&#xff0c;not4、True和False…
阅读更多...
代码随想录算法训练营Day24 | 122.买卖股票的最佳时期、55.跳跃游戏、45.跳跃游戏||

代码随想录算法训练营Day24 | 122.买卖股票的最佳时期、55.跳跃游戏、45.跳跃游戏||

LeetCode 122 买卖股票的最佳时期 本题思路&#xff1a;记录每天的利润值&#xff0c;第一天的为 0 &#xff0c;第二天的当天的减去前一天的。然后遍历相加为正的利润值。最后得到的结果就是最大利润。 class Solution {public int maxProfit(int[] prices) {int res 0;for(…
阅读更多...
使用Nginx和Fancyindex组合搭建文件下载站点详细教程

使用Nginx和Fancyindex组合搭建文件下载站点详细教程

目录 简介 TIPS 1.下载Nginx 2. 安装Fancyindex和Nginx-Fancyindex-Theme模块 2.1 安装编译工具和依赖 2.2 下载Fancyindex和Nginx-Fancyindex-Theme 2.3 编译Nginx并包括Fancyindex 3. 配置Nginx 4.体验 4.1light主题 4.2dark主题 后记 简介 当使用Nginx和Fancyinde…
阅读更多...
Unity3D学习之Unity基础

Unity3D学习之Unity基础

文章目录 1. 第一部分&#xff1a;MONO中的重要内容2. 延时函数2.1 什么是延时函数2.2 延时函数的使用2.2.1 延时重复函数2.2.2 取消延迟函数2.2.3 判断是否有延迟函数开启2.2.4 延迟函数和依附对象的关系 3 协同程序3.1 Unity中的多线程3.2 协同程序3.3 协程的使用3.3.1 关闭协…
阅读更多...
第三课：GPT

第三课：GPT

文章目录 第三课&#xff1a;GPT1、学习总结&#xff1a;GPT出现的原因GPT的方法原理目前存在的问题无监督的预训练优化目标模型结构 监督微调课程ppt及代码地址 2、学习心得&#xff1a;3、经验分享&#xff1a;4、课程反馈&#xff1a;5、使用MindSpore昇思的体验和反馈&…
阅读更多...
【Spring源码分析】扫描并注册BeanDefinition逻辑

【Spring源码分析】扫描并注册BeanDefinition逻辑

扫描并注册BeanDefinition逻辑 一、ClassPathBeanDefinitionScanner 扫描源码分析doScan 扫描的具体逻辑findCandidateComponents 方法解析generateBeanName 方法解析checkCandidate 方法解析 二、总结 阅读此需阅读下面这些博客先【Spring源码分析】Bean的元数据和一些Spring的…
阅读更多...
SpringBoot项目中简单使用虚拟机Redis

SpringBoot项目中简单使用虚拟机Redis

目录 步骤大致如下&#xff1a; 一.在pom文件中加入redis依赖 二.在虚拟机上打开我们下载好的Redis。开启服务器端并获取虚拟机ip地址 三.在项目配置。 四&#xff1a;使用redis 测试 redis是一个以键值对存储的NoSQL。被数百万开发人员用作缓存、矢量数据库、文档数据库、…
阅读更多...
AcWing 68：0到n-1中缺失的数字 ← 二分

AcWing 68：0到n-1中缺失的数字 ← 二分

【题目来源】https://www.acwing.com/problem/content/64/【题目描述】 一个长度为 n−1 的递增排序数组中的所有数字都是唯一的&#xff0c;并且每个数字都在范围 0 到 n−1 之内。 在范围 0 到 n−1 的 n 个数字中有且只有一个数字不在该数组中&#xff0c;请找出这个数字。【…
阅读更多...
vivado 定义板文件板

vivado 定义板文件板

定义板文件板 &#xff1c;board&#xff1e;标记是板文件的根。它包括识别基本信息的属性关于董事会。 <board schema_version"2.1" vendor"xilinx.com" name"kc705" display_name"Kintex-7 KC705 Evaluation Platform" url&qu…
阅读更多...
用C语言实现简单的三子棋游戏

用C语言实现简单的三子棋游戏

目录 1 -> 模块简介 2 -> test.c 3 -> game.c 4 -> game.h 1 -> 模块简介 test.c:测试游戏逻辑 game.c: 函数的实现 game.h:函数的声明 2 -> test.c #define _CRT_SECURE_NO_WARNINGS 1#include "game.h";void menu() {printf("****…
阅读更多...
Harmony Ble蓝牙App（四）描述符

Harmony Ble蓝牙App（四）描述符

Harmony Ble蓝牙App&#xff08;四&#xff09;描述符 前言正文一、优化二、描述① 概念② 描述提供者③ 显示描述符 三、源码 前言 上一篇中了解了特性和属性&#xff0c;同时显示设备蓝牙服务下的特性和属性&#xff0c;本文中就需要来使用这些特性和属性来完成一些功能。 正…
阅读更多...
FaFu--练习复盘--1

FaFu--练习复盘--1

1、输出图形及二维数组应用 1.1.输出图形 描述 编写程序打印n行如下图形&#xff0c;其中1≤n≤500。 输入用例 7 输出用例 具体实现 #include"stdio.h" int main(){int n,i,j;scanf("%d",&n);for(i 1; i< n;…
阅读更多...
软考系分之计算机网络通信方向、同步和交换

软考系分之计算机网络通信方向、同步和交换

文章目录 1、概述2、通信方向和同步3、交换方式4、总结 1、概述 本篇依旧是一图概括主要考察的知识点&#xff0c;包括通信方向&#xff08;单工、半双工、全双工&#xff09;&#xff0c;同步方式和数据交换等。 2、通信方向和同步 通信方向&#xff08;单工、半双工、全双工…
阅读更多...
【react】创建react项目+项目结构

【react】创建react项目+项目结构

使用create-react-app快速搭建开发环境 create-react-app是一个快速创建React开发环境的工具&#xff0c;底层由Webpack构建&#xff0c;封装了配置细节 npx create-react-app react_hm执行命令后开始创建 创建好执行cd react_hm npm start 当看到webpack compiled successfu…
阅读更多...
spring boot shardingsphere mybatis-plus druid mysql 搭建mysql数据库读写分离架构

spring boot shardingsphere mybatis-plus druid mysql 搭建mysql数据库读写分离架构

spring boot shardingsphere mybatis-plus druid mysql 搭建mysql数据库读写分离架构 ##关于window mysql主从搭建简单教程 传送门 window mysql5.7 搭建主从同步环境-CSDN博客 ##父pom.xml <?xml version"1.0" encoding"UTF-8"?> <project…
阅读更多...
傲空间私有部署Windows指南

傲空间私有部署Windows指南

推荐阅读 智能化校园&#xff1a;深入探讨云端管理系统设计与实现&#xff08;一&#xff09; 智能化校园&#xff1a;深入探讨云端管理系统设计与实现&#xff08;二&#xff09; 安装 docker 请下载对应的 Docker&#xff0c;安装完成后启动。 Docker Desktop for Windows…
阅读更多...
领略指针之妙

领略指针之妙

&#x1d649;&#x1d65e;&#x1d658;&#x1d65a;!!&#x1f44f;&#x1f3fb;‧✧̣̥̇‧✦&#x1f44f;&#x1f3fb;‧✧̣̥̇‧✦ &#x1f44f;&#x1f3fb;‧✧̣̥̇:Solitary-walk ⸝⋆ ━━━┓ - 个性标签 - &#xff1a;来于“云”的“羽球人”。…
阅读更多...
MySQL主从集群

MySQL主从集群

MySQL主从集群 主从模式、集群模式&#xff0c;都是在一个项目中使用多个mysql节点进行存储和读取数据。 当单机模式部署&#xff0c;不满足安全性、高可用、高并发等需求的时候&#xff0c;就需要考虑主从模式或者集群模式部署。 什么是主从模式&#xff1f; 主从模式&…
阅读更多...
[3D]菜板上的鱼

[3D]菜板上的鱼

本来想画条鲨鱼&#xff0c;结果成了条菜板上的鱼。 VS&#xff01; ** VS&#xff01; ** 【扭曲】更像菜板上的鱼了。
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023