kubernetes indexer源码解析
kubernetes indexer是实现了多索引的本地缓存
1.背景
以db中学校学生表做本地缓存为例
type Student struct {
ID uint32
Name string
Class uint32
}
// 本地缓存，id和学生的映射
var idCache = map[uint32]Student{}当需要以学生名来取值时，此时没有合适的缓存，可以再添加一份
// 本地缓存，名称和学生的映射
var idCache = map[string]Student{}但这样缓存就存了两份，浪费了内存。可以维护一个学生名和id的索引，在拖过id的缓存取值即可
// 索引，名称和id的映射
var index = map[string]id{}这样要通过名称查缓存，通过idCache[index[name]]即可。kubernetes indexer就是类似的思路，indexer支持任意类型，线程安全
2.使用
// 取pod中标签foo的值作为索引值
func testIndexFunc(obj interface{}) ([]string, error) {
pod := obj.(*v1.Pod)
return []string{pod.Labels["foo"]}, nil
}
func TestMultiIndexKeys(t *testing.T) {
// 创建索引器
index := NewIndexer(MetaNamespaceKeyFunc, Indexers{"byUser": testUsersIndexFunc})
pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "one", Annotations: map[string]string{"users": "ernie,bert"}}}
pod2 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "two", Annotations: map[string]string{"users": "bert,oscar"}}}
pod3 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "tre", Annotations: map[string]string{"users": "ernie,elmo"}}}
// 添加索引
index.Add(pod1)
index.Add(pod2)
index.Add(pod3)
expected := map[string]sets.String{}
expected["ernie"] = sets.NewString("one", "tre")
expected["bert"] = sets.NewString("one", "two")
expected["elmo"] = sets.NewString("tre")
expected["oscar"] = sets.NewString("two")
expected["elmo1"] = sets.NewString()
{
for k, v := range expected {
found := sets.String{}
// 根据索引名+缓存key获取索引结果
indexResults, err := index.ByIndex("byUser", k)
if err != nil {
t.Errorf("Unexpected error %v", err)
}
for _, item := range indexResults {
found.Insert(item.(*v1.Pod).Name)
}
if !found.Equal(v) {
t.Errorf("missing items, index %s, expected %v but found %v", k, v.List(), found.List())
}
}
}
// 验证删除索引
index.Delete(pod3)
erniePods, err := index.ByIndex("byUser", "ernie")
if err != nil {
t.Errorf("unexpected error: %v", err)
}
if len(erniePods) != 1 {
t.Errorf("Expected 1 pods but got %v", len(erniePods))
}
for _, erniePod := range erniePods {
if erniePod.(*v1.Pod).Name != "one" {
t.Errorf("Expected only 'one' but got %s", erniePod.(*v1.Pod).Name)
}
}
elmoPods, err := index.ByIndex("byUser", "elmo")
if err != nil {
t.Errorf("unexpected error: %v", err)
}
if len(elmoPods) != 0 {
t.Errorf("Expected 0 pods but got %v", len(elmoPods))
}
copyOfPod2 := pod2.DeepCopy()
copyOfPod2.Annotations["users"] = "oscar"
// 验证更新索引
index.Update(copyOfPod2)
bertPods, err := index.ByIndex("byUser", "bert")
if err != nil {
t.Errorf("unexpected error: %v", err)
}
if len(bertPods) != 1 {
t.Errorf("Expected 1 pods but got %v", len(bertPods))
}
for _, bertPod := range bertPods {
if bertPod.(*v1.Pod).Name != "one" {
t.Errorf("Expected only 'one' but got %s", bertPod.(*v1.Pod).Name)
}
}
}3.源码解析
func NewIndexer(keyFunc KeyFunc, indexers Indexers) Indexer {
return &cache{
// 线程安全的存储
cacheStorage: NewThreadSafeStore(indexers, Indices{}),
// 缓存key计算函数
keyFunc: keyFunc,
}
}// 线程安全的存储
func NewThreadSafeStore(indexers Indexers, indices Indices) ThreadSafeStore {
return &threadSafeMap{
// 缓存
items: map[string]interface{}{},
index: &storeIndex{
// Indexers是map[string]IndexFunc类型，保存索引名与索引值计算函数的映射
indexers: indexers,
// Indices是map[string]Index类型，通过索引名找到Index，在通过索引值找到缓存key
indices: indices,
},
}
}func (c *cache) Add(obj interface{}) error {
// 计算缓存key
key, err := c.keyFunc(obj)
if err != nil {
return KeyError{obj, err}
}
c.cacheStorage.Add(key, obj)
return nil
}func (c *threadSafeMap) Add(key string, obj interface{}) {
c.Update(key, obj)
}
func (c *threadSafeMap) Update(key string, obj interface{}) {
// 更新操作，加锁
c.lock.Lock()
defer c.lock.Unlock()
// 获取旧缓存对象
oldObject := c.items[key]
// 设置新缓存对象
c.items[key] = obj
// 更新索引
c.index.updateIndices(oldObject, obj, key)
}func (i *storeIndex) updateIndices(oldObj interface{}, newObj interface{}, key string) {
var oldIndexValues, indexValues []string
var err error
// 遍历所有索引器
for name, indexFunc := range i.indexers {
// 在缓存对象第一次添加的时候，oldObj为nil
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 := i.indices[name]
if index == nil {
index = Index{}
i.indices[name] = index
}
// 一个小优化，当添加的新值和旧值都为1并且相同时，无需处理
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
}
// 从索引中删除key
for _, value := range oldIndexValues {
i.deleteKeyFromIndex(key, value, index)
}
// 从索引中添加key
for _, value := range indexValues {
i.addKeyToIndex(key, value, index)
}
}
}
func (i *storeIndex) addKeyToIndex(key, indexValue string, index Index) {
set := index[indexValue]
if set == nil {
set = sets.String{}
index[indexValue] = set
}
// 索引名+索引值可能对应多个缓存key，通过set去重
set.Insert(key)
}
func (i *storeIndex) deleteKeyFromIndex(key, indexValue string, index Index) {
set := index[indexValue]
if set == nil {
return
}
// 删除缓存key
set.Delete(key)
// 当缓存key的set为空时，删除索引值的映射，避免oom
if len(set) == 0 {
delete(index, indexValue)
}
}4.总结
kubernetes indexer在实现多索引的本地缓存的思路，特别在代码可复用方面，值得我们学习借鉴
