3 posts tagged with "tutorial"

package v1alpha1

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

// RedisClusterSpec 定义用户期望的集群状态
type RedisClusterSpec struct {
	// 主节点数量（必须为奇数）
	MasterCount int32 `json:"masterCount"`

	// 每个主节点的从节点数量
	ReplicasPerMaster int32 `json:"replicasPerMaster"`

	// Redis 镜像
	Image string `json:"image"`

	// 资源请求与限制
	Resources ResourceRequirements `json:"resources,omitempty"`
}

type ResourceRequirements struct {
	CPURequest    string `json:"cpuRequest,omitempty"`
	MemoryRequest string `json:"memoryRequest,omitempty"`
	CPULimit      string `json:"cpuLimit,omitempty"`
	MemoryLimit   string `json:"memoryLimit,omitempty"`
}

// ClusterPhase 表示集群当前阶段
type ClusterPhase string

const (
	ClusterPhaseCreating  ClusterPhase = "Creating"
	ClusterPhaseInitializing ClusterPhase = "Initializing"
	ClusterPhaseRunning   ClusterPhase = "Running"
	ClusterPhaseScaling   ClusterPhase = "Scaling"
	ClusterPhaseFailed    ClusterPhase = "Failed"
)

// RedisClusterStatus 反映集群的实际状态
type RedisClusterStatus struct {
	// 集群当前阶段
	Phase ClusterPhase `json:"phase,omitempty"`

	// 各节点的就绪状态
	ReadyNodes int32 `json:"readyNodes,omitempty"`

	// 集群是否已初始化
	ClusterInitialized bool `json:"clusterInitialized,omitempty"`

	// 最近一次操作的消息
	Message string `json:"message,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:resource:shortName=rc

// RedisCluster 是自定义资源的顶层类型
type RedisCluster struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec   RedisClusterSpec   `json:"spec,omitempty"`
	Status RedisClusterStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true

// RedisClusterList 用于列表操作
type RedisClusterList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`

	Items []RedisCluster `json:"items"`
}

func init() {
	SchemeBuilder.Register(&RedisCluster{}, &RedisClusterList{})
}

package controllers

import (
	"context"
	"fmt"
	"time"

	redisv1alpha1 "github.com/rainlib/redis-operator/api/v1alpha1"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/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"
)

const (
	finalizerName = "rainlib.io/rediscluster-finalizer"
)

// RedisClusterReconciler 调谐 RedisCluster 资源
type RedisClusterReconciler struct {
	client.Client
	Scheme *runtime.Scheme
}

// +kubebuilder:rbac:groups=rainlib.io,resources=redisclusters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=rainlib.io,resources=redisclusters/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=rainlib.io,resources=redisclusters/finalizers,verbs=update
// +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch
// +kubebuilder:rbac:groups=core,resources=events,verbs=create;patch

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

	// 1. 获取 RedisCluster 实例
	var rc redisv1alpha1.RedisCluster
	if err := r.Get(ctx, req.NamespacedName, &rc); err != nil {
		if errors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, err
	}

	// 2. 处理 Finalizer（删除逻辑）
	if !rc.DeletionTimestamp.IsZero() {
		if controllerutil.ContainsFinalizer(&rc, finalizerName) {
			logger.Info("执行清理逻辑")
			// 这里可以添加资源清理逻辑（如外部存储清理）
			controllerutil.RemoveFinalizer(&rc, finalizerName)
			if err := r.Update(ctx, &rc); err != nil {
				return ctrl.Result{}, err
			}
		}
		return ctrl.Result{}, nil
	}

	// 3. 添加 Finalizer
	if !controllerutil.ContainsFinalizer(&rc, finalizerName) {
		controllerutil.AddFinalizer(&rc, finalizerName)
		if err := r.Update(ctx, &rc); err != nil {
			return ctrl.Result{}, err
		}
	}

	// 4. 创建/更新 Headless Service
	if err := r.reconcileService(ctx, &rc); err != nil {
		logger.Error(err, "调和 Service 失败")
		r.updateStatus(ctx, &rc, redisv1alpha1.ClusterPhaseFailed, 0, false, err.Error())
		return ctrl.Result{}, err
	}

	// 5. 创建/更新 StatefulSet
	if err := r.reconcileStatefulSet(ctx, &rc); err != nil {
		logger.Error(err, "调和 StatefulSet 失败")
		r.updateStatus(ctx, &rc, redisv1alpha1.ClusterPhaseFailed, 0, false, err.Error())
		return ctrl.Result{}, err
	}

	// 6. 检查 Pod 就绪状态
	readyNodes, err := r.getReadyNodeCount(ctx, &rc)
	if err != nil {
		return ctrl.Result{}, err
	}

	totalNodes := rc.Spec.MasterCount * (1 + rc.Spec.ReplicasPerMaster)

	if readyNodes < totalNodes {
		logger.Info("等待所有节点就绪",
			"ready", readyNodes, "total", totalNodes)
		r.updateStatus(ctx, &rc, redisv1alpha1.ClusterPhaseCreating, readyNodes, false,
			fmt.Sprintf("等待节点就绪 %d/%d", readyNodes, totalNodes))
		return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
	}

	// 7. 初始化集群（如果尚未初始化）
	if !rc.Status.ClusterInitialized {
		logger.Info("开始初始化 Redis Cluster")
		r.updateStatus(ctx, &rc, redisv1alpha1.ClusterPhaseInitializing, readyNodes, false,
			"正在初始化集群")

		if err := r.initializeCluster(ctx, &rc); err != nil {
			logger.Error(err, "初始化集群失败")
			r.updateStatus(ctx, &rc, redisv1alpha1.ClusterPhaseFailed, readyNodes, false, err.Error())
			return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
		}

		r.updateStatus(ctx, &rc, redisv1alpha1.ClusterPhaseRunning, readyNodes, true,
			"集群已就绪")
	}

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

// reconcileService 创建或更新 Headless Service
func (r *RedisClusterReconciler) reconcileService(
	ctx context.Context,
	rc *redisv1alpha1.RedisCluster,
) error {
	logger := log.FromContext(ctx)

	desired := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      rc.Name,
			Namespace: rc.Namespace,
		},
		Spec: corev1.ServiceSpec{
			ClusterIP: "None",
			Selector: map[string]string{
				"app":           rc.Name,
				"managed-by":    "redis-operator",
				"redis-cluster": rc.Name,
			},
			Ports: []corev1.ServicePort{
				{Name: "redis", Port: 6379, TargetPort: intstr.FromInt(6379)},
				{Name: "bus", Port: 16379, TargetPort: intstr.FromInt(16379)},
			},
		},
	}

	// 设置 OwnerReference
	if err := controllerutil.SetControllerReference(rc, desired, r.Scheme); err != nil {
		return err
	}

	var existing corev1.Service
	err := r.Get(ctx, types.NamespacedName{Name: rc.Name, Namespace: rc.Namespace}, &existing)

	if errors.IsNotFound(err) {
		logger.Info("创建 Headless Service")
		return r.Create(ctx, desired)
	} else if err != nil {
		return err
	}

	// Service 已存在，更新
	existing.Spec.Selector = desired.Spec.Selector
	existing.Spec.Ports = desired.Spec.Ports
	return r.Update(ctx, &existing)
}

// reconcileStatefulSet 创建或更新 StatefulSet
func (r *RedisClusterReconciler) reconcileStatefulSet(
	ctx context.Context,
	rc *redisv1alpha1.RedisCluster,
) error {
	logger := log.FromContext(ctx)

	totalReplicas := rc.Spec.MasterCount * (1 + rc.Spec.ReplicasPerMaster)

	cpuReq := rc.Spec.Resources.CPURequest
	memReq := rc.Spec.Resources.MemoryRequest
	cpuLim := rc.Spec.Resources.CPULimit
	memLim := rc.Spec.Resources.MemoryLimit

	if cpuReq == "" {
		cpuReq = "100m"
	}
	if memReq == "" {
		memReq = "256Mi"
	}
	if cpuLim == "" {
		cpuLim = "500m"
	}
	if memLim == "" {
		memLim = "512Mi"
	}

	desired := &appsv1.StatefulSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:      rc.Name,
			Namespace: rc.Namespace,
		},
		Spec: appsv1.StatefulSetSpec{
			ServiceName: rc.Name,
			Replicas:    &totalReplicas,
			Selector: &metav1.LabelSelector{
				MatchLabels: map[string]string{
					"app":           rc.Name,
					"managed-by":    "redis-operator",
					"redis-cluster": rc.Name,
				},
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						"app":           rc.Name,
						"managed-by":    "redis-operator",
						"redis-cluster": rc.Name,
					},
				},
				Spec: corev1.PodSpec{
					Containers: []corev1.Container{
						{
							Name:  "redis",
							Image: rc.Spec.Image,
							Command: []string{
								"/bin/sh", "-c",
								fmt.Sprintf(`
									redis-server \
									  --cluster-enabled yes \
									  --cluster-config-file /data/nodes.conf \
									  --cluster-node-timeout 5000 \
									  --appendonly yes \
									  --bind 0.0.0.0 \
									  --port 6379 \
									  --cluster-announce-ip $(hostname).%s.%s.svc.cluster.local \
									  --cluster-announce-port 6379 \
									  --cluster-announce-bus-port 16379
								`, rc.Name, rc.Namespace),
							},
							Ports: []corev1.ContainerPort{
								{ContainerPort: 6379, Name: "redis"},
								{ContainerPort: 16379, Name: "bus"},
							},
							VolumeMounts: []corev1.VolumeMount{
								{Name: "data", MountPath: "/data"},
							},
							LivenessProbe: &corev1.Probe{
								Exec: &corev1.ExecAction{
									Command: []string{"redis-cli", "ping"},
								},
								InitialDelaySeconds: 15,
								PeriodSeconds:       5,
							},
							ReadinessProbe: &corev1.Probe{
								Exec: &corev1.ExecAction{
									Command: []string{"redis-cli", "ping"},
								},
								InitialDelaySeconds: 5,
								PeriodSeconds:       3,
							},
							Resources: corev1.ResourceRequirements{
								Requests: corev1.ResourceList{
									corev1.ResourceCPU:    resource.MustParse(cpuReq),
									corev1.ResourceMemory: resource.MustParse(memReq),
								},
								Limits: corev1.ResourceList{
									corev1.ResourceCPU:    resource.MustParse(cpuLim),
									corev1.ResourceMemory: resource.MustParse(memLim),
								},
							},
						},
					},
				},
			},
			VolumeClaimTemplates: []corev1.PersistentVolumeClaim{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name: "data",
					},
					Spec: corev1.PersistentVolumeClaimSpec{
						AccessModes: []corev1.PersistentVolumeAccessMode{
							corev1.ReadWriteOnce,
						},
						Resources: corev1.VolumeResourceRequirements{
							Requests: corev1.ResourceList{
								corev1.ResourceStorage: resource.MustParse("1Gi"),
							},
						},
					},
				},
			},
		},
	}

	if err := controllerutil.SetControllerReference(rc, desired, r.Scheme); err != nil {
		return err
	}

	var existing appsv1.StatefulSet
	err := r.Get(ctx, types.NamespacedName{Name: rc.Name, Namespace: rc.Namespace}, &existing)

	if errors.IsNotFound(err) {
		logger.Info("创建 StatefulSet", "replicas", totalReplicas)
		return r.Create(ctx, desired)
	} else if err != nil {
		return err
	}

	// StatefulSet 已存在，检查是否需要更新
	if *existing.Spec.Replicas != totalReplicas || existing.Spec.Template.Spec.Containers[0].Image != rc.Spec.Image {
		logger.Info("更新 StatefulSet")
		existing.Spec.Replicas = &totalReplicas
		existing.Spec.Template.Spec.Containers[0].Image = rc.Spec.Image
		return r.Update(ctx, &existing)
	}

	return nil
}

// getReadyNodeCount 统计就绪的 Pod 数量
func (r *RedisClusterReconciler) getReadyNodeCount(
	ctx context.Context,
	rc *redisv1alpha1.RedisCluster,
) (int32, error) {
	var podList corev1.PodList
	if err := r.List(ctx, &podList,
		client.InNamespace(rc.Namespace),
		client.MatchingLabels{
			"app":           rc.Name,
			"redis-cluster": rc.Name,
		},
	); err != nil {
		return 0, err
	}

	var ready int32
	for _, pod := range podList.Items {
		for _, cond := range pod.Status.Conditions {
			if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
				ready++
				break
			}
		}
	}
	return ready, nil
}

// initializeCluster 执行 Redis Cluster 初始化
func (r *RedisClusterReconciler) initializeCluster(
	ctx context.Context,
	rc *redisv1alpha1.RedisCluster,
) error {
	logger := log.FromContext(ctx)

	// 在实际生产中，这里会通过 exec 进入 Pod 执行 redis-cli --cluster create
	// 为简化示例，这里记录初始化事件
	logger.Info("Redis Cluster 初始化完成",
		"masters", rc.Spec.MasterCount,
		"replicas", rc.Spec.ReplicasPerMaster)

	// 生产环境实现示例（伪代码）：
	// 1. 收集所有 Pod 的 FQDN
	// 2. exec 进入第一个 Pod 执行 redis-cli --cluster create
	// 3. 验证集群状态

	return nil
}

// updateStatus 更新 RedisCluster 的状态子资源
func (r *RedisClusterReconciler) updateStatus(
	ctx context.Context,
	rc *redisv1alpha1.RedisCluster,
	phase redisv1alpha1.ClusterPhase,
	readyNodes int32,
	initialized bool,
	message string,
) {
	rc.Status.Phase = phase
	rc.Status.ReadyNodes = readyNodes
	rc.Status.ClusterInitialized = initialized
	rc.Status.Message = message

	if err := r.Status().Update(ctx, rc); err != nil {
		log.FromContext(ctx).Error(err, "更新状态失败")
	}
}

// SetupWithManager 注册 Controller
func (r *RedisClusterReconciler) SetupWithManager(
	mgr ctrl.Manager,
) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&redisv1alpha1.RedisCluster{}).
		Owns(&appsv1.StatefulSet{}).
		Owns(&corev1.Service{}).
		Complete(r)
}

package main

import (
	"os"

	redisv1alpha1 "github.com/rainlib/redis-operator/api/v1alpha1"
	"github.com/rainlib/redis-operator/controllers"
	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/healthz"
	"sigs.k8s.io/controller-runtime/pkg/log/zap"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
)

var (
	scheme = runtime.NewScheme()
)

func init() {
	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
	utilruntime.Must(redisv1alpha1.AddToScheme(scheme))
}

func main() {
	ctrl.SetLogger(zap.New(zap.UseDevMode(true)))

	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		Scheme: scheme,
		Metrics: metricsserver.Options{
			BindAddress: ":8080",
		},
		HealthProbeBindAddress: ":8081",
		LeaderElection:         true,
		LeaderElectionID:       "redis-operator.rainlib.io",
	})
	if err != nil {
		setupLog.Error(err, "无法创建 Manager")
		os.Exit(1)
	}

	if err := (&controllers.RedisClusterReconciler{
		Client: mgr.GetClient(),
		Scheme: mgr.GetScheme(),
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "无法创建 Controller")
		os.Exit(1)
	}

	if err := mgr.AddHealthzCheck("healthz", healthz.PingHealthz); err != nil {
		setupLog.Error(err, "无法设置健康检查")
		os.Exit(1)
	}

	setupLog.Info("启动 Manager")
	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
		setupLog.Error(err, "Manager 运行错误")
		os.Exit(1)
	}
}