Kubernetes SharedInformerFactory共享Informer机制源码深刻剖析-Kubernetes商业环境实战

专一于大数据及容器云核心技术解密，可提供全栈的大数据+云原平生台咨询方案，请持续关注本套博客。若有任何学术交流，可随时联系。更多内容请关注《数据云技术社区》公众号。

1 SharedInformerFactory 前因后果(组概念)

• sharedInformerFactory就是构造各类Informer的地方。为何是各类Informer呢，每一个SharedInformer其实只负责一种对象，在构造SharedInformer的时候指定了对象类型。
• SharedInformerFactory能够构造Kubernetes里全部对象的Informer，并且主要用在controller-manager这个服务中。
• 由于controller-manager负责管理绝大部分controller，每类controller不只须要本身关注的对象的informer，同时也可能须要其余对象的Informer(好比ReplicationController也须要PodInformer,不然他没法感知Pod的启动和关闭，也就达不到监控的目的了)。
• SharedInformerFactory可让全部的controller共享使用同一个类对象的Informer。
• 为了方便管理，Kubernetes对Informer进行了分组。

// 代码源自client-go/informers/factory.go
// SharedInformerFactory是个interfaces，因此确定有具体的实现类 
type SharedInformerFactory interface {
    // 在informers这个包中又定义了一个SharedInformerFactory，这个主要是包内抽象，因此此处继承了这个接口
    internalinterfaces.SharedInformerFactory
    ForResource(resource schema.GroupVersionResource) (GenericInformer, error)
    
    // 等待全部的Informer都已经同步完成，这里同步其实就是遍历调用SharedInformer.HasSynced()
    // 因此函数须要周期性的调用指导全部的Informer都已经同步完毕
    WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool
 
    Admissionregistration() admissionregistration.Interface // 返回admissionregistration相关的Informer组
    Apps() apps.Interface                                   // 返回app相关的Informer组
    Autoscaling() autoscaling.Interface                     // 返回autoscaling相关的Informer组
    Batch() batch.Interface                                 // 返回job相关的Informer组
    Certificates() certificates.Interface                   // 返回certificates相关的Informer组
    Coordination() coordination.Interface                   // 返回coordination相关的Informer组
    Core() core.Interface                                   // 返回core相关的Informer组
    Events() events.Interface                               // 返回event相关的Informer组
    Extensions() extensions.Interface                       // 返回extension相关的Informer组
    Networking() networking.Interface                       // 返回networking相关的Informer组
    Policy() policy.Interface                               // 返回policy相关的Informer组
    Rbac() rbac.Interface                                   // 返回rbac相关的Informer组
    Scheduling() scheduling.Interface                       // 返回scheduling相关的Informer组
    Settings() settings.Interface                           // 返回settings相关的Informer组
    Storage() storage.Interface                             // 返回storage相关的Informer组
}

// 代码源自client-go/informers/internalinterfaces/factory_interfaces.go 
type SharedInformerFactory interface {
    // 核心逻辑函数，相似于不少类的Run()函数
    Start(stopCh <-chan struct{})
    
    // 这个很关键，经过对象类型，返回SharedIndexInformer
    InformerFor(obj runtime.Object, newFunc NewInformerFunc) cache.SharedIndexInformer
}


// 代码源自client-go/informers/factory.go
type sharedInformerFactory struct {

    // apiserver的客户端，暂时不用关心怎么实现的，只要知道他能列举和监听资源就能够了
    client           kubernetes.Interface
    
    // 并非全部的使用者都须要指定namesapce，好比kubectl，他就能够列举全部namespace的资源，因此他没有指定namesapce   namespace        string
    
    // 这是个函数指针，用来调整列举选项的，这个选项用来client列举对象使用
    tweakListOptions internalinterfaces.TweakListOptionsFunc
    // 互斥锁
    lock             sync.Mutex
    // 默认的同步周期，这个在SharedInformer须要用
    defaultResync    time.Duration
    // 每一个类型的Informer有本身自定义的同步周期
    customResync     map[reflect.Type]time.Duration
    
    // 每类对象一个Informer，但凡使用SharedInformerFactory构建的Informer同一个类型其实都是同一个Informer
    informers map[reflect.Type]cache.SharedIndexInformer
    
    // 各类Informer启动的标记
    startedInformers map[reflect.Type]bool
}
复制代码

2 SharedInformerFactory 启动运行

// 代码源自client-go/informers/factory.go
// 其实sharedInformerFactory的Start()函数就是启动全部具体类型的Informer的过程
// 由于每一个类型的Informer都是SharedIndexInformer，须要须要把每一个SharedIndexInformer都要启动起来
func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {
    // 加锁操做
    f.lock.Lock()
    defer f.lock.Unlock()
    // 遍历informers这个map
    for informerType, informer := range f.informers {
        // 看看这个Informer是否已经启动过
        if !f.startedInformers[informerType] {
            // 若是没启动过，那就启动一个协程执行SharedIndexInformer的Run()函数，咱们在分析SharedIndexInformer的时候
            // 咱们知道Run()是整个Informer的启动入口点
            go informer.Run(stopCh)
            // 设置Informer已经启动的标记
            f.startedInformers[informerType] = true
        }
    }
}
复制代码

3 Informer注册机制(接收器sharedInformerFactory)

• InformerFor()是给内部使用的，其实就是给具体类型的Informer使用的，这也就好理解为何要传入构造函数了。SharedInformerFactory主要就是解决了各种型Informer的共用问题，避免了重复构造。
• 解决了SharedIndexInformer是如何添加到sharedInformerFactory

// 代码源自client-go/informers/factory.go
// InformerFor()至关于每一个类型Informer的构造函数了，即使具体实现构造的地方是使用者提供的
// 这个函数须要使用者传入对象类型，由于在sharedInformerFactory里面是按照对象类型组织的Informer
// 更有趣的是这些Informer不是sharedInformerFactory建立的，须要使用者传入构造函数
// 这样作既保证了每一个类型的Informer只构造一次，同时又保证了具体Informer构造函数的私有化能力
func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
    // 加锁操做
    f.lock.Lock()
    defer f.lock.Unlock()
    // 经过反射获取obj的类型
    informerType := reflect.TypeOf(obj)
    // 看看这个类型的Informer是否已经建立了？
    informer, exists := f.informers[informerType]
    // 若是Informer已经建立，那么就复用这个Informer
    if exists {
        return informer
    }
    // 获取这个类型定制的同步周期，若是定制的同步周期那就用统一的默认周期
    resyncPeriod, exists := f.customResync[informerType]
    if !exists {
        resyncPeriod = f.defaultResync
    }
    // 调用使用者提供构造函数，而后把建立的Informer保存起来
    informer = newFunc(f.client, resyncPeriod)
    
    f.informers[informerType] = informer
 
    return informer
}
// 代码源自client-go/informers/internalinterfaces/factory_interfaces.go
// 这个函数定义就是具体类型Informer的构造函数，后面会有地方说明如何使用
type NewInformerFunc func(kubernetes.Interface, time.Duration) cache.SharedIndexInformer
复制代码

4 POD案例：获取内核Informer的分组(&group)

// 代码源自client-go/informers/factory.go
func (f *sharedInformerFactory) Core() core.Interface {
    // 调用了内核包里面的New()函数，详情见下文分析
    return core.New(f, f.namespace, f.tweakListOptions)
}

// 构造Interface的接口
func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) Interface {
    // 代码也挺简单的，很少说了
    return &group{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
}
复制代码

5 POD案例：获取Core分组中的V1版本

// 代码源自client-go/informers/core/v1/interface.go
// 仍是抽象类
type Interface interface {
    // 获取ComponentStatusInformer
    ComponentStatuses() ComponentStatusInformer
    // 获取ConfigMapInformer
    ConfigMaps() ConfigMapInformer
    // 获取EndpointsInformer
    Endpoints() EndpointsInformer
    // 获取EventInformer
    Events() EventInformer
    // 获取LimitRangeInformer
    LimitRanges() LimitRangeInformer
    // 获取NamespaceInformer
    Namespaces() NamespaceInformer
    // 获取NodeInformer
    Nodes() NodeInformer
    // 获取PersistentVolumeInformer
    PersistentVolumes() PersistentVolumeInformer
    // 获取PersistentVolumeClaimInformer
    PersistentVolumeClaims() PersistentVolumeClaimInformer
    // 获取PodInformer
    Pods() PodInformer
    // 获取PodTemplateInformer
    PodTemplates() PodTemplateInformer
    // 获取ReplicationControllerInformer
    ReplicationControllers() ReplicationControllerInformer
    // 获取ResourceQuotaInformer
    ResourceQuotas() ResourceQuotaInformer
    // 获取SecretInformer
    Secrets() SecretInformer
    // 获取ServiceInformer
    Services() ServiceInformer
    // 获取ServiceAccountInformer
    ServiceAccounts() ServiceAccountInformer
}

// 这个就是上面抽象类的实现了，这个和Core分组的命名都是挺有意思，分组用group做为实现类名
// 这个用version做为实现类名，确实这个是V1版本
type version struct {
    // 工厂的对象指针
    factory          internalinterfaces.SharedInformerFactory
    // 两个选项，很少说了，说了好多遍了
    namespace        string
    tweakListOptions internalinterfaces.TweakListOptionsFunc
}

// 这个就是Core分组V1版本的构造函数啦
func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) Interface {
    // 应该好理解吧？
    return &version{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
}

// 实现V1()这个接口的函数
func (g *group) V1() v1.Interface {
    // 经过调用v1包的New()函数实现的，下面会有相应代码的分析
    return v1.New(g.factory, g.namespace, g.tweakListOptions)
}

// 代码源自client-go/informers/core/interface.go
// Interface又是一个被玩坏的名字，若是没有报名，根本不知道干啥的
type Interface interface {
    V1() v1.Interface // 只有V1一个版本
}
// 这个是Interface的实现类，从名字上没任何关联吧？其实开发者命名也是挺有意思的，Interface定义的是接口
// 供外部使用，group也有意义，由于Core确实是内核Informer的分组
type group struct {
    // 须要工厂对象的指针
    factory          internalinterfaces.SharedInformerFactory
    // 这两个变量决定了Core这个分组对于SharedInformerFactory来讲只有如下两个选项
    namespace        string
    tweakListOptions internalinterfaces.TweakListOptionsFunc
}
复制代码

6 POD案例：PodInformer是经过Core分组Pods()建立

// 代码源自client-go/informers/core/v1/interface.go
// 上面咱们已经说过了version是v1.Interface的实现
func (v *version) Pods() PodInformer {
    // 返回了podInformer的对象，说明podInformer是PodInformer 实现类
    return &podInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
}


// 代码源自client-go/informers/core/v1/pod.go
// PodInformer定义了两个接口，分别为Informer()和Lister()，Informer()用来获取SharedIndexInformer对象
// Lister()用来获取PodLister对象，这个后面会有说明，当前能够不用关心
type PodInformer interface {
    Informer() cache.SharedIndexInformer
    Lister() v1.PodLister
}
// PodInformer的实现类，参数都是上面层层传递下来的，这里不说了
type podInformer struct {
    factory          internalinterfaces.SharedInformerFactory
    tweakListOptions internalinterfaces.TweakListOptionsFunc
    namespace        string
}

// 实现了PodInformer.Informer()接口函数
func (f *podInformer) Informer() cache.SharedIndexInformer {
    // 此处调用了工厂实现了Informer的建立
    return f.factory.InformerFor(&corev1.Pod{}, f.defaultInformer)
}
复制代码

7 POD案例：其余综合扩展

// 代码源自client-go/informers/core/v1/pod.go
// PodInformer定义了两个接口，分别为Informer()和Lister()，Informer()用来获取SharedIndexInformer对象
// Lister()用来获取PodLister对象，这个后面会有说明，当前能够不用关心
// 实现了PodInformer.Lister()接口函数
func (f *podInformer) Lister() v1.PodLister {
    return v1.NewPodLister(f.Informer().GetIndexer())
}
// 真正建立PodInformer的函数
func NewFilteredPodInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
    // 还有谁记得构造SharedIndexInformer须要写啥？本身温习《深刻浅出kubernetes之client-go的SharedInformer》
    return cache.NewSharedIndexInformer(
        // 须要ListWatch两个函数，就是用apiserver的client实现的，此处不重点解释每一个代码什么意思
        // 读者应该可以看懂是利用client实现了Pod的List和Watch
        &cache.ListWatch{
            ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                if tweakListOptions != nil {
                    tweakListOptions(&options)
                }
                return client.CoreV1().Pods(namespace).List(options)
            },
            WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                if tweakListOptions != nil {
                    tweakListOptions(&options)
                }
                return client.CoreV1().Pods(namespace).Watch(options)
            },
        },
        // 这个是要传入对象的类型，确定是Pod了
        &corev1.Pod{},
        // 同步周期
        resyncPeriod,
        // 对象键的计算函数
        indexers,
    )
}
复制代码

8 总结

为了方便管理，Kubernetes对Informer进行了分组，而且SharedInformerFactory能够构造Kubernetes里全部对象的Informer。

专一于大数据及容器云核心技术解密，可提供全栈的大数据+云原平生台咨询方案，请持续关注本套博客。若有任何学术交流，可随时联系。更多内容请关注《数据云技术社区》公众号。