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

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

概览

 

控制器模式与声明式 API

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

 

控制器模式

K8s 作为一个“容器编排”平台，其核心的功能是编排，Pod 作为 K8s 调度的最小单位,具备很多属性和字段，K8s 的编排正是通过一个个控制器根据被控制对象的属性和字段来实现。 下面我们看一个例子：

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），其伪代码逻辑为：

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，而命令式一次只能处理一个写请求。   声明式 API 让 K8s 的“容器编排”世界看起来温柔美好，而控制器（以及容器运行时，存储，网络模型等）才是这太平盛世的幕后英雄。说到这里，就会有人希望也能像 build-in 资源一样构建自己的自定义资源（CRD-Customize Resource Definition），然后为自定义资源写一个对应的控制器，推出自己的声明式 API。K8s 提供了 CRD 的扩展方式来满足用户这一需求，而且由于这种扩展方式十分灵活，在最新的 1.15 版本对 CRD 做了相当大的增强。对于用户来说，实现 CRD 扩展主要做两件事：

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

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

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 模板文件，经过 1、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() {
    ...
        // 1、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)
    }
        // 2、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")
        // 3、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 的：

1. 初始化了一个 Manager；
2. 将 Manager 的 Client 传给 Controller，并且调用 SetupWithManager 方法传入 Manager 进行 Controller 的初始化；
3. 启动 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