1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

// Indexer extends Store with multiple indices and restricts each
// accumulator to simply hold the current object (and be empty after
// Delete).
//
// There are three kinds of strings here:
// 1. a storage key, as defined in the Store interface,
// 2. a name of an index, and
// 3. an "indexed value", which is produced by an IndexFunc and
//    can be a field value or any other string computed from the object.
type Indexer interface {
    //嵌入了stroe，擁有stroe全部的方法
	Store
	
    // 回傳在indexName內與給定的Object有交集的所有object
	Index(indexName string, obj interface{}) ([]interface{}, error)
	
    // 回傳在indexname與透過indexed取的所有關聯的Object  Key
	IndexKeys(indexName, indexedValue string) ([]string, error)
    
    // 回傳在indexName所有的indexed
	ListIndexFuncValues(indexName string) []string
    
    // 回傳在indexName與透過indexed回傳所有關聯的object
	ByIndex(indexName, indexedValue string) ([]interface{}, error)
    
    // 回傳indexer
	GetIndexers() Indexers

    // 新增indexer
	AddIndexers(newIndexers Indexers) error
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

type Store interface {

    //新增物件
	Add(obj interface{}) error

    //更新物件
	Update(obj interface{}) error

    //刪除物件
	Delete(obj interface{}) error

    //列出所有物件
	List() []interface{}

    //列出所有物件的key
	ListKeys() []string

    // 透過物件取得object key
	Get(obj interface{}) (item interface{}, exists bool, err error)

    //透過object key取得object
	GetByKey(key string) (item interface{}, exists bool, err error)

    //一次性替換Object key 所代表的所有Object
	Replace([]interface{}, string) error

    //重新同步
	Resync() error
}

1
2
3
4
5
6
7

type cache struct {
	// cacheStorage bears the burden of thread safety for the cache
	cacheStorage ThreadSafeStore    //一個 thread safe 的 interface ，稍後會解釋。
	// keyFunc is used to make the key for objects stored in and retrieved from items, and
	// should be deterministic.
	keyFunc KeyFunc        //計算object key的方法，稍後會解釋
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

type ThreadSafeStore interface {
	Add(key string, obj interface{})
	Update(key string, obj interface{})
	Delete(key string)
	Get(key string) (item interface{}, exists bool)
	List() []interface{}
	ListKeys() []string
	Replace(map[string]interface{}, string)
	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 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
	// Resync is a no-op and is deprecated
	Resync() error
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// threadSafeMap implements ThreadSafeStore
type threadSafeMap struct {
	lock  sync.RWMutex                          //讀寫鎖（可能讀多寫少？）
        items map[string]interface{}               //儲存Object Key:Object

	// indexers maps a name to an IndexFunc
	indexers Indexers                           //用以計算indexed的function
	// indices maps a name to an Index
	indices Indices                             //indexed map透過這個表獲取Onject Key
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// IndexFunc knows how to compute the set of indexed values for an object.
type IndexFunc func(obj interface{}) ([]string, error)        //傳入Obaject會幫你算出多種[]indexed-name

// Index maps the indexed value to a set of keys in the store that match on that value
type Index map[string]sets.String                             //每個indexed-name會對應到一個[set]object key set
                                                              //inedxed->object key set

// Indexers maps a name to a IndexFunc
type Indexers map[string]IndexFunc                            //計算 indexed 的 function 們（透過 name 去分類）
                                                              //index name->index function->index

// Indices maps a name to an Index
type Indices map[string]Index                                 //index name(function name)->index[indexed-name]->[set]object key

1
2
3
4
5
6
7

func NewThreadSafeStore(indexers Indexers, indices Indices) ThreadSafeStore {
	return &threadSafeMap{
		items:    map[string]interface{}{},
		indexers: indexers,
		indices:  indices,
	}
}

1
2
3
4
5
6
7

// 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,
	}
}

1
2
3
4
5
6
7

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)         //更新indexed索引 ，稍後會解釋更新 indexed 索引的動作
}

1
2
3
4
5
6
7
8

//跟Add不是一模一樣...為什麼不重用add不明白
func (c *threadSafeMap) Update(key string, obj interface{}) {
	c.lock.Lock()                                //鎖不解釋
	defer c.lock.Unlock()                        //退出解鎖
	oldObject := c.items[key]                    //透過Object key 取出舊的物件     
	c.items[key] = obj                           //透過Object key寫入新的物件
	c.updateIndices(oldObject, obj, key)         //更新indexed索引
}

1
2
3
4
5
6
7
8

func (c *threadSafeMap) Delete(key string) {
	c.lock.Lock()                                    //鎖不解釋
	defer c.lock.Unlock()                            //退出解鎖
	if obj, exists := c.items[key]; exists {         //如果物件存在
		c.deleteFromIndices(obj, key)               //刪除indexed索引
		delete(c.items, key)                        //刪除物件
	}
}

1
2
3
4
5
6
7

func (c *threadSafeMap) Get(key string) (item interface{}, exists bool) {
	c.lock.RLock()                //讀操作，用讀鎖
	defer c.lock.RUnlock()        //解鎖
	item, exists = c.items[key]    //透過Object key拿取物件
	return item, exists            //回傳
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

func (c *threadSafeMap) List() []interface{} {
	c.lock.RLock()                //讀操作，用讀鎖
	defer c.lock.RUnlock()        //解鎖
        //for跑一次map全部加入到slice就好
	list := make([]interface{}, 0, len(c.items))
	for _, item := range c.items {
		list = append(list, item)
	}
	return list                    //回傳slice
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// ListKeys returns a list of all the keys of the objects currently
// in the threadSafeMap.
func (c *threadSafeMap) ListKeys() []string {
	c.lock.RLock()                //讀操作，用讀鎖
	defer c.lock.RUnlock()        //解鎖
        //for跑一次map全部加入到slice就好
	list := make([]string, 0, len(c.items))
	for key := range c.items {
		list = append(list, key)
	}
	return list                    //回傳slice
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

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{}                            //indexed 索引重建
	for key, item := range c.items {                 //更新indexed索引列表
		c.updateIndices(nil, item, key)
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

// 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.
//透過index name取得indexed後
func (c *threadSafeMap) Index(indexName string, obj interface{}) ([]interface{}, error) {
	c.lock.Lock()                                    //鎖不解釋
	defer c.lock.Unlock()                            //退出解鎖
    
	indexFunc := c.indexers[indexName]               //透過index name從indexer中取得對應的infexed function
	if indexFunc == nil {
		return nil, fmt.Errorf("Index with name %s does not exist", indexName)
	}

	indexedValues, err := indexFunc(obj)            //透過indexed function 計算indexed
	if err != nil {
		return nil, err
	}
	index := c.indices[indexName]                  //透過index name 從 indices 拿對應的 index

	var storeKeySet sets.String                    //object key的集合
	if len(indexedValues) == 1 {                   //大多數情狂indexed 只會有一個
		// In majority of cases, there is exactly one value matching.
		// Optimize the most common path - deduping is not needed here.
		storeKeySet = index[indexedValues[0]]    //透過indexed從index拿到對應的 object key set
	} else {                                     //不知道什麼情況會跑進去這裡，我用dlv跑測試沒有跑進這裡過...以下是看註解猜測的
		// Need to de-dupe the return list.
		// Since multiple keys are allowed, this can happen.
		storeKeySet = sets.String{}
		for _, indexedValue := range indexedValues {    //遞迴index所有的indexed,並且將數值插入set中
			for key := range index[indexedValue] {
				storeKeySet.Insert(key)
			}
		}
	}

	list := make([]interface{}, 0, storeKeySet.Len())    //Objest set 取出所有的 Object key 
    //將map有對應到的 Object 放入list 回傳
	for storeKey := range storeKeySet {
		list = append(list, c.items[storeKey])
	}
	return list, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// IndexKeys returns a list of the Store keys of the objects whose indexed values in the given index include the given indexed value.
// IndexKeys is thread-safe so long as you treat all items as immutable.
func (c *threadSafeMap) IndexKeys(indexName, indexedValue string) ([]string, error) {
	c.lock.RLock()                //讀操作，用讀鎖
	defer c.lock.RUnlock()        //解鎖

	indexFunc := c.indexers[indexName]        //透過 index name從indexer中取得對應的infexed function
	if indexFunc == nil {
		return nil, fmt.Errorf("Index with name %s does not exist", indexName)
	}

	index := c.indices[indexName]            //透過 index name 從 indices 拿對應的 index

	set := index[indexedValue]                //透過indexed 從 index 拿到所有的 Object key
	return set.List(), nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

func (c *threadSafeMap) ListIndexFuncValues(indexName string) []string {
	c.lock.RLock()                //讀操作，用讀鎖
	defer c.lock.RUnlock()        //解鎖

	index := c.indices[indexName]            //透過 index name 從 indices 拿對應的 index
	names := make([]string, 0, len(index))
    //拿出 index 中所有的 indexed key
	for key := range index {
		names = append(names, key)
	}
	return names
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// 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]        //透過index name從indexer 中取的 indexed function
	if indexFunc == nil {
		return nil, fmt.Errorf("Index with name %s does not exist", indexName)
	}

	index := c.indices[indexName]            //透過index name從Indices 中取得 index

	set := index[indexedValue]               //透過indexed name從index取的 Object key set
	list := make([]interface{}, 0, set.Len())   
    //遞迴儲存的object key set 對應的object倒出
	for key := range set {
		list = append(list, c.items[key])
	}

	return list, nil
}

1
2
3

func (c *threadSafeMap) GetIndexers() Indexers {
	return c.indexers                //沒啥好說的回傳一個indexer 
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

func (c *threadSafeMap) AddIndexers(newIndexers Indexers) error {
	c.lock.Lock()                                    //鎖不解釋
	defer c.lock.Unlock()                            //退出解鎖

	if len(c.items) > 0 {                            //現在有存資料了，不能加入indexer 
		return fmt.Errorf("cannot add indexers to running index")
	}

	oldKeys := sets.StringKeySet(c.indexers)        //把舊的indexer轉乘一個可以比對的資料
	newKeys := sets.StringKeySet(newIndexers)        //把新的indexer轉乘一個可以比對的資料

	if oldKeys.HasAny(newKeys.List()...) {           //比對兩個indexer 是否有重複
		return fmt.Errorf("indexer conflict: %v", oldKeys.Intersection(newKeys))
	}
    //上述不引響核心邏輯，不展開解析
    

    //把新的indexer加入indexer中
	for k, v := range newIndexers {
		c.indexers[k] = v
	}
	return nil
}

1
2
3
4

func (c *threadSafeMap) Resync() error {
	// Nothing to do
	return nil            //沒幹啥
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

// updateIndices modifies the objects location in the managed indexes, if this is an update, you must provide an oldObj
// updateIndices must be called from a function that already has a lock on the cache
//傳入舊的 object 為了確認有沒有這個 index （更新用）
//傳入新的 object 
//私有方法只能從add 或是 update呼叫，add/update 有上鎖，保證thread safe
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
    //如果有代舊的object的話需要先刪除舊的index，下一個段落會說到如何刪除
	if oldObj != nil {
		c.deleteFromIndices(oldObj, key)
	}
    //將所有的indexer跑過一次，拿到所有的index name以及indexed function
	for name, indexFunc := range c.indexers {
            //透過indexed function 算出indexed
		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 name 從 indices 拿到對應的 index
		index := c.indices[name]
            //確認 index 是否存在，不存在表示這個 indices 還沒有這個 index
            //建立 index 並且對應到 indecies 上
		if index == nil {
			index = Index{}
			c.indices[name] = index
		}
            
		for _, indexValue := range indexValues {
                //透過 indexed 檢查 index 對應的 Object key set
			set := index[indexValue]
                //若是set 為空表示 index 還沒建立object key set
                //建立 set 並且對應到 index 上
			if set == nil {
				set = sets.String{}
				index[indexValue] = set
			}
                    //object set set 插入 object key
			set.Insert(key)
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

// 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) {
    //將indexer 中全部 indexed function 取出
	for name, indexFunc := range c.indexers {
            //透過indexed function 算出 indexed 
		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 name 從indices拿到對應的index
		index := c.indices[name]
            //有可能 index name 在indeces中找不到（不太可能...） 
		if index == nil {
			continue
		}
            //將indexed 全部跑一次
		for _, indexValue := range indexValues {
                    //從index拿到對應的object key set 
			set := index[indexValue]
                    //如果set 存在的話就刪除對應的 Object key
			if set != nil {
				set.Delete(key)
                        
				    //刪除 object key 為零的 set ，為了避免佔用空間（kubernetes 上有相關的issuekubernetes/kubernetes/issues/84959）
				if len(set) == 0 {
					delete(index, indexValue)
				}
			}
		}
	}
}