fix: Unassigned slots and abnormal status in single-node cluster (#1483)

* Fix the issue where slots are not allocated and the cluster is not initialized when only one cluster shard is started in single instance

Signed-off-by: tcxdgit <[email protected]>

* File is not properly formatted

Signed-off-by: tcxdgit <[email protected]>

* refactor code for improved organization and clarity

Signed-off-by: tcxdgit <[email protected]>

* refactor code for improved organization and clarity

Signed-off-by: tcxdgit <[email protected]>

---------

Signed-off-by: tcxdgit <[email protected]>

Author: tcxdgit

internal/cmd/manager/cmd.go

@@ -242,6 +242,7 @@ func setupControllers(mgr ctrl.Manager, k8sClient kubernetes.Interface, maxConcu
 		Client:      mgr.GetClient(),
 		K8sClient:   k8sClient,
 		Healer:      healer,
+		Checker:     redis.NewChecker(k8sClient),
 		Recorder:    mgr.GetEventRecorderFor("rediscluster-controller"),
 		StatefulSet: k8sutils.NewStatefulSetService(k8sClient),
 	}).SetupWithManager(mgr, controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}); err != nil {

internal/controller/common/redis/check.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	commonapi "github.com/OT-CONTAINER-KIT/redis-operator/api/common/v1beta2"
+	rcvb2 "github.com/OT-CONTAINER-KIT/redis-operator/api/rediscluster/v1beta2"
 	rr "github.com/OT-CONTAINER-KIT/redis-operator/api/redisreplication/v1beta2"
 	"github.com/OT-CONTAINER-KIT/redis-operator/internal/service/redis"
 	corev1 "k8s.io/api/core/v1"
@@ -17,6 +18,7 @@ import (
 type Checker interface {
 	GetMasterFromReplication(ctx context.Context, rr *rr.RedisReplication) (corev1.Pod, error)
 	GetPassword(ctx context.Context, ns string, secret *commonapi.ExistingPasswordSecret) (string, error)
+	CheckClusterSlotsAssigned(ctx context.Context, cr *rcvb2.RedisCluster) (bool, error)
 }
 
 type checker struct {
@@ -101,3 +103,28 @@ func (c *checker) GetMasterFromReplication(ctx context.Context, rr *rr.RedisRepl
 	}
 	return realMasterPod, nil
 }
+
+// CheckClusterSlotsAssigned verifies if all Redis cluster slots (16384 total) are properly assigned
+func (c *checker) CheckClusterSlotsAssigned(ctx context.Context, cr *rcvb2.RedisCluster) (bool, error) {
+	leaderPodName := cr.Name + "-leader-0"
+	pod, err := c.k8s.CoreV1().Pods(cr.Namespace).Get(ctx, leaderPodName, metav1.GetOptions{})
+	if err != nil {
+		return false, err
+	}
+
+	password, err := c.GetPassword(ctx, cr.Namespace, cr.Spec.KubernetesConfig.ExistingPasswordSecret)
+	if err != nil {
+		return false, err
+	}
+
+	connInfo := createConnectionInfo(ctx, *pod, password, cr.Spec.TLS, c.k8s, cr.Namespace, "6379")
+
+	clusterStatus, err := c.redis.Connect(connInfo).GetClusterInfo(ctx)
+	if err != nil {
+		return false, err
+	}
+
+	allAssigned := clusterStatus.SlotsAssigned == 16384
+
+	return allAssigned, nil
+}

internal/controller/rediscluster/rediscluster_controller.go

@@ -48,6 +48,7 @@ type Reconciler struct {
 	client.Client
 	k8sutils.StatefulSet
 	Healer    redis.Healer
+	Checker   redis.Checker
 	K8sClient kubernetes.Interface
 	Recorder  record.EventRecorder
 }
@@ -190,6 +191,19 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		}
 	}
 
+	// When the number of leader replicas is 1 (single-node cluster)
+	if leaderReplicas == 1 {
+		// Check if the Redis cluster has no unassigned slots (i.e., all slots are properly allocated)
+		if slotsAssigned, err := r.Checker.CheckClusterSlotsAssigned(ctx, instance); err != nil {
+			return intctrlutil.RequeueE(ctx, err, "failed to get cluster slots")
+		} else {
+			if !slotsAssigned {
+				logger.Info("Start creating a single-node redis cluster")
+				k8sutils.ExecuteRedisClusterCommand(ctx, r.K8sClient, instance)
+			}
+		}
+	}
+
 	if nc := k8sutils.CheckRedisNodeCount(ctx, r.K8sClient, instance, ""); nc != totalReplicas {
 		logger.Info("Creating redis cluster by executing cluster creation commands")
 		leaderCount := k8sutils.CheckRedisNodeCount(ctx, r.K8sClient, instance, "leader")

internal/service/redis/client.go

@@ -24,6 +24,11 @@ type ConnectionInfo struct {
 	TLSConfig *tls.Config
 }
 
+// ClusterStatus cluster status information, including the number of assigned slots
+type ClusterStatus struct {
+	SlotsAssigned int // Total number of assigned slots
+}
+
 // GetAddress returns the connection address
 func (c *ConnectionInfo) GetAddress() string {
 	return net.JoinHostPort(c.Host, c.Port)
@@ -50,6 +55,7 @@ type Service interface {
 	GetAttachedReplicaCount(ctx context.Context) (int, error)
 	SentinelMonitor(ctx context.Context, master *ConnectionInfo, masterGroupName, quorum string) error
 	SentinelReset(ctx context.Context, masterGroupName string) error
+	GetClusterInfo(ctx context.Context) (*ClusterStatus, error)
 }
 
 type service struct {
@@ -175,3 +181,29 @@ func (c *service) GetAttachedReplicaCount(ctx context.Context) (int, error) {
 	}
 	return count, nil
 }
+
+// GetClusterInfo get cluster information by checking slot allocation
+func (c *service) GetClusterInfo(ctx context.Context) (*ClusterStatus, error) {
+	client := c.createClient()
+	if client == nil {
+		return nil, nil
+	}
+	defer client.Close()
+
+	slots, err := client.ClusterSlots(ctx).Result()
+	if err != nil {
+		return nil, err
+	}
+
+	status := &ClusterStatus{
+		SlotsAssigned: 0,
+	}
+
+	for _, slot := range slots {
+		if slot.Start <= slot.End && len(slot.Nodes) > 0 {
+			status.SlotsAssigned += slot.End - slot.Start + 1
+		}
+	}
+
+	return status, nil
+}