feat: add redis-role label to redis replication (#925)

* feat: add redis-role label to redis replication Signed-off-by: drivebyer <[email protected]> * fix lint Signed-off-by: drivebyer <[email protected]> * fix lint Signed-off-by: drivebyer <[email protected]> * fix test Signed-off-by: drivebyer <[email protected]> * fix test Signed-off-by: drivebyer <[email protected]> * fix test Signed-off-by: drivebyer <[email protected]> * fix test Signed-off-by: drivebyer <[email protected]> * fix test Signed-off-by: drivebyer <[email protected]> * fix test Signed-off-by: drivebyer <[email protected]> * fix test Signed-off-by: drivebyer <[email protected]> --------- Signed-off-by: drivebyer <[email protected]>
OT-CONTAINER-KIT · May 14, 2024 · 09a80f4 · 09a80f4
1 parent a9fa699
commit 09a80f4
Show file tree

Hide file tree

Showing 6 changed files with 126 additions and 8 deletions.
diff --git a/controllers/redisreplication_controller.go b/controllers/redisreplication_controller.go
@@ -8,6 +8,7 @@ import (
 	redisv1beta2 "github.com/OT-CONTAINER-KIT/redis-operator/api/v1beta2"
 	"github.com/OT-CONTAINER-KIT/redis-operator/k8sutils"
 	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/dynamic"
@@ -19,6 +20,7 @@ import (
 // RedisReplicationReconciler reconciles a RedisReplication object
 type RedisReplicationReconciler struct {
 	client.Client
+	k8sutils.Pod
 	K8sClient  kubernetes.Interface
 	Dk8sClient dynamic.Interface
 	Log        logr.Logger
@@ -86,13 +88,15 @@ func (r *RedisReplicationReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		if len(slaveNodes) == 0 {
 			realMaster = masterNodes[0]
 		}
-		err := k8sutils.CreateMasterSlaveReplication(ctx, r.K8sClient, r.Log, instance, masterNodes, realMaster)
-		if err != nil {
+		if err = k8sutils.CreateMasterSlaveReplication(ctx, r.K8sClient, r.Log, instance, masterNodes, realMaster); err != nil {
 			return ctrl.Result{RequeueAfter: time.Second * 60}, err
 		}
 	}
 	realMaster = k8sutils.GetRedisReplicationRealMaster(ctx, r.K8sClient, r.Log, instance, masterNodes)
-	if err := r.UpdateRedisReplicationMaster(ctx, instance, realMaster); err != nil {
+	if err = r.UpdateRedisReplicationMaster(ctx, instance, realMaster); err != nil {
+		return ctrl.Result{}, err
+	}
+	if err = r.UpdateRedisPodRoleLabel(ctx, instance, realMaster); err != nil {
 		return ctrl.Result{}, err
 	}
 	reqLogger.Info("Will reconcile redis operator in again 10 seconds")
@@ -110,6 +114,31 @@ func (r *RedisReplicationReconciler) UpdateRedisReplicationMaster(ctx context.Co
 	return nil
 }
 
+func (r *RedisReplicationReconciler) UpdateRedisPodRoleLabel(ctx context.Context, cr *redisv1beta2.RedisReplication, masterNode string) error {
+	labels := k8sutils.GetRedisReplicationLabels(cr)
+	pods, err := r.ListPods(ctx, cr.GetNamespace(), labels)
+	if err != nil {
+		return err
+	}
+	updateRoleLabelFunc := func(ctx context.Context, namespace string, pod corev1.Pod, role string) error {
+		if pod.Labels[k8sutils.RedisRoleLabelKey] != role {
+			return r.PatchPodLabels(ctx, namespace, pod.GetName(), map[string]string{k8sutils.RedisRoleLabelKey: role})
+		}
+		return nil
+	}
+	for _, pod := range pods.Items {
+		if masterNode == pod.GetName() {
+			err = updateRoleLabelFunc(ctx, cr.GetNamespace(), pod, k8sutils.RedisRoleLabelMaster)
+		} else {
+			err = updateRoleLabelFunc(ctx, cr.GetNamespace(), pod, k8sutils.RedisRoleLabelSlave)
+		}
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // SetupWithManager sets up the controller with the Manager.
 func (r *RedisReplicationReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).

diff --git a/k8sutils/const.go b/k8sutils/const.go
@@ -7,3 +7,9 @@ const (
 const (
 	EnvOperatorSTSPVCTemplateName = "OPERATOR_STS_PVC_TEMPLATE_NAME"
 )
+
+const (
+	RedisRoleLabelKey    = "redis-role"
+	RedisRoleLabelMaster = "master"
+	RedisRoleLabelSlave  = "slave"
+)
diff --git a/k8sutils/labels.go b/k8sutils/labels.go
@@ -149,3 +149,7 @@ func getRedisLabels(name string, st setupType, role string, labels map[string]st
 	}
 	return lbls
 }
+
+func GetRedisReplicationLabels(cr *redisv1beta2.RedisReplication) map[string]string {
+	return getRedisLabels(cr.GetName(), replication, "replication", cr.GetLabels())
+}
diff --git a/k8sutils/pod.go b/k8sutils/pod.go
@@ -0,0 +1,69 @@
+package k8sutils
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+)
+
+type Pod interface {
+	ListPods(ctx context.Context, namespace string, labels map[string]string) (*corev1.PodList, error)
+	PatchPodLabels(ctx context.Context, namespace, name string, labels map[string]string) error
+}
+
+type PodService struct {
+	kubeClient kubernetes.Interface
+	log        logr.Logger
+}
+
+func NewPodService(kubeClient kubernetes.Interface, log logr.Logger) *PodService {
+	log = log.WithValues("service", "k8s.pod")
+	return &PodService{
+		kubeClient: kubeClient,
+		log:        log,
+	}
+}
+
+func (s *PodService) ListPods(ctx context.Context, namespace string, labels map[string]string) (*corev1.PodList, error) {
+	selector := make([]string, 0, len(labels))
+	for key, value := range labels {
+		selector = append(selector, fmt.Sprintf("%s=%s", key, value))
+	}
+	return s.kubeClient.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: strings.Join(selector, ","),
+	})
+}
+
+type patchStringValue struct {
+	Op    string      `json:"op"`
+	Path  string      `json:"path"`
+	Value interface{} `json:"value"`
+}
+
+func (s *PodService) PatchPodLabels(ctx context.Context, namespace, podName string, labels map[string]string) error {
+	s.log.Info("Patch pod labels", "namespace", namespace, "podName", podName, "labels", labels)
+
+	var payloads []interface{}
+	for labelKey, labelValue := range labels {
+		payload := patchStringValue{
+			Op:    "replace",
+			Path:  "/metadata/labels/" + labelKey,
+			Value: labelValue,
+		}
+		payloads = append(payloads, payload)
+	}
+	payloadBytes, _ := json.Marshal(payloads)
+
+	_, err := s.kubeClient.CoreV1().Pods(namespace).Patch(ctx, podName, types.JSONPatchType, payloadBytes, metav1.PatchOptions{})
+	if err != nil {
+		s.log.Error(err, "Patch pod labels failed", "namespace", namespace, "podName", podName)
+	}
+	return err
+}
diff --git a/main.go b/main.go
@@ -134,12 +134,14 @@ func main() {
 		setupLog.Error(err, "unable to create controller", "controller", "RedisCluster")
 		os.Exit(1)
 	}
+	rrLog := ctrl.Log.WithName("controllers").WithName("RedisReplication")
 	if err = (&controllers.RedisReplicationReconciler{
 		Client:     mgr.GetClient(),
 		K8sClient:  k8sclient,
 		Dk8sClient: dk8sClient,
-		Log:        ctrl.Log.WithName("controllers").WithName("RedisReplication"),
+		Log:        rrLog,
 		Scheme:     mgr.GetScheme(),
+		Pod:        k8sutils.NewPodService(k8sclient, rrLog),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "RedisReplication")
 		os.Exit(1)

diff --git a/tests/e2e-chainsaw/v1beta2/ha-failover/chainsaw-test.yaml b/tests/e2e-chainsaw/v1beta2/ha-failover/chainsaw-test.yaml
@@ -19,18 +19,22 @@ spec:
             selector: control-plane=redis-operator
             container: manager
             tail: -1  # tail all logs
+
     - name: Sleep for 3 minutes
       try:
         - sleep:
             duration: 3m
+
     - name: Test sentinel monitoring
       try:
         - script:
             timeout: 10s
             content: |
-              kubectl exec --namespace ${NAMESPACE} redis -- redis-cli -h redis-sentinel-sentinel.${NAMESPACE}.svc -p 26379 sentinel master myMaster | grep -A 1 'flags' | tail -n 1
+              export MASTER_IP_FROM_SENTINEL=$(kubectl exec --namespace ${NAMESPACE} redis-sentinel-sentinel-0 -- redis-cli -p 26379 sentinel get-master-addr-by-name myMaster | head -n 1);
+              export MASTER_IP_FROM_LABEL=$(kubectl -n ${NAMESPACE} get pod -l app=redis-replication,redis-role=master,redis_setup_type=replication -o jsonpath='{.items[0].status.podIP}');
+              if [ "$MASTER_IP_FROM_SENTINEL" = "$MASTER_IP_FROM_LABEL" ]; then echo "OK"; else echo "FAIL"; fi
             check:
-              ($stdout=='master'): true
+              ($stdout=='OK'): true
       catch:
         - description: Redis Operator Logs
           podLogs:
@@ -53,18 +57,22 @@ spec:
             selector: control-plane=redis-operator
             container: manager
             tail: -1  # tail all logs
+
     - name: Sleep for 3 minutes
       try:
         - sleep:
             duration: 3m
+
     - name: Test sentinel monitoring
       try:
         - script:
             timeout: 10s
             content: |
-              kubectl exec --namespace ${NAMESPACE} redis -- redis-cli -h redis-sentinel-sentinel.${NAMESPACE}.svc -p 26379 sentinel master myMaster | grep -A 1 'flags' | tail -n 1
+              export MASTER_IP_FROM_SENTINEL=$(kubectl exec --namespace ${NAMESPACE} redis-sentinel-sentinel-0 -- redis-cli -p 26379 sentinel get-master-addr-by-name myMaster | head -n 1);
+              export MASTER_IP_FROM_LABEL=$(kubectl -n ${NAMESPACE} get pod -l app=redis-replication,redis-role=master,redis_setup_type=replication -o jsonpath='{.items[0].status.podIP}');
+              if [ $MASTER_IP_FROM_SENTINEL = $MASTER_IP_FROM_LABEL ]; then echo "OK"; else echo "FAIL"; fi
             check:
-              ($stdout=='master'): true
+              ($stdout=='OK'): true
       catch:
         - description: Redis Operator Logs
           podLogs: