# Kubebuilder 介绍 Operator 是 Kubernetes 的扩展软件，它利用 定制资源管理应用及其组件。 Operator 遵循 Kubernetes 的理念，特别是在控制器方面。 通过 Operator 的方案，可以对 Kubernetes 的功能进行友好地扩展。`Operatpr = CRD + Controller`。首先通过 yaml 定义，生成 CRD ，然后 Controller 不断地监听 etcd 中的数据，执行相应动作。开发 Operator 时，有很多繁琐且重复的事情。KubeBuilder 可以帮助我们快速生成骨架代码，开发一个 Kubernetes 的扩展功能。  # Kubebuilder 命令 ```bash # 初始化项目 kubebuilder init --domain zhangshun.io --license apache2 --owner zhangshun # 创建API kubebuilder create api --group apps --version v1beta1 --kind MyWeb # 部署CRD到k8s make install # 本地运行 make run # 构建镜像并上传至仓库 make docker-build docker-push IMG=zhangshunzz/myweb:v0.1 # 部署controller到k8s make deploy IMG=zhangshunzz/myweb:v0.1 # 从集群中删除CRD make uninstall # 从集群中卸载控制器 make undeploy ``` # Kubebuilder 简单的示例 **定义CRD** ```go type MyWebSpec struct { // 业务服务对应的镜像，包括名称:tag Image string `json:"image"` // service占用的宿主机端口，外部请求通过此端口访问pod的服务 Port *int32 `json:"port"` // 单个pod的QPS上限 SinglePodQPS *int32 `json:"singlePodQPS"` // 当前整个业务的总QPS TotalQPS *int32 `json:"totalQPS"` // 资源限制 Resources v1.ResourceRequirements `json:"resources"` } type MyWebStatus struct { // 当前kubernetes中实际支持的总QPS RealQPS *int32 `json:"realQPS"` } ``` **方法getExpectReplicas** ```go / 根据单个QPS和总QPS计算pod数量 func getExpectReplicas(myWeb *appsv1beta1.MyWeb) int32 { // 单个pod的QPS singlePodQPS := *(myWeb.Spec.SinglePodQPS) // 期望的总QPS totalQPS := *(myWeb.Spec.TotalQPS) // Replicas就是要创建的副本数 replicas := totalQPS / singlePodQPS if totalQPS%singlePodQPS > 0 { replicas++ } return replicas } ``` **方法createServiceIfNotExists** ```go // 新建service func createServiceIfNotExists(ctx context.Context, r *MyWebReconciler, myWeb *appsv1beta1.MyWeb, req ctrl.Request) error { log := r.Log.WithValues("func", "createService") service := &corev1.Service{} err := r.Get(ctx, req.NamespacedName, service) // 如果查询结果没有错误，证明service正常，就不做任何操作 if err == nil { log.Info("service exists") return nil } // 如果错误不是NotFound，就返回错误 if !errors.IsNotFound(err) { log.Error(err, "query service error") return err } // 实例化一个数据结构 service = &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ Namespace: myWeb.Namespace, Name: myWeb.Name, }, Spec: corev1.ServiceSpec{ Ports: []corev1.ServicePort{{ Name: "http", Port: 8080, NodePort: *myWeb.Spec.Port, }, }, Selector: map[string]string{ "app": APP_NAME, }, Type: corev1.ServiceTypeNodePort, }, } // 这一步非常关键！ // 建立关联后，删除elasticweb资源时就会将deployment也删除掉 log.Info("set reference") if err := controllerutil.SetControllerReference(myWeb, service, r.Scheme); err != nil { log.Error(err, "SetControllerReference error") return err } // 创建service log.Info("start create service") if err := r.Create(ctx, service); err != nil { log.Error(err, "create service error") return err } log.Info("create service success") return nil } ``` **方法createDeployment** ```go // 新建deployment func createDeployment(ctx context.Context, r *MyWebReconciler, myWeb *appsv1beta1.MyWeb) error { log := r.Log.WithValues("func", "createDeployment") // 计算期望的pod数量 expectReplicas := getExpectReplicas(myWeb) log.Info(fmt.Sprintf("expectReplicas [%d]", expectReplicas)) // 实例化一个数据结构 deployment := &appsv1.Deployment{ ObjectMeta: metav1.ObjectMeta{ Namespace: myWeb.Namespace, Name: myWeb.Name, }, Spec: appsv1.DeploymentSpec{ // 副本数是计算出来的 Replicas: pointer.Int32Ptr(expectReplicas), Selector: &metav1.LabelSelector{ MatchLabels: map[string]string{ "app": APP_NAME, }, }, Template: corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ Labels: map[string]string{ "app": APP_NAME, }, }, Spec: corev1.PodSpec{ Containers: []corev1.Container{ { Name: APP_NAME, // 用指定的镜像 Image: myWeb.Spec.Image, ImagePullPolicy: "IfNotPresent", Ports: []corev1.ContainerPort{ { Name: "http", Protocol: corev1.ProtocolSCTP, ContainerPort: CONTAINER_PORT, }, }, Resources: corev1.ResourceRequirements{ Requests: corev1.ResourceList{ "cpu": resource.MustParse(CPU_REQUEST), "memory": resource.MustParse(MEM_REQUEST), }, Limits: corev1.ResourceList{ "cpu": resource.MustParse(CPU_LIMIT), "memory": resource.MustParse(MEM_LIMIT), }, }, }, }, }, }, }, } // 这一步非常关键！ // 建立关联后，删除elasticweb资源时就会将deployment也删除掉 log.Info("set reference") if err := controllerutil.SetControllerReference(myWeb, deployment, r.Scheme); err != nil { log.Error(err, "SetControllerReference error") return err } // 创建deployment log.Info("start create deployment") if err := r.Create(ctx, deployment); err != nil { log.Error(err, "create deployment error") return err } log.Info("create deployment success") return nil } ``` **方法updateStatus** ```go // 完成了pod的处理后，更新最新状态 func updateStatus(ctx context.Context, r *MyWebReconciler, myWeb *appsv1beta1.MyWeb) error { log := r.Log.WithValues("func", "updateStatus") // 单个pod的QPS singlePodQPS := *(myWeb.Spec.SinglePodQPS) // pod总数 replicas := getExpectReplicas(elasticWeb) // 当pod创建完毕后，当前系统实际的QPS：单个pod的QPS * pod总数 // 如果该字段还没有初始化，就先做初始化 if nil == myWeb.Status.RealQPS { myWeb.Status.RealQPS = new(int32) } *(myWeb.Status.RealQPS) = singlePodQPS * replicas log.Info(fmt.Sprintf("singlePodQPS [%d], replicas [%d], realQPS[%d]", singlePodQPS, replicas, *(elasticWeb.Status.RealQPS))) if err := r.Update(ctx, myWeb); err != nil { log.Error(err, "update instance error") return err } return nil } ``` **Reconcile主干代码** ```go func (r *MyWebReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) { // 会用到context ctx := context.Background() log := r.Log.WithValues("myweb", req.NamespacedName) // your logic here log.Info("1. start reconcile logic") // 实例化数据结构 instance := &appsv1beta1.MyWeb{} // 通过客户端工具查询，查询条件是 err := r.Get(ctx, req.NamespacedName, instance) if err != nil { // 如果没有实例，就返回空结果，这样外部就不再立即调用Reconcile方法了 if errors.IsNotFound(err) { log.Info("2.1. instance not found, maybe removed") return reconcile.Result{}, nil } log.Error(err, "2.2 error") // 返回错误信息给外部 return ctrl.Result{}, err } log.Info("3. instance : " + instance.String()) // 查找deployment deployment := &appsv1.Deployment{} // 用客户端工具查询 err = r.Get(ctx, req.NamespacedName, deployment) // 查找时发生异常，以及查出来没有结果的处理逻辑 if err != nil { // 如果没有实例就要创建了 if errors.IsNotFound(err) { log.Info("4. deployment not exists") // 如果对QPS没有需求，此时又没有deployment，就啥事都不做了 if *(instance.Spec.TotalQPS) < 1 { log.Info("5.1 not need deployment") // 返回 return ctrl.Result{}, nil } // 先要创建service if err = createServiceIfNotExists(ctx, r, instance, req); err != nil { log.Error(err, "5.2 error") // 返回错误信息给外部 return ctrl.Result{}, err } // 立即创建deployment if err = createDeployment(ctx, r, instance); err != nil { log.Error(err, "5.3 error") // 返回错误信息给外部 return ctrl.Result{}, err } // 如果创建成功就更新状态 if err = updateStatus(ctx, r, instance); err != nil { log.Error(err, "5.4. error") // 返回错误信息给外部 return ctrl.Result{}, err } // 创建成功就可以返回了 return ctrl.Result{}, nil } else { log.Error(err, "7. error") // 返回错误信息给外部 return ctrl.Result{}, err } } // 如果查到了deployment，并且没有返回错误，就走下面的逻辑 // 根据单QPS和总QPS计算期望的副本数 expectReplicas := getExpectReplicas(instance) // 当前deployment的期望副本数 realReplicas := *deployment.Spec.Replicas log.Info(fmt.Sprintf("9. expectReplicas [%d], realReplicas [%d]", expectReplicas, realReplicas)) // 如果相等，就直接返回了 if expectReplicas == realReplicas { log.Info("10. return now") return ctrl.Result{}, nil } // 如果不等，就要调整 *(deployment.Spec.Replicas) = expectReplicas log.Info("11. update deployment's Replicas") // 通过客户端更新deployment if err = r.Update(ctx, deployment); err != nil { log.Error(err, "12. update deployment replicas error") // 返回错误信息给外部 return ctrl.Result{}, err } log.Info("13. update status") // 如果更新deployment的Replicas成功，就更新状态 if err = updateStatus(ctx, r, instance); err != nil { log.Error(err, "14. update status error") // 返回错误信息给外部 return ctrl.Result{}, err } return ctrl.Result{}, nil } ``` # Kubebuilder webhook # Kubebuilder 知识点小记