fix: tls connection broken in replication with sentinel (#1472)

* feat: support tls in replication with sentinel

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

* fix lint

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

* fix lint

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

* fix monitor

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

* tls flag

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

---------

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

Author: drivebyer

.github/workflows/ci.yaml

@@ -238,11 +238,14 @@ jobs:
 
       # NOTE: This is a workaround for the issue where the default storage class does not support volume expansion.
       # Since we don't require PVC resizing (unlike physical disks), we can simply ensure that the requested PVC size is met.
-      - name: Set allowVolumeExpansion to true
+      - name: Setup k8s Kind Cluster
         run: |
           DEFAULT_SC=$(kubectl get storageclass -o=jsonpath='{.items[?(@.metadata.annotations.storageclass\.kubernetes\.io/is-default-class=="true")].metadata.name}')
           kubectl patch storageclass $DEFAULT_SC -p '{"allowVolumeExpansion": true}'          
 
+          # install cert-manager
+          kubectl apply -f https://github.com/cert-manager/cert-manager/releases/download/v1.18.2/cert-manager.yaml
+
       - name: Load Docker image into Kind
         run: |
           kubectl cluster-info --context kind-kind

.yamllint.yml

@@ -2,7 +2,7 @@ extends: default
 
 rules:
   line-length:
-    max: 200
+    max: 300
     level: warning
     allow-non-breakable-words: true
     allow-non-breakable-inline-mappings: true

internal/agent/bootstrap/sentinel/config.go

@@ -108,6 +108,8 @@ func GenerateConfig() error {
 			cfg.Append("tls-key-file", redisTLSCertKey)
 			cfg.Append("tls-ca-cert-file", redisTLSCAKey)
 			cfg.Append("tls-auth-clients", "optional")
+			// Sentinel should use tls for replication connection.
+			cfg.Append("tls-replication", "yes")
 		} else {
 			fmt.Println("Running sentinel without TLS mode")
 		}

internal/controller/common/redis/check.go

@@ -71,11 +71,7 @@ func (c *checker) GetMasterFromReplication(ctx context.Context, rr *rr.RedisRepl
 
 	var masterPods []corev1.Pod
 	for _, pod := range pods.Items {
-		connInfo := &redis.ConnectionInfo{
-			IP:       pod.Status.PodIP,
-			Port:     "6379",
-			Password: password,
-		}
+		connInfo := createConnectionInfo(ctx, pod, password, rr.Spec.TLS, c.k8s, rr.Namespace, "6379")
 		isMaster, err := c.redis.Connect(connInfo).IsMaster(ctx)
 		if err != nil {
 			return corev1.Pod{}, err
@@ -87,11 +83,7 @@ func (c *checker) GetMasterFromReplication(ctx context.Context, rr *rr.RedisRepl
 
 	var realMasterPod corev1.Pod
 	for _, pod := range masterPods {
-		connInfo := &redis.ConnectionInfo{
-			IP:       pod.Status.PodIP,
-			Port:     "6379",
-			Password: password,
-		}
+		connInfo := createConnectionInfo(ctx, pod, password, rr.Spec.TLS, c.k8s, rr.Namespace, "6379")
 		count, err := c.redis.Connect(connInfo).GetAttachedReplicaCount(ctx)
 		if err != nil {
 			continue

internal/controller/common/redis/heal.go

@@ -2,13 +2,16 @@ package redis
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"fmt"
 	"strings"
 
 	commonapi "github.com/OT-CONTAINER-KIT/redis-operator/api/common/v1beta2"
 	rsvb2 "github.com/OT-CONTAINER-KIT/redis-operator/api/redissentinel/v1beta2"
 	"github.com/OT-CONTAINER-KIT/redis-operator/internal/controller/common"
 	"github.com/OT-CONTAINER-KIT/redis-operator/internal/service/redis"
+	"github.com/OT-CONTAINER-KIT/redis-operator/internal/util/cryptutil"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -22,7 +25,7 @@ type Healer interface {
 	SentinelReset(ctx context.Context, rs *rsvb2.RedisSentinel) error
 
 	// UpdatePodRoleLabel connect to all redis pods and update pod role label `redis-role` to `master` or `slave` according to their role.
-	UpdateRedisRoleLabel(ctx context.Context, ns string, labels map[string]string, secret *commonapi.ExistingPasswordSecret) error
+	UpdateRedisRoleLabel(ctx context.Context, ns string, labels map[string]string, secret *commonapi.ExistingPasswordSecret, tlsConfig *commonapi.TLSConfig) error
 }
 
 type healer struct {
@@ -37,7 +40,7 @@ func NewHealer(clientset kubernetes.Interface) Healer {
 	}
 }
 
-func (h *healer) UpdateRedisRoleLabel(ctx context.Context, ns string, labels map[string]string, secret *commonapi.ExistingPasswordSecret) error {
+func (h *healer) UpdateRedisRoleLabel(ctx context.Context, ns string, labels map[string]string, secret *commonapi.ExistingPasswordSecret, tlsConfig *commonapi.TLSConfig) error {
 	selector := make([]string, 0, len(labels))
 	for key, value := range labels {
 		selector = append(selector, fmt.Sprintf("%s=%s", key, value))
@@ -53,11 +56,7 @@ func (h *healer) UpdateRedisRoleLabel(ctx context.Context, ns string, labels map
 		return err
 	}
 	for _, pod := range pods.Items {
-		connInfo := &redis.ConnectionInfo{
-			IP:       pod.Status.PodIP,
-			Port:     "6379",
-			Password: password,
-		}
+		connInfo := createConnectionInfo(ctx, pod, password, tlsConfig, h.k8s, ns, "6379")
 		isMaster, err := h.redis.Connect(connInfo).IsMaster(ctx)
 		if err != nil {
 			return err
@@ -98,11 +97,8 @@ func (h *healer) SentinelReset(ctx context.Context, rs *rsvb2.RedisSentinel) err
 	}
 
 	for _, pod := range pods.Items {
-		connInfo := &redis.ConnectionInfo{
-			IP:       pod.Status.PodIP,
-			Port:     "26379",
-			Password: sentinelPass,
-		}
+		connInfo := createConnectionInfo(ctx, pod, sentinelPass, rs.Spec.TLS, h.k8s, rs.Namespace, "26379")
+
 		err = h.redis.Connect(connInfo).SentinelReset(ctx, rs.Spec.RedisSentinelConfig.MasterGroupName)
 		if err != nil {
 			return err
@@ -135,13 +131,10 @@ func (h *healer) SentinelMonitor(ctx context.Context, rs *rsvb2.RedisSentinel, m
 	}
 
 	for _, pod := range pods.Items {
-		connInfo := &redis.ConnectionInfo{
-			IP:       pod.Status.PodIP,
-			Port:     "26379",
-			Password: sentinelPass,
-		}
+		connInfo := createConnectionInfo(ctx, pod, sentinelPass, rs.Spec.TLS, h.k8s, rs.Namespace, "26379")
+
 		masterConnInfo := &redis.ConnectionInfo{
-			IP:       master,
+			Host:     master,
 			Port:     "6379",
 			Password: masterPass,
 		}
@@ -177,3 +170,63 @@ func (h *healer) getSentinelPods(ctx context.Context, rs *rsvb2.RedisSentinel) (
 	}
 	return pods, nil
 }
+
+// getRedisTLSConfig creates a TLS configuration for Redis connections
+func getRedisTLSConfig(ctx context.Context, client kubernetes.Interface, namespace, tlsSecretName string) *tls.Config {
+	// This is a wrapper to access the k8sutils internal function
+	// We'll implement a simplified version here for now
+	secret, err := client.CoreV1().Secrets(namespace).Get(ctx, tlsSecretName, metav1.GetOptions{})
+	if err != nil {
+		log.FromContext(ctx).Error(err, "Failed in getting TLS secret", "secretName", tlsSecretName, "namespace", namespace)
+		return nil
+	}
+
+	tlsClientCert, certExists := secret.Data["tls.crt"]
+	tlsClientKey, keyExists := secret.Data["tls.key"]
+	tlsCACert, caExists := secret.Data["ca.crt"]
+
+	if !certExists || !keyExists || !caExists {
+		log.FromContext(ctx).Error(fmt.Errorf("TLS secret missing required keys"), "TLS secret is missing required keys", "secretName", tlsSecretName)
+		return nil
+	}
+
+	cert, err := tls.X509KeyPair(tlsClientCert, tlsClientKey)
+	if err != nil {
+		log.FromContext(ctx).Error(err, "Failed to load TLS key pair", "secretName", tlsSecretName)
+		return nil
+	}
+
+	caCertPool := x509.NewCertPool()
+	caCertPool.AppendCertsFromPEM(tlsCACert)
+
+	return &tls.Config{
+		Certificates:       []tls.Certificate{cert},
+		RootCAs:            caCertPool,
+		MinVersion:         tls.VersionTLS12,
+		InsecureSkipVerify: true,
+		VerifyPeerCertificate: func(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error {
+			_, _, err := cryptutil.VerifyCertificateExceptServerName(rawCerts, &tls.Config{RootCAs: caCertPool})
+			return err
+		},
+	}
+}
+
+// createConnectionInfo creates a Redis connection info with TLS support
+func createConnectionInfo(ctx context.Context, pod v1.Pod, password string, tlsConfig *commonapi.TLSConfig, k8sClient kubernetes.Interface, namespace, port string) *redis.ConnectionInfo {
+	connInfo := &redis.ConnectionInfo{
+		Host:     pod.Status.PodIP,
+		Port:     port,
+		Password: password,
+	}
+
+	// Configure TLS if enabled
+	if tlsConfig != nil && tlsConfig.Secret.SecretName != "" {
+		serviceName := common.GetHeadlessServiceNameFromPodName(pod.Name)
+		connInfo.Host = fmt.Sprintf("%s.%s.%s.svc.cluster.local", pod.Name, serviceName, namespace)
+		// Get TLS configuration
+		tlsCfg := getRedisTLSConfig(ctx, k8sClient, namespace, tlsConfig.Secret.SecretName)
+		connInfo.TLSConfig = tlsCfg
+	}
+
+	return connInfo
+}

internal/controller/common/service.go

@@ -0,0 +1,30 @@
+package common
+
+import (
+	"strconv"
+	"strings"
+)
+
+// GetHeadlessServiceNameFromPodName trims the trailing ordinal (e.g. "-0", "-1") from a pod name to
+// derive the headless service name. If the pod name does not contain a trailing
+// ordinal segment the original name is returned unchanged.
+//
+// Examples for different Redis modes:
+// - RedisReplication: "redis-replication-0" -> "redis-replication-headless"
+// - RedisCluster Leader: "redis-cluster-leader-0" -> "redis-cluster-leader-headless"
+// - RedisCluster Follower: "redis-cluster-follower-1" -> "redis-cluster-follower-headless"
+// - RedisSentinel: "redis-sentinel-sentinel-0" -> "redis-sentinel-sentinel-headless"
+func GetHeadlessServiceNameFromPodName(podName string) string {
+	// Find the last dash in the pod name. If there is none, return the whole name.
+	idx := strings.LastIndex(podName, "-")
+	if idx == -1 {
+		return podName
+	}
+	// Check whether the suffix after the last dash is a number (the StatefulSet
+	// ordinal). If it is, trim it to get the service name; otherwise return the
+	// original pod name.
+	if _, err := strconv.Atoi(podName[idx+1:]); err == nil {
+		return podName[:idx] + "-headless"
+	}
+	return podName
+}

internal/controller/rediscluster/rediscluster_controller.go

@@ -287,7 +287,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 
 	for _, fakeRole := range []string{"leader", "follower"} {
 		labels := common.GetRedisLabels(instance.GetName()+"-"+fakeRole, common.SetupTypeCluster, fakeRole, instance.GetLabels())
-		if err = r.Healer.UpdateRedisRoleLabel(ctx, instance.GetNamespace(), labels, instance.Spec.KubernetesConfig.ExistingPasswordSecret); err != nil {
+		if err = r.Healer.UpdateRedisRoleLabel(ctx, instance.GetNamespace(), labels, instance.Spec.KubernetesConfig.ExistingPasswordSecret, instance.Spec.TLS); err != nil {
 			return intctrlutil.RequeueE(ctx, err, "")
 		}
 	}

internal/controller/redisreplication/redisreplication_controller.go

@@ -141,8 +141,14 @@ func (r *Reconciler) reconcileRedis(ctx context.Context, instance *rrvb2.RedisRe
 	defer cancel()
 
 	var realMaster string
-	masterNodes := k8sutils.GetRedisNodesByRole(ctx, r.K8sClient, instance, "master")
-	slaveNodes := k8sutils.GetRedisNodesByRole(ctx, r.K8sClient, instance, "slave")
+	masterNodes, err := k8sutils.GetRedisNodesByRole(ctx, r.K8sClient, instance, "master")
+	if err != nil {
+		return intctrlutil.RequeueE(ctx, err, "")
+	}
+	slaveNodes, err := k8sutils.GetRedisNodesByRole(ctx, r.K8sClient, instance, "slave")
+	if err != nil {
+		return intctrlutil.RequeueE(ctx, err, "")
+	}
 	if len(masterNodes) > 1 {
 		log.FromContext(ctx).Info("Creating redis replication by executing replication creation commands")
 
@@ -171,17 +177,23 @@ func (r *Reconciler) reconcileStatus(ctx context.Context, instance *rrvb2.RedisR
 	var err error
 	var realMaster string
 
-	masterNodes := k8sutils.GetRedisNodesByRole(ctx, r.K8sClient, instance, "master")
+	masterNodes, err := k8sutils.GetRedisNodesByRole(ctx, r.K8sClient, instance, "master")
+	if err != nil {
+		return intctrlutil.RequeueE(ctx, err, "")
+	}
 	realMaster = k8sutils.GetRedisReplicationRealMaster(ctx, r.K8sClient, instance, masterNodes)
 	if err = r.UpdateRedisReplicationMaster(ctx, instance, realMaster); err != nil {
 		return intctrlutil.RequeueE(ctx, err, "")
 	}
 	labels := common.GetRedisLabels(instance.GetName(), common.SetupTypeReplication, "replication", instance.GetLabels())
-	if err = r.Healer.UpdateRedisRoleLabel(ctx, instance.GetNamespace(), labels, instance.Spec.KubernetesConfig.ExistingPasswordSecret); err != nil {
+	if err = r.Healer.UpdateRedisRoleLabel(ctx, instance.GetNamespace(), labels, instance.Spec.KubernetesConfig.ExistingPasswordSecret, instance.Spec.TLS); err != nil {
 		return intctrlutil.RequeueE(ctx, err, "")
 	}
 
-	slaveNodes := k8sutils.GetRedisNodesByRole(ctx, r.K8sClient, instance, "slave")
+	slaveNodes, err := k8sutils.GetRedisNodesByRole(ctx, r.K8sClient, instance, "slave")
+	if err != nil {
+		return intctrlutil.RequeueE(ctx, err, "")
+	}
 	if realMaster != "" {
 		monitoring.RedisReplicationConnectedSlavesTotal.WithLabelValues(instance.Namespace, instance.Name).Set(float64(len(slaveNodes)))
 	} else {

internal/controller/redissentinel/redissentinel_controller.go

@@ -54,9 +54,9 @@ func (r *RedisSentinelReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 	reconcilers := []reconciler{
 		{typ: "finalizer", rec: r.reconcileFinalizer},
 		{typ: "replication", rec: r.reconcileReplication},
-		{typ: "sentinel", rec: r.reconcileSentinel},
 		{typ: "pdb", rec: r.reconcilePDB},
 		{typ: "service", rec: r.reconcileService},
+		{typ: "sentinel", rec: r.reconcileSentinel},
 	}
 
 	for _, reconciler := range reconcilers {
@@ -133,7 +133,7 @@ func (r *RedisSentinelReconciler) reconcileSentinel(ctx context.Context, instanc
 		return intctrlutil.RequeueE(ctx, err, "")
 	} else {
 		if instance.Spec.RedisSentinelConfig.ResolveHostnames == "yes" {
-			monitorAddr = fmt.Sprintf("%s.%s-headless.%s.svc", master.Name, rr.Name, rr.Namespace)
+			monitorAddr = fmt.Sprintf("%s.%s.%s.svc.cluster.local", master.Name, common.GetHeadlessServiceNameFromPodName(master.Name), rr.Namespace)
 		} else {
 			monitorAddr = master.Status.PodIP
 		}

internal/k8sutils/redis-sentinel.go

@@ -331,7 +331,11 @@ func getRedisReplicationMasterPod(ctx context.Context, client kubernetes.Interfa
 		log.FromContext(ctx).V(1).Info("Successfully got RedisReplication", "replication name", replicationName, "namespace", replicationNamespace)
 	}
 
-	masterPods := GetRedisNodesByRole(ctx, client, &replicationInstance, "master")
+	masterPods, err := GetRedisNodesByRole(ctx, client, &replicationInstance, "master")
+	if err != nil {
+		log.FromContext(ctx).Error(err, "Failed to get RedisReplication master pods", "replication name", replicationName, "namespace", replicationNamespace)
+		return emptyRedisInfo
+	}
 	if len(masterPods) == 0 {
 		log.FromContext(ctx).Error(errors.New("no master pods found"), "")
 		return emptyRedisInfo