24.3、informer机制

kubernetes使用informer机制来保证消息的实时性、可靠性、顺序性等，Kubernetes的其他组件都是通过client-go的Informer机制与Kubernetes API Server进行通信的。

其架构原理如下：

主要流程如下:

1. Reflector通过List获取集群的所有对象，并放到本地store缓存起来
2. 然后Reflector通过Watch时刻监听集群的事件
3. 如果获得集群事件就把这个事件发送到Delta FIFO
4. controller从Delta FIFO中依次pop出事件并交给informer的回调函数HandleDeltas
5. indexer将事件存储在本地的缓存中
6. distribute负责将资源对象分发到具体的处理函数
7. Rsync会把本地的缓存的资源对象同步到DeltaFIFO

比如我们要删除一个叫demo的pod，那么其主要流程如下：

1. 首先是Reflector通过Watch收到要删除名叫demo的pod事件，然后将这个事件发送给Delta FIFO
2. 从DeltaFIFO中pop这个事件并交给indexer进行本地处理，首先会存储这个事件，其次会从本地store中删除这个pod
3. 回调函数会调用distribute去处理不同的对象

从上面可以看到其主要组件有以下几个。
（1）、Reflector
Reflector负责List&Watch指定的kubernetes资源，当Watch到监控的资源发生变化时，就会触发相应的变更事件。比如Added（资源添加）事件、Updated（资源更新）事件、Deleted（资源删除）事件，并将其资源对象存放到本地缓存DeltaFIFO中。
（2）、DeltaFIFO
DeltaFIFO就是一个资源对象存储的一个队列。
（3）、indexer
Indexer是client-go用来存储资源对象并自带索引功能的本地存储，Reflector从DeltaFIFO中将消费出来的资源对象存储至Indexer。Indexer与Etcd集群中的数据完全保持一致。client-go可以很方便地从本地存储中读取相应的资源对象数据，而无须每次从远程Etcd集群中读取，以减轻Kubernetes API Server和Etcd集群的压力。
（4）、informer
Informer 是 Client-go 中的一个核心工具包。在Kubernetes源码中，如果 Kubernetes 的某个组件，需要 List/Get Kubernetes 中的 Object，在绝大多 数情况下，会直接使用Informer实例中的Lister()方法（该方法包含 了 Get 和 List 方法），而很少直接请求Kubernetes API。Informer 最基本 的功能就是List/Get Kubernetes中的 Object。

Informer

每一个Kubernetes资源上都实现了Informer机制。每一个Informer上都会实现Informer和Lister方法，例如PodInformer，代码示例如下(staging\src\k8s.io\client-go\informers\core\v1\pod.go)：

type PodInformer interface {
    Informer() cache.SharedIndexInformer
    Lister() v1.PodLister
}

定义不同资源的Informer，允许监控不同资源的资源事件。比如监控Pod的资源事件，当Pod有新增、删除、修改等事件的时候，client-go中的Watch能及时收到资源对象的变更信息。

1、Shared Informer

同一个资源的informer可能会被实例化多次，比如一个Pod可能会被deployment的Informer实例化，也可能会被replicaset实例化，如果一个informer一个Reflector，就会导致有许许多多的Reflector做相同的事情，会导致api server负载过重。所以client-go提供了一种shared informer机制，就是同一资源的Informer共享一个Reflector。

其中Informers字段的定义如下（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
}

可以看到informers字段是一个map类型，其中Key是资源类型，value是SharedIndexInformer。

而InformerFor方法添加了不同资源的infomer，如下（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
}

• 首先获取资源类型，如果存在则直接返回
• 获取rsync的时间，如果没有自定义则使用默认的时间
• 然后创建一个新的informer，更新Informers字段

最后通过Shared Informer的Start方法使f.informers中的每个informer通过goroutine持久运行。

// Start initializes all requested informers.
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
        }
    }
}

Reflector

Reflector主要通过List&Watch来获取监控集群资源，其主要有两部分：List和Watch。其中List就是列出集群的资源，基于HTTP短链接实现。Watch是监听资源事件，是基于HTTP长连接实现。

Reflector的结构体定义如下，代码路径staging\src\k8s.io\client-go\tools\cache\reflector.go：

type Reflector struct {
    name string
    expectedTypeName string
    expectedType reflect.Type
    expectedGVK *schema.GroupVersionKind
    store Store
    listerWatcher ListerWatcher
    backoffManager wait.BackoffManager
    resyncPeriod time.Duration
    ShouldResync func() bool
    clock clock.Clock
    paginatedResult bool
    lastSyncResourceVersion string
    isLastSyncResourceVersionUnavailable bool
    lastSyncResourceVersionMutex sync.RWMutex
    WatchListPageSize int64
    watchErrorHandler WatchErrorHandler
}

可以看到Reflector定义了一个listerWatcher，而ListerWatcher是一个接口，定义如下，代码路径：staging\src\k8s.io\client-go\tools\cache\listwatch.go：

// ListerWatcher is any object that knows how to perform an initial list and start a watch on a resource.
type ListerWatcher interface {
    Lister
    Watcher
}

而Lister和Watcher也是一个接口，其分别需要实现List和Watch方法，定义如下：

// Lister is any object that knows how to perform an initial list.
type Lister interface {
    List(options metav1.ListOptions) (runtime.Object, error)
}
// Watcher is any object that knows how to start a watch on a resource.
type Watcher interface {
    Watch(options metav1.ListOptions) (watch.Interface, error)
}

**
而ListWatch实现了这两个方法，定义如下：

type ListWatch struct {
    ListFunc  ListFunc
    WatchFunc WatchFunc
    // DisableChunking requests no chunking for this list watcher.
    DisableChunking bool
}
// List a set of apiserver resources
func (lw *ListWatch) List(options metav1.ListOptions) (runtime.Object, error) {
    // ListWatch is used in Reflector, which already supports pagination.
    // Don't paginate here to avoid duplication.
    return lw.ListFunc(options)
}
// Watch a set of apiserver resources
func (lw *ListWatch) Watch(options metav1.ListOptions) (watch.Interface, error) {
    return lw.WatchFunc(options)
}

而ListFunc和WatchFunc是定义的两个类型，如下：

// ListFunc knows how to list resources
type ListFunc func(options metav1.ListOptions) (runtime.Object, error)
// WatchFunc knows how to watch resources
type WatchFunc func(options metav1.ListOptions) (watch.Interface, error)

在Reflector中是通过ListAndWatch方法来实现的，代码如下（staging\src\k8s.io\client-go\tools\cache\reflector.go）：

（1）、List集群资源的代码简要如下：

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
    ....
    var resourceVersion string
    options := metav1.ListOptions{ResourceVersion: r.relistResourceVersion()}
    if err := func() error {
        initTrace := trace.New("Reflector ListAndWatch", trace.Field{"name", r.name})
        defer initTrace.LogIfLong(10 * time.Second)
        var list runtime.Object
        var paginatedResult bool
        var err error
        listCh := make(chan struct{}, 1)
        panicCh := make(chan interface{}, 1)
        go func() {
            defer func() {
                if r := recover(); r != nil {
                    panicCh <- r
                }
            }()
            pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
                return r.listerWatcher.List(opts)
            }))
            switch {
            case r.WatchListPageSize != 0:
                pager.PageSize = r.WatchListPageSize
            case r.paginatedResult:
            case options.ResourceVersion != "" && options.ResourceVersion != "0":
                pager.PageSize = 0
            }
            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)
        }()
        select {
        case <-stopCh:
            return nil
        case r := <-panicCh:
            panic(r)
        case <-listCh:
        }
        if err != nil {
            return fmt.Errorf("failed to list %v: %v", r.expectedTypeName, err)
        }
        if options.ResourceVersion == "0" && paginatedResult {
            r.paginatedResult = true
        }
        r.setIsLastSyncResourceVersionUnavailable(false) // list was successful
        initTrace.Step("Objects listed")
        listMetaInterface, err := meta.ListAccessor(list)
        if err != nil {
            return fmt.Errorf("unable to understand list result %#v: %v", list, err)
        }
        resourceVersion = listMetaInterface.GetResourceVersion()
        initTrace.Step("Resource version extracted")
        items, err := meta.ExtractList(list)
        if err != nil {
            return fmt.Errorf("unable to understand list result %#v (%v)", list, err)
        }
        initTrace.Step("Objects extracted")
        if err := r.syncWith(items, resourceVersion); err != nil {
            return fmt.Errorf("unable to sync list result: %v", err)
        }
        initTrace.Step("SyncWith done")
        r.setLastSyncResourceVersion(resourceVersion)
        initTrace.Step("Resource version updated")
        return nil
    }(); err != nil {
        return err
    }
    ......
}

（1）、首先通过r.listerWatcher.List(opts)获取所有资源
（2）、通过listMetaInterface.GetResourceVersion()获取资源版本
（3）、通过meta.ExtractList(list)将资源对象转换为列表形式，将runtime.Object对象转换成[]runtime.Object对象。
（4）、通过r.syncWith(items, resourceVersion)将资源对象列表和版本号存储到DeltaFIFO中
（5）、通过r``.``setLastSyncResourceVersion``(resourceVersion)更新最新的版本

所以List的作用就是在初始化之初获取到集群中指定的所有资源对象并将他们存储到本地。

（2）、Watch集群资源的简要代码

for {
        select {
        case <-stopCh:
            return nil
        default:
        }
        timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
        options = metav1.ListOptions{
            ResourceVersion: resourceVersion,
            TimeoutSeconds: &timeoutSeconds,
            AllowWatchBookmarks: true,
        }
        // start the clock before sending the request, since some proxies won't flush headers until after the first watch event is sent
        start := r.clock.Now()
        w, err := r.listerWatcher.Watch(options)
        if err != nil {
            if utilnet.IsConnectionRefused(err) {
                time.Sleep(time.Second)
                continue
            }
            return err
        }
        if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh); err != nil {
            if err != errorStopRequested {
                switch {
                case isExpiredError(err):
                    klog.V(4).Infof("%s: watch of %v closed with: %v", r.name, r.expectedTypeName, err)
                default:
                    klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedTypeName, err)
                }
            }
            return nil
        }
    }

（1）、通过metav1.ListOptions获取最新的资源
（2）、通过r.listerWatcher.Watch(options)来监控资源，实际调用了Pod Informer下的WatchFunc函数，它通过ClientSet客户端与Kubernetes API Server建立长连接，监控指定资源的变更事件
（3）、通过r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh)来处理资源的变更，当触发Added（资源添加）事件、Updated （资源更新）事件、Deleted（资源删除）事件时，将对应的资源对象更新到本地缓存DeltaFIFO中并更新ResourceVersion资源版本号，代码如下：

func (r *Reflector) watchHandler(start time.Time, w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) 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)
            }
            if r.expectedType != nil {
                if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a {
                    utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a))
                    continue
                }
            }
            if r.expectedGVK != nil {
                if e, a := *r.expectedGVK, event.Object.GetObjectKind().GroupVersionKind(); e != a {
                    utilruntime.HandleError(fmt.Errorf("%s: expected gvk %v, but watch event object had gvk %v", r.name, e, a))
                    continue
                }
            }
            meta, err := meta.Accessor(event.Object)
            if err != nil {
                utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
                continue
            }
            newResourceVersion := meta.GetResourceVersion()
            switch event.Type {
            case watch.Added:
                err := r.store.Add(event.Object)
                if err != nil {
                    utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", r.name, event.Object, err))
                }
            case watch.Modified:
                err := r.store.Update(event.Object)
                if err != nil {
                    utilruntime.HandleError(fmt.Errorf("%s: unable to update watch event object (%#v) to store: %v", r.name, event.Object, err))
                }
            case watch.Deleted:
                // 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 := r.store.Delete(event.Object)
                if err != nil {
                    utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", r.name, event.Object, err))
                }
            case watch.Bookmark:
                // A `Bookmark` means watch has synced here, just update the resourceVersion
            default:
                utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
            }
            *resourceVersion = newResourceVersion
            r.setLastSyncResourceVersion(newResourceVersion)
            eventCount++
        }
    watchDuration := r.clock.Since(start)
    if watchDuration < 1*time.Second && eventCount == 0 {
        return fmt.Errorf("very short watch: %s: Unexpected watch close - watch lasted less than a second and no items received", r.name)
    }
    klog.V(4).Infof("%s: Watch close - %v total %v items received", r.name, r.expectedTypeName, eventCount)
    return nil
}

DeltaFIFO

DeltaFIFO可以分开理解，FIFO是一个先进先出的队列，它拥有队列操作的基本方法，例如Add、Update、Delete、List、Pop、Close等，而Delta是一个资源对象存储，它可以保存资源对象的操作类型，例如Added（添加）操作类型、Updated（更新）操作类型、Deleted（删除）操作类型、Sync（同步）操作类型等。

DeltaFIFO的结构体定义如下（staging\src\k8s.io\client-go\tools\cache\delta_fifo.go）：

type DeltaFIFO struct {
    lock sync.RWMutex
    cond sync.Cond
    items map[string]Deltas
    queue []string
    populated bool
    initialPopulationCount int
    keyFunc KeyFunc
    knownObjects KeyListerGetter
    closed bool
    emitDeltaTypeReplaced bool
}

其中queue存储资源对象的key，items以map数据结构存储key对应的Deltas，Deltas是Delta的集合，Delta的数据结构如下：

type Delta struct {
    Type   DeltaType
    Object interface{}
}

DeltaFIFO是一个先进先出的队列，有数据的生产者和消费者，其中生成者是Reflector调用的Add方法，消费者是Controller调用的Pop方法。

1、生产者

DeltaFIFO队列中的资源对象在Added（资源添加）事件、Updated（资源更新）事件、Deleted（资源删除）事件中都调用了queueActionLocked函数，它是DeltaFIFO实现的关键，代码示例如下（staging\src\k8s.io\client-go\tools\cache\delta_fifo.go）：

func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
    id, err := f.KeyOf(obj)
    if err != nil {
        return KeyError{obj, err}
    }
    newDeltas := append(f.items[id], Delta{actionType, obj})
    newDeltas = dedupDeltas(newDeltas)
    if len(newDeltas) > 0 {
        if _, exists := f.items[id]; !exists {
            f.queue = append(f.queue, id)
        }
        f.items[id] = newDeltas
        f.cond.Broadcast()
    } else {
        delete(f.items, id)
    }
    return nil
}

（1）、首先通过f.KeyOf(obj)获取资源对象的key
（2）、然后使用append(f.items[id], Delta{actionType, obj})将Delta加入到对应的items中
（3）、使用dedupDeltas(newDeltas)对数据进行去重操作
（4）、使用f.items[id] = newDeltas更新新的Deltas，并通过cond.Broadcast通知所有消费者解除阻塞

2、消费者

Pop方法作为消费者方法使用，从DeltaFIFO的头部取出最早进入队列中的资源对象数据。Pop方法须传入process函数，用于接收并处理对象的回调方法，代码示例如下：

func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
    f.lock.Lock()
    defer f.lock.Unlock()
    for {
        for len(f.queue) == 0 {
            if f.closed {
                return nil, ErrFIFOClosed
            }
            f.cond.Wait()
        }
        id := f.queue[0]
        f.queue = f.queue[1:]
        if f.initialPopulationCount > 0 {
            f.initialPopulationCount--
        }
        item, ok := f.items[id]
        if !ok {
            continue
        }
        delete(f.items, id)
        err := process(item)
        if e, ok := err.(ErrRequeue); ok {
            f.addIfNotPresent(id, item)
            err = e.Err
        }
        return item, err
    }
}

（1）、如果队列为空，则进入阻塞状态
（2）、通过f.queue[0]取队列头部，并通过f.queue = f.queue[1:]更新队列
（3）、通过item, ok := f.items[id]判断Key是否存在，如果存在则通过process(item)进行回调处理
（4）、如果处理失败，则重新加入队列

其实queue中的数据是通过Controller的processLoop方法来实现的，如下(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))
        if err != nil {
            if err == ErrFIFOClosed {
                return
            }
            if c.config.RetryOnError {
                // This is the safe way to re-enqueue.
                c.config.Queue.AddIfNotPresent(obj)
            }
        }
    }
}

取出来后交给process的回调函数HandleDeltas，代码如下(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, Replaced, Added, Updated:
            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
                }
                isSync := false
                switch {
                case d.Type == Sync:
                    // Sync events are only propagated to listeners that requested resync
                    isSync = true
                case d.Type == Replaced:
                    if accessor, err := meta.Accessor(d.Object); err == nil {
                        if oldAccessor, err := meta.Accessor(old); err == nil {
                            // Replaced events that didn't change resourceVersion are treated as resync events
                            // and only propagated to listeners that requested resync
                            isSync = accessor.GetResourceVersion() == oldAccessor.GetResourceVersion()
                        }
                    }
                }
                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}, false)
            }
        case Deleted:
            if err := s.indexer.Delete(d.Object); err != nil {
                return err
            }
            s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
        }
    }
    return nil
}

HandleDeltas函数作为process回调函数，当资源对象的操作类型为Added、Updated、Deleted时，将该资源对象存储至Indexer（它是并发安全的存储），并通过distribute函数将资源对象分发至SharedInformer。

3、Rsync机制

Resync机制会将Indexer本地存储中的资源对象同步到DeltaFIFO中，并将这些资源对象设置为Sync的操作类型。Resync函数在Reflector中定时执行，它的执行周期由NewReflector函数传入的resyncPeriod参数设定。代码如下(staging\src\k8s.io\client-go\tools\cache\delta_fifo.go)：

func (f *DeltaFIFO) Resync() error {
    f.lock.Lock()
    defer f.lock.Unlock()
    if f.knownObjects == nil {
        return nil
    }
    keys := f.knownObjects.ListKeys()
    for _, k := range keys {
        if err := f.syncKeyLocked(k); err != nil {
            return err
        }
    }
    return nil
}
func (f *DeltaFIFO) syncKeyLocked(key string) error {
    obj, exists, err := f.knownObjects.GetByKey(key)
    if err != nil {
        klog.Errorf("Unexpected error %v during lookup of key %v, unable to queue object for sync", err, key)
        return nil
    } else if !exists {
        klog.Infof("Key %v does not exist in known objects store, unable to queue object for sync", key)
        return nil
    }
    id, err := f.KeyOf(obj)
    if err != nil {
        return KeyError{obj, err}
    }
    if len(f.items[id]) > 0 {
        return nil
    }
    if err := f.queueActionLocked(Sync, obj); err != nil {
        return fmt.Errorf("couldn't queue object: %v", err)
    }
    return nil
}

（1）、首先通过f.knownObjects.ListKeys()把本地所有的资源对象都找出来
（2）、然后调用f.syncKeyLocked(k)来判断在DeltaFIFO中是否存在这个对象，如果不存在则调用f.queueActionLocked(Sync, obj)把它加进来

Indexer

Indexer是client-go用来存储资源对象并自带索引功能的本地存储，Controller的processLoop从DeltaFIFO中将消费出来的资源对象存储至Indexer。Indexer中的数据与Etcd集群中的数据保持完全一致。client-go可以很方便地从本地存储中读取相应的资源对象数据，而无须每次都从远程Etcd集群中读取，这样可以减轻Kubernetes APIServer和Etcd集群的压力。

Indexer在threadSafeMap的基础上进行了封装，它继承了threadSafeMap的相关功能。

1、threadSafeMap

ThreadSafeMap是一个内存中的存储，其中的数据并不会写入本地磁盘中，每次的增、删、改、查操作都会加锁，以保证数据的一致性。ThreadSafeMap将资源对象数据存储于一个map数据结构中。其数据结构如下（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
}

其中items就是用来存储资源对象数据的。

每次增加、删除、修改都会调用updateIndices或者deleteFromIndices来执行操作，如下：

func (c *threadSafeMap) Add(key string, obj interface{}) {
    c.lock.Lock()
    defer c.lock.Unlock()
    oldObject := c.items[key]
    c.items[key] = obj
    c.updateIndices(oldObject, obj, key)
}
func (c *threadSafeMap) Update(key string, obj interface{}) {
    c.lock.Lock()
    defer c.lock.Unlock()
    oldObject := c.items[key]
    c.items[key] = obj
    c.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.deleteFromIndices(obj, key)
        delete(c.items, key)
    }
}

updateIndices或者deleteFromIndices的方法如下：

func (c *threadSafeMap) updateIndices(oldObj interface{}, newObj interface{}, key string) {
    // if we got an old object, we need to remove it before we add it again
    if oldObj != nil {
        c.deleteFromIndices(oldObj, key)
    }
    // name: 索引的名字    indexFunc：索引器的函数
    for name, indexFunc := range c.indexers {
        indexValues, err := indexFunc(newObj) // 获取索引的资源数据
        if err != nil {
            panic(fmt.Errorf("unable to calculate an index entry for key %q on index %q: %v", key, name, err))
        }
        index := c.indices[name]    // 通过索引的名字获取缓存的数据
        if index == nil {            // 如果数据是空的，则创建一个新的缓存map
            index = Index{}
            c.indices[name] = index
        }
        // indexValue: 具体的单个资源数据
        for _, indexValue := range indexValues {
            set := index[indexValue]    // 获取缓存数据的值
            if set == nil {                // 如果这个值不存在，创建一个新的map用于存放指
                set = sets.String{}
                index[indexValue] = set
            }
            set.Insert(key)                // 将值存进去
        }
    }
}
// deleteFromIndices removes the object from each of the managed indexes
// it is intended to be called from a function that already has a lock on the cache
func (c *threadSafeMap) deleteFromIndices(obj interface{}, key string) {
    for name, indexFunc := range c.indexers {
        indexValues, err := indexFunc(obj)
        if err != nil {
            panic(fmt.Errorf("unable to calculate an index entry for key %q on index %q: %v", key, name, err))
        }
        index := c.indices[name]
        if index == nil {
            continue
        }
        for _, indexValue := range indexValues {
            set := index[indexValue]
            if set != nil {
                set.Delete(key)
                // If we don't delete the set when zero, indices with high cardinality
                // short lived resources can cause memory to increase over time from
                // unused empty sets. See `kubernetes/kubernetes/issues/84959`.
                if len(set) == 0 {
                    delete(index, indexValue)
                }
            }
        }
    }
}

其中主要涉及Index,Indices,IndexFunc,Indexers这四个数据结构，它们的定义如下：

// Index maps the indexed value to a set of keys in the store that match on that value
type Index map[string]sets.String
// Indexers maps a name to a IndexFunc
type Indexers map[string]IndexFunc
// Indices maps a name to an Index
type Indices map[string]Index
// IndexFunc knows how to compute the set of indexed values for an object.
type IndexFunc func(obj interface{}) ([]string, error)

其中Indexers引用了IndexFunc，Indices引用了Index，它们关系如下图：

• Index存储了缓存数据，是map类型，其中sets.String是map [string] struct {}类型，以减少内存消耗
• Indices是存储缓存器，key是缓存器名字，value是缓存的数据，也就是上面的Index
• IndexFunc是索引器函数，返回的是一个资源列表
• Indexers是存储索引器，key是索引器名字，value是索引器函数，也就是上面的IndexFunc

它们的关系错综复杂，大概如下：

在缓存中获取索引的结果是通过ByIndex方法，调用方如下(staging\src\k8s.io\client-go\tools\cache\store.go)：

func (c *cache) ByIndex(indexName, indexKey string) ([]interface{}, error) {
    return c.cacheStorage.ByIndex(indexName, indexKey)
}

ByIndex的具体实现如下（staging\src\k8s.io\client-go\tools\cache\thread_safe_store.go）：

// ByIndex returns a list of the items whose indexed values in the given index include the given indexed value
func (c *threadSafeMap) ByIndex(indexName, indexedValue string) ([]interface{}, error) {
    c.lock.RLock()
    defer c.lock.RUnlock()
    indexFunc := c.indexers[indexName]
    if indexFunc == nil {
        return nil, fmt.Errorf("Index with name %s does not exist", indexName)
    }
    index := c.indices[indexName]
    set := index[indexedValue]
    list := make([]interface{}, 0, set.Len())
    for key := range set {
        list = append(list, c.items[key])
    }
    return list, nil
}

ByIndex接收两个参数：IndexName（索引器名称）和indexKey（需要检索的key）。首先从c.indexers中查找指定的索引器函数，从c.indices中查找指定的缓存器函数，然后根据需要检索的indexKey从缓存数据中查到并返回数据。

Index中的缓存数据为Set集合数据结构，Set本质与Slice相同，但Set中不存在相同元素。由于Go语言标准库没有提供Set数据结构，Go语言中的map结构类型是不能存在相同key的，所以Kubernetes将map结构类型的key作为Set数据结构，实现Set去重特性。