Determine if you need to operate resources

Author: polefishu

pkg/apis/redis/v1alpha1/distributedrediscluster_types.go

@@ -104,10 +104,12 @@ type DistributedRedisClusterStatus struct {
 	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
 	// Important: Run "operator-sdk generate k8s" to regenerate code after modifying this file
 	// Add custom validation using kubebuilder tags: https://book-v1.book.kubebuilder.io/beyond_basics/generating_crd.html
-	Status         ClusterStatus      `json:"status"`
-	Reason         string             `json:"reason,omitempty"`
-	NumberOfMaster int32              `json:"numberOfMaster,omitempty"`
-	Nodes          []RedisClusterNode `json:"nodes"`
+	Status               ClusterStatus      `json:"status"`
+	Reason               string             `json:"reason,omitempty"`
+	NumberOfMaster       int32              `json:"numberOfMaster,omitempty"`
+	MinReplicationFactor int32              `json:"minReplicationFactor,omitempty"`
+	MaxReplicationFactor int32              `json:"maxReplicationFactor,omitempty"`
+	Nodes                []RedisClusterNode `json:"nodes"`
 	// The number of restore which reached phase Succeeded.
 	// +optional
 	RestoreSucceeded int32 `json:"restoreSucceeded,omitempty"`

pkg/controller/distributedrediscluster/distributedrediscluster_controller.go

@@ -125,7 +125,8 @@ func (r *ReconcileDistributedRedisCluster) Reconcile(request reconcile.Request)
 	}
 
 	ctx := &syncContext{
-		cluster: instance,
+		cluster:   instance,
+		reqLogger: reqLogger,
 	}
 
 	err = r.ensureCluster(ctx)
@@ -185,6 +186,16 @@ func (r *ReconcileDistributedRedisCluster) Reconcile(request reconcile.Request)
 		}
 	}
 
+	requeue, err := ctx.healer.Heal(instance, clusterInfos, admin)
+	if err != nil {
+		return reconcile.Result{}, Redis.Wrap(err, "Heal")
+	}
+	if requeue {
+		return reconcile.Result{RequeueAfter: requeueAfter}, nil
+	}
+
+	ctx.admin = admin
+	ctx.clusterInfos = clusterInfos
 	err = r.waitForClusterJoin(ctx)
 	if err != nil {
 		switch GetType(err) {
@@ -215,13 +226,16 @@ func (r *ReconcileDistributedRedisCluster) Reconcile(request reconcile.Request)
 	reqLogger.V(4).Info("buildClusterStatus", "status", status)
 	r.updateClusterIfNeed(instance, status)
 
-	err = r.sync(ctx)
-	if err != nil {
-		new := instance.Status.DeepCopy()
-		SetClusterFailed(new, err.Error())
-		r.updateClusterIfNeed(instance, new)
-		return reconcile.Result{}, err
+	if needClusterOperation(instance, reqLogger) {
+		err = r.sync(ctx)
+		if err != nil {
+			new := instance.Status.DeepCopy()
+			SetClusterFailed(new, err.Error())
+			r.updateClusterIfNeed(instance, new)
+			return reconcile.Result{}, err
+		}
 	}
+
 	newClusterInfos, err := admin.GetClusterInfos()
 	if err != nil {
 		if clusterInfos.Status == redisutil.ClusterInfosPartial {
@@ -231,6 +245,5 @@ func (r *ReconcileDistributedRedisCluster) Reconcile(request reconcile.Request)
 	newStatus := buildClusterStatus(newClusterInfos, redisClusterPods.Items, &instance.Status)
 	SetClusterOK(newStatus, "OK")
 	r.updateClusterIfNeed(instance, newStatus)
-	//return reconcile.Result{RequeueAfter: requeueEnsure}, nil
-	return reconcile.Result{}, nil
+	return reconcile.Result{RequeueAfter: requeueEnsure}, nil
 }

pkg/controller/distributedrediscluster/helper.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"time"
 
+	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -151,3 +152,22 @@ func clusterPods(pods []corev1.Pod) []*corev1.Pod {
 	}
 	return podSlice
 }
+
+func needClusterOperation(cluster *redisv1alpha1.DistributedRedisCluster, reqLogger logr.Logger) bool {
+	if compareIntValue("NumberOfMaster", &cluster.Status.NumberOfMaster, &cluster.Spec.MasterSize) {
+		reqLogger.V(4).Info("needClusterOperation---NumberOfMaster")
+		return true
+	}
+
+	if compareIntValue("MinReplicationFactor", &cluster.Status.MinReplicationFactor, &cluster.Spec.ClusterReplicas) {
+		reqLogger.V(4).Info("needClusterOperation---MinReplicationFactor")
+		return true
+	}
+
+	if compareIntValue("MaxReplicationFactor", &cluster.Status.MaxReplicationFactor, &cluster.Spec.ClusterReplicas) {
+		reqLogger.V(4).Info("needClusterOperation---MaxReplicationFactor")
+		return true
+	}
+
+	return false
+}

pkg/controller/distributedrediscluster/init_from_backup.go

@@ -2,6 +2,7 @@ package distributedrediscluster
 
 import (
 	"fmt"
+	"math"
 	"reflect"
 
 	corev1 "k8s.io/api/core/v1"
@@ -82,6 +83,22 @@ func buildClusterStatus(clusterInfos *redisutil.ClusterInfos, pods []corev1.Pod,
 	}
 	status.NumberOfMaster = nbMaster
 
+	minReplicationFactor := math.MaxInt32
+	maxReplicationFactor := 0
+	for _, counter := range nbSlaveByMaster {
+		if counter > maxReplicationFactor {
+			maxReplicationFactor = counter
+		}
+		if counter < minReplicationFactor {
+			minReplicationFactor = counter
+		}
+	}
+	if len(nbSlaveByMaster) == 0 {
+		minReplicationFactor = 0
+	}
+	status.MaxReplicationFactor = int32(maxReplicationFactor)
+	status.MinReplicationFactor = int32(minReplicationFactor)
+
 	return status
 }
 

pkg/controller/distributedrediscluster/status.go

@@ -107,7 +107,8 @@ func (r *ReconcileDistributedRedisCluster) waitForClusterJoin(ctx *syncContext)
 	//if err != nil {
 	//	return Redis.Wrap(err, "SetConfigEpoch")
 	//}
-	if _, err := ctx.admin.GetClusterInfos(); err == nil {
+	if infos, err := ctx.admin.GetClusterInfos(); err == nil {
+		ctx.reqLogger.V(5).Info("debug waitForClusterJoin", "cluster infos", infos)
 		return nil
 	}
 	var firstNode *redisutil.Node
@@ -124,12 +125,7 @@ func (r *ReconcileDistributedRedisCluster) waitForClusterJoin(ctx *syncContext)
 	// waiting for cluster join will find all the nodes agree about
 	// the config as they are still empty with unassigned slots.
 	time.Sleep(1 * time.Second)
-	if _, err := ctx.healer.FixFailedNodes(ctx.cluster, ctx.clusterInfos, ctx.admin); err != nil {
-		return Cluster.Wrap(err, "FixFailedNodes")
-	}
-	if _, err := ctx.healer.FixUntrustedNodes(ctx.cluster, ctx.clusterInfos, ctx.admin); err != nil {
-		return Cluster.Wrap(err, "FixUntrustedNodes")
-	}
+
 	_, err = ctx.admin.GetClusterInfos()
 	if err != nil {
 		return Requeue.Wrap(err, "wait for cluster join")