歡迎訪問我的GitHub
https://github.com/zq2599/blog_demos
內容：所有原創文章分類匯總及配套源碼，涉及Java、Docker、Kubernetes、DevOPS等；
系列文章鏈接
- kubebuilder實戰之一：准備工作
- kubebuilder實戰之二：初次體驗kubebuilder
- kubebuilder實戰之三：基礎知識速覽
- kubebuilder實戰之四：operator需求說明和設計
- kubebuilder實戰之五：operator編碼
- kubebuilder實戰之六：構建部署運行
- kubebuilder實戰之七：webhook
- kubebuilder實戰之八：知識點小記
本篇概覽
- 本篇是《kubebuilder實戰》系列的第五篇，前面的一切努力(環境准備、知識儲備、需求分析、數據結構和業務邏輯設計)，都是為了將之前的設計用編碼實現；
- 既然已經充分准備，如今無需太多言語，咱們開始動手吧！
源碼下載
- 本篇實戰中的完整源碼可在GitHub下載到，地址和鏈接信息如下表所示(https://github.com/zq2599/blog_demos)：
| 名稱 | 鏈接 | 備注 |
|---|---|---|
| 項目主頁 | https://github.com/zq2599/blog_demos | 該項目在GitHub上的主頁 |
| git倉庫地址(https) | https://github.com/zq2599/blog_demos.git | 該項目源碼的倉庫地址，https協議 |
| git倉庫地址(ssh) | git@github.com:zq2599/blog_demos.git | 該項目源碼的倉庫地址，ssh協議 |
- 這個git項目中有多個文件夾，kubebuilder相關的應用在kubebuilder文件夾下，如下圖紅框所示：
- kubebuilder文件夾下有多個子文件夾，本篇對應的源碼在elasticweb目錄下，如下圖紅框所示：
新建項目elasticweb
- 新建名為elasticweb的文件夾，在里面執行以下命令即可創建名為elasticweb的項目，domain為com.bolingcavalry：
go mod init elasticweb
kubebuilder init --domain com.bolingcavalry
- 然后是CRD，執行以下命令即可創建相關資源：
kubebuilder create api \
--group elasticweb \
--version v1 \
--kind ElasticWeb
- 然后用IDE打開整個工程，我這里是goland：
CRD編碼
- 打開文件api/v1/elasticweb_types.go，做以下幾步改動：
- 修改數據結構ElasticWebSpec，增加前文設計的四個字段；
- 修改數據結構ElasticWebStatus，增加前文設計的一個字段；
- 增加String方法，這樣打印日志時方便我們查看，注意RealQPS字段是指針，因此可能為空，需要判空；
- 完整的elasticweb_types.go如下所示：
package v1
import (
"fmt"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"strconv"
)
// 期望狀態
type ElasticWebSpec struct {
// 業務服務對應的鏡像，包括名稱:tag
Image string `json:"image"`
// service占用的宿主機端口，外部請求通過此端口訪問pod的服務
Port *int32 `json:"port"`
// 單個pod的QPS上限
SinglePodQPS *int32 `json:"singlePodQPS"`
// 當前整個業務的總QPS
TotalQPS *int32 `json:"totalQPS"`
}
// 實際狀態，該數據結構中的值都是業務代碼計算出來的
type ElasticWebStatus struct {
// 當前kubernetes中實際支持的總QPS
RealQPS *int32 `json:"realQPS"`
}
// +kubebuilder:object:root=true
// ElasticWeb is the Schema for the elasticwebs API
type ElasticWeb struct {
metav1.TypeMeta `json:",inline"`
metav1.ObjectMeta `json:"metadata,omitempty"`
Spec ElasticWebSpec `json:"spec,omitempty"`
Status ElasticWebStatus `json:"status,omitempty"`
}
func (in *ElasticWeb) String() string {
var realQPS string
if nil == in.Status.RealQPS {
realQPS = "nil"
} else {
realQPS = strconv.Itoa(int(*(in.Status.RealQPS)))
}
return fmt.Sprintf("Image [%s], Port [%d], SinglePodQPS [%d], TotalQPS [%d], RealQPS [%s]",
in.Spec.Image,
*(in.Spec.Port),
*(in.Spec.SinglePodQPS),
*(in.Spec.TotalQPS),
realQPS)
}
// +kubebuilder:object:root=true
// ElasticWebList contains a list of ElasticWeb
type ElasticWebList struct {
metav1.TypeMeta `json:",inline"`
metav1.ListMeta `json:"metadata,omitempty"`
Items []ElasticWeb `json:"items"`
}
func init() {
SchemeBuilder.Register(&ElasticWeb{}, &ElasticWebList{})
}
- 在elasticweb目錄下執行make install即可部署CRD到kubernetes：
zhaoqin@zhaoqindeMBP-2 elasticweb % make install
/Users/zhaoqin/go/bin/controller-gen "crd:trivialVersions=true" rbac:roleName=manager-role webhook paths="./..." output:crd:artifacts:config=config/crd/bases
kustomize build config/crd | kubectl apply -f -
Warning: apiextensions.k8s.io/v1beta1 CustomResourceDefinition is deprecated in v1.16+, unavailable in v1.22+; use apiextensions.k8s.io/v1 CustomResourceDefinition
customresourcedefinition.apiextensions.k8s.io/elasticwebs.elasticweb.com.bolingcavalry created
- 部署成功后，用api-versions命令可以查到該GV：
回顧業務邏輯
-
核心數據結構設計編碼完畢，接下來該編寫業務邏輯代碼了，大家還記得前文設計的業務流程吧，簡單回顧一下，如下圖：
-
打開文件elasticweb_controller.go，接下來咱們逐漸添加內容；
添加資源訪問權限
- 咱們的elasticweb會對service、deployment這兩種資源做查詢、新增、修改等操作，因此需要這些資源的操作權限，增加下圖紅框中的兩行注釋，這樣代碼生成工具就會在RBAC配置中增加對應的權限：
常量定義
- 先把常量准備好，可見每個pod使用的CPU和內存都是在此固定的，您也可以改成在Spec中定義，這樣就可以從外部傳入了，另外這里為每個pod只分配了0.1個CPU，主要是因為我窮買不起好的CPU，您可以酌情調整該值：
const (
// deployment中的APP標簽名
APP_NAME = "elastic-app"
// tomcat容器的端口號
CONTAINER_PORT = 8080
// 單個POD的CPU資源申請
CPU_REQUEST = "100m"
// 單個POD的CPU資源上限
CPU_LIMIT = "100m"
// 單個POD的內存資源申請
MEM_REQUEST = "512Mi"
// 單個POD的內存資源上限
MEM_LIMIT = "512Mi"
)
方法getExpectReplicas
- 有個很重要的邏輯：根據單個pod的QPS和總QPS，計算需要多少個pod，咱們將這個邏輯封裝到一個方法中以便使用：
/ 根據單個QPS和總QPS計算pod數量
func getExpectReplicas(elasticWeb *elasticwebv1.ElasticWeb) int32 {
// 單個pod的QPS
singlePodQPS := *(elasticWeb.Spec.SinglePodQPS)
// 期望的總QPS
totalQPS := *(elasticWeb.Spec.TotalQPS)
// Replicas就是要創建的副本數
replicas := totalQPS / singlePodQPS
if totalQPS%singlePodQPS > 0 {
replicas++
}
return replicas
}
方法createServiceIfNotExists
- 將創建service的操作封裝到一個方法中，是的主干代碼的邏輯更清晰，可讀性更強；
- 創建service的時候，有幾處要注意：
- 先查看service是否存在，不存在才創建；
- 將service和CRD實例elasticWeb建立關聯(controllerutil.SetControllerReference方法)，這樣當elasticWeb被刪除的時候，service會被自動刪除而無需我們干預；
- 創建service的時候用到了client-go工具，推薦您閱讀《client-go實戰系列》,工具越熟練，編碼越盡興；
- 創建service的完整方法如下：
// 新建service
func createServiceIfNotExists(ctx context.Context, r *ElasticWebReconciler, elasticWeb *elasticwebv1.ElasticWeb, 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: elasticWeb.Namespace,
Name: elasticWeb.Name,
},
Spec: corev1.ServiceSpec{
Ports: []corev1.ServicePort{{
Name: "http",
Port: 8080,
NodePort: *elasticWeb.Spec.Port,
},
},
Selector: map[string]string{
"app": APP_NAME,
},
Type: corev1.ServiceTypeNodePort,
},
}
// 這一步非常關鍵！
// 建立關聯后，刪除elasticweb資源時就會將deployment也刪除掉
log.Info("set reference")
if err := controllerutil.SetControllerReference(elasticWeb, 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
- 將創建deployment的操作封裝在一個方法中，同樣是為了將主干邏輯保持簡潔；
- 創建deployment的方法也有幾處要注意：
- 調用getExpectReplicas方法得到要創建的pod的數量，該數量是創建deployment時的一個重要參數；
- 每個pod所需的CPU和內存資源也是deployment的參數；
- 將deployment和elasticweb建立關聯，這樣刪除elasticweb的時候deplyment就會被自動刪除了；
- 同樣是使用client-go客戶端工具創建deployment資源；
// 新建deployment
func createDeployment(ctx context.Context, r *ElasticWebReconciler, elasticWeb *elasticwebv1.ElasticWeb) error {
log := r.Log.WithValues("func", "createDeployment")
// 計算期望的pod數量
expectReplicas := getExpectReplicas(elasticWeb)
log.Info(fmt.Sprintf("expectReplicas [%d]", expectReplicas))
// 實例化一個數據結構
deployment := &appsv1.Deployment{
ObjectMeta: metav1.ObjectMeta{
Namespace: elasticWeb.Namespace,
Name: elasticWeb.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: elasticWeb.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(elasticWeb, 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
- 不論是創建deployment資源對象，還是對已有的deployment的pod數量做調整，這些操作完成后都要去修改Status，既實際的狀態，這樣外部才能隨時隨地知道當前elasticweb支持多大的QPS，因此需要將修改Status的操作封裝到一個方法中，給多個場景使用，Status的計算邏輯很簡單：pod數量乘以每個pod的QPS就是總QPS了，代碼如下：
// 完成了pod的處理后，更新最新狀態
func updateStatus(ctx context.Context, r *ElasticWebReconciler, elasticWeb *elasticwebv1.ElasticWeb) error {
log := r.Log.WithValues("func", "updateStatus")
// 單個pod的QPS
singlePodQPS := *(elasticWeb.Spec.SinglePodQPS)
// pod總數
replicas := getExpectReplicas(elasticWeb)
// 當pod創建完畢后，當前系統實際的QPS：單個pod的QPS * pod總數
// 如果該字段還沒有初始化，就先做初始化
if nil == elasticWeb.Status.RealQPS {
elasticWeb.Status.RealQPS = new(int32)
}
*(elasticWeb.Status.RealQPS) = singlePodQPS * replicas
log.Info(fmt.Sprintf("singlePodQPS [%d], replicas [%d], realQPS[%d]", singlePodQPS, replicas, *(elasticWeb.Status.RealQPS)))
if err := r.Update(ctx, elasticWeb); err != nil {
log.Error(err, "update instance error")
return err
}
return nil
}
主干代碼
- 前面細枝末節都處理完畢，可以開始主流程了，有了前面的流程圖的賦值，主流程的代碼很容就寫出來了，如下所示，已經添加了足夠的注釋，就不再贅述了：
func (r *ElasticWebReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
// 會用到context
ctx := context.Background()
log := r.Log.WithValues("elasticweb", req.NamespacedName)
// your logic here
log.Info("1. start reconcile logic")
// 實例化數據結構
instance := &elasticwebv1.ElasticWeb{}
// 通過客戶端工具查詢，查詢條件是
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
}
- 至此，整個elasticweb operator編碼就完成了，限於篇幅，咱們把部署、運行、鏡像制作等操作放在下一篇文章吧；
你不孤單，欣宸原創一路相伴
歡迎關注公眾號：程序員欣宸
微信搜索「程序員欣宸」，我是欣宸，期待與您一同暢游Java世界...
https://github.com/zq2599/blog_demos