关于云计算:Kubernetes-源码解析-Informer-的工作原理

上篇扒了 HPA 的源码，然而没深刻细节，明天往细节深刻。

Kubernetes 中的长久化数据保留在 etcd 中，各个组件并不会间接拜访 etcd，而是通过 api-server 裸露的 RESTful 接口对集群进行拜访和管制。

资源的控制器（图中右侧灰色的局部）读取数据也并不会间接从 api-server 中获取资源信息（这样会减少 api-server 的压力），而是从其“本地缓存”中读取。这个“本地缓存”只是表象的存在，加上缓存的同步逻辑就是明天要是说的Informer（灰色区域中的第一个蓝色块）所提供的性能。

从图中能够看到 Informer 的几个组件：

Reflector：与 api-server交互，监听资源的变更。
Delta FIFO Queue：增量的 FIFO 队列，保留 Reflector 监听到的资源变更（简略的封装）。
Indexer：Informer 的本地缓存，FIFO 队列中的数据依据不同的变更类型，在该缓存中进行操作。
- Local Store：

上篇提到了程度主动伸缩的控制器HorizontalController，其构造方法就须要提供 Informer。

//pkg/controller/podautoscaler/horizontal.go

type HorizontalController struct {
    scaleNamespacer scaleclient.ScalesGetter
    hpaNamespacer   autoscalingclient.HorizontalPodAutoscalersGetter
    mapper          apimeta.RESTMapper
    replicaCalc   *ReplicaCalculator
    eventRecorder record.EventRecorder
    downscaleStabilisationWindow time.Duration
    hpaLister       autoscalinglisters.HorizontalPodAutoscalerLister
    hpaListerSynced cache.InformerSynced
    podLister       corelisters.PodLister
    podListerSynced cache.InformerSynced
    queue workqueue.RateLimitingInterface
    recommendations map[string][]timestampedRecommendation}

func NewHorizontalController(
    evtNamespacer v1core.EventsGetter,
    scaleNamespacer scaleclient.ScalesGetter,
    hpaNamespacer autoscalingclient.HorizontalPodAutoscalersGetter,
    mapper apimeta.RESTMapper,
    metricsClient metricsclient.MetricsClient,
    // 从 HorizontalPodAutoscalerInformer 获取 hpa 实例信息
    hpaInformer autoscalinginformers.HorizontalPodAutoscalerInformer,
    // 从 PodInformer 中获取 pod 信息
    podInformer coreinformers.PodInformer,
    resyncPeriod time.Duration,
    downscaleStabilisationWindow time.Duration,
    tolerance float64,
    cpuInitializationPeriod,
    delayOfInitialReadinessStatus time.Duration,

) *HorizontalController {
    ......
        hpaInformer.Informer().AddEventHandlerWithResyncPeriod( // 增加事件处理器
        cache.ResourceEventHandlerFuncs{
            AddFunc:    hpaController.enqueueHPA,
            UpdateFunc: hpaController.updateHPA,
            DeleteFunc: hpaController.deleteHPA,
        },
        resyncPeriod,
    )
    ......
}

type HorizontalPodAutoscalerInformer interface {Informer() cache.SharedIndexInformer
    Lister() v1.HorizontalPodAutoscalerLister}

HorizontalPodAutoscalerInformer的实例化办法中就呈现了明天的正主cache.NewSharedIndexInformer()。

//staging/src/k8s.io/client-go/informers/autoscaling/v1/horizontalpodautoscaler.go
func NewFilteredHorizontalPodAutoscalerInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
    return cache.NewSharedIndexInformer(
           // 用于 list 和 watch api-server 中的资源。比方用来创立 Reflector
        &cache.ListWatch{ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                if tweakListOptions != nil {tweakListOptions(&options)
                }
                // 应用 HPA API 获取 HPA 资源
                return client.AutoscalingV1().HorizontalPodAutoscalers(namespace).List(options)
            },
            WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                if tweakListOptions != nil {tweakListOptions(&options)
                }
                // 应用 HPA API 监控 HPA 资源
                return client.AutoscalingV1().HorizontalPodAutoscalers(namespace).Watch(options)
            },
        },
        &autoscalingv1.HorizontalPodAutoscaler{},
        resyncPeriod,
        indexers,
    )
}

//staging/src/k8s.io/client-go/tools/cache/index.go
type Indexers map[string]IndexFunc
type IndexFunc func(obj interface{}) ([]string, error)

实例化 Indexers cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}

//staging/src/k8s.io/client-go/tools/cache/shared_informer.go
// ListerWatcher 用于 list 和 watch api-server 上的资源
//runtime.Object 要监控的资源的运行时对象
//time.Duration 同步的间隔时间
//Indexers 提供不同资源的索引数据的信息查询方法，如 namespace => MetaNamespaceIndexFunc
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), // 初始化 Indexer
        listerWatcher:                   lw,
        objectType:                      objType,
        resyncCheckPeriod:               defaultEventHandlerResyncPeriod,
        defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod,
        cacheMutationDetector:           NewCacheMutationDetector(fmt.Sprintf("%T", objType)),
        clock:                           realClock,
    }
    return sharedIndexInformer
}

Indexer提供了本地缓存的实现：计算 key 和对数据进行管制（通过调用 ThreadSafeStore 的接口）

type Indexer interface {
    Store
    Index(indexName string, obj interface{}) ([]interface{}, error)
    IndexKeys(indexName, indexedValue string) ([]string, error)
    ListIndexFuncValues(indexName string) []string
    ByIndex(indexName, indexedValue string) ([]interface{}, error)
    GetIndexers() Indexers
    AddIndexers(newIndexers Indexers) error
}

Indexer 的创立

//staging/src/k8s.io/client-go/tools/cache/store.go
//keyFunc：key 的生成规定
//indexers：提供了索引资源的不同信息的拜访办法，如用于查问命名空间的 namespace => MetaNamespaceIndexFunc
func NewIndexer(keyFunc KeyFunc, indexers Indexers) Indexer {
    return &cache{cacheStorage: NewThreadSafeStore(indexers, Indices{}),
        keyFunc:      keyFunc,
    }
}

`ThreadSafeStore`

ThreadSafeStore 提供了对存储的并发拜访接口

注意事项：不能批改 Get 或 List 返回的任何内容，因为它不仅会毁坏线程平安，还会毁坏索引性能。

//staging/src/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,
    }
}
type threadSafeMap struct {
    lock  sync.RWMutex
    items map[string]interface{} //key => value
    indexers Indexers //value 的信息的拜访办法
    indices Indices // 索引
}

Reflector通过 ListerWatcher（API）与api-server 交互，对资源进行监控。将资源实例的创立、更新、删除等工夫封装后保留在 Informer 的 FIFO 队列中。

//staging/src/k8s.io/client-go/tools/cache/reflector.go
func NewReflector(lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {return NewNamedReflector(naming.GetNameFromCallsite(internalPackages...), lw, expectedType, store, resyncPeriod)
}

// NewNamedReflector same as NewReflector, but with a specified name for logging
func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
    r := &Reflector{
        name:          name,
        listerWatcher: lw,
        store:         store, //FIFO 队列
        period:        time.Second,
        resyncPeriod:  resyncPeriod,
        clock:         &clock.RealClock{},}
    r.setExpectedType(expectedType)
    return r
}

通过 sharedIndexInformer#AddEventHandlerWithResyncPeriod() 注册事件监听器。

以后面的 HorizontalController 为例，创立 informer 的时候增加了三个解决办法：AddFunc、UpdateFunc、DeleteFunc。这三个办法的实现是将对应的元素的 key（固定格局 namespace/name）从 workequeue 中进行入队、出队的操作。（资源控制器监听了该 workqueue）

`controller-manager`

在通过 InformerFactory 创立 Informer 实现后，都会将新建的 Informer 退出到 InformerFactory 的一个 map 中。

在 controller-manager 在实现所有的控制器（各种 Controller，包含 CRD）后，会调用InformerFactory#Start() 来启动 InformerFactory 的map中的所有 Informer（调用Informer#Run() 办法）

`sharedIndexInformer#Run()`

//staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {defer utilruntime.HandleCrash()
      // 创立一个增量的 FIFO 队列：DeltaFIFO
    fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer)
    cfg := &Config{
        Queue:            fifo,
        ListerWatcher:    s.listerWatcher,
        ObjectType:       s.objectType,
        FullResyncPeriod: s.resyncCheckPeriod,
        RetryOnError:     false,
        ShouldResync:     s.processor.shouldResync,

        Process: s.HandleDeltas,
    }
     // 启动前的初始化，创立 Controller
    func() {s.startedLock.Lock()
        defer s.startedLock.Unlock()

        s.controller = New(cfg)
        s.controller.(*controller).clock = s.clock
        s.started = true
    }()
    processorStopCh := make(chan struct{})
    var wg wait.Group
    defer wg.Wait()              // Wait for Processor to stop
    defer close(processorStopCh) // Tell Processor to stop
    wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
    wg.StartWithChannel(processorStopCh, s.processor.run)
     // 退出时的状态清理
    defer func() {s.startedLock.Lock()
        defer s.startedLock.Unlock()
        s.stopped = true // Don't want any new listeners
    }()
    // 履行管制逻辑
    s.controller.Run(stopCh)
}

`controller#Run()`

//staging/src/k8s.io/client-go/tools/cache/controller.go
func (c *controller) Run(stopCh <-chan struct{}) {defer utilruntime.HandleCrash()
    go func() {
        <-stopCh
        c.config.Queue.Close()}()
    // 创立一个 Reflector，用于从 api-server list 和 watch 资源
    r := NewReflector(
        c.config.ListerWatcher,
        c.config.ObjectType,
        c.config.Queue,
        c.config.FullResyncPeriod,
    )
    r.ShouldResync = c.config.ShouldResync
    r.clock = c.clock
      // 为 controller 指定 Reflector
    c.reflectorMutex.Lock()
    c.reflector = r
    c.reflectorMutex.Unlock()
    var wg wait.Group
    defer wg.Wait()
     // 执行 Reflector#Run()：会启动一个 goroutine 开始监控资源，将 watch 到的数据写入到 queue（FIFO 队列）中
    wg.StartWithChannel(stopCh, r.Run)
      // 继续从 queue（FIFO 队列）获取数据并进行解决，解决的逻辑在 sharedIndexInformer#HandleDeltas()
    wait.Until(c.processLoop, time.Second, stopCh)
}

`sharedIndexInformer#HandleDeltas()`

//staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {s.blockDeltas.Lock()
    defer s.blockDeltas.Unlock()

    // from oldest to newest
    for _, d := range obj.(Deltas) { // 循环解决 FIFO 队列中取出的资源实例
        switch d.Type {
        case Sync, Added, Updated: // 同步（前面具体解读）、新增、更新事件
            isSync := d.Type == Sync
            s.cacheMutationDetector.AddObject(d.Object)
            if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {if err := s.indexer.Update(d.Object); err != nil { // 如果 indexer 中曾经存在，更掉用 update 办法进行更新
                    return err
                }
                // 更新胜利后发送“更新”告诉：蕴含了新、旧资源实例
                s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
            } else {
                  // 如果 indexer 中没有该资源实例，则放入 indexer 中
                if err := s.indexer.Add(d.Object); err != nil {return err}
                // 增加胜利后，发送“新增”告诉：蕴含了新加的资源实例
                s.processor.distribute(addNotification{newObj: d.Object}, isSync)
            }
        case Deleted: // 删除事件
            if err := s.indexer.Delete(d.Object); err != nil {// 从 indexer 中删除
                return err
            }
            // 删除胜利后，发送“删除告诉”：蕴含了删除的资源实例
            s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
        }
    }
    return nil
}

Informer 的实现不算简单，却在 Kubernetes 中很常见，每种资源的管制也都通过 Informer 来获取 api-server 的资源实例的变更。

文章对立公布在公众号 云原生指北

关于云计算:Kubernetes-源码解析-Informer-的工作原理

为什么要有 Informer？

初始化

Informer

Indexer

`ThreadSafeStore`

Reflector

增加同步事件监听器

运行

`controller-manager`

`sharedIndexInformer#Run()`

`controller#Run()`

`sharedIndexInformer#HandleDeltas()`

总结