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。

專一於大數據及容器雲核心技術解密，可提供全棧的大數據+雲原平生臺諮詢方案，請持續關注本套博客。若有任何學術交流，可隨時聯繫。更多內容請關注《數據雲技術社區》公衆號。