基于 MySQL Operator 进行二次开发

本教程将基于成熟的开源项目 MySQL Operator，完成一次完整的 Kubernetes Operator 二次开发实战。

Operator 开发 = 基于 Kubernetes 扩展机制，编写「自定义资源 CRD + 自定义控制器 Controller」，把人工运维经验写成程序，全自动管理复杂应用全生命周期的云原生开发工作。

简单类比：
K8s 自带控制器只会管 Pod、Deployment、Service 等通用资源；
Operator 是专属应用的智能运维机器人，懂数据库、中间件等业务专属运维逻辑，7×24 自动调谐集群状态。

概述

本练习将指导在现有的 MySQL Operator 项目基础上，开发一个新功能：MysqlBackupPolicy（备份策略管理）。通过这个练习，将学习完整的 Operator 开发流程。

练习目标

• ✅ 学习如何定义新的 CRD
• ✅ 掌握Controller控制器开发方法
• ✅ 理解同步器模式的应用
• ✅ 学会编写测试
• ✅ 理解代码生成流程

几个核心概念

• Operator：基于 CRD + 控制器，封装应用运维逻辑，用代码替代人工操作 K8s 资源；
• CRD(CustomResourceDefinition)：自定义资源类型，扩展 K8s API；
• Controller：无限循环 List-Watch-Reconcile，调谐资源期望状态与实际状态；
• Reconcile（调谐函数）：Operator 核心逻辑入口，每次资源变更触发执行；

主流开发框架：

• Kubebuilder（官方推荐，本次教程使用）
• Operator SDK（封装 kubebuilder 上层工具）
• Client-go（原生底层，无封装，适合深度定制）

功能设计

要做什么

添加一个新的自定义资源 MysqlBackupPolicy，让用户可以：

• 配置多个备份计划（每天、每周等）
• 设置备份保留时间
• 查看备份执行历史

API 设计

apiVersion: mysql.presslabs.org/v1alpha1
kind: MysqlBackupPolicy
metadata:
  name: my-backup-policy
spec:
  # 关联的 MySQL 集群
  clusterName: my-cluster
  
  # 备份计划列表
  schedules:
    - name: daily
      cron: "0 2 * * *"           # 每天凌晨2点
      backupURL: s3://my-bucket/daily/
      retention: 168h             # 保留7天 (168小时)
    
    - name: weekly
      cron: "0 3 * * 0"          # 每周日凌晨3点
      backupURL: s3://my-bucket/weekly/
      retention: 720h            # 保留30天

status:
  # 每个计划的最后执行时间
  lastBackupTimes:
    daily: "2024-01-15T02:00:00Z"
    weekly: "2024-01-14T03:00:00Z"
  
  # 整体状态
  conditions:
    - type: Ready
      status: "True"

---

开发步骤详解

---

第一步：环境准备

1.1 确保开发环境就绪

# 克隆项目（如果还没有）
git clone https://github.com/bitpoke/mysql-operator.git
cd mysql-operator

# 检查 Go 版本
go version  # 需要 1.16+

# 安装依赖工具
make

1.2 了解项目结构

花 10 分钟快速浏览以下目录：

• pkg/apis/mysql/v1alpha1/ - CRD 定义
• pkg/controller/ - 控制器实现
• pkg/internal/ - 业务逻辑

---

第二步：定义 CRD 类型

2.1 创建类型定义文件

开发 MysqlBackupPolicy自定义资源，API 版本 v1alpha1，Kind=MysqlBackupPolicy

# 新项目情况下，执行如下命令自动生成mysqlbackuppolicy_types.go文件，然后添加自定义内容
kubebuilder create api --group mysql --version v1alpha1 --kind MysqlBackupPolicy --force

手动在 pkg/apis/mysql/v1alpha1/ 目录下创建新文件 mysqlbackuppolicy_types.go：

/*
Copyright 2024 Pressinfra SRL

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package v1alpha1

import (
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.

// MysqlBackupPolicySpec defines the desired state of MysqlBackupPolicy
type MysqlBackupPolicySpec struct {
	// ClusterName is the name of the MysqlCluster to backup
	ClusterName string `json:"clusterName"`

	// Schedules is a list of backup schedules
	Schedules []BackupSchedule `json:"schedules"`
}

// BackupSchedule defines a backup schedule
type BackupSchedule struct {
	// Name is the unique name of this schedule
	Name string `json:"name"`

	// Cron is the cron expression for scheduling
	Cron string `json:"cron"`

	// BackupURL is the location where to store backups
	BackupURL string `json:"backupURL"`

	// Retention is how long to keep backups (e.g., 168h for 7 days)
	Retention string `json:"retention,omitempty"`

	// BackupSecretName is the secret with credentials for backup storage
	BackupSecretName string `json:"backupSecretName,omitempty"`
}

// MysqlBackupPolicyStatus defines the observed state of MysqlBackupPolicy
type MysqlBackupPolicyStatus struct {
	// LastBackupTimes records the last backup time for each schedule
	LastBackupTimes map[string]metav1.Time `json:"lastBackupTimes,omitempty"`

	// Conditions represents the latest available observations of the policy's current state
	Conditions []MysqlBackupPolicyCondition `json:"conditions,omitempty"`
}

// MysqlBackupPolicyCondition describes the state of a backup policy at a certain point
type MysqlBackupPolicyCondition struct {
	// Type of the condition
	Type MysqlBackupPolicyConditionType `json:"type"`

	// Status of the condition
	Status corev1.ConditionStatus `json:"status"`

	// LastTransitionTime is the last time the condition transitioned from one status to another
	LastTransitionTime metav1.Time `json:"lastTransitionTime"`

	// Reason is the reason for the condition's last transition
	Reason string `json:"reason,omitempty"`

	// Message is a human readable message indicating details about the transition
	Message string `json:"message,omitempty"`
}

// MysqlBackupPolicyConditionType is a valid condition type
type MysqlBackupPolicyConditionType string

const (
	// MysqlBackupPolicyReady means the backup policy is ready and active
	MysqlBackupPolicyReady MysqlBackupPolicyConditionType = "Ready"
)

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:printcolumn:name="Ready",type="string",JSONPath=".status.conditions[?(@.type=='Ready')].status"
// +kubebuilder:printcolumn:name="Cluster",type="string",JSONPath=".spec.clusterName"
// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"

// MysqlBackupPolicy is the Schema for the mysqlbackuppolicies API
type MysqlBackupPolicy struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec   MysqlBackupPolicySpec   `json:"spec,omitempty"`
	Status MysqlBackupPolicyStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true

// MysqlBackupPolicyList contains a list of MysqlBackupPolicy
type MysqlBackupPolicyList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`
	Items           []MysqlBackupPolicy `json:"items"`
}

func init() {
	SchemeBuilder.Register(&MysqlBackupPolicy{}, &MysqlBackupPolicyList{})
}

2.2 验证你理解了什么

概念	说明
Spec	用户期望的状态（输入）
Status	实际观察到的状态（输出）
+kubebuilder 注解	用于生成 CRD 和辅助代码
json:"..." 标签	控制序列化字段名

---

第三步：代码生成

3.1 生成 DeepCopy 函数

Kubernetes 要求所有 API 类型必须实现 DeepCopy 接口。让我们生成这些函数：

# 在项目根目录运行
make generate

查看生成的文件：

• pkg/apis/mysql/v1alpha1/zz_generated.deepcopy.go

3.2 生成 CRD 清单

make manifests

查看生成的文件：

• config/crd/bases/mysql.presslabs.org_mysqlbackuppolicies.yaml

3.3 检查生成结果

确保：

• ✅ zz_generated.deepcopy.go 包含了新类型的 DeepCopy 方法
• ✅ CRD YAML 文件已生成在 config/crd/bases/

---

第四步：添加内部业务逻辑

4.1 创建内部包装器

在 pkg/internal/ 目录下创建新目录 mysqlbackuppolicy/，并创建文件 mysqlbackuppolicy.go：

/*
Copyright 2024 Pressinfra SRL

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package mysqlbackuppolicy

import (
	"time"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	mysqlv1alpha1 "github.com/bitpoke/mysql-operator/pkg/apis/mysql/v1alpha1"
)

// MysqlBackupPolicy wraps a mysqlv1alpha1.MysqlBackupPolicy for adding business methods
type MysqlBackupPolicy struct {
	*mysqlv1alpha1.MysqlBackupPolicy
}

// New returns a wrapper for the given policy
func New(p *mysqlv1alpha1.MysqlBackupPolicy) *MysqlBackupPolicy {
	return &MysqlBackupPolicy{p}
}

// Unwrap returns the wrapped policy
func (p *MysqlBackupPolicy) Unwrap() *mysqlv1alpha1.MysqlBackupPolicy {
	return p.MysqlBackupPolicy
}

// SetDefaults sets default values on the policy spec
func (p *MysqlBackupPolicy) SetDefaults() {
	// 为每个计划设置默认值
	for i := range p.Spec.Schedules {
		schedule := &p.Spec.Schedules[i]
		
		// 默认保留 7 天
		if schedule.Retention == "" {
			schedule.Retention = "168h"
		}
	}
}

// Validate validates the policy spec
func (p *MysqlBackupPolicy) Validate() error {
	var errs []error

	// 验证集群名称
	if p.Spec.ClusterName == "" {
		errs = append(errs, ErrClusterNameRequired)
	}

	// 验证至少有一个计划
	if len(p.Spec.Schedules) == 0 {
		errs = append(errs, ErrAtLeastOneSchedule)
	}

	// 验证每个计划
	for _, schedule := range p.Spec.Schedules {
		if schedule.Name == "" {
			errs = append(errs, ErrScheduleNameRequired)
		}
		if schedule.Cron == "" {
			errs = append(errs, ErrCronRequired)
		}
		if schedule.BackupURL == "" {
			errs = append(errs, ErrBackupURLRequired)
		}
	}

	if len(errs) > 0 {
		return NewValidationError(errs...)
	}

	return nil
}

// UpdateStatusCondition updates a status condition
func (p *MysqlBackupPolicy) UpdateStatusCondition(
	condType mysqlv1alpha1.MysqlBackupPolicyConditionType,
	status corev1.ConditionStatus,
	reason, message string,
) {
	now := metav1.Now()

	// 查找现有条件
	existingIdx := -1
	for i, cond := range p.Status.Conditions {
		if cond.Type == condType {
			existingIdx = i
			break
		}
	}

	condition := mysqlv1alpha1.MysqlBackupPolicyCondition{
		Type:               condType,
		Status:             status,
		LastTransitionTime: now,
		Reason:             reason,
		Message:            message,
	}

	if existingIdx >= 0 {
		// 更新现有条件
		oldCond := p.Status.Conditions[existingIdx]
		if oldCond.Status == status {
			// 状态未变，保留原来的 transition time
			condition.LastTransitionTime = oldCond.LastTransitionTime
		}
		p.Status.Conditions[existingIdx] = condition
	} else {
		// 添加新条件
		p.Status.Conditions = append(p.Status.Conditions, condition)
	}
}

// UpdateLastBackupTime updates the last backup time for a schedule
func (p *MysqlBackupPolicy) UpdateLastBackupTime(scheduleName string, t time.Time) {
	if p.Status.LastBackupTimes == nil {
		p.Status.LastBackupTimes = make(map[string]metav1.Time)
	}
	p.Status.LastBackupTimes[scheduleName] = metav1.NewTime(t)
}

// GetLastBackupTime returns the last backup time for a schedule
func (p *MysqlBackupPolicy) GetLastBackupTime(scheduleName string) *time.Time {
	if p.Status.LastBackupTimes == nil {
		return nil
	}
	if t, ok := p.Status.LastBackupTimes[scheduleName]; ok {
		return &t.Time
	}
	return nil
}

4.2 创建错误定义

在同一个目录下创建 errors.go：

package mysqlbackuppolicy

import (
	"fmt"
	"strings"

	utilerrors "k8s.io/apimachinery/pkg/util/errors"
)

// Validation errors
var (
	ErrClusterNameRequired  = fmt.Errorf("clusterName is required")
	ErrAtLeastOneSchedule   = fmt.Errorf("at least one schedule is required")
	ErrScheduleNameRequired = fmt.Errorf("schedule name is required")
	ErrCronRequired         = fmt.Errorf("cron expression is required")
	ErrBackupURLRequired    = fmt.Errorf("backupURL is required")
)

// ValidationError represents a validation error
type ValidationError struct {
	Errors []error
}

// NewValidationError creates a new ValidationError
func NewValidationError(errs ...error) *ValidationError {
	return &ValidationError{Errors: errs}
}

// Error implements the error interface
func (e *ValidationError) Error() string {
	var msgs []string
	for _, err := range e.Errors {
		msgs = append(msgs, err.Error())
	}
	return fmt.Sprintf("validation failed: %s", strings.Join(msgs, "; "))
}

// Aggregate aggregates errors
func Aggregate(errs []error) error {
	if len(errs) == 0 {
		return nil
	}
	return utilerrors.NewAggregate(errs)
}

---

第五步：实现控制器

5.1 创建控制器目录结构

在 pkg/controller/ 下创建新目录 mysqlbackuppolicy/：

pkg/controller/mysqlbackuppolicy/
├── mysqlbackuppolicy_controller.go
└── mysqlbackuppolicy_controller_suite_test.go

5.2 创建控制器主文件

创建 mysqlbackuppolicy_controller.go：

/*
Copyright 2024 Pressinfra SRL

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package mysqlbackuppolicy

import (
	"context"
	"reflect"
	"time"

	"github.com/presslabs/controller-util/syncer"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/source"

	mysqlv1alpha1 "github.com/bitpoke/mysql-operator/pkg/apis/mysql/v1alpha1"
	"github.com/bitpoke/mysql-operator/pkg/controller/mysqlbackuppolicy/internal/syncer"
	"github.com/bitpoke/mysql-operator/pkg/internal/mysqlbackuppolicy"
)

const controllerName = "mysqlbackuppolicy-controller"

// Reconciler reconciles a MysqlBackupPolicy object
type Reconciler struct {
	client.Client
	Scheme   *runtime.Scheme
	Recorder record.EventRecorder
}

// +kubebuilder:rbac:groups=mysql.presslabs.org,resources=mysqlbackuppolicies,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=mysql.presslabs.org,resources=mysqlbackuppolicies/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=batch,resources=cronjobs,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=events,verbs=get;list;watch;create;update;patch

// Reconcile handles the reconciliation loop
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := log.FromContext(ctx).WithName(controllerName)
	log.Info("Reconciling MysqlBackupPolicy", "name", req.Name, "namespace", req.Namespace)

	// 1. 获取策略对象
	policy := mysqlbackuppolicy.New(&mysqlv1alpha1.MysqlBackupPolicy{})
	err := r.Get(ctx, req.NamespacedName, policy.Unwrap())
	if err != nil {
		if errors.IsNotFound(err) {
			// 对象已删除
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, err
	}

	// 2. 设置默认值
	policy.SetDefaults()

	// 3. 验证
	if err := policy.Validate(); err != nil {
		log.Error(err, "Policy validation failed")
		policy.UpdateStatusCondition(
			mysqlv1alpha1.MysqlBackupPolicyReady,
			corev1.ConditionFalse,
			"ValidationFailed",
			err.Error(),
		)
		r.Status().Update(ctx, policy.Unwrap())
		return ctrl.Result{}, err
	}

	// 4. 保存当前状态用于比较
	oldStatus := policy.Status.DeepCopy()
	defer func() {
		if !reflect.DeepEqual(oldStatus, policy.Status) {
			log.Info("Updating policy status")
			if err := r.Status().Update(ctx, policy.Unwrap()); err != nil {
				log.Error(err, "Failed to update status")
			}
		}
	}()

	// 5. 设置 Ready 条件为 True
	policy.UpdateStatusCondition(
		mysqlv1alpha1.MysqlBackupPolicyReady,
		corev1.ConditionTrue,
		"PolicyActive",
		"Backup policy is active",
	)

	// 6. 为每个备份计划创建 CronJob
	for _, schedule := range policy.Spec.Schedules {
		cronSyncer := syncer.NewCronJobSyncer(r.Client, r.Scheme, policy, schedule)
		if err := syncer.Sync(ctx, cronSyncer, r.Recorder); err != nil {
			return ctrl.Result{}, err
		}
	}

	// 7. 5分钟后重新检查（可选，用于处理保留策略等）
	return ctrl.Result{RequeueAfter: 5 * time.Minute}, nil
}

// SetupWithManager sets up the controller with the Manager
func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		Named(controllerName).
		For(&mysqlv1alpha1.MysqlBackupPolicy{}).
		Owns(&batchv1.CronJob{}).
		Complete(r)
}

5.3 创建 CronJob 同步器

在 pkg/controller/mysqlbackuppolicy/internal/syncer/ 目录下创建 cronjob.go：

package syncer

import (
	"fmt"

	"github.com/presslabs/controller-util/syncer"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"

	mysqlv1alpha1 "github.com/bitpoke/mysql-operator/pkg/apis/mysql/v1alpha1"
	"github.com/bitpoke/mysql-operator/pkg/internal/mysqlbackuppolicy"
)

// CronJobSyncer syncs a CronJob for a backup schedule
type CronJobSyncer struct {
	client.Client
	scheme   *runtime.Scheme
	policy   *mysqlbackuppolicy.MysqlBackupPolicy
	schedule mysqlv1alpha1.BackupSchedule
	cronJob  *batchv1.CronJob
}

// NewCronJobSyncer creates a new CronJob syncer
func NewCronJobSyncer(
	c client.Client,
	scheme *runtime.Scheme,
	policy *mysqlbackuppolicy.MysqlBackupPolicy,
	schedule mysqlv1alpha1.BackupSchedule,
) syncer.Interface {
	return &CronJobSyncer{
		Client:   c,
		scheme:   scheme,
		policy:   policy,
		schedule: schedule,
		cronJob:  newCronJobForSchedule(policy, schedule),
	}
}

// newCronJobForSchedule creates a CronJob for a backup schedule
func newCronJobForSchedule(
	policy *mysqlbackuppolicy.MysqlBackupPolicy,
	schedule mysqlv1alpha1.BackupSchedule,
) *batchv1.CronJob {
	labels := map[string]string{
		"app":                          "mysql-operator",
		"mysql.presslabs.org/policy":  policy.Name,
		"mysql.presslabs.org/schedule": schedule.Name,
	}

	// 创建备份 Job 的命令
	backupName := fmt.Sprintf("auto-%s-$(date +%%Y%%m%%d-%%H%%M%%S)", schedule.Name)
	cmd := []string{
		"kubectl", "create", "-f", "-",
	}
	backupYAML := fmt.Sprintf(`
apiVersion: mysql.presslabs.org/v1alpha1
kind: MysqlBackup
metadata:
  generateName: %s-
  namespace: %s
spec:
  clusterName: %s
  backupURL: %s
`, backupName, policy.Namespace, policy.Spec.ClusterName, schedule.BackupURL)

	return &batchv1.CronJob{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-%s", policy.Name, schedule.Name),
			Namespace: policy.Namespace,
			Labels:    labels,
		},
		Spec: batchv1.CronJobSpec{
			Schedule: schedule.Cron,
			JobTemplate: batchv1.JobTemplateSpec{
				Spec: batchv1.JobSpec{
					Template: corev1.PodTemplateSpec{
						Spec: corev1.PodSpec{
							RestartPolicy: corev1.RestartPolicyOnFailure,
							Containers: []corev1.Container{
								{
									Name:    "create-backup",
									Image:   "bitpoke/kubectl:latest",
									Command: cmd,
									Stdin:   true,
									// 这里简化处理，实际需要配置 ServiceAccount
								},
							},
						},
					},
				},
			},
		},
	}
}

// GetObject implements syncer.Interface
func (s *CronJobSyncer) GetObject() client.Object {
	key := client.ObjectKeyFromObject(s.cronJob)
	obj := &batchv1.CronJob{}
	if err := s.Get(context.TODO(), key, obj); err != nil {
		if errors.IsNotFound(err) {
			return nil
		}
	}
	return obj
}

// Object implements syncer.Interface
func (s *CronJobSyncer) Object() client.Object {
	return s.cronJob
}

// Tidy implements syncer.Interface
func (s *CronJobSyncer) Tidy() error {
	// 设置 OwnerReference
	return nil
}

// Sync implements syncer.Interface
func (s *CronJobSyncer) Sync(ctx context.Context) (bool, error) {
	// 简化实现，仅检查是否存在
	key := client.ObjectKeyFromObject(s.cronJob)
	existing := &batchv1.CronJob{}
	err := s.Get(ctx, key, existing)
	
	if err != nil {
		if errors.IsNotFound(err) {
			// 需要创建
			return true, nil
		}
		return false, err
	}
	
	// 已存在，检查是否需要更新
	needsUpdate := false
	if existing.Spec.Schedule != s.schedule.Cron {
		needsUpdate = true
		existing.Spec.Schedule = s.schedule.Cron
	}
	
	return needsUpdate, nil
}

---

第六步：注册控制器

6.1 创建控制器注册文件

在 pkg/controller/ 下创建 add_mysqlbackuppolicy.go：

/*
Copyright 2024 Pressinfra SRL

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controller

import (
	"sigs.k8s.io/controller-runtime/pkg/manager"

	"github.com/bitpoke/mysql-operator/pkg/controller/mysqlbackuppolicy"
)

func init() {
	// 将控制器添加到注册列表
	AddToManagerFuncs = append(AddToManagerFuncs, mysqlbackuppolicy.Add)
}

6.2 更新控制器的 Add 函数

在 pkg/controller/mysqlbackuppolicy/ 下添加 mysqlbackuppolicy_controller.go 的 Add 函数（在文件末尾添加）：

// Add creates a new MysqlBackupPolicy Controller and adds it to the Manager
func Add(mgr manager.Manager) error {
	return add(mgr, NewReconciler(mgr))
}

// NewReconciler creates a new reconciler
func NewReconciler(mgr manager.Manager) *Reconciler {
	return &Reconciler{
		Client:   mgr.GetClient(),
		Scheme:   mgr.GetScheme(),
		Recorder: mgr.GetEventRecorderFor(controllerName),
	}
}

// add adds a new Controller to mgr with r as the reconcile.Reconciler
func add(mgr manager.Manager, r *Reconciler) error {
	// 创建控制器
	c, err := controller.New(controllerName, mgr, controller.Options{Reconciler: r})
	if err != nil {
		return err
	}

	// 监听 MysqlBackupPolicy 资源
	err = c.Watch(&source.Kind{Type: &mysqlv1alpha1.MysqlBackupPolicy{}}, &handler.EnqueueRequestForObject{})
	if err != nil {
		return err
	}

	// 监听 CronJob 资源
	err = c.Watch(&source.Kind{Type: &batchv1.CronJob{}}, &handler.EnqueueRequestForOwner{
		IsController: true,
		OwnerType:    &mysqlv1alpha1.MysqlBackupPolicy{},
	})
	if err != nil {
		return err
	}

	return nil
}

---

第七步：创建示例资源

在 config/samples/ 目录下创建 mysql_v1alpha1_mysqlbackuppolicy.yaml：

apiVersion: mysql.presslabs.org/v1alpha1
kind: MysqlBackupPolicy
metadata:
  name: sample-backup-policy
spec:
  clusterName: sample-cluster
  schedules:
    - name: daily
      cron: "0 2 * * *"
      backupURL: s3://my-bucket/daily/
      retention: 168h
    
    - name: weekly
      cron: "0 3 * * 0"
      backupURL: s3://my-bucket/weekly/
      retention: 720h

---

第八步：编译和测试

8.1 运行代码生成和编译

# 生成代码
make generate
make manifests

# 编译检查
go build ./...

8.2 修复编译错误（常见问题）

常见问题 1：导入缺失

• 确保所有需要的包都已导入

常见问题 2：类型不匹配

• 检查 client.Object 的使用

8.3 运行单元测试（可选，高级练习）

创建 pkg/internal/mysqlbackuppolicy/mysqlbackuppolicy_test.go：

package mysqlbackuppolicy

import (
	"testing"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"

	mysqlv1alpha1 "github.com/bitpoke/mysql-operator/pkg/apis/mysql/v1alpha1"
)

func TestMysqlBackupPolicy(t *testing.T) {
	RegisterFailHandler(Fail)
	RunSpecs(t, "MysqlBackupPolicy Suite")
}

var _ = Describe("MysqlBackupPolicy", func() {
	Describe("SetDefaults", func() {
		It("should set default retention", func() {
			p := New(&mysqlv1alpha1.MysqlBackupPolicy{
				Spec: mysqlv1alpha1.MysqlBackupPolicySpec{
					Schedules: []mysqlv1alpha1.BackupSchedule{
						{Name: "test", Cron: "* * * * *", BackupURL: "s3://test/"},
					},
				},
			})
			p.SetDefaults()
			Expect(p.Spec.Schedules[0].Retention).To(Equal("168h"))
		})
	})

	Describe("Validate", func() {
		It("should fail if clusterName is empty", func() {
			p := New(&mysqlv1alpha1.MysqlBackupPolicy{})
			err := p.Validate()
			Expect(err).To(HaveOccurred())
		})

		It("should pass for valid policy", func() {
			p := New(&mysqlv1alpha1.MysqlBackupPolicy{
				Spec: mysqlv1alpha1.MysqlBackupPolicySpec{
					ClusterName: "test-cluster",
					Schedules: []mysqlv1alpha1.BackupSchedule{
						{Name: "daily", Cron: "0 2 * * *", BackupURL: "s3://test/"},
					},
				},
			})
			p.SetDefaults()
			err := p.Validate()
			Expect(err).NotTo(HaveOccurred())
		})
	})
})

运行测试：

go test ./pkg/internal/mysqlbackuppolicy/...

---

第九步：本地运行测试

9.1 安装 CRD 到集群

make install

9.2 本地运行 Operator

make run

9.3 测试创建资源

在另一个终端窗口：

# 首先需要一个 MysqlCluster
kubectl apply -f config/samples/mysql_v1alpha1_mysqlcluster.yaml

# 创建我们的备份策略
kubectl apply -f config/samples/mysql_v1alpha1_mysqlbackuppolicy.yaml

# 查看状态
kubectl get mysqlbackuppolicy
kubectl get cronjobs

---

练习总结与扩展

你学到了什么

1. CRD 定义 - 如何定义新的 API 类型
2. 代码生成 - 使用 Kubebuilder 工具生成辅助代码
3. 包装器模式 - 如何为 CRD 添加业务逻辑
4. 控制器开发 - 实现 Reconcile 循环
5. 同步器模式 - 使用 Syncer 管理从属资源
6. 控制器注册 - 如何将控制器添加到 Manager

进一步改进的建议（选做）

1. 完善 CronJob 同步器 - 正确实现 OwnerReference 和完整的同步逻辑
2. 备份保留策略 - 实现清理过期备份的逻辑
3. 备份验证 - 添加备份验证功能
4. Metrics - 添加 Prometheus 指标
5. Webhook - 添加验证 Webhook
6. E2E 测试 - 编写端到端测试

调试技巧

# 查看 Operator 日志
# （在运行 make run 的终端窗口）

# 查看资源详情
kubectl describe mysqlbackuppolicy sample-backup-policy

# 查看事件
kubectl get events --watch

---

常见问题解答

Q: 代码生成后没有看到我的新类型？

A: 检查以下几点：

1. 文件是否在正确的目录
2. init() 函数中是否调用了 SchemeBuilder.Register()
3. 运行 make generate 后检查 zz_generated.deepcopy.go

Q: 控制器不工作？

A: 调试步骤：

1. 检查控制器是否成功启动（看日志）
2. 检查 CRD 是否已安装
3. 检查 RBAC 权限是否足够
4. 查看 kubectl get events

Q: 如何使用真正的 Kubebuilder 脚手架？

A: 对于新项目，可以使用：

kubebuilder init --domain example.com --repo example.com/myoperator
kubebuilder create api --group mysql --version v1alpha1 --kind MysqlBackupPolicy

但在这个练习中，我们是手动添加到现有项目。

---

恭喜！你已经完成了一个完整的 Operator 功能开发！