threadSafeMap代码解析

// threadSafeMap implements ThreadSafeStore
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
}
 
// 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 an 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)

• lock用于加锁，使操作原子化
• items是实际存储的数据，它的key可以认为是普通数据库的主键，k8s一般对应资源的gvk
• Index可以认为是非主键索引，它保存的是 indexed value 到 相关的keys的映射
• Indexers是非主键索引的集合，每个非主键索引用indexName标识
• indexers是Indexers类型，即map[string]IndexFunc
• IndexFunc是将obj转换为indexed values的函数
• 一个indexName(非主键索引)对应一个IndexFunc, Index

那么现在可以尝试解读一下函数:

// Index returns a list of items that match the given object on the index function.
// Index is thread-safe so long as you treat all items as immutable.
func (c *threadSafeMap) Index(indexName string, obj interface{}) ([]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)
	}
 
	indexedValues, err := indexFunc(obj)
	if err != nil {
		return nil, err
	}
	index := c.indices[indexName]
 
	var storeKeySet sets.String
	if len(indexedValues) == 1 {
		// In majority of cases, there is exactly one value matching.
		// Optimize the most common path - deduping is not needed here.
		storeKeySet = index[indexedValues[0]]
	} else {
		// Need to de-dupe the return list.
		// Since multiple keys are allowed, this can happen.
		storeKeySet = sets.String{}
		for _, indexedValue := range indexedValues {
			for key := range index[indexedValue] {
				storeKeySet.Insert(key)
			}
		}
	}
 
	list := make([]interface{}, 0, storeKeySet.Len())
	for storeKey := range storeKeySet {
		list = append(list, c.items[storeKey])
	}
	return list, nil
}

该函数用于根据某个indexName(非主键索引)中某个obj对应的所有items。

1. 利用indexName查询indexFunc
2. 利用indexFunc计算obj的索引值(indexed values), 是一个[]string类型
3. 遍历indexed values， 并根据每个indexed value去对应的Index查找到真实的keys
4. 最后通过这些keys到items中查找到所有结果

该过成加速了查询，但是使更新变得麻烦， 更新的时候不仅要更新items中的数据，还要更新所有indices中的数据。

// updateIndices modifies the objects location in the managed indexes:
// - for create you must provide only the newObj
// - for update you must provide both the oldObj and the newObj
// - for delete you must provide only the oldObj
// updateIndices must be called from a function that already has a lock on the cache
func (c *threadSafeMap) updateIndices(oldObj interface{}, newObj interface{}, key string) {
	var oldIndexValues, indexValues []string
	var err error
	for name, indexFunc := range c.indexers {
		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))
		}
 
		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 := c.indices[name]
		if index == nil {
			index = Index{}
			c.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
		}
 
		for _, value := range oldIndexValues {
			c.deleteKeyFromIndex(key, value, index)
		}
		for _, value := range indexValues {
			c.addKeyToIndex(key, value, index)
		}
	}
}

以上就是k8s client-go 使用的缓存的最底层结构， 它用在deltafifo和store里面