深刻解析 Kubebuilder：让编写 CRD 变得更简单

时间 2019-11-21

标签深刻解析 kubebuilder 编写 crd 变得简单繁體版

原文原文链接

导读：自定义资源 CRD（Custom Resource Definition）能够扩展 Kubernetes API，掌握 CRD 是成为 Kubernetes 高级玩家的必备技能，本文将介绍 CRD 和 Controller 的概念，并对 CRD 编写框架 Kubebuilder 进行深刻分析，让您真正理解并能快速开发 CRD。

概览

控制器模式与声明式 API

在正式介绍 Kubebuidler 以前，咱们须要先了解下 K8s 底层实现大量使用的控制器模式，以及让用户大呼过瘾的声明式 API，这是介绍 CRDs 和 Kubebuidler 的基础。nginx

控制器模式

K8s 做为一个“容器编排”平台，其核心的功能是编排，Pod 做为 K8s 调度的最小单位,具有不少属性和字段，K8s 的编排正是经过一个个控制器根据被控制对象的属性和字段来实现。git

下面咱们看一个例子：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），其伪代码逻辑为：api

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

所谓声明式就是“告诉 K8s 你要什么，而不是告诉它怎么作的命令”，一个很熟悉的例子就是 SQL，你“告诉 DB 根据条件和各种算子返回数据，而不是告诉它怎么遍历，过滤，聚合”。在 K8s 里面，声明式的体现就是 kubectl apply 命令，在对象建立和后续更新中一直使用相同的 apply 命令，告诉 K8s 对象的终态便可，底层是经过执行了一个对原有 API 对象的 PATCH 操做来实现的，能够一次性处理多个写操做，具有 Merge 能力 diff 出最终的 PATCH，而命令式一次只能处理一个写请求。app

声明式 API 让 K8s 的“容器编排”世界看起来温柔美好，而控制器（以及容器运行时，存储，网络模型等）才是这太平盛世的幕后英雄。说到这里，就会有人但愿也能像 build-in 资源同样构建本身的自定义资源（CRD-Customize Resource Definition），而后为自定义资源写一个对应的控制器，推出本身的声明式 API。K8s 提供了 CRD 的扩展方式来知足用户这一需求，并且因为这种扩展方式十分灵活，在最新的 1.15 版本对 CRD 作了至关大的加强。对于用户来讲，实现 CRD 扩展主要作两件事：框架

1.编写 CRD 并将其部署到 K8s 集群里；dom

这一步的做用就是让 K8s 知道有这个资源及其结构属性，在用户提交该自定义资源的定义时（一般是 YAML 文件定义），K8s 可以成功校验该资源并建立出对应的 Go struct 进行持久化，同时触发控制器的调谐逻辑。ide

2.编写 Controller 并将其部署到 K8s 集群里。

这一步的做用就是实现调谐逻辑。

Kubebuilder 就是帮咱们简化这两件事的工具，如今咱们开始介绍主角。

Kubebuilder 是什么？

摘要

Kubebuilder 是一个使用 CRDs 构建 K8s API 的 SDK，主要是：

提供脚手架工具初始化 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 所示：

Kubebuilder 怎么用？

1. 建立脚手架工程

kubebuilder init --domain edas.io

这一步建立了一个 Go module 工程，引入了必要的依赖，建立了一些模板文件。

2. 建立 API

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

这一步建立了对应的 CRD 和 Controller 模板文件，通过一、2 两步，现有的工程结构如图 2 所示：

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）来帮助你们理解：

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 如今再也不是黑盒。

最佳实践

模式

1.使用 OwnerRefrence 来作资源关联，有两个特性：

Owner 资源被删除，被 Own 的资源会被级联删除，这利用了 K8s 的 GC；
被 Own 的资源对象的事件变动能够触发 Owner 对象的 Reconcile 方法；

2.使用 Finalizer 来作资源的清理。

注意点

不使用 Finalizer 时，资源被删除没法获取任何信息；
对象的 Status 字段变化也会触发 Reconcile 方法；
Reconcile 逻辑须要幂等；

优化

使用 IndexFunc 来优化资源查询的效率

总结

经过深刻分析，咱们能够看到 Kubebuilder 提供的功能对于快速编写 CRD 和 Controller 是十分有帮助的，不管是 Istio、Knative 等知名项目仍是各类自定义 Operators，都大量使用了 CRD，将各类组件抽象为 CRD，Kubernetes 变成控制面板将成为一个趋势，但愿本文可以帮助你们理解和把握这个趋势。

本文做者：刘洋(炎寻)

阅读原文

本文为云栖社区原创内容，未经容许不得转载。