k8s client-go源码分析 informer源码分析(2)-初始化与启动分析-6ren

k8s client-go源码分析 informer源码分析(2)-初始化与启动分析

转载作者：知者更新时间：2024-03-13 00:50:16

k8s client-go k8s informers实现了持续获取集群的所有资源对象、监听集群的资源对象变化功能，并在本地维护了全量资源对象的内存缓存，以减少对apiserver、对etcd的请求压力。Informers在启动的时候会首先在客户端调用List接口来获取全量的对象集合，然后通过Watch接口来获取增量的对象，然后更新本地缓存。

k8s client-go源码分析 informer源码分析(2)-初始化与启动分析

前面一篇文章对k8s informer做了概要分析，本篇文章将对informer的初始化与启动进行分析。

informer架构

先来回忆一下informer的架构。

k8s client-go informer主要包括以下部件：
（1）Reflector：Reflector从kube-apiserver中list&watch资源对象，然后调用DeltaFIFO的Add/Update/Delete/Replace方法将资源对象及其变化包装成Delta并将其丢到DeltaFIFO中；
（2）DeltaFIFO：DeltaFIFO中存储着一个map和一个queue，即map[object key]Deltas以及object key的queue，Deltas为Delta的切片类型，Delta装有对象及对象的变化类型（Added/Updated/Deleted/Sync），Reflector负责DeltaFIFO的输入，Controller负责处理DeltaFIFO的输出；
（3）Controller：Controller从DeltaFIFO的queue中pop一个object key出来，并获取其关联的 Deltas出来进行处理，遍历Deltas，根据对象的变化更新Indexer中的本地内存缓存，并通知Processor，相关对象有变化事件发生；
（4）Processor：Processor根据对象的变化事件类型，调用相应的ResourceEventHandler来处理对象的变化；
（5）Indexer：Indexer中有informer维护的指定资源对象的相对于etcd数据的一份本地内存缓存，可通过该缓存获取资源对象，以减少对apiserver、对etcd的请求压力；
（6）ResourceEventHandler：用户根据自身处理逻辑需要，注册自定义的的ResourceEventHandler，当对象发生变化时，将触发调用对应类型的ResourceEventHandler来做处理。

概述

...
	factory := informers.NewSharedInformerFactory(client, 30*time.Second)
	podInformer := factory.Core().V1().Pods()
	informer := podInformer.Informer()
	...
	go factory.Start(stopper)
	...
	if !cache.WaitForCacheSync(stopper, informer.HasSynced) {
		runtime.HandleError(fmt.Errorf("Timed out waiting for caches to sync"))
		return
	}
	...

上一节有列举了informer的使用代码，注意看到示例代码中的下面这段代码，做了informer初始化与启动，其中包括：
（1）informers.NewSharedInformerFactory：初始化informer factory；
（2）podInformer.Informer：初始化pod informer；
（3）factory.Start：启动informer factory；
（4）cache.WaitForCacheSync：等待list操作获取到的对象都同步到informer本地缓存Indexer中；

下面也将根据这四部分进行informer的初始化与启动分析。

基于k8s v1.17.4版本依赖的client-go

1.SharedInformerFactory的初始化

1.1 sharedInformerFactory结构体

先来看下sharedInformerFactory结构体，看下里面有哪些属性。

看到几个比较重要的属性：
（1）client：连接k8s的clientSet；
（2）informers：是个map，可以装各个对象的informer；
（3）startedInformers：记录已经启动的informer；

// staging/src/k8s.io/client-go/informers/factory.go
type sharedInformerFactory struct {
	client           kubernetes.Interface
	namespace        string
	tweakListOptions internalinterfaces.TweakListOptionsFunc
	lock             sync.Mutex
	defaultResync    time.Duration
	customResync     map[reflect.Type]time.Duration

	informers map[reflect.Type]cache.SharedIndexInformer
	// startedInformers is used for tracking which informers have been started.
	// This allows Start() to be called multiple times safely.
	startedInformers map[reflect.Type]bool
}

1.2 NewSharedInformerFactory

NewSharedInformerFactory方法用于初始化informer factory，主要是初始化并返回sharedInformerFactory结构体。

// staging/src/k8s.io/client-go/informers/factory.go
func NewSharedInformerFactory(client kubernetes.Interface, defaultResync time.Duration) SharedInformerFactory {
	return NewSharedInformerFactoryWithOptions(client, defaultResync)
}

func NewFilteredSharedInformerFactory(client kubernetes.Interface, defaultResync time.Duration, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) SharedInformerFactory {
	return NewSharedInformerFactoryWithOptions(client, defaultResync, WithNamespace(namespace), WithTweakListOptions(tweakListOptions))
}

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 options
	for _, opt := range options {
		factory = opt(factory)
	}

	return factory
}

2.对象informer的初始化

上一节有列举了informer的使用代码，注意看到示例代码中的下面这段代码，这里利用了工厂方法设计模式，podInformer.Informer()即初始化了sharedInformerFactory中的pod的informer，具体调用关系可自行看如下代码，比较简单，这里不再展开分析。

// 初始化informer factory以及pod informer
	factory := informers.NewSharedInformerFactory(client, 30*time.Second)
	podInformer := factory.Core().V1().Pods()
	informer := podInformer.Informer()

2.1 podInformer.Informer

Informer方法中调用了f.factory.InformerFor方法来做pod informer的初始化。

// k8s.io/client-go/informers/core/v1/pod.go
func (f *podInformer) Informer() cache.SharedIndexInformer {
	return f.factory.InformerFor(&corev1.Pod{}, f.defaultInformer)
}

2.2 f.factory.InformerFor

Informer方法中调用了f.factory.InformerFor方法来做pod informer的初始化，并传入f.defaultInformer作为newFunc，而在f.factory.InformerFor方法中，调用newFunc来初始化informer。

这里也可以看到，其实informer初始化后会存储进map f.informers[informerType]中，即存储进sharedInformerFactory结构体的informers属性中，方便共享使用。

// staging/src/k8s.io/client-go/informers/factory.go
func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
	f.lock.Lock()
	defer f.lock.Unlock()

	informerType := reflect.TypeOf(obj)
	informer, exists := f.informers[informerType]
	if exists {
		return informer
	}

	resyncPeriod, exists := f.customResync[informerType]
	if !exists {
		resyncPeriod = f.defaultResync
	}

	informer = newFunc(f.client, resyncPeriod)
	f.informers[informerType] = informer

	return informer
}

2.3 newFunc/f.defaultInformer

defaultInformer方法中，调用了NewFilteredPodInformer方法来初始化pod informer，最终初始化并返回sharedIndexInformer结构体。

// k8s.io/client-go/informers/core/v1/pod.go
func (f *podInformer) defaultInformer(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
	return NewFilteredPodInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
}

func NewFilteredPodInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
	return cache.NewSharedIndexInformer(
		&cache.ListWatch{
			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
				if tweakListOptions != nil {
					tweakListOptions(&options)
				}
				return client.CoreV1().Pods(namespace).List(options)
			},
			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
				if tweakListOptions != nil {
					tweakListOptions(&options)
				}
				return client.CoreV1().Pods(namespace).Watch(options)
			},
		},
		&corev1.Pod{},
		resyncPeriod,
		indexers,
	)
}

func NewSharedIndexInformer(lw ListerWatcher, objType runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer {
	realClock := &clock.RealClock{}
	sharedIndexInformer := &sharedIndexInformer{
		processor:                       &sharedProcessor{clock: realClock},
		indexer:                         NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers),
		listerWatcher:                   lw,
		objectType:                      objType,
		resyncCheckPeriod:               defaultEventHandlerResyncPeriod,
		defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod,
		cacheMutationDetector:           NewCacheMutationDetector(fmt.Sprintf("%T", objType)),
		clock: realClock,
	}
	return sharedIndexInformer
}

2.4 sharedIndexInformer结构体

sharedIndexInformer结构体中重点看到以下几个属性：
（1）indexer：对应着informer中的部件Indexer，Indexer中有informer维护的指定资源对象的相对于etcd数据的一份本地内存缓存，可通过该缓存获取资源对象，以减少对apiserver、对etcd的请求压力；
（2）controller：对应着informer中的部件Controller，Controller从DeltaFIFO中pop Deltas出来处理，根据对象的变化更新Indexer中的本地内存缓存，并通知Processor，相关对象有变化事件发生；
（3）processor：对应着informer中的部件Processor，Processor根据对象的变化事件类型，调用相应的ResourceEventHandler来处理对象的变化；

// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
type sharedIndexInformer struct {
	indexer    Indexer
	controller Controller

	processor             *sharedProcessor
	cacheMutationDetector CacheMutationDetector

	// This block is tracked to handle late initialization of the controller
	listerWatcher ListerWatcher
	objectType    runtime.Object

	// resyncCheckPeriod is how often we want the reflector's resync timer to fire so it can call
	// shouldResync to check if any of our listeners need a resync.
	resyncCheckPeriod time.Duration
	// defaultEventHandlerResyncPeriod is the default resync period for any handlers added via
	// AddEventHandler (i.e. they don't specify one and just want to use the shared informer's default
	// value).
	defaultEventHandlerResyncPeriod time.Duration
	// clock allows for testability
	clock clock.Clock

	started, stopped bool
	startedLock      sync.Mutex

	// blockDeltas gives a way to stop all event distribution so that a late event handler
	// can safely join the shared informer.
	blockDeltas sync.Mutex
}

Indexer接口与cache结构体

cache结构体为Indexer接口的实现；

// staging/src/k8s.io/client-go/tools/cache/store.go
type cache struct {
	cacheStorage ThreadSafeStore
	keyFunc KeyFunc
}

threadSafeMap struct是ThreadSafeStore接口的一个实现，其最重要的一个属性便是items了，items是用map构建的键值对，资源对象都存在items这个map中，key根据资源对象来算出，value为资源对象本身，这里的items即为informer的本地缓存了，而indexers与indices属性则与索引功能有关。

// staging/src/k8s.io/client-go/tools/cache/thread_safe_store.go
type threadSafeMap struct {
	lock  sync.RWMutex
	items map[string]interface{}

	// indexers maps a name to an IndexFunc
	indexers Indexers
	// indices maps a name to an Index
	indices Indices
}

关于Indexer的详细分析会在后续有专门的文章做分析，这里不展开分析；

controller结构体

而controller结构体则包含了informer中的主要部件Reflector以及DeltaFIFO；
（1）Reflector：Reflector从kube-apiserver中list&watch资源对象，然后将对象的变化包装成Delta并将其丢到DeltaFIFO中；
（2）DeltaFIFO：DeltaFIFO存储着map[object key]Deltas以及object key的queue，Delta装有对象及对象的变化类型，Reflector负责DeltaFIFO的输入，Controller负责处理DeltaFIFO的输出；

// staging/src/k8s.io/client-go/tools/cache/controller.go
type controller struct {
	config         Config
	reflector      *Reflector
	reflectorMutex sync.RWMutex
	clock          clock.Clock
}

type Config struct {
	// The queue for your objects; either a FIFO or
	// a DeltaFIFO. Your Process() function should accept
	// the output of this Queue's Pop() method.
	Queue
	...
}

3.启动sharedInformerFactory

sharedInformerFactory.Start为informer factory的启动方法，其主要逻辑为循环遍历informers，然后跑goroutine调用informer.Run来启动sharedInformerFactory中存储的各个informer。

// staging/src/k8s.io/client-go/informers/factory.go
func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {
	f.lock.Lock()
	defer f.lock.Unlock()

	for informerType, informer := range f.informers {
		if !f.startedInformers[informerType] {
			go informer.Run(stopCh)
			f.startedInformers[informerType] = true
		}
	}
}

sharedIndexInformer.Run

sharedIndexInformer.Run用于启动informer，主要逻辑为：
（1）调用NewDeltaFIFO，初始化DeltaFIFO；
（2）构建Config结构体，这里留意下Process属性，赋值了s.HandleDeltas，后面会分析到该方法；
（3）调用New，利用Config结构体来初始化controller；
（4）调用s.processor.run，启动processor；
（5）调用s.controller.Run，启动controller；

// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
    
    // 初始化DeltaFIFO
	fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer)
    
    // 构建Config结构体
	cfg := &Config{
		Queue:            fifo,
		ListerWatcher:    s.listerWatcher,
		ObjectType:       s.objectType,
		FullResyncPeriod: s.resyncCheckPeriod,
		RetryOnError:     false,
		ShouldResync:     s.processor.shouldResync,

		Process: s.HandleDeltas,
	}

	func() {
		s.startedLock.Lock()
		defer s.startedLock.Unlock()
        // 初始化controller
		s.controller = New(cfg)
		s.controller.(*controller).clock = s.clock
		s.started = true
	}()

	// Separate stop channel because Processor should be stopped strictly after controller
	processorStopCh := make(chan struct{})
	var wg wait.Group
	defer wg.Wait()              // Wait for Processor to stop
	defer close(processorStopCh) // Tell Processor to stop
	wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
	// 启动processor
	wg.StartWithChannel(processorStopCh, s.processor.run)

	defer func() {
		s.startedLock.Lock()
		defer s.startedLock.Unlock()
		s.stopped = true // Don't want any new listeners
	}()
	// 启动controller
	s.controller.Run(stopCh)
}

3.1 New

New函数初始化了controller并return。

// staging/src/k8s.io/client-go/tools/cache/controller.go
func New(c *Config) Controller {
	ctlr := &controller{
		config: *c,
		clock:  &clock.RealClock{},
	}
	return ctlr
}

3.2 s.processor.run

s.processor.run启动了processor，其中注意到listener.run与listener.pop两个核心方法即可，暂时没有用到，等下面用到他们的时候再做分析。

// 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)
			p.wg.Start(listener.pop)
		}
		p.listenersStarted = true
	}()
	<-stopCh
	p.listenersLock.RLock()
	defer p.listenersLock.RUnlock()
	for _, listener := range p.listeners {
		close(listener.addCh) // Tell .pop() to stop. .pop() will tell .run() to stop
	}
	p.wg.Wait() // Wait for all .pop() and .run() to stop
}

3.3 controller.Run

controller.Run为controller的启动方法，这里主要看到几个点：
（1）调用NewReflector，初始化Reflector；
（2）调用r.Run，实际上是调用了Reflector的启动方法来启动Reflector；
（3）调用c.processLoop，开始controller的核心处理；

// k8s.io/client-go/tools/cache/controller.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.clock = c.clock

	c.reflectorMutex.Lock()
	c.reflector = r
	c.reflectorMutex.Unlock()

	var wg wait.Group
	defer wg.Wait()

	wg.StartWithChannel(stopCh, r.Run)

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

3.3.1 Reflector结构体

先来看到Reflector结构体，这里重点看到以下属性：
（1）expectedType：放到Store中（即DeltaFIFO中）的对象类型；
（2）store：store会赋值为DeltaFIFO，具体可以看之前的informer初始化与启动分析即可得知，这里不再展开分析；
（3）listerWatcher：存放list方法和watch方法的ListerWatcher interface实现；

// k8s.io/client-go/tools/cache/reflector.go
type Reflector struct {
    ...
    expectedType reflect.Type
    store Store
    listerWatcher ListerWatcher
    ...
}

3.3.2 r.Run/Reflector.Run

Reflector.Run方法中启动了Reflector，而Reflector的核心处理逻辑为从kube-apiserver处做list&watch操作，然后将得到的对象封装存储进DeltaFIFO中。

// staging/src/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.expectedTypeName, r.resyncPeriod, r.name)
	wait.Until(func() {
		if err := r.ListAndWatch(stopCh); err != nil {
			utilruntime.HandleError(err)
		}
	}, r.period, stopCh)
}

3.3.3 controller.processLoop

controller的核心处理方法processLoop中，最重要的逻辑是循环调用c.config.Queue.Pop将DeltaFIFO中的队头元素给pop出来，然后调用c.config.Process方法来做处理，当处理出错时，再调用c.config.Queue.AddIfNotPresent将对象重新加入到DeltaFIFO中去。

// k8s.io/client-go/tools/cache/controller.go
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)
			}
		}
	}
}

3.3.4 c.config.Process/sharedIndexInformer.HandleDeltas

根据前面sharedIndexInformer.Run方法的分析中可以得知，c.config.Process其实就是sharedIndexInformer.HandleDeltas。

HandleDeltas方法中，将从DeltaFIFO中pop出来的对象以及类型，相应的在indexer中做添加、更新、删除操作，并调用s.processor.distribute通知自定义的ResourceEventHandler。

// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
	s.blockDeltas.Lock()
	defer s.blockDeltas.Unlock()

	// from oldest to newest
	for _, d := range obj.(Deltas) {
		switch d.Type {
		case Sync, Added, Updated:
			isSync := d.Type == Sync
			s.cacheMutationDetector.AddObject(d.Object)
			if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
				if err := s.indexer.Update(d.Object); err != nil {
					return err
				}
				s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
			} else {
				if err := s.indexer.Add(d.Object); err != nil {
					return err
				}
				s.processor.distribute(addNotification{newObj: d.Object}, isSync)
			}
		case Deleted:
			if err := s.indexer.Delete(d.Object); err != nil {
				return err
			}
			s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
		}
	}
	return nil
}

怎么通知到自定义的ResourceEventHandler呢？继续往下看。

3.3.5 sharedIndexInformer.processor.distribute

可以看到distribute方法最终是将构造好的addNotification、updateNotification、deleteNotification对象写入到p.addCh中。

// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
	p.listenersLock.RLock()
	defer p.listenersLock.RUnlock()

	if sync {
		for _, listener := range p.syncingListeners {
			listener.add(obj)
		}
	} else {
		for _, listener := range p.listeners {
			listener.add(obj)
		}
	}
}

func (p *processorListener) add(notification interface{}) {
	p.addCh <- notification
}

到这里，processor中的listener.pop以及listener.run方法终于派上了用场，继续往下看。

3.3.6 listener.pop

分析processorListener的pop方法可以得知，其逻辑实际上就是将p.addCh中的对象给拿出来，然后丢进了p.nextCh中。那么谁来处理p.nextCh呢？继续往下看。

// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (p *processorListener) pop() {
	defer utilruntime.HandleCrash()
	defer close(p.nextCh) // Tell .run() to stop

	var nextCh chan<- interface{}
	var notification interface{}
	for {
		select {
		case nextCh <- notification:
			// Notification dispatched
			var ok bool
			notification, ok = p.pendingNotifications.ReadOne()
			if !ok { // Nothing to pop
				nextCh = nil // Disable this select case
			}
		case notificationToAdd, ok := <-p.addCh:
			if !ok {
				return
			}
			if notification == nil { // No notification to pop (and pendingNotifications is empty)
				// Optimize the case - skip adding to pendingNotifications
				notification = notificationToAdd
				nextCh = p.nextCh
			} else { // There is already a notification waiting to be dispatched
				p.pendingNotifications.WriteOne(notificationToAdd)
			}
		}
	}
}

3.3.7 listener.run

在processorListener的run方法中，将循环读取p.nextCh，判断对象类型，是updateNotification则调用p.handler.OnUpdate方法，是addNotification则调用p.handler.OnAdd方法，是deleteNotification则调用p.handler.OnDelete方法做处理。

// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (p *processorListener) run() {
	// this call blocks until the channel is closed.  When a panic happens during the notification
	// we will catch it, **the offending item will be skipped!**, and after a short delay (one second)
	// the next notification will be attempted.  This is usually better than the alternative of never
	// delivering again.
	stopCh := make(chan struct{})
	wait.Until(func() {
		// this gives us a few quick retries before a long pause and then a few more quick retries
		err := wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
			for next := range p.nextCh {
				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
			return true, nil
		})

		// the only way to get here is if the p.nextCh is empty and closed
		if err == nil {
			close(stopCh)
		}
	}, 1*time.Minute, stopCh)
}

而p.handler.OnUpdate、p.handler.OnAdd、p.handler.OnDelete方法实际上就是自定义的的ResourceEventHandlerFuncs了。

informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
    AddFunc:    onAdd,
    UpdateFunc: onUpdate,
    DeleteFunc: onDelete,
  })

// staging/src/k8s.io/client-go/tools/cache/controller.go
type ResourceEventHandlerFuncs struct {
	AddFunc    func(obj interface{})
	UpdateFunc func(oldObj, newObj interface{})
	DeleteFunc func(obj interface{})
}

func (r ResourceEventHandlerFuncs) OnAdd(obj interface{}) {
	if r.AddFunc != nil {
		r.AddFunc(obj)
	}
}

func (r ResourceEventHandlerFuncs) OnUpdate(oldObj, newObj interface{}) {
	if r.UpdateFunc != nil {
		r.UpdateFunc(oldObj, newObj)
	}
}

func (r ResourceEventHandlerFuncs) OnDelete(obj interface{}) {
	if r.DeleteFunc != nil {
		r.DeleteFunc(obj)
	}
}

4.cache.WaitForCacheSync(stopper, informer.HasSynced)

可以看出在cache.WaitForCacheSync方法中，实际上是调用方法入参cacheSyncs ...InformerSynced来判断cache是否同步完成（即调用informer.HasSynced方法），而这里说的cache同步完成，意思是等待informer从kube-apiserver同步资源完成，即informer的list操作获取的对象都存入到informer中的indexer本地缓存中；

// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func WaitForCacheSync(stopCh <-chan struct{}, cacheSyncs ...InformerSynced) bool {
	err := wait.PollImmediateUntil(syncedPollPeriod,
		func() (bool, error) {
			for _, syncFunc := range cacheSyncs {
				if !syncFunc() {
					return false, nil
				}
			}
			return true, nil
		},
		stopCh)
	if err != nil {
		klog.V(2).Infof("stop requested")
		return false
	}

	klog.V(4).Infof("caches populated")
	return true
}

4.1 informer.HasSynced

HasSynced方法实际上是调用了sharedIndexInformer.controller.HasSynced方法；

// staging/src/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()
}

s.controller.HasSynced

这里的c.config.Queue.HasSynced()方法，实际上是指DeltaFIFO的HasSynced方法，会在DeltaFIFO的分析中再详细分析，这里只需要知道当informer的list操作获取的对象都存入到informer中的indexer本地缓存中则返回true即可；

// staging/src/k8s.io/client-go/tools/cache/controller.go
func (c *controller) HasSynced() bool {
	return c.config.Queue.HasSynced()
}

4.2 sharedInformerFactory.WaitForCacheSync

可以顺带看下sharedInformerFactory.WaitForCacheSync方法，其实际上是遍历factory中的所有informer，调用cache.WaitForCacheSync，然后传入每个informer的HasSynced方法作为入参；

// staging/src/k8s.io/client-go/informers/factory.go
func (f *sharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool {
	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
	}()

	res := map[reflect.Type]bool{}
	for informType, informer := range informers {
		res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced)
	}
	return res
}

至此，整个informer的初始化与启动的分析就结束了，后面会对informer中的各个核心部件进行详细分析，敬请期待。

总结

下面用两张图片总结一下informer的初始化与启动；

informer初始化

informer启动

文章推荐：一文开启无监督学习之旅

文章推荐：详解.NET Model ValidationAttribute的使用

文章推荐：本科课程【计算机网络】实验2 - 交换机的VLAN配置实验

文章推荐： Linux 服务器上部署web程序

c++ - 数字 xor K - K = 数字 + K xor K，为什么？
很难说出这里问的是什么。这个问题是含糊的、模糊的、不完整的、过于宽泛的或修辞性的，无法以目前的形式得到合理的回答。如需帮助澄清此问题以便重新打开它，visit the help center 。已关
algorithm - O(K + (N-K)logK) 是否等同于 O(K + N log K)？
我们可以说 O(K + (N-K)logK)相当于O(K + N logK)对于 1 < = K <= N ？最佳答案简短的回答是它们不等价，这取决于k 的值。如果k等于N，那么第一个复杂度是O(
algorithm - 合并 K 个排序链表，为什么复杂度是 O(N * K * K)，而不是 O(N * K)
我有以下解决方案，但我从其他评论者那里听说它是 O(N * K * K)，而不是 O(N * K)其中 N 是 K 列表的(最大)长度，K 是列表的数量。例如，给定列表 [1, 2, 3] 和 [4,
C++ 语法，i % k == l % k == 0 和 i % k == 0 && l % k == 0 之间的区别
我试图理解这些语法结构之间的语义差异。 if ((i% k) == (l % k) == 0) 和 if ((i % k) == 0 && (l % k) == 0) 最佳答案您的特定表达式((i
python - 将数组 (k,) 或 (k, n) 乘以一维数组 (k,)
我有时会使用一维数组: A = np.array([1, 2, 3, 4]) 或 2D 阵列(使用 scipy.io.wavfile 读取单声道或立体声信号): A = np.array([[1, 2
python - 用于确定 k 均值中的 k 的 k 折交叉验证？
在文档聚类过程中，作为数据预处理步骤，我首先应用奇异向量分解得到U、S和Vt 然后通过选择适当数量的特征值，我截断了 Vt，这让我从阅读的内容中得到了很好的文档-文档相关性 here .现在我正在对矩
c++ - Top K 最小选择算法 - O (n + k log n) vs O (n log k) for k << N
我问的是关于 Top K 算法的问题。我认为 O(n + k log n) 应该更快，因为……例如，如果您尝试插入 k = 300 和 n = 100000000，我们可以看到 O(n + k log
r - 列出 k 个数字的所有排列，取自 0 :k,，总和为 k
这个问题与另一个问题R:sample()密切相关。。我想在 R 中找到一种方法来列出 k 个数字的所有排列，总和为 k，其中每个数字都是从 0:k 中选择的。如果k=7，我可以从0,1,...,7中
machine-learning - 了解 Precision@K、AP@K、MAP@K
我目前正在评估基于隐式反馈的推荐系统。我对排名任务的评估指标有点困惑。具体来说，我希望通过精确度和召回率来进行评估。 Precision@k has the advantage of not requ
python - 生成所有可能的 n 维 k*k*...*k 数组，每个数组都有沿轴的行
我在 Python 中工作，需要找到一种算法来生成所有可能的 n 维 k,k,...,k 数组，每个数组都沿轴有一行 1。因此，该函数接受两个数字 - n 和 k，并且应该返回一个数组列表，其中包含沿
algorithm - 寻找最大数量 k 使得对于 k 对的所有组合，我们在每个组合中有 k 个不同的元素
我们有 N 对。每对包含两个数字。我们必须找到最大数 K，这样如果我们从给定的 N 对中取 J (1 2，如果我们选择三对 (1,2)，我们只有两个不同的数字，即 1 和 2。从一个开始检查每个可能
algorithm - 在 O(K*log(K)) 中打印给定堆中最大的 K 个元素？
鉴于以下问题，我不能完全确定我当前的解决方案: 问题: 给定一个包含 n 元素的最大堆，它存储在数组 A 中，是否可以打印所有最大的 K 元素在 O(K*log(K)) 中？我的回答: 是的，是的，
scala - Apache Spark - Scala - 如何将 FlatMap (k, {v1,v2,v3,...}) 到 ((k,v1),(k,v2),(k,v3),...)
我明白了: val vector: RDD[(String, Array[String])] = [("a", {v1,v2,..}),("b", {u1,u2,..})] 想转换成: RDD[(St
algorithm - 将 X 中的所有 x_i 分成 K 组 s.t. var(sum(x in k) for k in K) 被最小化
我有 X 个正数，索引为 x_i。每个 x_i 需要进入 K 组之一(其中 K 是预先确定的)。令 S_j 为 K_j 中所有 x_i 的总和。我需要分配所有 x_i 以使所有 S_j 的方差最小化。
c - 为什么对于长度为 k 的字符串需要 char[k + 1] 而不是 char[k] ？
关闭。这个问题是not reproducible or was caused by typos .它目前不接受答案。这个问题是由于错别字或无法再重现的问题引起的。虽然类似的问题可能是on-topi
algorithm - 为什么 k*k <= n 优于 k <= Math.sqrt(n)
我正在研究寻找原始数的算法，看到下面的语句，我不明白为什么。 while (k*k <= n) 优于 while (k <= Math.sqrt(n)) 是因为函数调用吗？该调用函数使用更多资源。更
c - k x k bool 矩阵的快速乘法，其中 8 <= k <= 16
我想找到一种尽可能快的方法来将两个小 bool 矩阵相乘，其中小意味着 8x8、9x9 ... 16x16。这个例程会被大量使用，所以需要非常高效，所以请不要建议直截了当的解决方案应该足够快。对于
java - Guava :Set + Function = Map？
有没有一种惯用的方法来获取 Set和 Function ，并获得 Map实时取景？ (即 Map 由 Set 和 Function 组合支持，例如，如果将元素添加到 Set ，则相应的条目也存在于 M
c - 函数 f1() 正在返回变量 k 的地址，但由于 k 在堆栈上，因此在括号后它应该从堆栈内存中展开变量 k
这个问题在这里已经有了答案: Can a local variable's memory be accessed outside its scope? (20 个答案) returning addr
matlab - 为什么替换矩阵的 NaN 不适用于 k(k==NaN) = SomeNumber ，其中 k 是要操作的矩阵
给定一个矩阵:- k = [1 2 3 ; 4 5 6 ; 7 8 NaN]; 如果我想用 0 替换一个数字，比如 2，我可以使用这个:k(k==2) =

知者

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城

k8s client-go源码分析 informer源码分析(2)-初始化与启动分析

k8s client-go源码分析 informer源码分析(2)-初始化与启动分析

informer架构

概述

基于k8s v1.17.4版本依赖的client-go

1.SharedInformerFactory的初始化

1.1 sharedInformerFactory结构体

1.2 NewSharedInformerFactory

2.对象informer的初始化

2.1 podInformer.Informer

2.2 f.factory.InformerFor

2.3 newFunc/f.defaultInformer

2.4 sharedIndexInformer结构体

Indexer接口与cache结构体

controller结构体

3.启动sharedInformerFactory

sharedIndexInformer.Run

3.1 New

3.2 s.processor.run

3.3 controller.Run

3.3.1 Reflector结构体

3.3.2 r.Run/Reflector.Run

3.3.3 controller.processLoop

3.3.4 c.config.Process/sharedIndexInformer.HandleDeltas

3.3.5 sharedIndexInformer.processor.distribute

3.3.6 listener.pop

3.3.7 listener.run

4.cache.WaitForCacheSync(stopper, informer.HasSynced)

4.1 informer.HasSynced

s.controller.HasSynced

4.2 sharedInformerFactory.WaitForCacheSync

总结

informer初始化

informer启动