深刻解析 Kubebuilder：讓編寫 CRD 變得更簡單

時間 2020-05-04

標籤深刻解析 kubebuilder 編寫 crd 變得簡單简体版

原文原文鏈接

做者 | 劉洋(炎尋) 阿里雲高級開發工程師nginx

導讀：自定義資源 CRD（Custom Resource Definition）能夠擴展 Kubernetes API，掌握 CRD 是成爲 Kubernetes 高級玩家的必備技能，本文將介紹 CRD 和 Controller 的概念，並對 CRD 編寫框架 Kubebuilder 進行深刻分析，讓您真正理解並能快速開發 CRD。

概覽

控制器模式與聲明式 API

在正式介紹 Kubebuidler 以前，咱們須要先了解下 K8s 底層實現大量使用的控制器模式，以及讓用戶大呼過癮的聲明式 API，這是介紹 CRDs 和 Kubebuidler 的基礎。
git

控制器模式

K8s 做爲一個「容器編排」平臺，其核心的功能是編排，Pod 做爲 K8s 調度的最小單位,具有不少屬性和字段，K8s 的編排正是經過一個個控制器根據被控制對象的屬性和字段來實現。

下面咱們看一個例子：
github

apiVersion: apps/v1
kind: Deployment
metadata:
  name: test
spec:
  selector:
    matchLabels:
      app: test
  replicas: 2
  template:
    metadata:
      labels:
        app: test
    spec:
      containers:
      - name: nginx
        image: nginx:1.7.9
        ports:
        - containerPort: 80

K8s 集羣在部署時包含了 Controllers 組件，裏面對於每一個 build-in 的資源類型（好比 Deployments, Statefulset, CronJob, ...）都有對應的 Controller，基本是 1:1 的關係。上面的例子中，Deployment 資源建立以後，對應的 Deployment Controller 編排動做很簡單，確保攜帶了 app=test 的 Pod 個數永遠等於 2，Pod 由 template 部分定義，具體來講，K8s 裏面是 kube-controller-manager 這個組件在作這件事，能夠看下 K8s 項目的 pkg/controller 目錄，裏面包含了全部控制器，都以獨有的方式負責某種編排功能，可是它們都遵循一個通用編排模式，即：調諧循環（Reconcile loop），其僞代碼邏輯爲：
segmentfault

for {
actualState := GetResourceActualState(rsvc)
expectState := GetResourceExpectState(rsvc)
if actualState == expectState {
// do nothing
} else {
Reconcile(rsvc)
}
}

就是一個無限循環（實際是事件驅動+定時同步來實現，不是無腦循環）不斷地對比指望狀態和實際狀態，若是有出入則進行 Reconcile（調諧）邏輯將實際狀態調整爲指望狀態。指望狀態就是咱們的對象定義（一般是 YAML 文件），實際狀態是集羣裏面當前的運行狀態（一般來自於 K8s 集羣內外相關資源的狀態彙總），控制器的編排邏輯主要是第三步作的，這個操做被稱爲調諧（Reconcile），整個控制器調諧的過程稱爲「Reconcile Loop」，調諧的最終結果通常是對被控制對象的某種寫操做，好比增/刪/改 Pod。

在控制器中定義被控制對象是經過「模板」完成的，好比 Deployment 裏面的 template 字段裏的內容跟一個標準的 Pod 對象的 API 定義同樣，全部被這個 Deployment 管理的 Pod 實例，都是根據這個 template 字段的建立的，這就是 PodTemplate，一個控制對象的定義通常是由上半部分的控制定義（指望狀態），加上下半部分的被控制對象的模板組成。
api

聲明式 API

所謂聲明式就是「告訴 K8s 你要什麼，而不是告訴它怎麼作的命令」，一個很熟悉的例子就是 SQL，你「告訴 DB 根據條件和各種算子返回數據，而不是告訴它怎麼遍歷，過濾，聚合」。在 K8s 裏面，聲明式的體現就是 kubectl apply 命令，在對象建立和後續更新中一直使用相同的 apply 命令，告訴 K8s 對象的終態便可，底層是經過執行了一個對原有 API 對象的 PATCH 操做來實現的，能夠一次性處理多個寫操做，具有 Merge 能力 diff 出最終的 PATCH，而命令式一次只能處理一個寫請求。

聲明式 API 讓 K8s 的「容器編排」世界看起來溫柔美好，而控制器（以及容器運行時，存儲，網絡模型等）纔是這太平盛世的幕後英雄。說到這裏，就會有人但願也能像 build-in 資源同樣構建本身的自定義資源（CRD-Customize Resource Definition），而後爲自定義資源寫一個對應的控制器，推出本身的聲明式 API。K8s 提供了 CRD 的擴展方式來知足用戶這一需求，並且因爲這種擴展方式十分靈活，在最新的 1.15 版本對 CRD 作了至關大的加強。對於用戶來講，實現 CRD 擴展主要作兩件事：
緩存

編寫 CRD 並將其部署到 K8s 集羣裏；

這一步的做用就是讓 K8s 知道有這個資源及其結構屬性，在用戶提交該自定義資源的定義時（一般是 YAML 文件定義），K8s 可以成功校驗該資源並建立出對應的 Go struct 進行持久化，同時觸發控制器的調諧邏輯。微信

編寫 Controller 並將其部署到 K8s 集羣裏。

這一步的做用就是實現調諧邏輯。

Kubebuilder 就是幫咱們簡化這兩件事的工具，如今咱們開始介紹主角。
網絡

Kubebuilder 是什麼？

摘要

Kubebuilder 是一個使用 CRDs 構建 K8s API 的 SDK，主要是：
app

提供腳手架工具初始化 CRDs 工程，自動生成 boilerplate 代碼和配置；
提供代碼庫封裝底層的 K8s go-client；

方便用戶從零開始開發 CRDs，Controllers 和 Admission Webhooks 來擴展 K8s。
框架

核心概念

GVKs&GVRs

GVK = GroupVersionKind，GVR = GroupVersionResource。

API Group & Versions（GV）

API Group 是相關 API 功能的集合，每一個 Group 擁有一或多個 Versions，用於接口的演進。

Kinds & Resources

每一個 GV 都包含多個 API 類型，稱爲 Kinds，在不一樣的 Versions 之間同一個 Kind 定義可能不一樣， Resource 是 Kind 的對象標識（resource type），通常來講 Kinds 和 Resources 是 1:1 的，好比 pods Resource 對應 Pod Kind，可是有時候相同的 Kind 可能對應多個 Resources，好比 Scale Kind 可能對應不少 Resources：deployments/scale，replicasets/scale，對於 CRD 來講，只會是 1:1 的關係。

每個 GVK 都關聯着一個 package 中給定的 root Go type，好比 apps/v1/Deployment 就關聯着 K8s 源碼裏面 k8s.io/api/apps/v1 package 中的 Deployment struct，咱們提交的各種資源定義 YAML 文件都須要寫：

apiVersion：這個就是 GV 。
kind：這個就是 K。

根據 GVK K8s 就能找到你到底要建立什麼類型的資源，根據你定義的 Spec 建立好資源以後就成爲了 Resource，也就是 GVR。GVK/GVR 就是 K8s 資源的座標，是咱們建立/刪除/修改/讀取資源的基礎。

Scheme

每一組 Controllers 都須要一個 Scheme，提供了 Kinds 與對應 Go types 的映射，也就是說給定 Go type 就知道他的 GVK，給定 GVK 就知道他的 Go type，好比說咱們給定一個 Scheme: "tutotial.kubebuilder.io/api/v1".CronJob{} 這個 Go type 映射到 batch.tutotial.kubebuilder.io/v1 的 CronJob GVK，那麼從 Api Server 獲取到下面的 JSON:

{
    "kind": "CronJob",
    "apiVersion": "batch.tutorial.kubebuilder.io/v1",
    ...
}

就能構造出對應的 Go type了，經過這個 Go type 也能正確地獲取 GVR 的一些信息，控制器能夠經過該 Go type 獲取到指望狀態以及其餘輔助信息進行調諧邏輯。

Manager

Kubebuilder 的核心組件，具備 3 個職責：

負責運行全部的 Controllers；
初始化共享 caches，包含 listAndWatch 功能；
初始化 clients 用於與 Api Server 通訊。

Cache

Kubebuilder 的核心組件，負責在 Controller 進程裏面根據 Scheme 同步 Api Server 中全部該 Controller 關心 GVKs 的 GVRs，其核心是 GVK -> Informer 的映射，Informer 會負責監聽對應 GVK 的 GVRs 的建立/刪除/更新操做，以觸發 Controller 的 Reconcile 邏輯。

Controller

Kubebuidler 爲咱們生成的腳手架文件，咱們只須要實現 Reconcile 方法便可。

Clients

在實現 Controller 的時候不可避免地須要對某些資源類型進行建立/刪除/更新，就是經過該 Clients 實現的，其中查詢功能實際查詢是本地的 Cache，寫操做直接訪問 Api Server。

Index

因爲 Controller 常常要對 Cache 進行查詢，Kubebuilder 提供 Index utility 給 Cache 加索引提高查詢效率。

Finalizer

在通常狀況下，若是資源被刪除以後，咱們雖然可以被觸發刪除事件，可是這個時候從 Cache 裏面沒法讀取任何被刪除對象的信息，這樣一來，致使不少垃圾清理工做由於信息不足沒法進行，K8s 的 Finalizer 字段用於處理這種狀況。在 K8s 中，只要對象 ObjectMeta 裏面的 Finalizers 不爲空，對該對象的 delete 操做就會轉變爲 update 操做，具體說就是 update deletionTimestamp 字段，其意義就是告訴 K8s 的 GC「在deletionTimestamp 這個時刻以後，只要 Finalizers 爲空，就立馬刪除掉該對象」。

因此通常的使用姿式就是在建立對象時把 Finalizers 設置好（任意 string），而後處理 DeletionTimestamp 不爲空的 update 操做（實際是 delete），根據 Finalizers 的值執行完全部的 pre-delete hook（此時能夠在 Cache 裏面讀取到被刪除對象的任何信息）以後將 Finalizers 置爲空便可。

OwnerReference

K8s GC 在刪除一個對象時，任何 ownerReference 是該對象的對象都會被清除，與此同時，Kubebuidler 支持全部對象的變動都會觸發 Owner 對象 controller 的 Reconcile 方法。

全部概念集合在一塊兒如圖 1 所示：

圖 1-Kubebuilder 核心概念

Kubebuilder 怎麼用？

1. 建立腳手架工程

kubebuilder init --domain edas.io

這一步建立了一個 Go module 工程，引入了必要的依賴，建立了一些模板文件。

2. 建立 API

kubebuilder create api --group apps --version v1alpha1 --kind Application

這一步建立了對應的 CRD 和 Controller 模板文件，通過一、2 兩步，現有的工程結構如圖 2 所示：

圖 2-Kubebuilder 生成的工程結構說明

3. 定義 CRD

在圖 2 中對應的文件定義 Spec 和 Status。

4. 編寫 Controller 邏輯

在圖 3 中對應的文件實現 Reconcile 邏輯。

5. 測試發佈

本地測試完以後使用 Kubebuilder 的 Makefile 構建鏡像，部署咱們的 CRDs 和 Controller 便可。

Kubebuilder 出現的意義？

讓擴展 K8s 變得更簡單，K8s 擴展的方式不少，Kubebuilder 目前專一於 CRD 擴展方式。

深刻

在使用 Kubebuilder 的過程當中有些問題困擾着我：

如何同步自定義資源以及 K8s build-in 資源？
Controller 的 Reconcile 方法是如何被觸發的？
Cache 的工做原理是什麼？
...

帶着這些問題咱們去看看源碼 :D。

源碼閱讀

從 main.go 開始

Kubebuilder 建立的 main.go 是整個項目的入口，邏輯十分簡單：

var (
    scheme   = runtime.NewScheme()
    setupLog = ctrl.Log.WithName("setup")
)
func init() {
    appsv1alpha1.AddToScheme(scheme)
    // +kubebuilder:scaffold:scheme
}
func main() {
    ...
        // 一、init Manager
    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{Scheme: scheme, MetricsBindAddress: metricsAddr})
    if err != nil {
        setupLog.Error(err, "unable to start manager")
        os.Exit(1)
    }
        // 二、init Reconciler（Controller）
    err = (&controllers.ApplicationReconciler{
        Client: mgr.GetClient(),
        Log:    ctrl.Log.WithName("controllers").WithName("Application"),
        Scheme: mgr.GetScheme(),
    }).SetupWithManager(mgr)
    if err != nil {
        setupLog.Error(err, "unable to create controller", "controller", "EDASApplication")
        os.Exit(1)
    }
    // +kubebuilder:scaffold:builder
    setupLog.Info("starting manager")
        // 三、start Manager
    if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
        setupLog.Error(err, "problem running manager")
        os.Exit(1)
    }

能夠看到在 init 方法裏面咱們將 appsv1alpha1 註冊到 Scheme 裏面去了，這樣一來 Cache 就知道 watch 誰了，main 方法裏面的邏輯基本都是 Manager 的：

初始化了一個 Manager；
將 Manager 的 Client 傳給 Controller，而且調用 SetupWithManager 方法傳入 Manager 進行 Controller 的初始化；
啓動 Manager。

咱們的核心就是看這 3 個流程。

Manager 初始化

Manager 初始化代碼以下：

// New returns a new Manager for creating Controllers.
func New(config *rest.Config, options Options) (Manager, error) {
    ...
    // Create the cache for the cached read client and registering informers
    cache, err := options.NewCache(config, cache.Options{Scheme: options.Scheme, Mapper: mapper, Resync: options.SyncPeriod, Namespace: options.Namespace})
    if err != nil {
        return nil, err
    }
    apiReader, err := client.New(config, client.Options{Scheme: options.Scheme, Mapper: mapper})
    if err != nil {
        return nil, err
    }
    writeObj, err := options.NewClient(cache, config, client.Options{Scheme: options.Scheme, Mapper: mapper})
    if err != nil {
        return nil, err
    }
    ...
    return &controllerManager{
        config:           config,
        scheme:           options.Scheme,
        errChan:          make(chan error),
        cache:            cache,
        fieldIndexes:     cache,
        client:           writeObj,
        apiReader:        apiReader,
        recorderProvider: recorderProvider,
        resourceLock:     resourceLock,
        mapper:           mapper,
        metricsListener:  metricsListener,
        internalStop:     stop,
        internalStopper:  stop,
        port:             options.Port,
        host:             options.Host,
        leaseDuration:    *options.LeaseDuration,
        renewDeadline:    *options.RenewDeadline,
        retryPeriod:      *options.RetryPeriod,
    }, nil
}

能夠看到主要是建立 Cache 與 Clients：

建立 Cache

Cache 初始化代碼以下：

// New initializes and returns a new Cache.
func New(config *rest.Config, opts Options) (Cache, error) {
    opts, err := defaultOpts(config, opts)
    if err != nil {
        return nil, err
    }
    im := internal.NewInformersMap(config, opts.Scheme, opts.Mapper, *opts.Resync, opts.Namespace)
    return &informerCache{InformersMap: im}, nil
}
// newSpecificInformersMap returns a new specificInformersMap (like
// the generical InformersMap, except that it doesn't implement WaitForCacheSync).
func newSpecificInformersMap(...) *specificInformersMap {
    ip := &specificInformersMap{
        Scheme:            scheme,
        mapper:            mapper,
        informersByGVK:    make(map[schema.GroupVersionKind]*MapEntry),
        codecs:            serializer.NewCodecFactory(scheme),
        resync:            resync,
        createListWatcher: createListWatcher,
        namespace:         namespace,
    }
    return ip
}
// MapEntry contains the cached data for an Informer
type MapEntry struct {
    // Informer is the cached informer
    Informer cache.SharedIndexInformer
    // CacheReader wraps Informer and implements the CacheReader interface for a single type
    Reader CacheReader
}
func createUnstructuredListWatch(gvk schema.GroupVersionKind, ip *specificInformersMap) (*cache.ListWatch, error) {
        ...
    // Create a new ListWatch for the obj
    return &cache.ListWatch{
        ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
            if ip.namespace != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot {
                return dynamicClient.Resource(mapping.Resource).Namespace(ip.namespace).List(opts)
            }
            return dynamicClient.Resource(mapping.Resource).List(opts)
        },
        // Setup the watch function
        WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
            // Watch needs to be set to true separately
            opts.Watch = true
            if ip.namespace != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot {
                return dynamicClient.Resource(mapping.Resource).Namespace(ip.namespace).Watch(opts)
            }
            return dynamicClient.Resource(mapping.Resource).Watch(opts)
        },
    }, nil
}

能夠看到 Cache 主要就是建立了 InformersMap，Scheme 裏面的每一個 GVK 都建立了對應的 Informer，經過 informersByGVK 這個 map 作 GVK 到 Informer 的映射，每一個 Informer 會根據 ListWatch 函數對對應的 GVK 進行 List 和 Watch。

建立 Clients

建立 Clients 很簡單：

// defaultNewClient creates the default caching client
func defaultNewClient(cache cache.Cache, config *rest.Config, options client.Options) (client.Client, error) {
    // Create the Client for Write operations.
    c, err := client.New(config, options)
    if err != nil {
        return nil, err
    }
    return &client.DelegatingClient{
        Reader: &client.DelegatingReader{
            CacheReader:  cache,
            ClientReader: c,
        },
        Writer:       c,
        StatusClient: c,
    }, nil
}

讀操做使用上面建立的 Cache，寫操做使用 K8s go-client 直連。

Controller 初始化

下面看看 Controller 的啓動：

func (r *EDASApplicationReconciler) SetupWithManager(mgr ctrl.Manager) error {
    err := ctrl.NewControllerManagedBy(mgr).
        For(&appsv1alpha1.EDASApplication{}).
        Complete(r)
return err
}

使用的是 Builder 模式，NewControllerManagerBy 和 For 方法都是給 Builder 傳參，最重要的是最後一個方法 Complete，其邏輯是：

func (blder *Builder) Build(r reconcile.Reconciler) (manager.Manager, error) {
...
    // Set the Manager
    if err := blder.doManager(); err != nil {
        return nil, err
    }
    // Set the ControllerManagedBy
    if err := blder.doController(r); err != nil {
        return nil, err
    }
    // Set the Watch
    if err := blder.doWatch(); err != nil {
        return nil, err
    }
...
    return blder.mgr, nil
}

主要是看看 doController 和 doWatch 方法：

doController 方法

func New(name string, mgr manager.Manager, options Options) (Controller, error) {
    if options.Reconciler == nil {
        return nil, fmt.Errorf("must specify Reconciler")
    }
    if len(name) == 0 {
        return nil, fmt.Errorf("must specify Name for Controller")
    }
    if options.MaxConcurrentReconciles <= 0 {
        options.MaxConcurrentReconciles = 1
    }
    // Inject dependencies into Reconciler
    if err := mgr.SetFields(options.Reconciler); err != nil {
        return nil, err
    }
    // Create controller with dependencies set
    c := &controller.Controller{
        Do:                      options.Reconciler,
        Cache:                   mgr.GetCache(),
        Config:                  mgr.GetConfig(),
        Scheme:                  mgr.GetScheme(),
        Client:                  mgr.GetClient(),
        Recorder:                mgr.GetEventRecorderFor(name),
        Queue:                   workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), name),
        MaxConcurrentReconciles: options.MaxConcurrentReconciles,
        Name:                    name,
    }
    // Add the controller as a Manager components
    return c, mgr.Add(c)
}

該方法初始化了一個 Controller，傳入了一些很重要的參數：

Do：Reconcile 邏輯；
Cache：找 Informer 註冊 Watch；
Client：對 K8s 資源進行 CRUD；
Queue：Watch 資源的 CUD 事件緩存；
Recorder：事件收集。

doWatch 方法

func (blder *Builder) doWatch() error {
    // Reconcile type
    src := &source.Kind{Type: blder.apiType}
    hdler := &handler.EnqueueRequestForObject{}
    err := blder.ctrl.Watch(src, hdler, blder.predicates...)
    if err != nil {
        return err
    }
    // Watches the managed types
    for _, obj := range blder.managedObjects {
        src := &source.Kind{Type: obj}
        hdler := &handler.EnqueueRequestForOwner{
            OwnerType:    blder.apiType,
            IsController: true,
        }
        if err := blder.ctrl.Watch(src, hdler, blder.predicates...); err != nil {
            return err
        }
    }
    // Do the watch requests
    for _, w := range blder.watchRequest {
        if err := blder.ctrl.Watch(w.src, w.eventhandler, blder.predicates...); err != nil {
            return err
        }
    }
    return nil
}

能夠看到該方法對本 Controller 負責的 CRD 進行了 watch，同時底下還會 watch 本 CRD 管理的其餘資源，這個 managedObjects 能夠經過 Controller 初始化 Buidler 的 Owns 方法傳入，說到 Watch 咱們關心兩個邏輯：

註冊的 handler

type EnqueueRequestForObject struct{}
// Create implements EventHandler
func (e *EnqueueRequestForObject) Create(evt event.CreateEvent, q workqueue.RateLimitingInterface) {
        ...
    q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
        Name:      evt.Meta.GetName(),
        Namespace: evt.Meta.GetNamespace(),
    }})
}
// Update implements EventHandler
func (e *EnqueueRequestForObject) Update(evt event.UpdateEvent, q workqueue.RateLimitingInterface) {
    if evt.MetaOld != nil {
        q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
            Name:      evt.MetaOld.GetName(),
            Namespace: evt.MetaOld.GetNamespace(),
        }})
    } else {
        enqueueLog.Error(nil, "UpdateEvent received with no old metadata", "event", evt)
    }
    if evt.MetaNew != nil {
        q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
            Name:      evt.MetaNew.GetName(),
            Namespace: evt.MetaNew.GetNamespace(),
        }})
    } else {
        enqueueLog.Error(nil, "UpdateEvent received with no new metadata", "event", evt)
    }
}
// Delete implements EventHandler
func (e *EnqueueRequestForObject) Delete(evt event.DeleteEvent, q workqueue.RateLimitingInterface) {
        ...
    q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
        Name:      evt.Meta.GetName(),
        Namespace: evt.Meta.GetNamespace(),
    }})
}

能夠看到 Kubebuidler 爲咱們註冊的 Handler 就是將發生變動的對象的 NamespacedName 入隊列，若是在 Reconcile 邏輯中須要判斷建立/更新/刪除，須要有本身的判斷邏輯。

註冊的流程

// Watch implements controller.Controller
func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prct ...predicate.Predicate) error {
    ...
    log.Info("Starting EventSource", "controller", c.Name, "source", src)
    return src.Start(evthdler, c.Queue, prct...)
}
// Start is internal and should be called only by the Controller to register an EventHandler with the Informer
// to enqueue reconcile.Requests.
func (is *Informer) Start(handler handler.EventHandler, queue workqueue.RateLimitingInterface,
    ...
    is.Informer.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
    return nil
}

咱們的 Handler 實際註冊到 Informer 上面，這樣整個邏輯就串起來了，經過 Cache 咱們建立了全部 Scheme 裏面 GVKs 的 Informers，而後對應 GVK 的 Controller 註冊了 Watch Handler 到對應的 Informer，這樣一來對應的 GVK 裏面的資源有變動都會觸發 Handler，將變動事件寫到 Controller 的事件隊列中，以後觸發咱們的 Reconcile 方法。

Manager 啓動

func (cm *controllerManager) Start(stop <-chan struct{}) error {
    ...
    go cm.startNonLeaderElectionRunnables()
    ...
}
func (cm *controllerManager) startNonLeaderElectionRunnables() {
    ...
    // Start the Cache. Allow the function to start the cache to be mocked out for testing
    if cm.startCache == nil {
        cm.startCache = cm.cache.Start
    }
    go func() {
        if err := cm.startCache(cm.internalStop); err != nil {
            cm.errChan <- err
        }
    }()
        ...
        // Start Controllers
    for _, c := range cm.nonLeaderElectionRunnables {
        ctrl := c
        go func() {
            cm.errChan <- ctrl.Start(cm.internalStop)
        }()
    }
    cm.started = true
}

主要就是啓動 Cache，Controller，將整個事件流運轉起來，咱們下面來看看啓動邏輯。

Cache 啓動

func (ip *specificInformersMap) Start(stop <-chan struct{}) {
    func() {
        ...
        // Start each informer
        for _, informer := range ip.informersByGVK {
            go informer.Informer.Run(stop)
        }
    }()
}
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
        ...
        // informer push resource obj CUD delta to this fifo queue
    fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer)
    cfg := &Config{
        Queue:            fifo,
        ListerWatcher:    s.listerWatcher,
        ObjectType:       s.objectType,
        FullResyncPeriod: s.resyncCheckPeriod,
        RetryOnError:     false,
        ShouldResync:     s.processor.shouldResync,
                // handler to process delta
        Process: s.HandleDeltas,
    }
    func() {
        s.startedLock.Lock()
        defer s.startedLock.Unlock()
                // this is internal controller process delta generate by reflector
        s.controller = New(cfg)
        s.controller.(*controller).clock = s.clock
        s.started = true
    }()
        ...
    wg.StartWithChannel(processorStopCh, s.processor.run)
    s.controller.Run(stopCh)
}
func (c *controller) Run(stopCh <-chan struct{}) {
    ...
    r := NewReflector(
        c.config.ListerWatcher,
        c.config.ObjectType,
        c.config.Queue,
        c.config.FullResyncPeriod,
    )
    ...
        // reflector is delta producer
    wg.StartWithChannel(stopCh, r.Run)
        // internal controller's processLoop is comsume logic
    wait.Until(c.processLoop, time.Second, stopCh)
}

Cache 的初始化核心是初始化全部的 Informer，Informer 的初始化核心是建立了 reflector 和內部 controller，reflector 負責監聽 Api Server 上指定的 GVK，將變動寫入 delta 隊列中，能夠理解爲變動事件的生產者，內部 controller 是變動事件的消費者，他會負責更新本地 indexer，以及計算出 CUD 事件推給咱們以前註冊的 Watch Handler。

Controller 啓動

// Start implements controller.Controller
func (c *Controller) Start(stop <-chan struct{}) error {
    ...
    for i := 0; i < c.MaxConcurrentReconciles; i++ {
        // Process work items
        go wait.Until(func() {
            for c.processNextWorkItem() {
            }
        }, c.JitterPeriod, stop)
    }
    ...
}
func (c *Controller) processNextWorkItem() bool {
    ...
    obj, shutdown := c.Queue.Get()
    ...
    var req reconcile.Request
    var ok bool
    if req, ok = obj.(reconcile.Request); 
        ...
    // RunInformersAndControllers the syncHandler, passing it the namespace/Name string of the
    // resource to be synced.
    if result, err := c.Do.Reconcile(req); err != nil {
        c.Queue.AddRateLimited(req)
        ...
    } 
        ...
}

Controller 的初始化是啓動 goroutine 不斷地查詢隊列，若是有變動消息則觸發到咱們自定義的 Reconcile 邏輯。

總體邏輯串連

上面咱們經過源碼閱讀已經十分清楚整個流程，可是正所謂一圖勝千言，我製做了一張總體邏輯串連圖（圖 3）來幫助你們理解：

圖 3-Kubebuidler 總體邏輯串連圖

Kubebuilder 做爲腳手架工具已經爲咱們作了不少，到最後咱們只須要實現 Reconcile 方法便可，這裏再也不贅述。

守得雲開見月明

剛開始使用 Kubebuilder 的時候，由於封裝程度很高，不少事情都是懵逼狀態，剖析完以後不少問題就很明白了，好比開頭提出的幾個：

如何同步自定義資源以及 K8s build-in 資源？

須要將自定義資源和想要 Watch 的 K8s build-in 資源的 GVKs 註冊到 Scheme 上，Cache 會自動幫咱們同步。

Controller 的 Reconcile 方法是如何被觸發的？

經過 Cache 裏面的 Informer 獲取資源的變動事件，而後經過兩個內置的 Controller 以生產者消費者模式傳遞事件，最終觸發 Reconcile 方法。

Cache 的工做原理是什麼？

GVK -> Informer 的映射，Informer 包含 Reflector 和 Indexer 來作事件監聽和本地緩存。

還有不少問題我就不一一說了，總之，如今 Kubebuilder 如今再也不是黑盒。

同類工具對比

Operator Framework 與 Kubebuilder 很相似，這裏由於篇幅關係再也不展開。

最佳實踐

模式

使用 OwnerRefrence 來作資源關聯，有兩個特性：

Owner 資源被刪除，被 Own 的資源會被級聯刪除，這利用了 K8s 的 GC；
被 Own 的資源對象的事件變動能夠觸發 Owner 對象的 Reconcile 方法；

使用 Finalizer 來作資源的清理。

注意點

不使用 Finalizer 時，資源被刪除沒法獲取任何信息；
對象的 Status 字段變化也會觸發 Reconcile 方法；
Reconcile 邏輯須要冪等；

優化

使用 IndexFunc 來優化資源查詢的效率

總結

經過深刻分析，咱們能夠看到 Kubebuilder 提供的功能對於快速編寫 CRD 和 Controller 是十分有幫助的，不管是 Istio、Knative 等知名項目仍是各類自定義 Operators，都大量使用了 CRD，將各類組件抽象爲 CRD，Kubernetes 變成控制面板將成爲一個趨勢，但願本文可以幫助你們理解和把握這個趨勢。

「阿里巴巴雲原生微信公衆號（ID：Alicloudnative）關注微服務、Serverless、容器、Service Mesh等技術領域、聚焦雲原生流行技術趨勢、雲原生大規模的落地實踐，作最懂雲原生開發者的技術公衆號。」

搜索「阿里巴巴雲原生公衆號」獲取更多K8s容器技術內容

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。