Operator实战1：使用kubebuilder开发一个部署web服务的Operator

云原生企业级实战笔记专栏

使用Kind部署本地k8s测试集群

使用brew install kind安装kind
使用 kind create cluster基于docker创建一个k8s集群，要求先安装docker
电脑重启需要启动kindest/node容器
让kind里的node能够拉取到本地镜像，可以使用命令：kind load docker-image ${IMG}

安装kubebuilder工具

gitlab查看已发布版本：https://github.com/kubernetes-sigs/kubebuilder/releases

Mac操作系统可执行下面脚本安装：

# 其中v3.9.0是版本，可以指定其它
# darwin是操作系统，通过“go env GOOS”命令获取
# amd64是cpu架构，通过“go env GOARCH”命令获取
curl -L https://github.com/kubernetes-sigs/kubebuilder/releases/download/v3.9.0/kubebuilder_darwin_amd64 ./kubebuilder
chmod -R "+x" ./kubebuilder
sudo mv ./kubebuilder /usr/local/bin/

需求场景

举一个非常简单的需求场景，仅用于介绍如何使用kubebuilder开发一个Operator，非真实需求场景。Operator的定义是CRD+CRD Controller，因此这个例子需要包含自定义CRD和实现自定义CRD的控制器。

假设有一个需求，当用户给定一个web服务镜像地址时，自动部署这个web服务，然后通过NodePort暴露服务给外部可以访问。

实现需求

官方使用文档：https://cloudnative.to/kubebuilder/introduction.html

一、初始化项目

在$GOPATH/src目录下创建一个空项目，然后进入项目目录，使用kubebuilder脚手架工具初始化项目。

mkdir $GOPATH/src/operator-example
cd $GOPATH/src/operator-example
kubebuilder init --domain wujiuye.com

其中operator-example为项目名；--domain是指定域名，也是CRD的Group的后缀。

二、创建CRD

执行kubebuilder create api命令创建一个名为DeployWebService的crd，group指定为operator，version指定为v1beta1。

kubebuilder create api \
--kind DeployWebService \
--plural deploywebservice \
--group operator \
--version v1beta1 \
--resource true --controller true

--resource true用于指定为资源生成CRD
--controller true用于指定为资源生成Controller

此时项目包含如下代码文件。

-- operator-example
---- api
------ v1bate1
-------- deploywebservice_types.go ## CRD
---- controllers
------ deploywebservice_controller.go ## Controller
---- main.go

三、根据需求完善CRD

需求描述“当用户给定一个web服务镜像地址时，自动部署这个web服务”，因此CRD需要提供一个字段给用户填写镜像，另外“然后通过NodePort暴露服务给外部可以访问”，需要有一个字段给用户填写web服务镜像暴露的端口，然后还要有一个副本数的字段，让用户填写部署多少个Pod。

需要修改deploywebservice_types.go代码文件，在DeployWebServiceSpec结构体中添加自定义资源字段：

type DeployWebServiceSpec struct {
    Image       string `json:"image"`
    Replicas    int32  `json:"replicas"`
    ExposedPort int    `json:"exposed_port"`
}

此外，我们还需要给资源添加状态，用于描述资源当前处理什么状态，是否部署成功，遇到了什么问题，方便用户排查问题。

虽然API上并没有对如何设计资源的状态做任何约定，包括kubernetes在1.19版本之前提供的内置资源，它们的状态也并不都遵循约定，但状态的设计建议还是遵循官方新提出的约定，可参考这篇《kubernetes设计与实现-API设计约定-condition设计约定：https://renhongcai.gitbook.io/kubernetes/di-shi-liu-zhang-api-she-ji-yue-ding/1.2-api_convention_condition》。

通常一个condition必须包含type（状态类型）和status（状态值）两个信息，在Kubernetesv1.19版本社区提供了一个标准的condition类型定义：

type ConditionStatus string
 
const (
    ConditionTrue    ConditionStatus = "True"
    ConditionFalse   ConditionStatus = "False"
    ConditionUnknown ConditionStatus = "Unknown"
)

type Condition struct {
    // 类型（使用驼峰风格），如”Available“。
    Type string `json:"type"`
    // 状态（枚举值：”True“、”False“和”Unknown“）。
    Status ConditionStatus `json:"status"`
    // 观察到的generation。
    // 如果ObservedGeneration 比metada.generation小，说明不是最新状态。
    // +optional
    ObservedGeneration int64 `json:"observed_generation,omitempty"`
    // 上次变化时间
    LastTransitionTime metav1.Time `json:"last_transition_time"`
    // 状态变化原因（使用驼峰风格），如”NewReplicaSetAvailable“
    Reason string `json:"reason"`
    // 描述信息，如”Deployment has minimum availability“
    Message string `json:"message" protobuf:"bytes,6,opt,name=message"`
}

我们基于这一标准来设计我们的自定义资源DeployWebService的状态定义，修改deploywebservice_types.go代码文件，在DeployWebServiceStatus结构体中添加Conditions字段：

type DeployWebServiceStatus struct {
    Conditions []Condition `json:"conditions"`
}

四、实现控制器逻辑

控制器需要在监听到DeployWebService资源CRUD事件时，调协DeployWebService的部署/更新/卸载，并且要更新资源的状态。

由于是demo，不会实现的很完善，这里仅实现以下逻辑：

查询DeployWebService资源
DeployWebService资源存在： 2.1. 查看Deployment是否存在，不存在创建Deployment来部署web服务，如果存在根据Deployment的状态来更新DeployWebService资源的状态。 2.2. 查看Service是否存在，不存在则创建Service来暴露服务给k8s集群外部访问，如果存在则根据Service的状态来更新DeployWebService资源的状态。

需要注意，创建Deployment、Service资源需要设置owen指向DeployWebService资源，以实现当DeployWebService资源被删除时，Deployment、Service资源会自动被删除。

首先是实现整体的框架：

func (r *DeployWebServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    _ = log.FromContext(ctx)

    // 1. 查询DeployWebService资源
    dws := &operatorv1beta1.DeployWebService{}
    if err := r.Get(ctx, req.NamespacedName, dws); err != nil {
       return ctrl.Result{}, client.IgnoreNotFound(err)
    }

    // 2.1 查看Deployment是否存在，不存在创建Deployment来部署web服务，如果存在根据Deployment的状态来更新DeployWebService资源的状态
    deployment := &v1.Deployment{}
    if err := r.Get(ctx, req.NamespacedName, deployment); err != nil {
       if errors.IsNotFound(err) {
          // 不存在创建Deployment来部署web服务
       } else {
          // 如果存在根据Deployment的状态来更新DeployWebService资源的状态
       }
    }

    // 2.2 查看Service是否存在，不存在则创建Service来暴露服务给k8s集群外部访问，如果存在则根据Service的状态来更新DeployWebService资源的状态
    service := &v12.Service{}
    if err := r.Get(ctx, req.NamespacedName, service); err != nil {
       if errors.IsNotFound(err) {
          // 不存在则创建Service来暴露服务给k8s集群外部访问
       } else {
          // 如果存在则根据Service的状态来更新DeployWebService资源的状态
       }
    }

    return ctrl.Result{}, nil
}

实现当不存在Deployment则创建Deployment来部署web服务。

// 不存在创建Deployment来部署web服务
deployment = &v1.Deployment{
    TypeMeta: metav1.TypeMeta{
       APIVersion: "apps/v1",
       Kind:       "Deployment",
    },
    ObjectMeta: metav1.ObjectMeta{
       Name:      req.Name + "-deployment",
       Namespace: req.Namespace,
       OwnerReferences: []metav1.OwnerReference{
          *metav1.NewControllerRef(dws, dws.GroupVersionKind()),
       },
    },
    Spec: v1.DeploymentSpec{
       Replicas: &dws.Spec.Replicas,
       Selector: &metav1.LabelSelector{
          MatchLabels: map[string]string{
             "webservice": req.Name,
          },
       },
       Template: v12.PodTemplateSpec{
          ObjectMeta: metav1.ObjectMeta{
             Labels: map[string]string{
                "webservice": req.Name,
             },
          },
          Spec: v12.PodSpec{
             Containers: []v12.Container{
                v12.Container{
                   Image: dws.Spec.Image,
                   // ....
                },
             },
          },
       },
    },
}
if err := r.Create(ctx, deployment); err != nil {
    return ctrl.Result{}, err
}
// 更新状态
dws.Status.Conditions = append(dws.Status.Conditions, operatorv1beta1.Condition{
    Type:               "DeploymentProgressing",
    Status:             "True",
    LastTransitionTime: metav1.Now(),
    Reason:             "CreateDeployment",
    Message:            "success to create deployment.",
})
dwsJson, _ := json.Marshal(dws)
if err := r.Status().Patch(ctx, dws, client.RawPatch(types.MergePatchType, dwsJson)); err != nil {
    return ctrl.Result{}, err
}

实现如果存在则根据Deployment的状态来更新DeployWebService资源的状态。

if availableCond, ok := GetDeploymentCondition(deployment, "Available"); ok {
    if availableCond.Status == v12.ConditionTrue {
       dws.Status.Conditions = append(dws.Status.Conditions, operatorv1beta1.Condition{
          Type:               "DeploymentAvailable",
          Status:             "True",
          LastTransitionTime: metav1.Now(),
          Reason:             "DeploymentStatusUpdate",
          Message:            "deployment is available",
       })
    } else {
       dws.Status.Conditions = append(dws.Status.Conditions, operatorv1beta1.Condition{
          Type:               "DeploymentAvailable",
          Status:             "False",
          LastTransitionTime: metav1.Now(),
          Reason:             "DeploymentStatusUpdate",
          Message:            "deployment is not available",
       })
    }
    dwsJson, _ := json.Marshal(dws)
    if err := r.Status().Patch(ctx, dws, client.RawPatch(types.MergePatchType, dwsJson)); err != nil {
       return ctrl.Result{}, err
    }
}

实现不存在Service则创建Service来暴露服务给k8s集群外部访问。

service = &v12.Service{
    TypeMeta: metav1.TypeMeta{
       APIVersion: "v1",
       Kind:       "Service",
    },
    ObjectMeta: metav1.ObjectMeta{
       Name:      req.Name + "-service",
       Namespace: req.Namespace,
       OwnerReferences: []metav1.OwnerReference{
          *metav1.NewControllerRef(dws, dws.GroupVersionKind()),
       },
    },
    Spec: v12.ServiceSpec{
       Ports: []v12.ServicePort{
          v12.ServicePort{
             Name:       "webhttp",
             Protocol:   "TCP",
             Port:       80,
             TargetPort: intstr.FromInt(8080),
          },
       },
       Selector: map[string]string{
          "webservice": req.Name,
       },
       Type: "NodePort",
    },
}
if err := r.Create(ctx, service); err != nil {
    return ctrl.Result{}, err
}
// 更新状态
dws.Status.Conditions = append(dws.Status.Conditions, operatorv1beta1.Condition{
    Type:               "ServiceProgressing",
    Status:             "True",
    LastTransitionTime: metav1.Now(),
    Reason:             "CreateService",
    Message:            "success to create service.",
})
dwsJson, _ := json.Marshal(dws)
if err := r.Status().Patch(ctx, dws, client.RawPatch(types.MergePatchType, dwsJson)); err != nil {
    return ctrl.Result{}, err
}

实现如果存在则根据Service的状态来更新DeployWebService资源的状态。

// 这里我们直接设置为可用
dws.Status.Conditions = append(dws.Status.Conditions, operatorv1beta1.Condition{
    Type:               "ServiceAvailable",
    Status:             "True",
    LastTransitionTime: metav1.Now(),
    Reason:             "ServiceStatusUpdate",
    Message:            "service is available",
})
dwsJson, _ := json.Marshal(dws)
if err := r.Status().Patch(ctx, dws, client.RawPatch(types.MergePatchType, dwsJson)); err != nil {
    return ctrl.Result{}, err
}

当然，这里的状态更新实现比较粗糙，会导致Conditions数组不断增大，应该做一下过滤重复，type相同的Condition应该是更新覆盖，而不是新增。这些留给大家自己动手去实现。

更完善的，还需要监听Deployment、Service资源的事件，在Deployment、Service资源变更后更新DeployWebService资源的状态，以及当Deployment、Service资源被误删除后，能够实现自动创建出来。另外应该使用Finalizer特性，在DeployWebService资源删除之前，完成一些资源清理操作。

更优化的实现，应该将Deployment、Service定义为模版，通过模版生成，这样可以少写很多的代码，提升代码的可读性，然后像更新状态这些重复出现的代码应该抽象为一个updateStatus方法。

五、为控制器生成RBAC

这一步本地debug是不需要的。

由于controller实现的逻辑需要操作Deployment、Service资源，以及DeployWebService资源，因此我们在controller添加以下注解来告诉kubebuilder怎么帮我们生成role.yaml。

//+kubebuilder:rbac:groups=operator.wujiuye.com,resources=deploywebservice,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=operator.wujiuye.com,resources=deploywebservice/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=operator.wujiuye.com,resources=deploywebservice/finalizers,verbs=update
//+kubebuilder:rbac:groups="",resources=services,verbs=get;update;create;delete;list;watch;patch
//+kubebuilder:rbac:groups=extensions;apps,resources=deployments,verbs=get;update;create;delete;list;watch;patch
func (r *DeployWebServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {

}

现在执行make manifests命令后，config/rbac/role.yaml文件就会更新，生成的role.yaml内容如下。

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  creationTimestamp: null
  name: manager-role
rules:
- apiGroups:
  - ""
  resources:
  - services
  verbs:
  - create
  - delete
  - get
  - list
  - patch
  - update
  - watch
- apiGroups:
  - apps
  - extensions
  resources:
  - deployments
  verbs:
  - create
  - delete
  - get
  - list
  - patch
  - update
  - watch
- apiGroups:
  - operator.wujiuye.com
  resources:
  - deploywebservice
  verbs:
  - create
  - delete
  - get
  - list
  - patch
  - update
  - watch
- apiGroups:
  - operator.wujiuye.com
  resources:
  - deploywebservice/finalizers
  verbs:
  - update
- apiGroups:
  - operator.wujiuye.com
  resources:
  - deploywebservice/status
  verbs:
  - get
  - patch
  - update

调试Operator

在项目路径下，打开一个终端，执行make generate命令，用于为资源生成DeepCopy方法。
然后执行make manifests命令，用于生成crd的yaml文件、生成RBAC的role.yaml。默认生成的crd文件存在config/crd/bases/目录下，默认生成的role.yaml存在config/rbac目录下。
本地debug也需要将crd部署到k8s集群，执行kubectl将crd部署到k8s集群。

kubectl apply -f config/crd/bases/operator.wujiuye.com_deploywebservice.yaml

在IDEA中点击debug就可以启动Operator。
修改kubebuilder生成的sample，用于测试controller的逻辑。sample路径是config/samples/operator_v1beta1_deploywebservice.yaml。

apiVersion: operator.wujiuye.com/v1beta1
kind: DeployWebService
metadata:
  labels:
    app.kubernetes.io/name: deploywebservice
    app.kubernetes.io/instance: deploywebservice-sample
    app.kubernetes.io/part-of: operator-example
    app.kubernetes.io/managed-by: kustomize
    app.kubernetes.io/created-by: operator-example
  name: deploywebservice-sample
spec:
  image: "webservice:1.0.0"
  replicas: 1
  exposed_port: 8080

将sample安装到k8s集群就会触发Controller的Reconcile方法执行。

kubectl apply -f config/samples/operator_v1beta1_deploywebservice.yaml