Kubernetes Operator入门

Operator = Controller + CRD

用户 apply 一个 CR -> API Server 存进 etcd
    -> Operator Watch 到这个 CR 的变更
    -> Controller 创建/更新原生子资源（Deployment / ConfigMap / Service / StatefulSet）
    -> 监听创建结果 -> 根据结果更新 CR 的 status 字段

                     kubebuilder              operator-sdk
──────────────      ───────────────────      ─────────────────────
主导方               Kubernetes SIG           Red Hat
语言支持             Go                        Go / Ansible / Helm
底层依赖             controller-runtime       复用 kubebuilder + 更多封装
与 OLM 集成          弱                       强（OLM 本来就是它家的）
部署承载方式          打成 kustomize / helm    可直接打成 OCI bundle，配合 OLM
定位                 核心脚手架                更完整的一站式能力

case 1  成功，无需重试     -> Result{} , nil           （从队列删除）
case 2  失败，需要重试     -> Result{},  err           （立即重新入队）
case 3  成功，但要重试     -> Result{Requeue:true}, nil（重新入队，可查次数）
case 4  延迟一段时间重试   -> Result{RequeueAfter:5s}, nil（延迟队列）

# 按官方文档 install kind（前提有 docker）
kind create cluster
kubectl get nodes   # 能看到一个单节点集群

# 安装 kubebuilder CLI
# 按官方文档安装即可（示例 3.x 版本）
kubebuilder version

mkdir application && cd application

# 初始化 Go module
go mod init application.co/application

# 用 kubebuilder 初始化项目布局，--domain 决定 API group 的后缀
# 这里演示用 co 作为 domain，生产环境换成你自己的域名段
kubebuilder init --domain co \
    --repo application.co/application \
    --license apache2

kubebuilder create api --group application --version v1 --kind Application

Create Resource under [application.co/application]?  # 生成 CRD -> y
Create Controller under [application.co/application]? # 生成 Controller -> y

package v1

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

// ApplicationDeploymentSpec 自定义的 Deployment 定义（简化）
type ApplicationDeploymentSpec struct {
	Image   string `json:"image"`
	Replicas *int32 `json:"replicas,omitempty"`
	Port    int32  `json:"port"`
}

// ApplicationSpec 定义 Spec：deployment / service / ingress
type ApplicationSpec struct {
	// Foo 是 kubebuilder 给的示例，删掉
	Deployment *ApplicationDeploymentSpec `json:"deployment"`
	// 直接复用原生 Service 的 Spec
	Service   *corev1.ServiceSpec         `json:"service,omitempty"`
	// 直接复用原生 Ingress 的 Spec
	Ingress   *networkingv1.IngressSpec    `json:"ingress,omitempty"`
}

// ApplicationStatus 记录实际状态，这里放一个可用副本数
type ApplicationStatus struct {
	AvailableReplicas int32 `json:"availableReplicas"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
type Application struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec   ApplicationSpec   `json:"spec,omitempty"`
	Status ApplicationStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true
type ApplicationList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`
	Items []Application `json:"items"`
}

func init() {
	SchemeBuilder.Register(&Application{}, &ApplicationList{})
}

make manifests

package controller

import (
	"context"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	networkingv1 "k8s.io/api/networking/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	applicationv1 "application.co/application/api/v1"
	"github.com/go-logr/logr"
)

type ApplicationReconciler struct {
	client.Client
	Scheme *runtime.Scheme
}

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

	// 1. 获取我们的 CR（Application）
	var app applicationv1.Application
	if err := r.Get(ctx, req.NamespacedName, &app); err != nil {
		// 拿不到可能是被删，不用重试
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}
	log.Info("application fetched", "name", app.Name)

	// 给子资源准备统一的 label
	labels := map[string]string{"app": app.Name}

	// 2. 创建或更新 Deployment
	if app.Spec.Deployment != nil {
		dep := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: app.Name, Namespace: app.Namespace}}
		_, err := controllerutil.CreateOrUpdate(ctx, r.Client, dep, func() error {
			replicas := app.Spec.Deployment.Replicas
			if replicas == nil {
				t := int32(1)
				replicas = &t
			}
			dep.Spec.Replicas = replicas
			dep.Spec.Selector = &metav1.LabelSelector{MatchLabels: labels}
			dep.Spec.Template = corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{Labels: labels},
				Spec: corev1.PodSpec{
					Containers: []corev1.Container{{
						Name:  app.Name,
						Image: app.Spec.Deployment.Image,
						Ports: []corev1.ContainerPort{{ContainerPort: app.Spec.Deployment.Port}},
					}},
				},
			}
			// 标记归属：让 CR 成为这个 Deployment 的所有者
			return controllerutil.SetControllerReference(&app, dep, r.Scheme)
		})
		if err != nil {
			return ctrl.Result{}, err
		}
		log.Info("deployment created/updated")
	}

	// 3. 创建或更新 Service（直接复用 CRD 里定义的 ServiceSpec）
	if app.Spec.Service != nil {
		svc := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: app.Name, Namespace: app.Namespace}}
		_, err := controllerutil.CreateOrUpdate(ctx, r.Client, svc, func() error {
			svc.Spec = *app.Spec.Service
			return controllerutil.SetControllerReference(&app, svc, r.Scheme)
		})
		if err != nil {
			return ctrl.Result{}, err
		}
		log.Info("service created/updated")
	}

	// 4. 创建或更新 Ingress
	if app.Spec.Ingress != nil {
		ing := &networkingv1.Ingress{ObjectMeta: metav1.ObjectMeta{Name: app.Name, Namespace: app.Namespace}}
		_, err := controllerutil.CreateOrUpdate(ctx, r.Client, ing, func() error {
			ing.Spec = *app.Spec.Ingress
			return controllerutil.SetControllerReference(&app, ing, r.Scheme)
		})
		if err != nil {
			return ctrl.Result{}, err
		}
		log.Info("ingress created/updated")
	}

	// 5. 更新 status（把 Deployment 的实际副本数写进去）
	_ = r.updateStatus(ctx, &app)

	return ctrl.Result{}, nil
}

func (r *ApplicationReconciler) updateStatus(ctx context.Context, app *applicationv1.Application) error {
	var dep appsv1.Deployment
	if err := r.Get(ctx, types.NamespacedName{
		Name: app.Name, Namespace: app.Namespace}, &dep); err == nil {
		app.Status.AvailableReplicas = dep.Status.AvailableReplicas
		_ = r.Status().Update(ctx, app)
	}
	return nil
}

// SetupWithManager 注册 controller：盯 Application 这个 kind，并 Owns Deployment
func (r *ApplicationReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&applicationv1.Application{}).
		Owns(&appsv1.Deployment{}).
		Complete(r)
}

# 1. 把 CRD 安装进 kind 集群（用 kustomize 方式部署 config/crd）
make install

# 2. 本地跑 operator（还没打成镜像，就在本机 make run）
make run

# 3. 写一个示例 CR：config/samples/application_v1_application.yaml

apiVersion: application.co/v1
kind: Application
metadata:
  name: application-sample
spec:
  deployment:
    image: nginx:latest
    replicas: 1
    port: 80
  service:
    ports:
      - port: 80
        targetPort: 80
        protocol: TCP
  ingress:
    ingressClassName: nginx
    rules:
      - host: aiops.example.com
        http:
          paths:
            - path: /
              pathType: Prefix
              backend:
                service:
                  name: application-sample
                  port:
                    number: 80

kubectl apply -f config/samples/application_v1_application.yaml
kubectl get deployment
kubectl get service
kubectl get ingress
# 三个名为 application-sample 的资源都出来了，说明 operator 符合预期

# 更新：改镜像版本后再次 apply，Deployment 的 image 应该跟着变
kubectl get deployment -o yaml

# 级联删除：这时 operator 并没写删除逻辑，但删 CR 三个资源会一起消失
kubectl delete -f config/samples/application_v1_application.yaml
kubectl get deployment   # 没了
kubectl get service      # 没了
kubectl get ingress      # 没了

kind: CronHPA
spec:
  scaleTargetRef:      # 要操作哪个工作负载
    apiVersion: apps/v1
    kind: Deployment
    name: nginx
  jobs:                # 定时伸缩任务（数组）
    - name: scaleUp
      schedule: "*/1 * * * *"   # 每分钟执行（测试用，生产可写每天11点等）
      targetSize: 3

mkdir cronhpa && cd cronhpa
go mod init autoscaling.co/cronhpa
kubebuilder init --domain co --repo autoscaling.co/cronhpa --license apache2

kubebuilder create api --group autoscaling --version v1 --kind CronHPA
# Create Resource -> y, Create Controller -> y

package v1

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

// ScaleTargetReference 指定目标工作负载
type ScaleTargetReference struct {
	APIVersion string `json:"apiVersion"`
	Kind       string `json:"kind"`
	Name       string `json:"name"`
}

// CronJobSpec 一个定时任务
type CronJobSpec struct {
	Name       string `json:"name"`
	Schedule   string `json:"schedule"`   // cron 表达式
	TargetSize int32  `json:"targetSize"` // 目标副本数
}

// CronHPASpec 定义 spec
type CronHPASpec struct {
	ScaleTargetRef ScaleTargetReference `json:"scaleTargetRef"`
	Jobs           []CronJobSpec         `json:"jobs"`
}

// CronHPAStatus 记录实际状态
type CronHPAStatus struct {
	CurrentReplicas int32               `json:"currentReplicas"`
	LastScaleTime   *metav1.Time        `json:"lastScaleTime,omitempty"`
	// 每个 job 各自的最后运行时间
	LastRunTime     map[string]string    `json:"lastRunTime,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:printcolumn:name="Target",type=string,JSONPath=".spec.scaleTargetRef.name",description="目标工作负载"
// +kubebuilder:printcolumn:name="Schedule",type=string,JSONPath=".spec.jobs[0].schedule",description="Cron 表达式"
// +kubebuilder:printcolumn:name="TargetSize",type=integer,JSONPath=".spec.jobs[0].targetSize",description="目标副本数"
type CronHPA struct { ... }

make manifests

go get github.com/robfig/cron/v3

func (r *CronHPAReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := log.FromContext(ctx)
	log.Info("cronhpa reconciling")

	// 获取 CR
	var chp autoscalingv1.CronHPA
	if err := r.Get(ctx, req.NamespacedName, &chp); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	now := time.Now()
	// 记录下一次应该运行的时间
	var nextRunTime *time.Time

	// 遍历每个 job，检查调度时间并更新目标副本数
	for _, job := range chp.Spec.Jobs {
		// 取上次运行时间（可能有不同 job 分别记录）
		var lastRunTime *time.Time
		if t := chp.Status.LastRunTime[job.Name]; t != "" {
			if parsed, err := time.Parse(time.RFC3339, t); err == nil {
				lastRunTime = &parsed
			}
		}
		log.Info("check job", "job", job.Name, "lastRunTime", lastRunTime)

		nextScheduleTime, err := r.getNextScheduleTime(job.Schedule, lastRunTime)
		if err != nil {
			return ctrl.Result{}, err
		}

		// 当前时间已到达/超过计划时间 -> 更新副本数
		if now.After(*nextScheduleTime) || now.Equal(*nextScheduleTime) {
			log.Info("update job", "name", job.Name, "targetSize", job.TargetSize)
			if err := r.updateDeploymentReplicas(ctx, &chp, &job); err != nil {
				log.Error(err, "failed to update replicas")
				return ctrl.Result{}, err
			}
			chp.Status.LastScaleTime = &metav1.Time{Time: now}
			chp.Status.CurrentReplicas = job.TargetSize
			// 记录该 job 的最后运行时间
			if chp.Status.LastRunTime == nil {
				chp.Status.LastRunTime = map[string]string{}
			}
			chp.Status.LastRunTime[job.Name] = now.Format(time.RFC3339)
		}

		// 计算下一次运行时间（从现在起）
		next := r.getNextScheduleTime(job.Schedule, &now)
		if next == nil { continue }
		if nextRunTime == nil || next.Before(nextRunTime) {
			nextRunTime = next
		}
	}

	_ = r.Status().Update(ctx, &chp)

	// 关键！用 RequeueAfter 实现"定时器"：到了时间再触发一次
	requeueAfter := time.Hour
	if nextRunTime != nil {
		d := nextRunTime.Sub(now)
		if d < 0 {
			d = time.Second // 已经过期，1 秒后就再试
		}
		requeueAfter = d
	}
	log.Info("will requeue", "after", requeueAfter)
	return ctrl.Result{RequeueAfter: requeueAfter}, nil
}

// getNextScheduleTime 用 robfig/cron 解析并计算下次调度时间
func (r *CronHPAReconciler) getNextScheduleTime(schedule string, after *time.Time) (*time.Time, error) {
	parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
	sched, err := parser.Parse(schedule)
	if err != nil {
		return nil, err
	}
	t := sched.Next(*after)
	return &t, nil
}

// updateDeploymentReplicas 更新目标 Deployment 的副本数
func (r *CronHPAReconciler) updateDeploymentReplicas(ctx context.Context, chp *autoscalingv1.CronHPA, job *autoscalingv1.CronJobSpec) error {
	dep := &appsv1.Deployment{}
	if err := r.Get(ctx, types.NamespacedName{
		Name: chp.Spec.ScaleTargetRef.Name, Namespace: chp.Namespace}, dep); err != nil {
		return err
	}
	// 已经是目标值就什么都不做（幂等）
	if dep.Spec.Replicas != nil && *dep.Spec.Replicas == job.TargetSize {
		return nil
	}
	dep.Spec.Replicas = &job.TargetSize
	return r.Update(ctx, dep)
}

make install
make run

apiVersion: autoscaling.co/v1
kind: CronHPA
metadata:
  name: cronhpa-sample
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: nginx
  jobs:
    - name: scaleUp
      schedule: "*/1 * * * *"   # 每分钟
      targetSize: 3

kubectl get cronhpa            # 能看到 Target / Schedule / TargetSize 列（printcolumn 生效）
kubectl get deploy nginx       # 副本从 1 -> 3（定时扩容）
kubectl scale deploy nginx --replicas=1   # 人为缩到 1
# 等一分钟，它又被拉回 3 —— 定时伸缩能力达成

# 1. 设 IMG 环境变量（含镜像仓库路径 + 版本号）。
#    省略 docker.io 前缀时默认推 Docker Hub。
export IMG=docker.io/<your-name>/application-operator:v0.1

# 2. 构建并推送 operator 镜像（项目根目录有自动生成的 Dockerfile，make 帮你简化）
make docker-build # 若本机有 docker-daemon
make docker-push

# 3. 生成一个"一键部署"的 install 清单文件，输出到 dist/ 目录
make build-installer

# dist/install.yaml 里包含 namespace、CRD、ServiceAccount、ClusterRole、
# ClusterRoleBinding、operator 自身的工作负载（Deployment）……整套都齐了
kubectl apply -f dist/install.yaml

# 先安装 operator-sdk CLI
# 创建一个 redis-operator 目录
mkdir redis-operator && cd redis-operator
go mod init redis.co/redis-operator

# 初始化，注意插件要指定为 helm
operator-sdk init --domain co --plugins=helm

# 创建 API：_--helm-chart 指定一个 Helm Chart。这里先给仓库地址下载
# 第一次可能因依赖构建报错：进 chart 目录手动 build 依赖
operator-sdk create api --group app --version v1 \
    --kind Redis --helm-chart=embed/prefix?  # 略，实际写 chart 源

operator-sdk create api --group app --version v1 \
    --kind Redis --helm-chart=helm-charts/redis

# 构建 operator 镜像并 push（直接通过 --image 传 IMG）
operator-sdk build --image docker.io/<name>/redis-operator:v0.1
docker push docker.io/<name>/redis-operator:v0.1

# 部署 operator 到集群
make deploy IMG=docker.io/<name>/redis-operator:v0.1
kubectl get ns     # 会新增一个 redis-system，operator 工作负载部署在这里

# 部署示例 CR（相当于声明一个 redis 实例）
kubectl apply -f config/samples/app_v1_redis.yaml
kubectl get pods    # redis 应该被拉起来

is forbidden: usercannot get resource ... API group ...

kubectl get deployment -n redis-system -o yaml | grep serviceAccount

# 把该 SA 绑到集群级权限上（演示用，正式环境应最小授权）
kubectl create clusterrolebinding redis-operator-cluster-admin \
    --clusterrole=cluster-admin \
    --serviceaccount=redis-system:redis-operator-controller-manager

kubectl rollout restart deployment -n redis-system

# 0. 先把之前手动部署的 redis-operator 卸载干净
make uninstall

# 1. 在集群里安装 OLM（会安装一批工作负载）
operator-sdk olm install

# 2. 设两个环境变量：operator 镜像 + bundle 镜像
export IMG=docker.io/<your-name>/redis-operator:v0.1
export BUNDLE_IMG=quay.io/<your-name>/redis-operator-bundle:v0.0.1

# 3. 生成 bundle（交互式，会问你一堆元信息）
make bundle
#    它会问你 operator 名称(redis-operator)、description、keywords、developer、email
#    生成完成后 bundle 名为 docker.io/xxx/redis-operator-bundle:v0.0.1

# 4. 构建并推送 bundle 镜像
make bundle-build bundle-push

# 5. 校验 bundle 是否合法
operator-sdk bundle validate "$BUNDLE_IMG"

# 6. 一条命令在任意集群安装 operator
operator-sdk run bundle "$BUNDLE_IMG"

# 1. 安装并 setup（会装 kube-apiserver 和 etcd 两个二进制）
setup-envtest use latest

# 2. 让测试环境能读取到 envtest 的二进制目录
#    默认从某个目录读 kube-apiserver 和 etcd，用 setup-envtest 设置 KUBEBUILDER_ASSETS 环境变量

internal/controller/
├── suite_test.go     # 测试入口（BeforeSuite / AfterSuite）
└── <name>_controller_test.go   # 写测试用例的地方

package controller

import (
	"path/filepath"
	"testing"

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

	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/envtest"
	applicationv1 "application.co/application/api/v1"
	ctrl "sigs.k8s.io/controller-runtime"
)

var (
	cfg       *rest.Config
	k8sClient client.Client
	testEnv   *envtest.Environment
)

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

var _ = BeforeSuite(func() {
	ctrl.SetLogger(zap.New(zap.UseDevMode(true)))

	// 初始化 envtest 环境，把我们的 CRD 目录挂进来
	testEnv = &envtest.Environment{
		CRDDirectoryPaths:     []string{filepath.Join("..", "..", "config", "crd", "bases")},
		ErrorIfCRDPathMissing: true,
	}

	var err error
	cfg, err = testEnv.Start() // 启动了一个 K8s 集群（无控制器）
	Expect(err).NotTo(HaveOccurred())

	Expect(applicationv1.AddToScheme(scheme.Scheme)).NotTo(HaveOccurred())

	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
	Expect(err).NotTo(HaveOccurred())
})

var _ = AfterSuite(func() {
	_ = testEnv.Stop()
})

var _ = Describe("CRD e2e", func() {
	Context("create and fetch", func() {
		It("should create and fetch CRD", func() {
			myResource := &applicationv1.Application{
				ObjectMeta: metav1.ObjectMeta{Name: "test", Namespace: "default"},
				Spec: applicationv1.ApplicationSpec{ /* 填 spec */ },
			}
			Expect(k8sClient.Create(context.Background(), myResource)).To(Succeed())

			var got applicationv1.Application
			Expect(k8sClient.Get(context.Background(), client.ObjectKeyFromObject(myResource), &got)).To(Succeed())
			// 这里还可以继续验证 spec 字段 / status 字段
		})
	}
})

make test   # 或按项目 Makefile 里的 target