[[399769]]

在前面的文章當中我們已經完整的完成了一個 Operator 的開發，涉及到了 CURD、預刪除、Status、Event、OwnerReference、WebHook，也算是將一個 Operator 開發中會涉及到的點大部分都了解了一下。kubebuilder 幫我們做了很多事情，讓我們的開發基本上只需要關注一個 Reconcile 函數就可以了，但是從另外一個方面來講，kubebuilder 目前對我們來說它還是一個黑盒，會產生很多的疑問：

• Reconcile 方法是怎么被觸發的?
• 怎么識別到不同的資源?
• 整體是如何進行工作的?
• ……

架構

我們先來看一下來自官方文檔的這個架構圖[1]

arch

• Process 進程通過 main.go啟動，一般來說一個 Controller 只有一個進程，如果做了高可用的話，會有多個
• Manager 每個進程會有一個 Manager，這是核心組件，主要負責
  • metrics 的暴露
  • webhook 證書
  • 初始化共享的 cache
  • 初始化共享的 clients 用于和 APIServer 進行通信
  • 所有的 Controller 的運行
• Client 一般來說，我們 創建、更新、刪除某個資源的時候會直接調用 Client 和 APIServer 進行通信
• Cache 負責同步 Controller 關心的資源，其核心是 GVK -> Informer 的映射，一般我們的 Get 和 List 操作都會從 Cache 中獲取數據
• Controller 控制器的業務邏輯所在的地方，一個 Manager 可能會有多個 Controller，我們一般只需要實現 Reconcile 方法就行。圖上的 Predicate 是事件過濾器，我們可以在 Controller 中過濾掉我們不關心的事件信息
• WebHook 就是我們準入控制實現的地方了，主要是有兩類接口，一個是 MutatingAdmissionWebhook 需要實現 Defaulter 接口，一個是 ValidatingAdmissionWebhook 需要實現 Validator 接口

源碼分析

了解了基本的架構之后，我們就從入口 main.go 開始，看一看 kubebuilder 究竟在后面偷偷的做了哪些事情吧。

main.go

// 省略了參數綁定和 error check 的代碼 
func main() { 
    var metricsAddr string 
    var enableLeaderElection bool 
    var probeAddr string 
 
    ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts))) 
 
    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{ 
        Scheme:                 scheme, 
        MetricsBindAddress:     metricsAddr, 
        Port:                   9443, 
        HealthProbeBindAddress: probeAddr, 
        LeaderElection:         enableLeaderElection, 
        LeaderElectionID:       "97acaccf.lailin.xyz", 
        // CertDir:                "config/cert/", // 手動指定證書位置用于測試 
    }) 
     
 
    (&controllers.NodePoolReconciler{ 
        Client:   mgr.GetClient(), 
        Log:      ctrl.Log.WithName("controllers").WithName("NodePool"), 
        Scheme:   mgr.GetScheme(), 
        Recorder: mgr.GetEventRecorderFor("NodePool"), 
    }).SetupWithManager(mgr) 
 
    (&nodesv1.NodePool{}).SetupWebhookWithManager(mgr) 
   
    //+kubebuilder:scaffold:builder 
 
    mgr.AddHealthzCheck("healthz", healthz.Ping) 
    mgr.AddReadyzCheck("readyz", healthz.Ping) 
 
    setupLog.Info("starting manager") 
    mgr.Start(ctrl.SetupSignalHandler()) 
} 

可以看到 main.go 主要是做了一些啟動的工作包括：

• 創建一個 Manager
• 使用剛剛創建的 Manager 創建了一個 Controller
• 啟動 WebHook
• 添加健康檢查
• 啟動 Manager

下面我們就順著 main 函數里面的邏輯一步步的往下看看

NewManger

// New returns a new Manager for creating Controllers. 
func New(config *rest.Config, options Options) (Manager, error) { 
    // 省略配置初始化相關代碼 
 
    // 創建 cache 
    cache, err := options.NewCache(config,  
                                 cache.Options{ 
                                   Scheme: options.Scheme, // main 中傳入的 scheme 
                                   Mapper: mapper,         // k8s api 和 go type 的轉換器 
                                   Resync: options.SyncPeriod, // 默認 10 小時，一般不要改 
                                   Namespace: options.Namespace, // 需要監聽的 namespace 
                                 }) 
 
  // 創建和 APIServer 交互的 client，讀寫分離 
    clientOptions := client.Options{Scheme: options.Scheme, Mapper: mapper} 
    apiReader, err := client.New(config, clientOptions) 
 
 
    writeObj, err := options.ClientBuilder. 
        WithUncached(options.ClientDisableCacheFor...). 
        Build(cache, config, clientOptions) 
 
    if options.DryRunClient { 
        writeObj = client.NewDryRunClient(writeObj) 
    } 
 
    // 創建事件記錄器 
    recorderProvider, err := options.newRecorderProvider(config, options.Scheme, options.Logger.WithName("events"), options.makeBroadcaster) 
 
    // 需要需要高可用的話，創建選舉相關的配置 
    leaderConfig := config 
    if options.LeaderElectionConfig != nil { 
        leaderConfig = options.LeaderElectionConfig 
    } 
    resourceLock, err := options.newResourceLock(leaderConfig, recorderProvider, leaderelection.Options{ 
        LeaderElection:             options.LeaderElection, 
        LeaderElectionResourceLock: options.LeaderElectionResourceLock, 
        LeaderElectionID:           options.LeaderElectionID, 
        LeaderElectionNamespace:    options.LeaderElectionNamespace, 
    }) 
 
    // 創建 metric 和 健康檢查的接口 
    metricsListener, err := options.newMetricsListener(options.MetricsBindAddress) 
 
    // By default we have no extra endpoints to expose on metrics http server. 
    metricsExtraHandlers := make(map[string]http.Handler) 
 
    // Create health probes listener. This will throw an error if the bind 
    // address is invalid or already in use. 
    healthProbeListener, err := options.newHealthProbeListener(options.HealthProbeBindAddress) 
    if err != nil { 
        return nil, err 
    } 
 
  // 最后將這些配置放到 manager 中 
    return &controllerManager{ 
        config:                  config, 
        scheme:                  options.Scheme, 
        cache:                   cache, 
        fieldIndexes:            cache, 
        client:                  writeObj, 
        apiReader:               apiReader, 
        recorderProvider:        recorderProvider, 
        resourceLock:            resourceLock, 
        mapper:                  mapper, 
        metricsListener:         metricsListener, 
        metricsExtraHandlers:    metricsExtraHandlers, 
        logger:                  options.Logger, 
        elected:                 make(chan struct{}), 
        port:                    options.Port, 
        host:                    options.Host, 
        certDir:                 options.CertDir, 
        leaseDuration:           *options.LeaseDuration, 
        renewDeadline:           *options.RenewDeadline, 
        retryPeriod:             *options.RetryPeriod, 
        healthProbeListener:     healthProbeListener, 
        readinessEndpointName:   options.ReadinessEndpointName, 
        livenessEndpointName:    options.LivenessEndpointName, 
        gracefulShutdownTimeout: *options.GracefulShutdownTimeout, 
        internalProceduresStop:  make(chan struct{}), 
    }, nil 
} 

創建 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 
} 

這里主要是調用 NewInformersMap方法創建 Informer 的映射

func NewInformersMap(config *rest.Config, 
    scheme *runtime.Scheme, 
    mapper meta.RESTMapper, 
    resync time.Duration, 
    namespace string) *InformersMap { 
 
    return &InformersMap{ 
        structured:   newStructuredInformersMap(config, scheme, mapper, resync, namespace), 
        unstructured: newUnstructuredInformersMap(config, scheme, mapper, resync, namespace), 
        metadata:     newMetadataInformersMap(config, scheme, mapper, resync, namespace), 
 
        Scheme: scheme, 
    } 
} 

NewInformersMap會去分別創建，結構化、非結構化以及 metadata 的 InformerMap 而這些方法最后都會去調用 newSpecificInformersMap方法，區別就是不同的方法傳入的 createListWatcherFunc 參數不同

func newSpecificInformersMap(config *rest.Config, 
    scheme *runtime.Scheme, 
    mapper meta.RESTMapper, 
    resync time.Duration, 
    namespace string, 
    createListWatcher createListWatcherFunc) *specificInformersMap { 
    ip := &specificInformersMap{ 
        config:            config, 
        Scheme:            scheme, 
        mapper:            mapper, 
        informersByGVK:    make(map[schema.GroupVersionKind]*MapEntry), 
        codecs:            serializer.NewCodecFactory(scheme), 
        paramCodec:        runtime.NewParameterCodec(scheme), 
        resync:            resync, 
        startWait:         make(chan struct{}), 
        createListWatcher: createListWatcher, 
        namespace:         namespace, 
    } 
    return ip 
} 

newSpecificInformersMap 和常規的 InformersMap 類似，區別是沒實現 WaitForCacheSync方法

以結構化的傳入的 createStructuredListWatch 為例，主要是返回一個用于創建 SharedIndexInformer 的 ListWatch 對象

func createStructuredListWatch(gvk schema.GroupVersionKind, ip *specificInformersMap) (*cache.ListWatch, error) { 
 // Kubernetes APIs work against Resources, not GroupVersionKinds.  Map the 
 // groupVersionKind to the Resource API we will use. 
 mapping, err := ip.mapper.RESTMapping(gvk.GroupKind(), gvk.Version) 
 if err != nil { 
  return nil, err 
 } 
 
 client, err := apiutil.RESTClientForGVK(gvk, false, ip.config, ip.codecs) 
 if err != nil { 
  return nil, err 
 } 
 listGVK := gvk.GroupVersion().WithKind(gvk.Kind + "List") 
 listObj, err := ip.Scheme.New(listGVK) 
 if err != nil { 
  return nil, err 
 } 
 
 // TODO: the functions that make use of this ListWatch should be adapted to 
 //  pass in their own contexts instead of relying on this fixed one here. 
 ctx := context.TODO() 
 // Create a new ListWatch for the obj 
 return &cache.ListWatch{ 
  ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) { 
   res := listObj.DeepCopyObject() 
   isNamespaceScoped := ip.namespace != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot 
   err := client.Get().NamespaceIfScoped(ip.namespace, isNamespaceScoped).Resource(mapping.Resource.Resource).VersionedParams(&opts, ip.paramCodec).Do(ctx).Into(res) 
   return res, err 
  }, 
  // Setup the watch function 
  WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) { 
   // Watch needs to be set to true separately 
   opts.Watch = true 
   isNamespaceScoped := ip.namespace != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot 
   return client.Get().NamespaceIfScoped(ip.namespace, isNamespaceScoped).Resource(mapping.Resource.Resource).VersionedParams(&opts, ip.paramCodec).Watch(ctx) 
  }, 
 }, nil 
} 

小結: cache 主要是創建了一些 InformerMap，完成了 GVK 到 Informer 的映射，每個 Informer 會根據 ListWatch 函數對對應的 GVK 進行 List 和 Watch。

創建 Client

func New(config *rest.Config, options Options) (Client, error) { 
 if config == nil { 
  return nil, fmt.Errorf("must provide non-nil rest.Config to client.New") 
 } 
 
 // Init a scheme if none provided 
 if options.Scheme == nil { 
  options.Scheme = scheme.Scheme 
 } 
 
 // Init a Mapper if none provided 
 if options.Mapper == nil { 
  var err error 
  options.Mapper, err = apiutil.NewDynamicRESTMapper(config) 
  if err != nil { 
   return nil, err 
  } 
 } 
 
 clientcache := &clientCache{ 
  config: config, 
  scheme: options.Scheme, 
  mapper: options.Mapper, 
  codecs: serializer.NewCodecFactory(options.Scheme), 
 
  structuredResourceByType:   make(map[schema.GroupVersionKind]*resourceMeta), 
  unstructuredResourceByType: make(map[schema.GroupVersionKind]*resourceMeta), 
 } 
 
 rawMetaClient, err := metadata.NewForConfig(config) 
 if err != nil { 
  return nil, fmt.Errorf("unable to construct metadata-only client for use as part of client: %w", err) 
 } 
 
 c := &client{ 
  typedClient: typedClient{ 
   cache:      clientcache, 
   paramCodec: runtime.NewParameterCodec(options.Scheme), 
  }, 
  unstructuredClient: unstructuredClient{ 
   cache:      clientcache, 
   paramCodec: noConversionParamCodec{}, 
  }, 
  metadataClient: metadataClient{ 
   client:     rawMetaClient, 
   restMapper: options.Mapper, 
  }, 
  scheme: options.Scheme, 
  mapper: options.Mapper, 
 } 
 
 return c, nil 
} 

client 創建了兩個一個用于讀，一個用于寫，用于讀的會直接使用上面的 cache，用于寫的才會直接和 APIServer 進行交互

Controller

下面我們看一下核心的 Controller 是怎么初始化和工作的

if err = (&controllers.NodePoolReconciler{ 
  Client:   mgr.GetClient(), 
  Log:      ctrl.Log.WithName("controllers").WithName("NodePool"), 
  Scheme:   mgr.GetScheme(), 
  Recorder: mgr.GetEventRecorderFor("NodePool"), 
}).SetupWithManager(mgr); err != nil { 
  setupLog.Error(err, "unable to create controller", "controller", "NodePool") 
  os.Exit(1) 
} 

main.go 的方法里面主要是初始化了 Controller 的結構體，然后調用了 SetupWithManager方法

// SetupWithManager sets up the controller with the Manager. 
func (r *NodePoolReconciler) SetupWithManager(mgr ctrl.Manager) error { 
    return ctrl.NewControllerManagedBy(mgr). 
        For(&nodesv1.NodePool{}). 
        Watches(&source.Kind{Type: &corev1.Node{}}, handler.Funcs{UpdateFunc: r.nodeUpdateHandler}). 
        Complete(r) 
} 

SetupWithManager之前有講到過，主要是使用了建造者模式，去構建了我們需要監聽的對象，只有這些對象的相關事件才會觸發我們的 Reconcile 邏輯。這里面的 Complete 最后其實是調用了 Build 方法

func (blder *Builder) Build(r reconcile.Reconciler) (controller.Controller, error) { 
    // 省略參數校驗 
 
    // Set the Config 
    blder.loadRestConfig() 
 
    // 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.ctrl, nil 
} 

Build主要調用 doController 、doWatch兩個方法

func (blder *Builder) doController(r reconcile.Reconciler) error { 
    ctrlOptions := blder.ctrlOptions 
    if ctrlOptions.Reconciler == nil { 
        ctrlOptions.Reconciler = r 
    } 
 
    // Retrieve the GVK from the object we're reconciling 
    // to prepopulate logger information, and to optionally generate a default name. 
    gvk, err := getGvk(blder.forInput.object, blder.mgr.GetScheme()) 
    if err != nil { 
        return err 
    } 
 
    // Setup the logger. 
    if ctrlOptions.Log == nil { 
        ctrlOptions.Log = blder.mgr.GetLogger() 
    } 
    ctrlOptions.Log = ctrlOptions.Log.WithValues("reconciler group", gvk.Group, "reconciler kind", gvk.Kind) 
 
    // Build the controller and return. 
    blder.ctrl, err = newController(blder.getControllerName(gvk), blder.mgr, ctrlOptions) 
    return err 
} 

doController主要是初始化了一個 Controller，這里面傳入了我們實現 的Reconciler以及獲取到我們的 GVK 的名稱

func (blder *Builder) doWatch() error { 
    // Reconcile type 
    typeForSrc, err := blder.project(blder.forInput.object, blder.forInput.objectProjection) 
    if err != nil { 
        return err 
    } 
    src := &source.Kind{Type: typeForSrc} 
    hdler := &handler.EnqueueRequestForObject{} 
    allPredicates := append(blder.globalPredicates, blder.forInput.predicates...) 
    if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil { 
        return err 
    } 
 
    // Watches the managed types 
    for _, own := range blder.ownsInput { 
        typeForSrc, err := blder.project(own.object, own.objectProjection) 
        if err != nil { 
            return err 
        } 
        src := &source.Kind{Type: typeForSrc} 
        hdler := &handler.EnqueueRequestForOwner{ 
            OwnerType:    blder.forInput.object, 
            IsController: true, 
        } 
        allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...) 
        allPredicates = append(allPredicates, own.predicates...) 
        if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil { 
            return err 
        } 
    } 
 
    // Do the watch requests 
    for _, w := range blder.watchesInput { 
        allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...) 
        allPredicates = append(allPredicates, w.predicates...) 
 
        // If the source of this watch is of type *source.Kind, project it. 
        if srckind, ok := w.src.(*source.Kind); ok { 
            typeForSrc, err := blder.project(srckind.Type, w.objectProjection) 
            if err != nil { 
                return err 
            } 
            srckind.Type = typeForSrc 
        } 
 
        if err := blder.ctrl.Watch(w.src, w.eventhandler, allPredicates...); err != nil { 
            return err 
        } 
    } 
    return nil 
} 

Watch 主要是監聽我們想要的資源變化，blder.ctrl.Watch(src, hdler, allPredicates...)通過過濾源事件的變化，allPredicates是過濾器，只有所有的過濾器都返回 true 時，才會將事件傳遞給 EventHandler hdler，這里會將 Handler 注冊到 Informer 上

啟動

func (cm *controllerManager) Start(ctx context.Context) (err error) { 
    cm.internalCtx, cm.internalCancel = context.WithCancel(ctx) 
 
    // 這個用來表示所有的協程都已經退出了， 
    stopComplete := make(chan struct{}) 
    defer close(stopComplete) 
     
  // ...... 
 
    // 用于保存錯誤 
    cm.errChan = make(chan error) 
 
    // 如果需要 metric 就啟動 metric 服務 
    if cm.metricsListener != nil { 
        go cm.serveMetrics() 
    } 
 
    // 啟動健康檢查服務 
    if cm.healthProbeListener != nil { 
        go cm.serveHealthProbes() 
    } 
 
   
    go cm.startNonLeaderElectionRunnables() 
 
    go func() { 
        if cm.resourceLock != nil { 
            err := cm.startLeaderElection() 
            if err != nil { 
                cm.errChan <- err 
            } 
        } else { 
            // Treat not having leader election enabled the same as being elected. 
            close(cm.elected) 
            go cm.startLeaderElectionRunnables() 
        } 
    }() 
 
  // 判斷是否需要退出 
    select { 
    case <-ctx.Done(): 
        // We are done 
        return nil 
    case err := <-cm.errChan: 
        // Error starting or running a runnable 
        return err 
    } 
} 

無論是不是 leader 最后都會使用 startRunnable 啟動 Controller

func (cm *controllerManager) startNonLeaderElectionRunnables() { 
    cm.mu.Lock() 
    defer cm.mu.Unlock() 
 
    cm.waitForCache(cm.internalCtx) 
 
    // Start the non-leaderelection Runnables after the cache has synced 
    for _, c := range cm.nonLeaderElectionRunnables { 
        // Controllers block, but we want to return an error if any have an error starting. 
        // Write any Start errors to a channel so we can return them 
        cm.startRunnable(c) 
    } 
} 

實際上是調用了 Controller 的 Start方法

// Start implements controller.Controller 
func (c *Controller) Start(ctx context.Context) error { 
 
  // Controller 只能被執行一次 
    c.mu.Lock() 
    if c.Started { 
        return errors.New("controller was started more than once. This is likely to be caused by being added to a manager multiple times") 
    } 
 
    // Set the internal context. 
    c.ctx = ctx 
 
  // 獲取隊列 
    c.Queue = c.MakeQueue() 
    defer c.Queue.ShutDown() 
 
    err := func() error { 
        defer c.mu.Unlock() 
 
        defer utilruntime.HandleCrash() 
 
        // 嘗試等待緩存 
        for _, watch := range c.startWatches { 
            c.Log.Info("Starting EventSource", "source", watch.src) 
            if err := watch.src.Start(ctx, watch.handler, c.Queue, watch.predicates...); err != nil { 
                return err 
            } 
        } 
 
        // 啟動 Controller 
        c.Log.Info("Starting Controller") 
 
     
        for _, watch := range c.startWatches { 
            syncingSource, ok := watch.src.(source.SyncingSource) 
            if !ok { 
                continue 
            } 
            if err := syncingSource.WaitForSync(ctx); err != nil { 
                // This code is unreachable in case of kube watches since WaitForCacheSync will never return an error 
                // Leaving it here because that could happen in the future 
                err := fmt.Errorf("failed to wait for %s caches to sync: %w", c.Name, err) 
                c.Log.Error(err, "Could not wait for Cache to sync") 
                return err 
            } 
        } 
 
        // All the watches have been started, we can reset the local slice. 
        // 
        // We should never hold watches more than necessary, each watch source can hold a backing cache, 
        // which won't be garbage collected if we hold a reference to it. 
        c.startWatches = nil 
 
        if c.JitterPeriod == 0 { 
            c.JitterPeriod = 1 * time.Second 
        } 
 
        // Launch workers to process resources 
        c.Log.Info("Starting workers", "worker count", c.MaxConcurrentReconciles) 
        ctrlmetrics.WorkerCount.WithLabelValues(c.Name). 
                    Set(float64(c.MaxConcurrentReconciles)) 
        for i := 0; i < c.MaxConcurrentReconciles; i++ { 
            go wait.UntilWithContext(ctx, func(ctx context.Context) { 
                // 查詢隊列中有沒有關注的事件，有的話就觸發我們的 reconcile 邏輯 
                for c.processNextWorkItem(ctx) { 
                } 
            }, c.JitterPeriod) 
        } 
 
        c.Started = true 
        return nil 
    }() 
    if err != nil { 
        return err 
    } 
 
    <-ctx.Done() 
    c.Log.Info("Stopping workers") 
    return nil 
} 
 
// attempt to process it, by calling the reconcileHandler. 
func (c *Controller) processNextWorkItem(ctx context.Context) bool { 
    obj, shutdown := c.Queue.Get() 
    if shutdown { 
        // Stop working 
        return false 
    } 
 
    // We call Done here so the workqueue knows we have finished 
    // processing this item. We also must remember to call Forget if we 
    // do not want this work item being re-queued. For example, we do 
    // not call Forget if a transient error occurs, instead the item is 
    // put back on the workqueue and attempted again after a back-off 
    // period. 
    defer c.Queue.Done(obj) 
 
    ctrlmetrics.ActiveWorkers.WithLabelValues(c.Name).Add(1) 
    defer ctrlmetrics.ActiveWorkers.WithLabelValues(c.Name).Add(-1) 
 
    c.reconcileHandler(ctx, obj) 
    return true 
} 

總結

Reconcile 方法的觸發是通過 Cache 中的 Informer 獲取到資源的變更事件，然后再通過生產者消費者的模式觸發我們自己實現的 Reconcile 方法的。

Kubebuilder 是一個非常好用的 Operator 開發框架，不僅極大的簡化了 Operator 的開發過程，并且充分的利用了 go interface 的特性留下了足夠的擴展性，這個我們可以學習，如果我們的業務代碼開發框架能夠做到這個地步，我覺得也就不錯了

參考文獻

架構圖 https://master.book.kubebuilder.io/architecture.html?

本文轉載自微信公眾號「mohuishou」，可以通過以下二維碼關注。轉載本文請聯系mohuishou公眾號。

原文鏈接：https://lailin.xyz/post/operator-09-kubebuilder-code.html