informer

​ informer是client-go中的核心工具包，它带有本地缓存和索引机制，并支持注册EventHandler，本地缓存被称为store，索引被称为indexer。informer中主要包含controller、reflector、deltafifo、localstore、lister和processer六个组件，负责各组件与api server的资源和事件同步。

1.简介

1.1.设计目标

• informer依赖kubernetes的List/Watch API，通过Lister()方法List/Get对象，informer不会去请求kubernetes API，而是直接查询本地缓存，减少对kubernetes API的直接调用。informer初始化时，先调用kubernetes list api获取某种resource的全部object缓存在内存中，然后会通过watch api建立长连接监听resource变化实现缓存动态维护。它的特点包括以下几部分：

  1.实时性：informer通过长连接实时获取资源变化，客户端可及时感知最新资源状态

  2.缓存机制：informer将资源对象保存在本地缓存，提高查询效率，减少API服务器压力

  3.事件驱动：informer通过事件通知机制，将资源变化通知给监听器，实现事件驱动编程

• 

1.2.组件定义

• reflector：通过List/Watch监听api server，把增量的数据推到deltaFIFO增量事件队列

  deltaFIFO：增量事件，存储delta事件触发缓存刷新

  storeIndex：index是存储索引，它可以加速缓存数据检索，基于索引拿到资源的namespace/name后可以从threadSafeMap拿到对象

  threadSafeMap：本地缓存store，storeIndex作为索引，用于存储具体资源对象

  controller：和k8s中的控制器不同，它主要负责实例化并启动reflector反射器，调用processLoop消费deltaFIFO队列

• 

1.3.代码结构

• cache
  ├── controller.go  # 包含：Config、Run、processLoop、NewInformer、NewIndexerInformer
  ├── delta_fifo.go  # 包含：NewDeltaFIFO、DeltaFIFO、AddIfNotPresent
  ├── expiration_cache.go
  ├── expiration_cache_fakes.go
  ├── fake_custom_store.go
  ├── fifo.go   # 包含：Queue、FIFO、NewFIFO
  ├── heap.go
  ├── index.go    # 包含：Indexer、MetaNamespaceIndexFunc
  ├── listers.go
  ├── listwatch.go   # 包含：ListerWatcher、ListWatch、List、Watch
  ├── mutation_cache.go
  ├── mutation_detector.go
  ├── reflector.go   # 包含：Reflector、NewReflector、Run、ListAndWatch
  ├── reflector_metrics.go
  ├── shared_informer.go  # 包含：NewSharedInformer、WaitForCacheSync、Run、HasSynced
  ├── store.go  # 包含：Store、MetaNamespaceKeyFunc、SplitMetaNamespaceKey
  ├── testing
  │   ├── fake_controller_source.go
  ├── thread_safe_store.go  # 包含：ThreadSafeStore、threadSafeMap
  ├── undelta_store.go
  

2.源码分析

2.1.Reflector

2.1.1.定义

• reflector的主要职责是从api server获取全量及监听增量事件，把获取的相关资源类型的Add/Update/Delete事件写到deltafifo队列里，它的结构体核心定义如下。

• type Reflector struct {
      
  	store Store
  
  	listerWatcher ListerWatcher
  
  	lastSyncResourceVersion string
  }
  

2.1.2.运行原理

• Run()作为reflector的运行入口，会启动ListAndWatch监听，一直循环调用直到stopCh通知退出

• func (r *Reflector) Run(stopCh <-chan struct{}) {
  	klog.V(3).Infof("Starting reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
  	wait.BackoffUntil(func() {
          // ListAndWatch监听并缓存资源
  		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.expectedTypeName, r.resyncPeriod, r.name)
  }
  

• ListAndWatch首先会通过匿名函数内的listerWatcher.list()尝试获取某资源相关条件下的所有对象，并记录当前最新的resourceVersion版本，启动一个协程去处理resync定时同步逻辑，默认不开启resync功能。然后，根据resourceVersion和timeoutSeconds参数实例化一个watcher对象，该watcher会去监听api server提供的watch接口，并把获取的事件向resultChan输出，最后通过watchHandler方法监听watcher.resultChan管道，将获取到的事件推到deltaFIFO队列。

• 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 {
  		...
  		go func() {
  			defer func() {
  				if r := recover(); r != nil {
  					panicCh <- r
  				}
  			}()
  			// 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)
  			}))
  			...
              // 通过分页器List所有资源对象
  			list, paginatedResult, err = pager.List(context.Background(), options)
  			if isExpiredError(err) || isTooLargeResourceVersionError(err) {
  				r.setIsLastSyncResourceVersionUnavailable(true)
  				list, paginatedResult, err = pager.List(context.Background(), metav1.ListOptions{ResourceVersion: r.relistResourceVersion()})
  			}
  			close(listCh)
  		}()
  		...
  		return nil
  	}(); err != nil {
  		return err
  	}
  
  	resyncerrc := make(chan error, 1)
  	cancelCh := make(chan struct{})
  	defer close(cancelCh)
      // 启动协程处理同步逻辑
  	go func() {
  		resyncCh, cleanup := r.resyncChan()
  		defer func() {
  			cleanup() // Call the last one written into cleanup
  		}()
  		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 {
  		...
          // 通过watcher对象监听资源变化
  		w, err := r.listerWatcher.Watch(options)
  		...
  		// 利用
  		if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh); err != nil {
  			if err != errorStopRequested {
  				switch {
  				case isExpiredError(err):
  					// Don't set LastSyncResourceVersionUnavailable - LIST call with ResourceVersion=RV already
  					// has a semantic that it returns data at least as fresh as provided RV.
  					// So first try to LIST with setting RV to resource version of last observed object.
  					klog.V(4).Infof("%s: watch of %v closed with: %v", r.name, r.expectedTypeName, err)
  				case apierrors.IsTooManyRequests(err):
  					klog.V(2).Infof("%s: watch of %v returned 429 - backing off", r.name, r.expectedTypeName)
  					<-r.initConnBackoffManager.Backoff().C()
  					continue
  				default:
  					klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedTypeName, err)
  				}
  			}
  			return nil
  		}
  	}
  }
  

• list-watch中的list()并不是每次都拉取全量的数据，第一次拉取时由于resourceVersion为空，所以拉取的是全量数据。当list-watch出现异常进行重试连接时，list()拉取的resourceVersion为上次最新的版本，这样list会获取比该版本更新的所有数据。此外，list-watch能力是基于底层的RESTClient实现的。

• func NewFilteredListWatchFromClient(c Getter, resource string, namespace string, optionsModifier func(options *metav1.ListOptions)) *ListWatch {
  	listFunc := func(options metav1.ListOptions) (runtime.Object, error) {
  		optionsModifier(&options)
  		return c.Get().
  			Namespace(namespace).
  			Resource(resource).
  			VersionedParams(&options, metav1.ParameterCodec).
  			Do(context.TODO()).
          	// list()的底层实现
  			Get()
  	}
  	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()的底层实现
  			Watch(context.TODO())
  	}
  	return &ListWatch{ListFunc: listFunc, WatchFunc: watchFunc}
  }
  

• c.Get()中的Get()是Getter接口定义的一个方法，用于获取restclient.Request，restclient.Request内的RESTClient实现rest.Interface定义的方法规范，rest.Interface是一个相对底层的工具，封装了kubernetes rest apis的相应动作。因此listwatch获取资源信息时，其实走的就是rest.Interface定义的rest接口。

• --- Getter接口
      type Getter interface {
          Get() *restclient.Request
      }
  
  --- Interface接口
      type Interface interface {
          GetRateLimiter() flowcontrol.RateLimiter
          Verb(verb string) *Request
          Post() *Request
          Put() *Request
          Patch(pt types.PatchType) *Request
          Get() *Request
          Delete() *Request
          APIVersion() schema.GroupVersion
      }
  
  --- Getter接口和Interface接口的实现
      type RESTClient struct {
          base *url.URL
  
          versionedAPIPath string
  
          content ClientContentConfig
  
          createBackoffMgr func() BackoffManager
  
          rateLimiter flowcontrol.RateLimiter
  
          warningHandler WarningHandler
  
          Client *http.Client
      }
  

• watcherHandler从watcher.watch结果的resultChan管道获取变更事件，然后添加到deltaFIFO队列中，store的Add/Update/Delete操作其实在deltaFIFO里都是插入逻辑，只是插入的事件类型为Add/Update/Delete。

• func (r *Reflector) watchHandler(...) 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)
  			}
  			...
  			meta, err := meta.Accessor(event.Object)
  			if err != nil {
  				utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
  				continue
  			}
              // 获取当前对象的resourceVersion
  			newResourceVersion := meta.GetResourceVersion()
  			switch event.Type {
  			case watch.Added:
                  // 新增事件
  				err := r.store.Add(event.Object)
  				...
  			case watch.Modified:
                  // 更新事件
  				err := r.store.Update(event.Object)
  				...
  			case watch.Deleted:
  				// 删除事件
  				err := r.store.Delete(event.Object)
  				...
  			default:
  				...
  			}
              // 设置最新同步的resourceVersion
  			*resourceVersion = newResourceVersion
  			r.setLastSyncResourceVersion(newResourceVersion)
  			if rvu, ok := r.store.(ResourceVersionUpdater); ok {
  				rvu.UpdateResourceVersion(newResourceVersion)
  			}
  			eventCount++
  		}
  	}
  	...
  	return nil
  }
  

• 针对informer中的watch机制，当客户端通过watch api监听api server后，api server通过在返回http header中配置Transfer-Encoding: chunked实现分块编码传输，把需要的数据按照chunked块的方式流式写入。客户端收到api server经过chunked编码的数据量后，同样按照http chunked的方式进行解码，即watch的整个过程，除了基于reflector获取数据响应的decoder外，需要api server和rest client双方基于chunked加解码。

• --- api server加码
  	e := streaming.NewEncoder(framer, s.Encoder)
  
      // ensure the connection times out
      timeoutCh, cleanup := s.TimeoutFactory.TimeoutCh()
      defer cleanup()
  
      // begin the stream
      w.Header().Set("Content-Type", s.MediaType)
      w.Header().Set("Transfer-Encoding", "chunked")
      w.WriteHeader(http.StatusOK)
      flusher.Flush()
  
  --- 客户端解码
  	$ curl -i http://{kube-api-server-ip}:8080/api/v1/watch/pods?watch=yes
      HTTP/1.1 200 OK
      Content-Type: application/json
      Transfer-Encoding: chunked
      Date: Thu, 02 Jan 2020 20:22:59 GMT
      Transfer-Encoding: chunked
  
      {"type":"ADDED", "object":{"kind":"Pod","apiVersion":"v1",...}}
      {"type":"ADDED", "object":{"kind":"Pod","apiVersion":"v1",...}}
      {"type":"MODIFIED", "object":{"kind":"Pod","apiVersion":"v1",...}}
  

• 

2.1.3.工作总结

• reflector启动运行之后，会先执行一次资源对象全量拉取，通过单独的协程将数据更新到deltaFIFO队列中，然后持续监听资源对象的增量事件(创建、更新、删除等)，并将事件去重之后更新到deltaFIFO队列，最后deltaFIFO队列的数据会被消费者取出，执行具体的业务逻辑。
• 

2.2.DeltaFIFO

2.2.1.定义

• deltaFIFO主要作用是确保reflector和indexer之间的对象同步，具体工作时，reflector通过watcher.watch感知对象变化，并将新的状态存入deltaFIFO，controller会周期性读取deltaFIFO的数据，实时地了解到资源对象地变化，从而更新缓存中对象状态。

• type DeltaType string
  
  const (
      Added   DeltaType = "Added"
      Updated DeltaType = "Updated"
      Deleted DeltaType = "Deleted"
      Replaced DeltaType = "Replaced"
      Sync DeltaType = "Sync"
  )
  

• deltaFIFO结构体中主要包括以下核心内容，queue用于存对象格式化后的key，按照FIFO的定义先进先出；items作为字典存储了deltas数据，key为queue的元素，val为delta列表。

• type DeltaFIFO struct {
      lock sync.RWMutex
  
      cond sync.Cond
  
      items map[string]Deltas
  
      queue []string
  	...
  }
  
  type Deltas []Delta、
  type Delta struct {
      // 事件类型
      Type   DeltaType
      // K8S资源对象
      Object interface{}
  }
  

2.2.2.运行原理

• 默认情况下，deltaFIFO的items的key和queue的元素都是通过MetaNamespaceKeyFunc计算，该函数可以从k8s任意资源对象提取name和namespace，然后格式为key string，当有namespace不为空时，则使用namespace/name，否则使用name作为key。

• func MetaNamespaceKeyFunc(obj interface{}) (string, error) {
  	if key, ok := obj.(ExplicitKey); ok {
  		return string(key), nil
  	}
  	meta, err := meta.Accessor(obj)
  	if err != nil {
  		return "", fmt.Errorf("object has no meta: %v", err)
  	}
  	if len(meta.GetNamespace()) > 0 {
  		return meta.GetNamespace() + "/" + meta.GetName(), nil
  	}
  	return meta.GetName(), nil
  }
  

• watch机制监听到事件后，会把事件入队操作

• func (f *DeltaFIFO) Add(obj interface{}) error {
  	f.lock.Lock()
  	defer f.lock.Unlock()
  	f.populated = true
  	return f.queueActionLocked(Added, obj)
  }
  
  func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
      // 通过obj拼key--namespace/name
  	id, err := f.KeyOf(obj)
  	if err != nil {
  		return KeyError{obj, err}
  	}
      ...
  
      // 从items中获取已经存在的deltas列表
  	oldDeltas := f.items[id]
      
      // 把新增的事件加入已经存在的deltas
  	newDeltas := append(oldDeltas, Delta{actionType, obj})
  	newDeltas = dedupDeltas(newDeltas)
  
  	if len(newDeltas) > 0 {
          // 判断key对应的deltas是否存在，如果之前不存在，需要把key加入queue队列
  		if _, exists := f.items[id]; !exists {
  			f.queue = append(f.queue, id)
  		}
          // 刷新key对应的deltas
  		f.items[id] = newDeltas
  		f.cond.Broadcast()
  	} else {
  		// newDeltas为空，返回错误
  		f.items[id] = newDeltas
  		return fmt.Errorf("Impossible dedupDeltas for id=%q: oldDeltas=%#+v, obj=%#+v; broke DeltaFIFO invariant by storing empty Deltas", id, oldDeltas, obj)
  	}
  	return nil
  }
  

• Add新事件时，会进行去重操作，用倒数第一个delta和倒数第二个delta进行对比，如果两个都是delted类型，会把两个删除类型的delta去重合并，最终保留一个deleted类型的delta对象。

• func dedupDeltas(deltas Deltas) Deltas {
  	n := len(deltas)
  	if n < 2 {
  		return deltas
  	}
  	a := &deltas[n-1]
  	b := &deltas[n-2]
  	if out := isDup(a, b); out != nil {
  		deltas[n-2] = *out
  		return deltas[:n-1]
  	}
  	return deltas
  }
  
  func isDup(a, b *Delta) *Delta {
  	if out := isDeletionDup(a, b); out != nil {
  		return out
  	}
  	// TODO: Detect other duplicate situations? Are there any?
  	return nil
  }
  
  func isDeletionDup(a, b *Delta) *Delta {
  	if b.Type != Deleted || a.Type != Deleted {
  		return nil
  	}
  	// Do more sophisticated checks, or is this sufficient?
  	if _, ok := b.Object.(DeletedFinalStateUnknown); ok {
  		return a
  	}
  	return b
  }
  

• 

• 消费delta事件时，调用初始化deltaFIFO传入的process方法，该方法其实就是informer的HandleDeltas方法，它内部调用实例化informer时注册的eventHanlder方法，按照deltaType类型选择调用OnAdd/OnUpdate/OnDelete。

• func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
  	f.lock.Lock()
  	defer f.lock.Unlock()
  	for {
          // queue没有保存任何key时阻塞
  		for len(f.queue) == 0 {
  			if f.closed {
  				return nil, ErrFIFOClosed
  			}
  
  			f.cond.Wait()
  		}
          // 弹出队头元素
  		id := f.queue[0]
  		f.queue = f.queue[1:]
  		depth := len(f.queue)
  		if f.initialPopulationCount > 0 {
  			f.initialPopulationCount--
  		}
          // 获取deltas对象
  		item, ok := f.items[id]
  		if !ok {
  			// 缓存中不存在该deltas不处理
  			continue
  		}
          // 缓存中弹出该deltas元素
  		delete(f.items, id)
  		if depth > 10 {
  			trace := utiltrace.New("DeltaFIFO Pop Process",
  				utiltrace.Field{Key: "ID", Value: id},
  				utiltrace.Field{Key: "Depth", Value: depth},
  				utiltrace.Field{Key: "Reason", Value: "slow event handlers blocking the queue"})
  			defer trace.LogIfLong(100 * time.Millisecond)
  		}
          // deltas中的事件信息交给process处理
  		err := process(item)
  		if e, ok := err.(ErrRequeue); ok {
  			f.addIfNotPresent(id, item)
  			err = e.Err
  		}
  		return item, err
  	}
  }
  

• HandleDeltas内部会调用processDeltas按照类型处理处理事件，主要就是调用OnAdd/OnUpdate/OnDelete调整缓存中的对象

• func processDeltas(
  	// Object which receives event notifications from the given deltas
  	handler ResourceEventHandler,
  	clientState Store,
  	deltas Deltas,
  ) error {
  	// from oldest to newest
  	for _, d := range deltas {
  		obj := d.Object
  
  		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
  }
  

• 从deltaFIFO中消费对象事件时，会调用addIfNotPresent处理deltaFIFO的对象事件requeue操作，即如果该对象同步到缓存失败，则会再次放入items和queue，并通过condition通知唤醒其他协程处理。

• func (f *DeltaFIFO) addIfNotPresent(id string, deltas Deltas) {
  	f.populated = true
  	if _, exists := f.items[id]; exists {
  		return
  	}
  
  	f.queue = append(f.queue, id)
  	f.items[id] = deltas
  	f.cond.Broadcast()
  }
  

• 

2.2.3.工作总结

• deltaFIFO队列添加资源对象的delta事件时，需要判断队列是否存储过该事件，有则在关联的deltas列表后面追加新的delta，否则在queue和items中添加一遍。其中，添加已有的key时，不会调整该key在queue的顺序，deltaFIFO增量队列的FIFO先进先出只是对同一个key的deltas列表。deltaFIFO队列简单来说，就是一个生产者->消费者队列，生产者是reflector，消费者是process处理逻辑方法，用于将资源对象的增量事件信息delta存储到items，用来显示资源对象的具体操作类型，例如Add创建、Update更新和Delete删除等。

• 

2.3.Indexer

2.3.1.定义

• storeIndex实现了indexer对象索引存储功能，实例化indexer索引对象时，注册计算索引的方法，后面每次新增对象时会使用indexFunc计算出需要索引的值列表，通过倒排的方式来组织写入，读取的时候需要从指定indexFunc名字的index里读取。

• // 数据类型中的key表示通过索引函数计算出来的值，value表示对应的结果字符串集合
  type Index map[string]sets.String
  
  // 数据类型中的key表示索引函数名称，value表示索引函数
  type Indexers map[string]IndexFunc
  
  // 数据类型中的key表示索引函数名称，value表示index数据类型
  type Indices map[string]Index
  
  type storeIndex struct {
  	// indexers maps a name to an IndexFunc
  	indexers Indexers
  	// indices maps a name to an Index
  	indices Indices
  }
  

2.3.2.运行原理

• updateIndices方法用来更新索引，其内部会遍历所有注册的indexer索引方法集合，然后使用indexFunc计算出oldObj和newObj的索引值，后面删除旧的obj索引值，然后添加新的obj索引值。这里的key一般是资源对象的namespace/name值，indexer使用这个key描述具体的资源对象，后面可以使用这个key在threadSafeMap里获取真正的obj。

• func (i *storeIndex) updateIndices(oldObj interface{}, newObj interface{}, key string) {
  	var oldIndexValues, indexValues []string
  	var err error
  	for name, indexFunc := range i.indexers {
          // 通过indexFunc计算oldObj的索引值
  		if oldObj != nil {
  			oldIndexValues, err = indexFunc(oldObj)
  		} else {
  			oldIndexValues = oldIndexValues[:0]
  		}
  		if err != nil {
  			panic(fmt.Errorf("unable to calculate an index entry for key %q on index %q: %v", key, name, err))
  		}
  
          // 通过indexFunc计算newObj的索引值
  		if newObj != nil {
  			indexValues, err = indexFunc(newObj)
  		} else {
  			indexValues = indexValues[:0]
  		}
  		if err != nil {
  			panic(fmt.Errorf("unable to calculate an index entry for key %q on index %q: %v", key, name, err))
  		}
  
  		index := i.indices[name]
  		if index == nil {
  			index = Index{}
  			i.indices[name] = index
  		}
  
          // 如果新旧对象的索引值一致
  		if len(indexValues) == 1 && len(oldIndexValues) == 1 && indexValues[0] == oldIndexValues[0] {
  			// We optimize for the most common case where indexFunc returns a single value which has not been changed
  			continue
  		}
  
          // 否则删除oldObj的索引值，添加newObj的索引值
  		for _, value := range oldIndexValues {
  			i.deleteKeyFromIndex(key, value, index)
  		}
  		for _, value := range indexValues {
  			i.addKeyToIndex(key, value, index)
  		}
  	}
  }
  
  // 添加索引，直接在index关联的set集合里添加key
  func (i *storeIndex) addKeyToIndex(key, indexValue string, index Index) {
  	set := index[indexValue]
  	if set == nil {
  		set = sets.String{}
  		index[indexValue] = set
  	}
  	set.Insert(key)
  }
  
  // 删除索引，直接在index关联的set集合里删除key
  func (i *storeIndex) deleteKeyFromIndex(key, indexValue string, index Index) {
  	set := index[indexValue]
  	if set == nil {
  		return
  	}
  	set.Delete(key)
  	if len(set) == 0 {
  		delete(index, indexValue)
  	}
  }
  

• getKeysByIndex方法用于读取索引，调用时传入两个参数，indexName为索引函数，indexValue为索引值。读取索引时，判断indexName是否注册过，接着从indeces中获取indexName对应的index结构，再返回indexValue对应的names集合

• func (i *storeIndex) getKeysByIndex(indexName, indexedValue string) (sets.String, error) {
  	indexFunc := i.indexers[indexName]
  	if indexFunc == nil {
  		return nil, fmt.Errorf("Index with name %s does not exist", indexName)
  	}
  
  	index := i.indices[indexName]
  	return index[indexedValue], nil
  }
  

• 

2.3.3.工作总结

• storeIndex对象实现了资源对象存储索引功能，每次存储索引对象时，通过内部的索引函数计算出索引对象对应的索引值，然后写入内部的Indices对象。简单来说，storeIndex对象就是用于存储资源对象并自动完成索引功能的本地存储组件，完成存储后，对于接下来的查询操作，可以通过索引机制高效获取到对应资源对象。

  reflector通过监听机制将资源对象变更事件传入deltaFIFO队列后，经过process消费者函数处理后将资源对象索引存入到storeIndex对象，完成资源对象的索引功能，资源对象的具体存储由threadSafeMap对象实现。

  storeIndex和threadSafeMap对象的配合，可以保证本地拉取到资源对象和kubernetes etcd集群中的资源对象数据一致，这样就可以在获取指定资源对象时，避免每次都从api server中实时获取，降低控制平面节点的负载压力。

• 

2.4.ThreadSafeStore

2.4.1.定义

• threadSafeStore接口表示索引和资源对象缓存的相关操作，具体的实现由threadSafeMap对象来完成。它用于维护索引和缓存资源对象，索引使用storeIndex实现，资源对象的缓存则使用map[string]interface{}实现。threadSafeMap封装了storeIndex，内部实现了Add/Update/Delete方法，修改时不仅从缓存中操作对象，还会调整索引。另外，ByIndex实现了indexer中获取匹配索引的names,然后从缓存中获取匹配names的资源对象。

• // threadSafeMap implements ThreadSafeStore
  type threadSafeMap struct {
  	lock  sync.RWMutex
  	items map[string]interface{}
  
  	// index implements the indexing functionality
  	index *storeIndex
  }
  

2.4.2.运行原理

• threadSafeMap的对内部的map进一步封装，加了读写锁保证并发安全操作

• // NewThreadSafeStore creates a new instance of ThreadSafeStore.
  func NewThreadSafeStore(indexers Indexers, indices Indices) ThreadSafeStore {
  	return &threadSafeMap{
  		items: map[string]interface{}{},
  		index: &storeIndex{
  			indexers: indexers,
  			indices:  indices,
  		},
  	}
  }
  

• threadSafeMap的相关操作其实就是内部map的CRUD操作

• func (c *threadSafeMap) Add(key string, obj interface{}) {
  	c.Update(key, obj)
  }
  
  func (c *threadSafeMap) Update(key string, obj interface{}) {
  	c.lock.Lock()
  	defer c.lock.Unlock()
  	oldObject := c.items[key]
  	c.items[key] = obj
  	c.index.updateIndices(oldObject, obj, key)
  }
  
  func (c *threadSafeMap) Delete(key string) {
  	c.lock.Lock()
  	defer c.lock.Unlock()
  	if obj, exists := c.items[key]; exists {
  		c.index.updateIndices(obj, nil, key)
  		delete(c.items, key)
  	}
  }
  
  func (c *threadSafeMap) Get(key string) (item interface{}, exists bool) {
  	c.lock.RLock()
  	defer c.lock.RUnlock()
  	item, exists = c.items[key]
  	return item, exists
  }
  

• 

2.4.3.工作总结

• threadSafeMap是用于实现线程安全字典的一种工具，主要是为了在并发环境下提供安全的资源对象存储的功能。它依赖storeIndex的索引维护功能，通过storeIndex获取对象索引并进一步从items获取存储的资源对象。对象需要更新时，它会判断对象是否存在，不存在就是新增缓存对象，否则会清理旧缓存、重建新缓存并覆盖。
• 

2.5.controller

2.5.1.定义

• 作为控制中心，控制器集成了上文中提到的reflector、deltaFIFO、indexer、store组件，使各组件可以协调工作。Controller规定了控制器接口，controller则具体实现了这些接口，保证了控制器的协调功能。

• type Controller interface {
      
      // 1.初始化并启动reflector，通过listerwatcher监听变化资源事件并放入队列
      // 2.不断从队列中获取变化资源事件，执行对应操作
  	Run(stopCh <-chan struct{})
  
      // HasSynced的实现委托给了具体的队列，用于检测同步是否完成
  	HasSynced() bool
  
  	LastSyncResourceVersion() string
  }
  
  type Config struct {
  
      // deltaFIFO作为具体实现
  	Queue
  
      // 资源监听
  	ListerWatcher
  
      // 队列中资源事件处理方法
  	Process ProcessFunc
  
  	ObjectType runtime.Object
  
      // 全量同步周期
  	FullResyncPeriod time.Duration
  
      // 是否需要重新同步的检测方法
  	ShouldResync ShouldResyncFunc
  
  	RetryOnError bool
  
      // 监听错误处理回调方法
  	WatchErrorHandler WatchErrorHandler
  
      // 单词请求数据块大小
  	WatchListPageSize int64
  }
  
  // 实现Controller接口的对象
  type controller struct {
  	config         Config
  	reflector      *Reflector
  	reflectorMutex sync.RWMutex
  	clock          clock.Clock
  }
  

2.5.2.运行原理

• newInformer方法用于初始化一个控制器实例，组织各个组件配置及信息。

• func newInformer(
  	lw ListerWatcher,
  	objType runtime.Object,
  	resyncPeriod time.Duration,
  	h ResourceEventHandler,
  	clientState Store,
  	transformer TransformFunc,
  ) Controller {
  	fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
  		KnownObjects:          clientState,
  		EmitDeltaTypeReplaced: true,
  		Transformer:           transformer,
  	})
  
  	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, deltas)
  			}
  			return errors.New("object given as Process argument is not Deltas")
  		},
  	}
  	return New(cfg)
  }
  

• 控制器初始化完成后，通过controller.Run方法启动处理上下游的业务流程操作，即启动reflector调用ListAndWatch获取全量资源并监听资源变化事件，然后存储到deltaFIFO队列，另一方面，通过controller.processLoop方法不断从deltaFIFO队列中获取元素并执行对应的操作。processLoop方法其实会调用deltaFIFO.Pop进行消费，deltaFIFO.Pop又会调用初始化informer注册的process，它会作为eventHandler注册到deltaFIFO.Pop被调用进行事件处理。其实这形成的一个闭环，即controller.processLoop消费队列数据其实最终掉的就是自己初始化informer时注册的processDeltas。

• func (c *controller) Run(stopCh <-chan struct{}) {
  	// 启动单独的 goroutine 监听关闭 channel 信号
  	go func() {
  		<-stopCh
  		c.config.Queue.Close()
  	}()
  	
  	// 初始化一个 Reflector 
  	r := NewReflectorWithOptions(
          ...
  	)
  	
  	var wg wait.Group
  
  	// 调用 Reflector.Run 方法
  	// 具体细节请参考前文中的 [启动 Reflector] 小节
  	wg.StartWithChannel(stopCh, r.Run)
  
  	// 调用 controller.processLoop 消费队列元素
  	wait.Until(c.processLoop, time.Second, stopCh)
  	wg.Wait()
  }
  

• controller.processLoop其实是一个无限循环，不断从队列中取出元素，并执行对应操作。

• func (c *controller) processLoop() {
  	for {
  		// 这里配置对象中的回调方法也就是 processDeltas
  		obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
  		
  		...
  	}
  }
  

• 最后可以看一下队列后元素出队后的回调方法processDeltas，它会作为eventHandler被透传到deltaFIFO回调使用，原始的controller以及其另一实现shared_informer都会注册processDeltas作为队列元素处理模块。该方法内部会遍历资源的事件列表，然后根据不同的事件类型执行不同的操作。

• func processDeltas(
  	// Object which receives event notifications from the given deltas
  	handler ResourceEventHandler,
  	clientState Store,
  	deltas Deltas,
  ) error {
  	// from oldest to newest
  	for _, d := range deltas {
  		obj := d.Object
  
  		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
  }
  

2.5.3.工作总结

• kubernetes中的informer是client-go客户端库中实现的一种机制，它可以监听kubernetes集群中各类资源对象的变更，并将这些变更事件通知发送到控制器，同时进行资源对象信息的数据索引和存储。通俗来说，informer的关键作用就是充当kubernetes api server和资源控制器之间的中间层，类似的工作机制有消息队列、设计模式中的观察者模式等。
• 

2.6.informer

2.6.1.原理分析

• informer的实现其实依赖于controller，它内部会初始化indexer、deltaFIFO以及controller对象，另外会在config里准备好reflector启动所需的listWatcher、fifo等对象。

• func NewIndexerInformer(
      lw ListerWatcher,
      objType runtime.Object,
      resyncPeriod time.Duration,
      h ResourceEventHandler,
      indexers Indexers,
  ) (Indexer, Controller) {
  
      // 实例化 indexer 存储对象
      clientState := NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers)
  
      return clientState, newInformer(lw, objType, resyncPeriod, h, clientState, nil)
  }
  
  func newInformer(
      lw ListerWatcher,
      h ResourceEventHandler,
      clientState Store,
      ...
  ) Controller {
  
      // 实例化 DeltaFIFO 增量队列
      fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
          KnownObjects:          clientState,
          EmitDeltaTypeReplaced: true,
      })
  
      // 创建 config 集合对象, 内置 fifo, listerwatcher, process 对象.
      cfg := &Config{
          Queue:            fifo,
          ListerWatcher:    lw,
          ObjectType:       objType,
          ...
          Process: func(obj interface{}, isInInitialList bool) error {
              if deltas, ok := obj.(Deltas); ok {
                  // 消费队列数据的eventHandler
                  return processDeltas(h, clientState, transformer, deltas, isInInitialList)
              }
              return errors.New("object given as Process argument is not Deltas")
          },
      }
  
      // 实例化 controller 对象, 注意 newinformer 返回的是 controller 方法
      return New(cfg)
  }
  
  // 不再复述
  func New(c *Config) Controller {
      ctlr := &controller{
          config: *c,
      }
      return ctlr
  }
  

• informer启动其实就是controller.Run()的调用，Run()方法内会初始化reflector反射器对象，启动reflector的List/Watch监听api server事件，并将监听到的事件推入deltaFIFO队列，此外还会基于新的协程不停执行processLoop调用deltaFIFO的Pop方法消费deltaFIFO队列数据，更新store的资源对象信息和触发用户注册的resourceEventHandler事件回调方法。

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

• WaitForCacheSync用于等待informer同步完成，其内部逻辑是每隔100ms调用传入的HasSynced()方法，直到同步完成才会退出。

• 

2.6.2.工作总结

• NewIndexerInformer创建的informer其实是比较基础的实现，它的内部依赖controller实现informer的功能，controller又会关联reflector、deltaFIFO、store(包含indexer、threadSafeMap)组件之间的协调联动。client-go的informer过程实现还是颇为复杂，主要是其内部由多个组件构成，组件之间需要控制器来协调，所以client-go还提供的封装更完整的sharedInformer实现，提供了更丰富的功能。
•