PV Controller 的本地 cache 及其 indexer

controller := &PersistentVolumeController{
    // 根据accessmode添加索引
    volumes:                       newPersistentVolumeOrderedIndex(),
    claims:                        cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc),
    // ... 忽略其他代码
}

func newPersistentVolumeOrderedIndex() persistentVolumeOrderedIndex {
	return persistentVolumeOrderedIndex{cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{"accessmodes": accessModesIndexFunc})}
}

func (ctrl *PersistentVolumeController) initializeCaches(volumeLister corelisters.PersistentVolumeLister, claimLister corelisters.PersistentVolumeClaimLister) {
	volumeList, err := volumeLister.List(labels.Everything())
	if err != nil {	return }
	for _, volume := range volumeList {
        // 从informer的cache中取出来，放到私有缓存中
		volumeClone := volume.DeepCopy()
		if _, err = ctrl.storeVolumeUpdate(volumeClone); err != nil {
			klog.Errorf("error updating volume cache: %v", err)
		}
	}
}

func (ctrl *PersistentVolumeController) storeVolumeUpdate(volume interface{}) (bool, error) {
	return storeObjectUpdate(ctrl.volumes.store, volume, "volume")
}
// storeObjectUpdate 会在两种情况下更新本地缓存：
// 1. Informer回调函数（来自etcd的事件）
// 2. controller自己修改了资源，更新了pv之后，在update到apiserver之后也要同步更新到本地缓存
func storeObjectUpdate(store cache.Store, obj interface{}, className string) (bool, error) {
    // 首先尝试从缓存中获取
	oldObj, found, err := store.Get(obj)
	if err != nil {
		return false, fmt.Errorf("Error finding %s %q in controller cache: %v", className, objName, err)
	}
    // 返回 metav1.Object类型，用来访问资源的元信息
	objAccessor, _ := meta.Accessor(obj)
    // 本地缓存中不存在
	if !found {
		// 打印resourceVersion，并记录到本地缓存中，并直接返回
		klog.V(4).Infof("storeObjectUpdate: adding %s %q, version %s", className, objName, objAccessor.GetResourceVersion())
		_ = store.Add(obj)
		return true, nil
	}

	oldObjAccessor, err := meta.Accessor(oldObj)
	if err != nil {
		return false, err
	}
    // 获取新旧资源的ResourceVersion
	objResourceVersion, _ := strconv.ParseInt(objAccessor.GetResourceVersion(), 10, 64)
	oldObjResourceVersion, _ := strconv.ParseInt(oldObjAccessor.GetResourceVersion(), 10, 64)
	// 旧的资源RV更高，直接返回
	if oldObjResourceVersion > objResourceVersion {
		return false, nil
	}
    // 否则就更新本地缓存
	if err = store.Update(obj); err != nil {
		return false, fmt.Errorf("Error updating %s %q in controller cache: %v", className, objName, err)
	}
	return true, nil
}

	if claim.Spec.VolumeName == "" {
		// User did not care which PV they get.
		delayBinding, _ := ctrl.shouldDelayBinding(claim)

		// 从本地缓存取volume
		volume, err := ctrl.volumes.findBestMatchForClaim(claim, delayBinding)
		if err != nil {
			klog.V(2).Infof("synchronizing unbound PersistentVolumeClaim[%s]: Error finding PV for claim: %v", claimToClaimKey(claim), err)
			return fmt.Errorf("Error finding PV for claim %q: %v", claimToClaimKey(claim), err)
        }
    }

type PersistentVolumeController struct {
	volumeLister       corelisters.PersistentVolumeLister
    volumeListerSynced cache.InformerSynced
    // 省去其他代码
}

func newPersistentVolumeOrderedIndex() persistentVolumeOrderedIndex {
	return persistentVolumeOrderedIndex{cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{"accessmodes": accessModesIndexFunc})}
}

type IndexFunc func(obj interface{}) ([]string, error)

// persistentVolumeOrderedIndex 根据 accessModes 进行索引，并根据存储容量排序。
type persistentVolumeOrderedIndex struct {
	store cache.Indexer
}

// Indexer 是一个存储接口，允许通过多个索引函数来list资源
type Indexer interface {
	Store
	// 使用预配置的索引名，以及期望资源的模板，枚举所有匹配的资源
	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
}

// listByAccessModes 列出缓存中所有与给定的pv的accessModes相等的 pv
func (pvIndex *persistentVolumeOrderedIndex) listByAccessModes(modes []v1.PersistentVolumeAccessMode) ([]*v1.PersistentVolume, error) {
    // 使用pv构造期望的accessMode
	pv := &v1.PersistentVolume{
		Spec: v1.PersistentVolumeSpec{
			AccessModes: modes,
		},
	}
    // 枚举
	objs, err := pvIndex.store.Index("accessmodes", pv)
	if err != nil {
		return nil, err
	}
    // 类型转义
	volumes := make([]*v1.PersistentVolume, len(objs))
	for i, obj := range objs {
		volumes[i] = obj.(*v1.PersistentVolume)
	}
	return volumes, nil
}

func (c *threadSafeMap) Index(indexName string, obj interface{}) ([]interface{}, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()
    // 根据预配置的索引名，从map中取出索引函数
	indexFunc := c.indexers[indexName]
	if indexFunc == nil {
		return nil, fmt.Errorf("Index with name %s does not exist", indexName)
	}
    // 使用索引函数，以及给定的期望模板，计算出索引值，返回值类型是 []string
	indexKeys, err := indexFunc(obj)
	if err != nil {
		return nil, err
	}
	index := c.indices[indexName]

	var returnKeySet sets.String
	if len(indexKeys) == 1 {
        // 在大多数情况下，indexKeys里面只有一个元素，也就是说，我们自定义的索引函数，针对给定
        // 的资源，只生成一个索引值，[]string里面只含有一个元素。这里把这个path单独拎出来优化一下。
		returnKeySet = index[indexKeys[0]]
	} else {
		// 如果对应的资源有多个索引值，每个索引值都对应一个结果集合，表示这个结果集合匹配特定的索引值
		returnKeySet = sets.String{}
		for _, indexKey := range indexKeys {
			for key := range index[indexKey] {
				returnKeySet.Insert(key)
			}
		}
	}
    // 根据索引保存的元素的key，取出元素并返回。
	list := make([]interface{}, 0, returnKeySet.Len())
	for absoluteKey := range returnKeySet {
		list = append(list, c.items[absoluteKey])
	}
	return list, nil
}

type threadSafeMap struct {
    lock  sync.RWMutex
    // 存放资源
    items map[string]interface{}
    // 索引名字到索引函数的映射，这个是提前注册的
	indexers Indexers
    // 这个就是存放索引的集合了，是个二维的map，外层的map是根据索引名，取出该索引名下分类好的
    // 各个key集合，每个key集合对应该索引下的特定索引值
	indices Indices
}
// Index maps the indexed value to a set of keys in the store that match on that value
// Index 映射 索引值到 key的集合
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

// DeletionHandlingMetaNamespaceKeyFunc checks for
// DeletedFinalStateUnknown objects before calling
// MetaNamespaceKeyFunc.
func DeletionHandlingMetaNamespaceKeyFunc(obj interface{}) (string, error) {
	if d, ok := obj.(DeletedFinalStateUnknown); ok {
		return d.Key, nil
	}
	return MetaNamespaceKeyFunc(obj)
}