Helm Operator

发表于： 分类于：
字数： 2138 阅读：≈ 11分钟 浏览：
总结摘要
在 Kubernetes 生态中，Operator 模式已成为管理复杂应用生命周期的标准方式。本文将详细介绍如何使用 Kubebuilder 开发一个具有门禁检查功能的 Helm 部署 Operator，帮助你理解控制器调和逻辑、状态管理及事件处理等核心概念。

使用 Kubebuilder 开发 Helm 部署 Operator：从入门到实践

在 Kubernetes 生态中，Operator 模式已成为管理复杂应用生命周期的标准方式。本文将详细介绍如何使用 Kubebuilder 开发一个具有门禁检查功能的 Helm 部署 Operator，帮助你理解控制器调和逻辑、状态管理及事件处理等核心概念。

什么是 Operator？

Operator 是一种包装、部署和管理 Kubernetes 应用的方法，它通过自定义资源 (CR) 和控制器扩展 Kubernetes API，将领域知识编码到软件中，实现应用生命周期的自动化管理。

对于 Helm 应用来说，一个功能完善的 Operator 可以：

  • 自动处理 Helm Chart 的部署、升级和回滚

  • 实现自定义的部署门禁检查（如依赖检查、手动审批）

  • 维护应用状态并提供丰富的可观测性

  • 响应相关资源变化并自动调和状态

开发环境准备

必要工具安装 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 安装 Kubebuilder

curl -L -o kubebuilder https://github.com/kubernetes-sigs/kubebuilder/releases/download/v3.10.0/kubebuilder\_linux\_amd64

chmod +x kubebuilder && sudo mv kubebuilder /usr/local/bin/

# 安装 kustomize

go install sigs.k8s.io/kustomize/kustomize/v4@latest

# 确保已安装 Go (1.19+) 和 Docker

初始化项目 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 创建项目目录

mkdir helm-app-operator && cd helm-app-operator

# 初始化项目

kubebuilder init --domain demo.example.com --repo demo.example.com/helm-app-operator

# 创建 API 组和版本

kubebuilder create api --group demo --version v1alpha1 --kind HelmApp

定义自定义资源 (CRD)

编辑 api/v1alpha1/helmapp_types.go 文件，定义我们的 HelmApp 资源：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

// HelmAppSpec 定义了 Helm 应用的期望状态

type HelmAppSpec struct {

     // Chart 定义 Helm 图表信息

     Chart ChartSpec `json:"chart"`

     // 目标部署命名空间

     TargetNamespace string `json:"targetNamespace"`

     // Helm  Values 配置

     Values map[string]interface{} `json:"values,omitempty"`

     // 部署前需要通过的门禁条件

     Gates []GateSpec `json:"gates,omitempty"`

}

// ChartSpec 包含 Helm 图表的元数据

type ChartSpec struct {

       // 图表仓库地址

       Repository string `json:"repository"`

       // 图表名称

       Name string `json:"name"`

       // 图表版本

       Version string `json:"version"`

       // 门禁检查间隔(秒)

       CheckInterval int32 `json:"checkInterval,omitempty"`

}

// GateSpec 定义单个门禁检查项

type GateSpec struct {

       // 门禁名称

       Name string `json:"name"`

       // 门禁类型 (ConfigMapExists/DependencyReady/ManualApproval)

       Type string `json:"type"`

       // 门禁描述

       Description string `json:"description,omitempty"`

       // 是否为强制门禁

       Mandatory bool `json:"mandatory,omitempty"`

}

// HelmAppStatus 定义了 Helm 应用的实际状态

type HelmAppStatus struct {

       // 部署阶段 (Pending/Deploying/Deployed/DeploymentFailed)

       Phase string `json:"phase,omitempty"`

      

       // Helm 发布名称

       ReleaseName string `json:"releaseName,omitempty"`

      

       // 最后部署的版本

       LastDeployedVersion string `json:"lastDeployedVersion,omitempty"`

      

       // 状态条件集合

       Conditions []corev1.PodCondition `json:"conditions,omitempty"`

}

生成 CRD 清单并安装：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 生成 API 代码

make generate

# 生成 CRD 清单

make manifests

# 安装 CRD 到集群

make install

实现控制器逻辑

控制器是 Operator 的核心，负责协调资源的期望状态与实际状态。我们的控制器需要实现：

  1. 门禁检查逻辑

  2. Helm 部署功能

  3. 状态管理与更新

  4. 事件记录与错误处理

控制器核心结构

编辑 controllers/helmapp_controller.go：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

// HelmAppReconciler 处理 HelmApp 资源

type HelmAppReconciler struct {

       ctrl.Client

       Scheme   *runtime.Scheme

       Log      logr.Logger

       Recorder record.EventRecorder

}

// SetupWithManager 设置控制器

func (r *HelmAppReconciler) SetupWithManager(mgr manager.Manager) error {

       return ctrl.NewControllerManagedBy(mgr).

               For(\&demov1alpha1.HelmApp{}).

               Watches(

                       \&source.Kind{Type: \&corev1.ConfigMap{}},

                       handler.EnqueueRequestsFromMapFunc(r.findHelmAppsForConfigMap),

               ).

               WithOptions(controller.Options{

                       MaxConcurrentReconciles: 5,

               }).

               Complete(r)

}

调和逻辑实现

Reconcile 函数是控制器的核心，实现主要的调和逻辑：

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

// Reconcile 处理 HelmApp 资源的调和逻辑

func (r *HelmAppReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {

       log := log.FromContext(ctx)

       // 1. 获取 HelmApp 实例

       helmApp := \&demov1alpha1.HelmApp{}

       if err := r.Get(ctx, req.NamespacedName, helmApp); err != nil {

               log.Error(err, "无法获取 HelmApp 资源")

               return ctrl.Result{}, ctrl.IgnoreNotFound(err)

       }

       // 2. 初始化状态

       if helmApp.Status.Phase == "" {

               helmApp.Status.Phase = "Pending"

               if err := r.Status().Update(ctx, helmApp); err != nil {

                       log.Error(err, "无法更新 HelmApp 状态")

                       return r.handleStatusUpdateError(ctx, helmApp, err)

               }

               r.Recorder.Event(helmApp, corev1.EventTypeNormal, "Initializing", "开始处理 Helm 应用部署")

               return ctrl.Result{Requeue: true}, nil

       }

       // 3. 执行门禁检查

       allGatesPassed, err := r.checkAllGates(ctx, helmApp)

       if err != nil {

               log.Error(err, "门禁检查失败")

               return ctrl.Result{RequeueAfter: 30 * time.Second}, err

       }

       // 4. 如果所有门禁都通过，执行 Helm 部署

       if allGatesPassed && helmApp.Status.Phase != "Deployed" {

               log.Info("所有门禁检查通过，开始部署 Helm 应用")

               helmApp.Status.Phase = "Deploying"

               if err := r.Status().Update(ctx, helmApp); err != nil {

                       log.Error(err, "无法更新 HelmApp 状态为部署中")

                       return r.handleStatusUpdateError(ctx, helmApp, err)

               }

               // 执行 Helm 部署

               release, err := r.deployWithHelm(ctx, helmApp)

               if err != nil {

                       log.Error(err, "Helm 部署失败")

                       helmApp.Status.Phase = "DeploymentFailed"

                       r.updateCondition(helmApp, "Deployment", corev1.ConditionFalse, "DeploymentError", err.Error())

                       r.Recorder.Event(helmApp, corev1.EventTypeWarning, "DeploymentFailed", err.Error())

                       if err := r.Status().Update(ctx, helmApp); err != nil {

                               return r.handleStatusUpdateError(ctx, helmApp, err)

                       }

                       return ctrl.Result{RequeueAfter: 60 * time.Second}, nil

               }

               // 更新部署成功状态

               helmApp.Status.Phase = "Deployed"

               helmApp.Status.ReleaseName = release.Name

               helmApp.Status.LastDeployedVersion = release.Chart.Metadata.Version

               r.updateCondition(helmApp, "Ready", corev1.ConditionTrue, "Deployed", "应用部署成功并就绪")

               r.Recorder.Event(helmApp, corev1.EventTypeNormal, "Deployed", fmt.Sprintf("Helm 应用部署成功: %s", release.Name))

              

               if err := r.Status().Update(ctx, helmApp); err != nil {

                       log.Error(err, "无法更新 HelmApp 状态为已部署")

                       return r.handleStatusUpdateError(ctx, helmApp, err)

               }

               return ctrl.Result{}, nil

       }

       // 5. 如果门禁未通过，定期重试

       if !allGatesPassed {

               log.Info("部分门禁检查未通过，将在稍后重试")

               checkInterval := time.Duration(helmApp.Spec.Chart.CheckInterval) * time.Second

               if checkInterval <= 0 {

                       checkInterval = 30 * time.Second

               }

               return ctrl.Result{RequeueAfter: checkInterval}, nil

       }

       // 6. 已部署状态，定期检查

       return ctrl.Result{RequeueAfter: 5 * time.Minute}, nil

}

门禁检查实现

实现三种门禁检查类型：配置检查、依赖检查和手动审批：

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183

// 检查所有门禁条件

func (r *HelmAppReconciler) checkAllGates(ctx context.Context, helmApp *demov1alpha1.HelmApp) (bool, error) {

       log := log.FromContext(ctx)

       allPassed := true

       passedCount := 0

       // 检查每个定义的门禁

       for \_, gate := range helmApp.Spec.Gates {

               gatePassed, reason, message := r.checkGate(ctx, helmApp, gate)

              

               // 更新门禁状态 Condition

               status := corev1.ConditionFalse

               if gatePassed {

                       status = corev1.ConditionTrue

                       passedCount++

               } else {

                       allPassed = false

                       if gate.Mandatory {

                               log.Info("强制门禁检查未通过", "gate", gate.Name)

                       }

               }

              

               r.updateCondition(helmApp, gate.Name, status, reason, message)

              

               // 记录门禁检查事件

               eventType := corev1.EventTypeNormal

               if !gatePassed {

                       eventType = corev1.EventTypeWarning

               }

               eventReason := fmt.Sprintf("Gate%s", strings.Title(strings.ToLower(reason)))

               eventMessage := fmt.Sprintf("门禁检查 '%s' %s: %s", gate.Name,

                       map[bool]string{true: "通过", false: "未通过"}[gatePassed],

                       message)

               r.Recorder.Eventf(helmApp, eventType, eventReason, eventMessage)

       }

       // 更新所有门禁是否通过的汇总 Condition

       allGatesStatus := corev1.ConditionFalse

       if allPassed {

               allGatesStatus = corev1.ConditionTrue

       }

       r.updateCondition(helmApp, "AllGatesPassed", allGatesStatus,

               "GatesCheckComplete", "所有门禁检查已完成")

       return allPassed, nil

}

// 检查单个门禁条件

func (r *HelmAppReconciler) checkGate(ctx context.Context, helmApp *demov1alpha1.HelmApp, gate demov1alpha1.GateSpec) (bool, string, string) {

       switch gate.Type {

       case "ConfigMapExists":

               // 检查指定的 ConfigMap 是否存在

               cm := \&corev1.ConfigMap{}

               if err := r.Get(ctx, types.NamespacedName{

                       Name:      gate.Name,

                       Namespace: helmApp.Spec.TargetNamespace,

               }, cm); err != nil {

                       return false, "ConfigMapMissing", fmt.Sprintf("配置文件 %s 不存在", gate.Name)

               }

               return true, "ConfigMapFound", fmt.Sprintf("配置文件 %s 已找到", gate.Name)

              

       case "DependencyReady":

               // 检查依赖的 HelmApp 是否已就绪

               depApp := \&demov1alpha1.HelmApp{}

               if err := r.Get(ctx, types.NamespacedName{

                       Name:      gate.Name,

                       Namespace: helmApp.Namespace,

               }, depApp); err != nil {

                       return false, "DependencyMissing", fmt.Sprintf("依赖应用 %s 不存在", gate.Name)

               }

              

               if depApp.Status.Phase != "Deployed" {

                       return false, "DependencyNotReady", fmt.Sprintf("依赖应用 %s 尚未就绪", gate.Name)

               }

               return true, "DependencyReady", fmt.Sprintf("依赖应用 %s 已就绪", gate.Name)

              

       case "ManualApproval":

               // 检查手动审批（通过特定 ConfigMap 中的标记）

               approvalCM := \&corev1.ConfigMap{}

               cmName := fmt.Sprintf("approval-%s", helmApp.Name)

               if err := r.Get(ctx, types.NamespacedName{

                       Name:      cmName,

                       Namespace: helmApp.Namespace,

               }, approvalCM); err != nil {

                       return false, "ApprovalPending", "等待手动审批"

               }

              

               if approvalCM.Data["approved"] == "true" {

                       return true, "Approved", "已获得手动审批"

               }

               return false, "NotApproved", "尚未获得手动审批"

              

       default:

               return false, "UnknownGateType", fmt.Sprintf("未知的门禁类型: %s", gate.Type)

       }

}

状态更新错误处理

状态更新失败时的处理策略非常重要，需要确保状态最终能反映真实情况：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

// 处理状态更新错误

func (r *HelmAppReconciler) handleStatusUpdateError(ctx context.Context, helmApp *demov1alpha1.HelmApp, originalErr error) (ctrl.Result, error) {

       log := log.FromContext(ctx)

      

       // 记录错误事件

       r.Recorder.Event(helmApp, corev1.EventTypeWarning, "StatusUpdateFailed",

               fmt.Sprintf("状态更新失败: %v", originalErr))

      

       // 检查错误类型

       if apierrors.IsConflict(originalErr) {

               // 资源版本冲突，重新获取资源后重试

               log.Info("状态更新冲突，将重新获取资源后重试")

               if err := r.Get(ctx, types.NamespacedName{Name: helmApp.Name, Namespace: helmApp.Namespace}, helmApp); err != nil {

                       log.Error(err, "重新获取资源失败")

                       return ctrl.Result{}, err

               }

               return ctrl.Result{Requeue: true}, nil

       }

      

       if apierrors.IsNotFound(originalErr) {

               // 资源已被删除，无需继续处理

               log.Info("资源已被删除，停止状态更新")

               return ctrl.Result{}, nil

       }

      

       // 其他错误，重试几次

       maxRetries := 3

       retryInterval := 2 * time.Second

       var lastErr error

      

       for i := 0; i < maxRetries; i++ {

               time.Sleep(retryInterval)

               if err := r.Status().Update(ctx, helmApp); err != nil {

                       lastErr = err

                       log.Error(err, "状态更新重试失败", "重试次数", i+1)

                       continue

               }

               log.Info("状态更新重试成功")

               return ctrl.Result{}, nil

       }

      

       // 多次重试失败，记录并延迟重试

       log.Error(lastErr, "多次重试后状态更新仍失败")

       return ctrl.Result{RequeueAfter: 10 * time.Second}, lastErr

}

实现 Helm 部署功能

集成 Helm SDK 实现应用部署：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

// 使用 Helm 部署应用

func (r *HelmAppReconciler) deployWithHelm(ctx context.Context, helmApp *demov1alpha1.HelmApp) (*release.Release, error) {

       settings := cli.New()

       actionConfig := new(action.Configuration)

      

       if err := actionConfig.Init(settings.RESTClientGetter(), helmApp.Spec.TargetNamespace,

               os.Getenv("HELM\_DRIVER"), r.Log.Info); err != nil {

               return nil, fmt.Errorf("无法初始化 Helm 配置: %v", err)

       }

       // 添加 Helm 仓库

       addRepo := action.NewRepoAdd(actionConfig)

       addRepo.URL = helmApp.Spec.Chart.Repository

       addRepo.Update = true

       if \_, err := addRepo.Run(helmApp.Name, helmApp.Spec.Chart.Repository); err != nil {

               return nil, fmt.Errorf("无法添加 Helm 仓库: %v", err)

       }

       // 安装 Helm 图表

       install := action.NewInstall(actionConfig)

       install.ReleaseName = fmt.Sprintf("%s-%s", helmApp.Name, generateRandomString(5))

       install.Namespace = helmApp.Spec.TargetNamespace

       install.CreateNamespace = true

       install.Version = helmApp.Spec.Chart.Version

       // 加载图表

       chartPath, err := install.LocateChart(helmApp.Spec.Chart.Name, settings)

       if err != nil {

               return nil, fmt.Errorf("无法找到 Helm 图表: %v", err)

       }

       // 转换 values 配置

       values, err := structsToValues(helmApp.Spec.Values)

       if err != nil {

               return nil, fmt.Errorf("无法转换 values 配置: %v", err)

       }

       // 执行安装

       release, err := install.Run(chartPath, values)

       if err != nil {

               return nil, fmt.Errorf("Helm 安装失败: %v", err)

       }

       return release, nil

}

构建和部署 Operator

构建镜像 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 设置镜像仓库（替换为你的仓库）

export IMG=your-registry/helm-app-operator:v0.1.0

# 构建并推送镜像

make docker-build docker-push

# 部署 Operator 到集群

make deploy

创建示例资源

创建 config/samples/demo_v1alpha1_helmapp.yaml：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

apiVersion: demo.example.com/v1alpha1

kind: HelmApp

metadata:

name: example-webapp

namespace: default

spec:

chart:

   repository: https://charts.bitnami.com/bitnami

   name: nginx

   version: "13.2.25"

   checkInterval: 30  # 门禁检查间隔30秒

targetNamespace: webapps

values:

   replicaCount: 2

   service:

     type: ClusterIP

     port: 80

   resources:

     requests:

       cpu: 100m

       memory: 128Mi

gates:

   - name: required-config

     type: ConfigMapExists

     description: 检查是否存在必要的配置文件

     mandatory: true

   - name: database-service

     type: DependencyReady

     description: 确保数据库服务已部署

     mandatory: true

   - name: production-approval

     type: ManualApproval

     description: 生产环境部署需要手动审批

     mandatory: true

应用示例资源：

1

kubectl apply -f config/samples/demo\_v1alpha1\_helmapp.yaml

控制器重新入队机制详解

在控制器开发中，理解重新入队机制至关重要，它决定了控制器何时再次处理资源：

返回值是否重新入队重试时机典型场景
ctrl.Result{}, err是立即临时错误（如网络波动）
ctrl.Result{RequeueAfter: t}, nil是延迟 t 时间后需等待的场景（如门禁未通过）
ctrl.Result{Requeue: true}, nil是立即无错误但需重试（如资源未就绪）
ctrl.Result{}, nil否不重试状态已达成（如部署完成）

在我们的 Helm Operator 中：

  1. 当门禁未通过时，使用 RequeueAfter 按配置的间隔重试：
1

return ctrl.Result{RequeueAfter: checkInterval}, nil
  1. 当部署失败时，延迟 60 秒后重试：
1

return ctrl.Result{RequeueAfter: 60 * time.Second}, nil
  1. 当状态更新失败时，根据错误类型决定重试策略：
1

return r.handleStatusUpdateError(ctx, helmApp, err)
  1. 当所有操作成功完成时，不重新入队：
1

return ctrl.Result{}, nil

测试与验证

查看资源状态 

1
2
3
4
5
6
7

# 查看 HelmApp 资源

kubectl get helmapps -o wide

# 查看详细状态

kubectl describe helmapp example-webapp

模拟门禁通过

  1. 创建所需的 ConfigMap：
1

kubectl create configmap required-config -n webapps

  1. 创建依赖的 HelmApp（如果需要）

  2. 创建手动审批 ConfigMap：

1

kubectl create configmap approval-example-webapp -n default --from-literal=approved=true
  1. 观察部署过程：
1
2
3

kubectl get pods -n webapps

kubectl get helmapps example-webapp -o jsonpath='{.status.phase}'

总结

本文详细介绍了使用 Kubebuilder 开发 Helm 部署 Operator 的完整流程，包括：

  1. 环境搭建和项目初始化

  2. 自定义资源 (CRD) 设计与实现

  3. 控制器核心逻辑开发，包括：

  • 调和循环实现

  • 门禁检查机制

  • Helm 部署集成

  • 状态管理与错误处理

  1. 部署和测试 Operator

通过这个 Operator，我们可以实现复杂应用的自动化部署流程，包括各种前置检查和审批机制，大大提高了部署的可靠性和可维护性。

Kubebuilder 提供了丰富的工具和框架，使我们能够专注于业务逻辑的实现，而不必关心 Kubernetes API 的细节。掌握 Operator 开发技能，将为你的 Kubernetes 应用管理带来更大的灵活性和自动化能力。