展开全文

CRD定义范例

与其他资源对象一样，对CRD的定义也使用YAML配置进行声明。下面先来看下本文的Demo例子的CRD定义。

本文的crddemo源码为：https://github.com/domac/crddemo

mydemo.yaml

apiVersion: apiextensions.k8s.io/v1beta1

kind: CustomResourceDefinition

name: mydemos.crddemo.k8s.io

group: crddemo.k8s.io

version: v1

names:

kind: Mydemo

plural: mydemos

scope: Namespaced

CRD定义中的关键字段如下：

• group：设置API所属的组，将其映射为API URL中的“/apis/”下一级目录。它是逻辑上相关的Kinds集合
• scope：该API的生效范围，可选项为Namespaced和Cluster。
• version：每个Group可以存在多个版本。例如，v1alpha1，然后升为v1beta1，最后稳定为v1版本。
• names： CRD的名称，包括单数、复数、kind、所属组等名称定义

在这个CRD中，我指定了group: crddemo.k8s.io，version: v1这样的API信息，也指定了这个CR的资源类型叫作Mydemo，复数（plural）是 mydemos。

我们先别着急使用kubectl create创建资源定义，我们接下来要做的是再基于这个CRD的定义创建相应的具体自定义对象。

example-mydemo.yaml

apiVersion: crddemo.k8s.io/v1

kind: Mydemo

name: example-mydemo

ip: “127.0.0.1”

port: 8080

这个资源对象跟定义Pod差不多，它的主要信息都是来源上面的定义，Kind是Mydemo，版本是v1，资源组是crddemo.k8s.io

除了这些设置，还需要在spec端设置相应的参数，一般是开发者自定义定制的，在这里，我定制了两个属性：ip和port。所以整个对象我要告诉Kubernetes，我期待监听处理一个叫example-mydemo的程序，它的地址是127.0.0.1，端口是8080。当然，这里只是一个demo，没有什么严格的属性约束，开发者还是根据自己的业务需要自行定义吧。为了不影响本文的介绍，姑且认为这两个属性是非常重要的。

到这里为止，相对轻松的工作已经完成，我们已经完成CRD的“设计图”工作，下面我们开始动手构建这个CRD控制器的编码工作了。

CRD控制器原理

在正式编码之前，我们先理解一下自定义控制器的工作原理，如下图：

CRD控制器的工作流，可分为监听、同步、触发三个步骤：

一、Controller首先会通过Informer （所谓的Informer，就是一个自带缓存和索引机制），从Kubernetes的API Server中获取它所关心的对象，举个例子，也就是我编写的Controller获取的应该是Mydemo对象。值得注意的是Informer在构建之前，会使用我们生成的client（下面编码阶段会提到），再透过Reflector的ListAndWatch机制跟API Server建立连接，不断地监听Mydemo对象实例的变化。在ListAndWatch 机制下，一旦 APIServer 端有新的 Mydemo 实例被创建、删除或者更新，Reflector都会收到“事件通知”。该事件及它对应的API对象会被放进一个Delta FIFO Queue中。

二、Local Store此时完成同步缓存操作。

三、Informer根据这些事件的类型，触发我们编写并注册号的ResourceEventHandler，完成业务动作的触发。

上面图中的Control Loop实际上可以通过code-generator生成，下面也会提到。总之Control Loop中我们只关心如何拿到“实际状态”，并与“期待状态”对比，从而具体的差异处理逻辑，只需要开发者自行编写即可。

CRD开发过程

下面会通过一个简单的例子，开始我们的CRD代码的编写， 完整代码：https://github.com/domac/crddemo

自定义资源代码编写

首先，Kubernetes涉及的代码生成对项目目录结构是有要求的，所以我们先创建一个结构如下的项目。

├── controller.go

├── crd

│ └── mydemo.yaml

├── example

│ └── example-mydemo.yaml

├── main.go

└── pkg

└── apis

└── crddemo

├── register.go

└── v1

├── doc.go

├── register.go

├── types.go

可见关键部分的pkg目录就是API组的URL结构，如下图：

v1下面的types.go文件里，则定义了Mydemo对象的完整描述。

1、我们首先开看pkg/apis/crddemo/register.go，这个文件主要用来存放全局变量，如下：

package crddemo

const (

GroupName = “crddemo.k8s.io”

Version = “v1”

)

2、pkg/apis/crddemo/v1下的doc.go 也是比较简单的：

// +k8s:deepcopy-gen=package

// +groupName=crddemo.k8s.io

package v1

在这个文件中，你会看到+k8s:deepcopy-gen=package和+groupName=crddemo.k8s.io，这就是Kubernetes进行代码生成要用的Annotation风格的注释。

• +k8s:deepcopy-gen=package意思是，请为整个v1包里的所有类型定义自动生成DeepCopy方法；
• +groupName=crddemo.k8s.io，则定义了这个包对应的crddemo API组的名字。

可以看到，这些定义在doc.go文件的注释，起到的是全局的代码生成控制的作用，所以也被称为Global Tags。

3、pkg/apis/crddemo/types.go的作用就是定义一个Mydemo类型到底有哪些字段（比如，spec字段里的内容）。这个文件的主要内容如下所示：

package v1

import (

metav1 “k8s.io/apimachinery/pkg/apis/meta/v1”

)

// +genclient

// +genclient:noStatus

// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

// Mydemo 描述一个Mydemo的资源字段

typeMydemo struct {

metav1.TypeMeta `json: “,inline”`

metav1.ObjectMeta `json: “metadata,omitempty”`

Spec MydemoSpec `json: “spec”`

}

//MydemoSpec 为Mydemo的资源的spec属性的字段

typeMydemoSpec struct {

Ip string `json: “ip”`

Port int `json: “port”`

}

// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

//复数形式

typeMydemoList struct {

metav1.TypeMeta `json: “,inline”`

metav1.ListMeta `json: “metadata”`

Items []Mydemo `json: “items”`

}

上面的代码，可以开的我们的Mydemo可续定义方法跟Kubernetes对象一样，都包含了TypeMeta和ObjectMeta字段，而其中比较重要的是 Spec 字段，就是需要我们自己定义的部分：定义了IP和Port两个字段。

此外，除了定义Mydemo类型，你还需要定义一个MydemoList类型，用来描述一组Mydemo对象应该包括哪些字段。之所以需要这样一个类型，是因为在Kubernetes中，获取所有某对象的List方法，返回值都是List类型，而不是某类型的数组。所以代码上一定要做区分。

关于上面代码的几个重要注解，下面说明一下：

• +genclient这段注解的意思是：请为下面资源类型生成对应的Client代码。
• +genclient:noStatus的意思是：这个API资源类型定义里，没有Status字段，因为Mydemo才是主类型，所以+genclient要写在Mydemo之上，不用写在MydemoList之上，这时要细心注意的。
• +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object的意思是，请在生成DeepCopy的时候，实现Kubernetes提供的runtime.Object接口。否则，在某些版本的Kubernetes里，你的这个类型定义会出现编译错误。

4、pkg/apis/crddemo/register.go作用就是注册一个类型（Type）给APIServer。

package v1

import (

“github.com/domac/crddemo/pkg/apis/crddemo”

metav1 “k8s.io/apimachinery/pkg/apis/meta/v1”

“k8s.io/apimachinery/pkg/runtime”

“k8s.io/apimachinery/pkg/runtime/schema”

)

var SchemeGroupVersion = schema.GroupVersion{

Group: crddemo.GroupName,

Version: crddemo.Version,

}

var (

SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes)

AddToScheme = SchemeBuilder.AddToScheme

)

func Resource(resource string) schema.GroupResource {

returnSchemeGroupVersion.WithResource(resource).GroupResource

}

func Kind(kind string) schema.GroupKind {

returnSchemeGroupVersion.WithKind(kind).GroupKind

}

//Mydemo资源类型在服务器端的注册的工作，APIServer会自动帮我们完成

//但与之对应的，我们还需要让客户端也能知道Mydemo资源类型的定义

func addKnownTypes(scheme *runtime.Scheme) error {

scheme.AddKnownTypes(

SchemeGroupVersion,

&Mydemo{},

&MydemoList{},

)

// register the typeinthe scheme

metav1.AddToGroupVersion(scheme, SchemeGroupVersion)

returnnil

}

有了addKnownTypes这个方法，Kubernetes就能够在后面生成客户端的时候，知道Mydemo以及MydemoList类型的定义了。

好了，到这里为止，我们有关定义的代码已经写好了，正如controller原理图所示，接下来我们需要通过Kubernetes提供的代码生成工具，为上面的Mydemo资源类型生成clientset、informer和lister。

关于如何使用代码生成，这里我已经编写了一个脚步，只需只需本脚步即可。

代码生成

具体可以调用我提供的shll脚本：code-gen.sh

#!/bin/bash

set-x

ROOT_PACKAGE= “github.com/domac/crddemo”

CUSTOM_RESOURCE_NAME= “crddemo”

CUSTOM_RESOURCE_VERSION= “v1”

GO111MODULE=off

# 安装k8s.io/code-generator

[[ -d $GOPATH/src/k8s.io/code-generator ]] || go get -u k8s.io/code-generator/…

# 执行代码自动生成，其中pkg/client是生成目标目录，pkg/apis是类型定义目录

cd$GOPATH/src/k8s.io/code-generator && ./generate-groups.sh all ” $ROOT_PACKAGE/pkg/client” ” $ROOT_PACKAGE/pkg/apis” ” $CUSTOM_RESOURCE_NAME: $CUSTOM_RESOURCE_VERSION”

当只需代码生成脚步后，可以发现我们的代码工作目录也发送了变化，多出了一个client目录。

client

├── clientset

│ └── versioned

│ ├── clientset.go

│ ├── doc.go

│ ├── fake

│ │ ├── clientset_generated.go

│ │ ├── doc.go

│ │ └── register.go

│ ├── scheme

│ │ ├── doc.go

│ │ └── register.go

│ └── typed

│ └── crddemo

│ └── v1

│ ├── crddemo_client.go

│ ├── doc.go

│ ├── fake

│ │ ├── doc.go

│ │ ├── fake_crddemo_client.go

│ │ └── fake_mydemo.go

│ ├── generated_expansion.go

│ └── mydemo.go

├── informers

│ └── externalversions

│ ├── crddemo

│ │ ├── interface.go

│ │ └── v1

│ │ ├── interface.go

│ │ └── mydemo.go

│ ├── factory.go

│ ├── generic.go

│ └── internalinterfaces

│ └── factory_interfaces.go

└── listers

└── crddemo

└── v1

├── expansion_generated.go

└── mydemo.go

其中，pkg/apis/crddemo/v1下面的zz_generated.deepcopy.go文件，就是自动生成的DeepCopy代码文件。下面的三个包（clientset、informers、 listers），都是Kubernetes为Mydemo类型生成的client库，这些库会在后面编写自定义控制器的时候用到。

自定义控制器代码编写

Kubernetes的声明式API并不像“命令式API”那样有着明显的执行逻辑。这就使得基于声明式API的业务功能实现，往往需要通过控制器模式来“监视”API 对象的变化，然后以此来决定实际要执行的具体工作。

main.go

package main

import (

“flag”

“os”

“os/signal”

“syscall”

“time”

“github.com/golang/glog”

“k8s.io/client-go/kubernetes”

“k8s.io/client-go/tools/clientcmd”

clientset “github.com/domac/crddemo/pkg/client/clientset/versioned”

informers “github.com/domac/crddemo/pkg/client/informers/externalversions”

)

//程序启动参数

var (

flagSet = flag.NewFlagSet( “crddemo”, flag.Exit)

master = flag.String( “master”, “”, “The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.”)

kubeconfig = flag.String( “kubeconfig”, “”, “Path to a kubeconfig. Only required if out-of-cluster.”)

onlyOneSignalHandler = make(chan struct{})

shutdownSignals = []os.Signal{os.Interrupt, syscall.SIGTERM}

)

//设置信号处理

func setupSignalHandler (stopCh <-chan struct{}) {

close(onlyOneSignalHandler)

stop := make(chan struct{})

c := make(chan os.Signal, 2)

signal.Notify(c, shutdownSignals…)

go func{

<-c

close(stop)

<-c

os.Exit(1)

}

returnstop

}

func main{

flag.Parse

//设置一个信号处理，应用于优雅关闭

stopCh := setupSignalHandler

cfg, err := clientcmd.BuildConfigFromFlags(*master, *kubeconfig)

iferr != nil {

glog.Fatalf( “Error building kubeconfig: %s”, err.Error)

}

kubeClient, err := kubernetes.NewForConfig(cfg)

iferr != nil {

glog.Fatalf( “Error building kubernetes clientset: %s”, err.Error)

}

mydemoClient, err := clientset.NewForConfig(cfg)

iferr != nil {

glog.Fatalf( “Error building example clientset: %s”, err.Error)

}

//informerFactory工厂类， 这里注入我们通过代码生成的client

//clent主要用于和API Server进行通信，实现ListAndWatch

mydemoInformerFactory := informers.NewSharedInformerFactory(mydemoClient, time.Second*30)

//生成一个crddemo组的Mydemo对象传递给自定义控制器

controller := NewController(kubeClient, mydemoClient,

mydemoInformerFactory.V1.Mydemos)

go mydemoInformerFactory(stopCh)

iferr = controller.Run(2, stopCh); err != nil {

glog.Fatalf( “Error running controller: %s”, err.Error)

}

}

接下来，我们来看跟业务最紧密的控制器Controller的编写。

controller.go部分重要代码：

… …

func NewController(

kubeclientset kubernetes.Interface,

mydemoslientset clientset.Interface,

mydemoInformer informers.MydemoInformer) *Controller {

// Create event broadcaster

// Add sample-controller types to the default Kubernetes Scheme so Events can be

// logged forsample-controller types.

utilruntime.Must(mydemoscheme.AddToScheme(scheme.Scheme))

glog.V(4).Info( “Creating event broadcaster”)

eventBroadcaster := record.NewBroadcaster

eventBroadcaster.StartLogging(glog.Infof)

eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1.Events( “”)})

recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})

//使用client和前面创建的Informer，初始化了自定义控制器

controller := &Controller{

kubeclientset: kubeclientset,

mydemoslientset: mydemoslientset,

demoInformer: mydemoInformer.Lister,

mydemosSynced: mydemoInformer.Informer.HasSynced,

workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter, “Mydemos”), //WorkQueue的实现，负责同步Informer和控制循环之间的数据

recorder: recorder,

}

glog.Info(” Setting up mydemo event handlers”)

//mydemoInformer注册了三个Handler（AddFunc、UpdateFunc和DeleteFunc）

// 分别对应API对象的“添加”“更新”和“删除”事件。而具体的处理操作，都是将该事件对应的API对象加入到工作队列中

mydemoInformer.Informer.AddEventHandler(cache.ResourceEventHandlerFuncs{

AddFunc: controller.enqueueMydemo,

UpdateFunc: func(old, new interface{}) {

oldMydemo := old.(*samplecrdv1.Mydemo)

newMydemo := new.(*samplecrdv1.Mydemo)

ifoldMydemo.ResourceVersion == newMydemo.ResourceVersion {

return

}

controller.enqueueMydemo(new)

},

DeleteFunc: controller.enqueueMydemoForDelete,

})

returncontroller

}

… …

通过上面Controller的代码实现，我们基本实现了控制器ListAndWatch的事件注册逻辑：通过APIServer的LIST API获取所有最新版本的API对象；然后，再通过WATCH-API来监听所有这些API对象的变化。通过监听到的事件变化，Informer就可以实时地更新本地缓存，并且调用这些事件对应的EventHandler了。

下面，我们再来看原理图中的Control Loop的部分。

func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {

defer runtime.HandleCrash

defer c.workqueue.ShutDown

// 记录开始日志

glog.Info( “Starting Mydemo control loop”)

glog.Info( “Waiting for informer caches to sync”)

ifok := cache.WaitForCacheSync(stopCh, c.mydemosSynced); !ok {

returnfmt.Errorf( “failed to wait for caches to sync”)

}

glog.Info( “Starting workers”)

fori := 0; i < threadiness; i++ {

go wait.Until(c.runWorker, time.Second, stopCh)

}

glog.Info( “Started workers”)

<-stopCh

glog.Info( “Shutting down workers”)

returnnil

}

可以看到，启动控制循环的逻辑非常简单，就是同步+循环监听任务。而这个循环监听任务就是我们真正的业务实现部分了。

//runWorker是一个不断运行的方法，并且一直会调用c.processNextWorkItem从workqueue读取和读取消息

func (c *Controller) runWorker{

forc. processNextWorkItem{

}

}

//从workqueue读取和读取消息

func (c *Controller) processNextWorkItem bool {

obj, shutdown := c.workqueue.Get

ifshutdown {

returnfalse

}

err := func(obj interface{}) error {

defer c.workqueue.Done(obj)

var key string

var ok bool

ifkey, ok = obj.(string); !ok {

c.workqueue.Forget(obj)

runtime.HandleError(fmt.Errorf( “expected string in workqueue but got %#v”, obj))

returnnil

}

iferr := c.syncHandler(key); err != nil {

returnfmt.Errorf( “error syncing ‘%s’: %s”, key, err.Error)

}

c.workqueue.Forget(obj)

glog.Infof( “Successfully synced ‘%s'”, key)

returnnil

}(obj)

iferr != nil {

runtime.HandleError(err)

returntrue

}

returntrue

}

//尝试从Informer维护的缓存中拿到了它所对应的Mydemo对象

func (c *Controller) syncHandler(key string) error {

namespace, name, err := cache.SplitMetaNamespaceKey(key)

iferr != nil {

runtime.HandleError(fmt.Errorf( “invalid resource key: %s”, key))

returnnil

}

mydemo, err := c.demoInformer.Mydemos(namespace).Get(name)

//从缓存中拿不到这个对象，那就意味着这个Mydemo对象的Key是通过前面的“删除”事件添加进工作队列的。

iferr != nil {

iferrors.IsNotFound(err) {

//对应的Mydemo对象已经被删除了

glog.Warningf( “DemoCRD: %s/%s does not exist in local cache, will delete it from Mydemo …”,

namespace, name)

glog.Infof( “[DemoCRD] Deleting mydemo: %s/%s …”, namespace, name)

returnnil

}

runtime.HandleError(fmt.Errorf( “failed to list mydemo by: %s/%s”, namespace, name))

returnerr

}

glog.Infof( “[DemoCRD] Try to process mydemo: %#v …”, mydemo)

c.recorder.Event(mydemo, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)

returnnil

}

代码中的demoInformer，从namespace中通过key获取Mydemo对象这个操作，其实就是在访问本地缓存的索引，实际上，在Kubernetes的源码中，你会经常看到控制器从各种Lister里获取对象，比如：podLister、nodeLister等等，它们使用的都是Informer和缓存机制。

而如果控制循环从缓存中拿不到这个对象（demoInformer返回了IsNotFound错误），那就意味着这个Mydemo对象的Key是通过前面的“删除”事件添加进工作队列的。所以，尽管队列里有这个Key，但是对应的Mydemo对象已经被删除了。而如果能够获取到对应的Mydemo对象，就可以执行控制器模式里的对比“期望状态（用户通过YAML文件提交到APIServer里的信息）”和“实际状态（我们的控制循环需要通过查询实际的Mydemo资源情况”的功能逻辑了。不过在本例子中，就不做过多的业务假设了。

至此，一个完整的自定义API对象和它所对应的自定义控制器，就编写完毕了。

部署测试

代码编码后，我们准备开始代码的发布，可以使用提供Makefile进行编译。

$ make

… …

gofmt -w .

go test-v .

? github.com/domac/crddemo [no testfiles]

mkdir -p releases

GOOS=linux GOARCH=amd64 go build -mod=vendor -ldflags “-s -w”-v -o releases/crddemo *.go

github.com/golang/groupcache/lru

k8s.io/apimachinery/third_party/forked/golang/json

k8s.io/apimachinery/pkg/util/mergepatch

k8s.io/kube-openapi/pkg/util/proto

k8s.io/client-go/tools/record/util

k8s.io/apimachinery/pkg/util/strategicpatch

k8s.io/client-go/tools/record

command-line-arguments

go clean -i

… …

编译完成后，会生成crddemo的二进制文件，我们要做把crddemo放到Kubernetes集群中，或者本地也行，只要能访问到apiserver和具备kubeconfig。

可以看到，程序运行的时候，一开始会报错。这是因为，此时Mydemo对象的CRD还没有被创建出来，所以Informer去APIServer里获取Mydemos对象时，并不能找到Mydemo这个API资源类型的定义。

接下来，我们执行我们自定义资源的定义文件：

$ kubectl apply -f crd/mydemo.yaml

customresourcedefinition.apiextensions.k8s.io/mydemos.crddemo.k8s.io created

此时，观察crddemo的日志输出，可以看到Controller的日志恢复了正常，控制循环启动成功。

然后，我们可以对我们的Mydemo对象进行增删改查操作了。

提交我们的自定义资源对象：

$ kubectl apply -f example-mydemo.yaml

mydemo.crddemo.k8s.io/example-mydemo created

创建成功够，看Kubernetes集群是否成功存储起来：

$ kubectl get Mydemo

NAME AGE

example-mydemo 2s

这时候，查看一下控制器的输出：

可以看到，我们上面创建example-mydemo.yaml的操作，触发了EventHandler的添加事件，从而被放进了工作队列。紧接着，控制循环就从队列里拿到了这个对象，并且打印出了正在处理这个Mydemo对象的日志。

我们这时候，尝试修改资源，对对应的port属性进行修改。

apiVersion: crddemo.k8s.io/v1

kind: Mydemo

name: example-mydemo

ip: “127.0.0.1”

port: 9090

手动执行修改：

$ kubectl apply -f example-mydemo.yaml

此时，crddemo新增出来的日志如下：

可以看到，这一次，Informer 注册的更新事件被触发，更新后的Mydemo对象的Key被添加到了工作队列之中。

所以，接下来控制循环从工作队列里拿到的Mydemo对象，与前一个对象是不同的：它的ResourceVersion的值从10818363变成了10818457；而Spec里的Port字段，则变成了9080。最后，我再把这个对象删除掉：

$ kubectl delete -f example-mydemo.yaml

mydemo.crddemo.k8s.io “example-mydemo”deleted

这一次，在控制器的输出里，我们就可以看到，Informer注册的“删除”事件被触发，输出如下：

然后，Kubernetes集群的资源也被清除了：

$ kubectl get Mydemo

No resources found indefault namespace.

以上就是使用自定义控制器的基本开发流程。

原文链接：https://lihaoquan.me/2020/3/8/k8s-crd-develop.html