init

Author: Zhang Jun

client-go/1.store-indexer-cache.md

@@ -0,0 +1,384 @@
+Store 是 KV 类型的缓存。
+Indexer 在 Store 接口的基础上，添加了对象索引功能。索引是索引值和对象 Key 集合的映射表，不同 IndexFunc 为对象产生不同的索引值列表。通过索引值可以快速获取对象的 Key。
+cache 是实现了 Store 和 Indexer 接口的内置对象，它内部使用了实现 ThreadSafeStore 接口的对象和 KeyFunc 类型的函数。
+后续创建的 Informer 使用 DeletionHandlingMetaNamespaceKeyFunc 作为 KeyFunc 的 cache 实现对象缓存。
+
+## Store
+
+Store 是 KV 类型的缓存，用于保存和查询对象。但它没有定义生成对象标识 key 的方式，由具体的实现来定的（见后文分析的 `cache` 对象）。
+
+// 来源于 k8s.io/client-go/tools/cache/store.go
+type Store interface {
+	Add(obj interface{}) error
+	Update(obj interface{}) error
+	Delete(obj interface{}) error
+	List() []interface{}
+	ListKeys() []string
+	Get(obj interface{}) (item interface{}, exists bool, err error)
+	GetByKey(key string) (item interface{}, exists bool, err error)
+
+    // 使用传入的对象列表替换 Store 中的对象
+	Replace([]interface{}, string) error
+	Resync() error
+}
+
+`NewStore` 函数返回一个实现该接口的 `cache` 类型对象（见后文分析）。
+
+由于 `Queue` 接口是 `Store` 的超级，所以实现 `Queue` 接口的 `FIFO` 类型(fifo.go)、`DeltaFIFO` 类型(delta_file.go) 也实现了 `Store`接口。（详见: [2.queue-fifo-delta_fifo.md](2.queue-fifo-delta_fifo.md）
+
+## 索引接口 Indexer
+Index 是在 Store 的基础上，添加了索引功能，方便后续按照索引来快速获取（一批）对象。
+
+// 来源于 k8s.io/client-go/tools/cache/index.go
+type Indexer interface {
+	// Index 实现了 Store 的接口
+	Store
+	// 返回注册的、名为 indexName 的索引函数
+	// Retrieve list of objects that match on the named indexing function
+	Index(indexName string, obj interface{}) ([]interface{}, error)
+	// IndexKeys returns the set of keys that match on the named indexing function.
+	IndexKeys(indexName, indexKey string) ([]string, error)
+	// ListIndexFuncValues returns the list of generated values of an Index func
+	ListIndexFuncValues(indexName string) []string
+	// ByIndex lists object that match on the named indexing function with the exact key
+	ByIndex(indexName, indexKey string) ([]interface{}, error)
+	// GetIndexer return the indexers
+	GetIndexers() Indexers
+
+	// AddIndexers adds more indexers to this store.  If you call this after you already have data
+	// in the store, the results are undefined.
+	AddIndexers(newIndexers Indexers) error
+}
+
+`NewIndexer` 函数返回一个实现该接口的 `cache` 类型对象（见后文分析）。
+
+## 为对象生成索引值列表的 IndexFunc 和包含命名 IndexFunc 集合的 Indexers
+对象的索引是一个字符串列表，可以使用 `IndexFunc` 类型的函数来获取。
+
+client-go 中提供了一个该类型的函数 `MetaNamespaceIndexFunc`，它使用对象所在的 `Namespace` 作为索引：
+
+	// 来源于 k8s.io/client-go/tools/cache/index.go
+	func MetaNamespaceIndexFunc(obj interface{}) ([]string, error) {
+		meta, err := meta.Accessor(obj)
+		if err != nil {
+			return []string{""}, fmt.Errorf("object has no meta: %v", err)
+		}
+		return []string{meta.GetNamespace()}, nil
+	}
+
+同时为该 IndexFunc 定义了一个名称 `NamespaceIndex string = "namespace"`，该名称后续可以作为 `Indexes` 的 map key。
+
+除了 client-go 预定义的 IndexFunc，开发者也可以定义其它 IndexFunc，并给它们分别命名后保存到 Indexers 中：
+
+	// 来源于 k8s.io/client-go/tools/cache/index.go
+	// map[索引函数名称]索引函数
+	type Indexers map[string]IndexFunc
+
+对象通过 IndexFunc 可以产生多个索引，而通过 KeyFunc 只能生产一个唯一表示该对象的 key。
+
+
+## 索引数据结构 Index 和 Indices
+
+前面说过，一个对象通过 `IndexFunc` 可以产生多个索引，所以一个索引匹配一组对象。
+
+`Index` 是用于保存某个索引匹配的对象 Key 集合。
+
+`Indices` 是用命名的 IndexFunc 作为 map key 的 Index 集合，它保存各命名函数产生的所有索引，即各索引匹配的一组对象的集合。
+
+	// 来源于 k8s.io/client-go/tools/cache/index.go
+	// map[索引字符串]set{对象 Key 集合}
+	type Index map[string]sets.String
+
+	// 对象的索引。map[索引函数名称]Index
+	type Indices map[string]Index
+
+
+调用 AddIndexers() 方法向 Index 添加各种索引函数，进而为 Store 中的对象创建多种索引。
+
+## 多线程安全的、带有索引的缓存 ThreadSafeStore
+
+`ThreadSafeStore` 通过锁机制，实现多 goroutine 可以并发访问的带有索引(Indexer 接口方法)功能的 K-V 数据库（Store 接口方法）。
+
+`ThreadSafeStore` 后续用于实现 Indexer 和 Store 接口，所以它的方法集中包含 Store/Indexer 接口中定义的同名方法，但和 Store 接口中的同名方法相比，多了一个 obj key 参数，少了一个 GetByKey 方法。
+
+	// 来源于 k8s.io/client-go/tools/cache/thread_safe_store.go
+	type ThreadSafeStore interface {
+		// 下面这些方法和 Store 接口方法同名，差别在于多了唯一标识 obj 的 key 参数
+		Add(key string, obj interface{})
+		Update(key string, obj interface{})
+		Delete(key string)
+		List() []interface{}
+		ListKeys() []string
+		Get(key string) (item interface{}, exists bool) // 相比 Store 接口，缺少了 GetByKey(key string) 方法
+
+		Replace(map[string]interface{}, string)
+		Resync() error
+
+		// 下面这些是 Indexer 的接口方法
+		Index(indexName string, obj interface{}) ([]interface{}, error) 
+		IndexKeys(indexName, indexKey string) ([]string, error)
+		ListIndexFuncValues(name string) []string
+		ByIndex(indexName, indexKey string) ([]interface{}, error)
+		GetIndexers() Indexers
+		AddIndexers(newIndexers Indexers) error
+	}
+
+后文会具体分析这些方法的功能。
+
+`NewThreadSafeStore` 返回一个实现该接口的对象，该对象的类型是 `threadSafeMap`：
+
+	// 来源于 k8s.io/client-go/tools/cache/thread_safe_store.go
+	func NewThreadSafeStore(indexers Indexers, indices Indices) ThreadSafeStore {
+		return &threadSafeMap{
+			items:    map[string]interface{}{},
+			indexers: indexers,
+			indices:  indices,
+		}
+	}
+
+`threadSafeMap` 使用内置的 items 保存所有对象。
+
+	// 来源于 k8s.io/client-go/tools/cache/thread_safe_store.go
+	// threadSafeMap implements ThreadSafeStore
+	type threadSafeMap struct {
+		lock  sync.RWMutex
+		// 对象集合。使用对象的 Key 作为 map key；
+		items map[string]interface{}
+		// 命名的索引函数集合。map[索引函数名称]索引函数
+		indexers Indexers
+		// 对象的索引。map[索引函数名称][索引字符串]set{对象 Key 集合}
+		indices Indices
+	}
+
+我们看看 `threadSafeMap`  的方法实现：
+
+	// 来源于 k8s.io/client-go/tools/cache/thread_safe_store.go
+	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)
+	}
+
+当 Add/Update 一个 obj 时，先使用传入的 obj 更新缓存，然后调用 `updateIndices` 方法更新索引。
+`updateIndices` 方法分别使用 indexers 中的索引函数，为对象创建索引值(多个)，然后将这些索引以及该对象的 Key 更新到索引中（c.indices)。
+
+	// 来源于 k8s.io/client-go/tools/cache/thread_safe_store.go
+	func (c *threadSafeMap) updateIndices(oldObj interface{}, newObj interface{}, key string) {
+		// 从索引中移除老的 obj
+		if oldObj != nil {
+			c.deleteFromIndices(oldObj, key)
+		}
+		// 遍历 c.indexers 中的索引函数，调用它们为对象产生索引列表
+		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 {
+				index = Index{}
+				c.indices[name] = index
+			}
+			// 将所有索引值和对象 Key 更新到缓存 c.indices 中
+			for _, indexValue := range indexValues {
+				set := index[indexValue]
+				if set == nil {
+					set = sets.String{}
+					index[indexValue] = set
+				}
+				set.Insert(key)
+			}
+		}
+	}
+
+`Delete` 方法的实现：
+
+	// 来源于 k8s.io/client-go/tools/cache/thread_safe_store.go
+	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)
+		}
+	}
+
+先后从索引和内部缓存中删除元素。 `deleteFromIndices` 方法遍历 c.indexers 中的索引函数，为 ojb 计算索引值列表，然后再从缓存 c.indices 的索引值的对象 Key 集合中删除该对象 Key。
+注意：因为一个索引值可能匹配多个对象，所以不能直接删除 c.indics 的索引值。
+
+`Replace` 方法使用传入的 items 更新内部缓存，然后重建索引：
+
+	func (c *threadSafeMap) Replace(items map[string]interface{}, resourceVersion string) {
+		c.lock.Lock()
+		defer c.lock.Unlock()
+		c.items = items
+
+		// rebuild any index
+		c.indices = Indices{}
+		for key, item := range c.items {
+			c.updateIndices(nil, item, key)
+		}
+	}
+
+`ThreadSafeStore` 的其它方法功能介绍：
+
+1. `Index(indexName string, obj interface{}) ([]interface{}, error)`
+	indexName 为索引函数名称（下同）。使用对应的索引函数为 obj 创建索引值列表，然后查询索引，返回匹配这些索引值的对象列表（去重）。
+2. `ByIndex(indexName, indexKey string) ([]interface{}, error)`
+	indexKey 为索引值，查询索引，返回这个索引值匹配的对象列表；
+3. `IndexKeys(indexName, indexKey string) ([]string, error)`
+	indexKey 为索引值，查询索引，返回这个索引匹配的对象 Key 列表；
+4. `ListIndexFuncValues(indexName string) []string`
+	查询索引，返回 indexName 对应的索引函数创建的所有索引包含的对象 Key 列表；
+5. `GetIndexers() Indexers`
+	返回命名的索引函数集合 `c.indexers`
+6. `AddIndexers(newIndexers Indexers) error`
+	将 nexIndexers 中的命名函数添加到 c.indexers 中。
+	必须在添加任何对象前调用该方法，否则会出错返回。
+	newIndexers 中的命名方法不能与 c.indexers 中保存的命名冲突，否则出错返回。
+7. `Resync() error`
+	直接返回。因为 Add/Update/Delete/Replace 方法都会同时更新缓存和索引，两者时刻是同步的。
+
+## KeyFunc 和实现 Store/Indexer 接口的 cache
+
+在分析 `cache` 类型之前，我们先分析为对象生成唯一标识字符串的函数类型 `KeyFunc`：
+
+	// 来源于 k8s.io/client-go/tools/cache/store.go
+	// KeyFunc knows how to make a key from an object. Implementations should be deterministic.
+	type KeyFunc func(obj interface{}) (string, error)
+
+client-go 提供了两个 KeyFunc 类型函数 `MetaNamespaceKeyFunc` 和 `DeletionHandlingMetaNamespaceKeyFunc` (实际用的最多)：
++ `MetaNamespaceKeyFunc`：提取对象的 `<namespace>/<object-name>` 或 `<object-name>` 作为 Key；
++ `DeletionHandlingMetaNamespaceKeyFunc`：先检查对象是不是 DeletedFinalStateUnknown 类型，如果是直接返回对象的 Key 字段，否则调用 `MetaNamespaceKeyFunc`
+
+	// 来源于 k8s.io/client-go/tools/cache/store.go
+	func MetaNamespaceKeyFunc(obj interface{}) (string, error) {
+		// 如果对象时字符串，则直接使用它作为 Key
+		if key, ok := obj.(ExplicitKey); ok {
+			return string(key), nil
+		}
+		// 否则提取对象的 Meta 信息
+		meta, err := meta.Accessor(obj)
+		if err != nil {
+			return "", fmt.Errorf("object has no meta: %v", err)
+		}
+		// 如果对象有 Namespace 则使用  `<namespace>/<object-name>` 作为 key
+		// 否则使用 `<object-name>` 作为 key
+		if len(meta.GetNamespace()) > 0 {
+			return meta.GetNamespace() + "/" + meta.GetName(), nil
+		}
+		return meta.GetName(), nil
+	}
+
+	// 来源于 k8s.io/client-go/tools/cache/controller.go
+	func DeletionHandlingMetaNamespaceKeyFunc(obj interface{}) (string, error) {
+		// DeletedFinalStateUnknown 是封装了删除对象 key 和对象本身的类型，由 DeltaFIFO.Replace() 方法产生
+		if d, ok := obj.(DeletedFinalStateUnknown); ok {
+			return d.Key, nil
+		}
+		return MetaNamespaceKeyFunc(obj)
+	}
+与 `MetaNamespaceKeyFunc` 功能相反的是 `SplitMetaNamespaceKey` 函数，它将传入的 Key 分解，返回对应的 namespace 和 obj name。
+
+然后我们再来分析内置类型 `cache`。
+
+`cache` 实现了 `Indexer` 和 `Store` 接口，函数 `NewIndexer` 和 `NewStore` 均返回该类型的对象：
+
+	// 来源于 k8s.io/client-go/tools/cache/store.go
+	// NewIndexer returns an Indexer implemented simply with a map and a lock.
+	func NewIndexer(keyFunc KeyFunc, indexers Indexers) Indexer {
+		return &cache{
+			cacheStorage: NewThreadSafeStore(indexers, Indices{}),
+			keyFunc:      keyFunc,
+		}
+	}
+
+	// 来源于 k8s.io/client-go/tools/cache/store.go
+	// NewStore returns a Store implemented simply with a map and a lock.
+	func NewStore(keyFunc KeyFunc) Store {
+		return &cache{
+			cacheStorage: NewThreadSafeStore(Indexers{}, Indices{}),
+			keyFunc:      keyFunc,
+		}
+	}
+
+在分析 `ThreadSafeStore` 时提到过，它的缓存相关方法如 `Add/Update/Delete/Get` 都**需要传入**对象的标识字符串 Key。而 `cache` 封装了 `ThreadSafeStore` 和 `KeyFunc` 类型成员，它使用成员 `keyFunc` 函数为添加到 `cacheStorage` 的对象生成标识字符串 Key。以 cache 的 `Add` 方法为例：
+
+	// 来源于 k8s.io/client-go/tools/cache/store.go
+	func (c *cache) Add(obj interface{}) error {
+		key, err := c.keyFunc(obj)
+		if err != nil {
+			return KeyError{obj, err}
+		}
+		c.cacheStorage.Add(key, obj)
+		return nil
+	}
+
+cache 类型实现了 `Store` 和 `Indexer` 接口，在很多地方都有应用，如 Informer 和 DeltaFIFO：
+
+1. NewInformer 函数：
+	// 来源于 k8s.io/client-go/tools/cache/controller.go
+	func NewInformer(
+		lw ListerWatcher,
+		objType runtime.Object,
+		resyncPeriod time.Duration,
+		h ResourceEventHandler,
+	) (Store, Controller){
+		clientState := NewStore(DeletionHandlingMetaNamespaceKeyFunc)
+		fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, clientState)
+		cfg := &Config{
+			Queue:            fifo,
+			...
+		}
+		...
+		return clientState, New(cfg)
+	}
+
+2. NewIndexerInformer 函数：
+	// 来源于 k8s.io/client-go/tools/cache/controller.go
+	func NewIndexerInformer(
+		lw ListerWatcher,
+		objType runtime.Object,
+		resyncPeriod time.Duration,
+		h ResourceEventHandler,
+		indexers Indexers,
+	) (Indexer, Controller) {
+		clientState := NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers)
+		fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, clientState)
+		cfg := &Config{
+			Queue:            fifo,
+			...
+		}
+		...
+		return clientState, New(cfg)
+	}
+3.  NewSharedIndexInformer 函数：
+	// k8s.io/client-go/tools/cache/shared_informer.go
+	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
+	}
+
+
+后面文章我们讨论 Informer 使用的 Queue 和 FIFO。