feat: support create redis cluster by nodeport service

.github/workflows/e2e-chainsaw.yml

@@ -20,6 +20,7 @@ jobs:
           - ./tests/e2e-chainsaw/v1beta2/hostnetwork/
           - ./tests/e2e-chainsaw/v1beta2/password/
           - ./tests/e2e-chainsaw/v1beta2/ha-setup/
+          - ./tests/e2e-chainsaw/v1beta2/nodeport/
 
     steps:
     - name: Checkout code

controllers/rediscluster_controller.go

@@ -116,16 +116,16 @@
 		}
 	}
 
-	err = k8sutils.CreateRedisLeader(instance)
-	if err != nil {
-		return ctrl.Result{}, err
-	}
 	if leaderReplicas != 0 {
 		err = k8sutils.CreateRedisLeaderService(instance)
 		if err != nil {
 			return ctrl.Result{}, err
 		}
 	}
+	err = k8sutils.CreateRedisLeader(instance)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
 
 	err = k8sutils.ReconcileRedisPodDisruptionBudget(instance, "leader", instance.Spec.RedisLeader.PodDisruptionBudget)
 	if err != nil {
@@ -149,18 +149,17 @@
 				return ctrl.Result{}, err
 			}
 		}
-
-		err = k8sutils.CreateRedisFollower(instance)
-		if err != nil {
-			return ctrl.Result{}, err
-		}
 		// if we have followers create their service.
 		if followerReplicas != 0 {
 			err = k8sutils.CreateRedisFollowerService(instance)
 			if err != nil {
 				return ctrl.Result{}, err
 			}
 		}
+		err = k8sutils.CreateRedisFollower(instance)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
 		err = k8sutils.ReconcileRedisPodDisruptionBudget(instance, "follower", instance.Spec.RedisFollower.PodDisruptionBudget)
 		if err != nil {
 			return ctrl.Result{}, err

k8sutils/redis-cluster.go

@@ -1,6 +1,11 @@
 package k8sutils
 
 import (
+	"strconv"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/util/intstr"
+
 	commonapi "github.com/OT-CONTAINER-KIT/redis-operator/api"
 	redisv1beta2 "github.com/OT-CONTAINER-KIT/redis-operator/api/v1beta2"
 	"github.com/OT-CONTAINER-KIT/redis-operator/pkg/util"
@@ -31,6 +36,7 @@
 	res := statefulSetParameters{
 		Replicas:                      &replicas,
 		ClusterMode:                   true,
+		ClusterVersion:                cr.Spec.ClusterVersion,
 		NodeConfVolume:                cr.Spec.Storage.NodeConfVolume,
 		NodeSelector:                  params.NodeSelector,
 		PodSecurityContext:            cr.Spec.PodSecurityContext,
@@ -94,7 +100,7 @@
 }
 
 // generateRedisClusterContainerParams generates Redis container information
-func generateRedisClusterContainerParams(cr *redisv1beta2.RedisCluster, securityContext *corev1.SecurityContext, readinessProbeDef *commonapi.Probe, livenessProbeDef *commonapi.Probe) containerParameters {
+func generateRedisClusterContainerParams(cr *redisv1beta2.RedisCluster, securityContext *corev1.SecurityContext, readinessProbeDef *commonapi.Probe, livenessProbeDef *commonapi.Probe, role string) containerParameters {
 	trueProperty := true
 	falseProperty := false
 	containerProp := containerParameters{
@@ -108,6 +114,50 @@
 	if cr.Spec.EnvVars != nil {
 		containerProp.EnvVars = cr.Spec.EnvVars
 	}
+	if cr.Spec.KubernetesConfig.Service != nil && cr.Spec.KubernetesConfig.Service.ServiceType == "NodePort" {
+		envVars := util.Coalesce(containerProp.EnvVars, &[]corev1.EnvVar{})
+		*envVars = append(*envVars, corev1.EnvVar{
+			Name:  "NODEPORT",
+			Value: "true",
+		})
+		*envVars = append(*envVars, corev1.EnvVar{
+			Name: "HOST_IP",
+			ValueFrom: &corev1.EnvVarSource{
+				FieldRef: &corev1.ObjectFieldSelector{
+					FieldPath: "status.hostIP",
+				},
+			},
+		})
+
+		type ports struct {
+			announcePort    int
+			announceBusPort int
+		}
+		nps := map[string]ports{} // pod name to ports
+		replicas := cr.Spec.GetReplicaCounts(role)
+		for i := 0; i < int(replicas); i++ {
+			svc, err := getService(cr.Namespace, cr.ObjectMeta.Name+"-"+role+"-"+strconv.Itoa(i))
+			if err != nil {
+				log.Error(err, "Cannot get service for Redis", "Setup.Type", role)
+			} else {
+				nps[svc.Name] = ports{
+					announcePort:    int(svc.Spec.Ports[0].NodePort),
+					announceBusPort: int(svc.Spec.Ports[1].NodePort),
+				}
+			}
+		}
+		for name, np := range nps {
+			*envVars = append(*envVars, corev1.EnvVar{
+				Name:  "announce_port_" + strings.ReplaceAll(name, "-", "_"),
+				Value: strconv.Itoa(np.announcePort),
+			})
+			*envVars = append(*envVars, corev1.EnvVar{
+				Name:  "announce_bus_port_" + strings.ReplaceAll(name, "-", "_"),
+				Value: strconv.Itoa(np.announceBusPort),
+			})
+		}
+		containerProp.EnvVars = envVars
+	}
 	if cr.Spec.Storage != nil {
 		containerProp.AdditionalVolume = cr.Spec.Storage.VolumeMount.Volume
 		containerProp.AdditionalMountPath = cr.Spec.Storage.VolumeMount.MountPath
@@ -223,7 +273,7 @@
 		generateRedisClusterParams(cr, service.getReplicaCount(cr), service.ExternalConfig, service),
 		redisClusterAsOwner(cr),
 		generateRedisClusterInitContainerParams(cr),
-		generateRedisClusterContainerParams(cr, service.SecurityContext, service.ReadinessProbe, service.LivenessProbe),
+		generateRedisClusterContainerParams(cr, service.SecurityContext, service.ReadinessProbe, service.LivenessProbe, service.RedisStateFulType),
 		cr.Spec.Sidecars,
 	)
 	if err != nil {
@@ -268,6 +318,15 @@
 	additionalServiceType := "ClusterIP"
 	if cr.Spec.KubernetesConfig.Service != nil {
 		additionalServiceType = cr.Spec.KubernetesConfig.Service.ServiceType
+		if additionalServiceType == "NodePort" {
+			// If NodePort is enabled, we need to create a service for every redis pod.
+			// Then use --cluster-announce-ip --cluster-announce-port --cluster-announce-bus-port to make cluster.
+			err = service.createOrUpdateClusterNodePortService(cr)
+			if err != nil {
+				logger.Error(err, "Cannot create nodeport service for Redis", "Setup.Type", service.RedisServiceRole)
+				return err
+			}
+		}
 	}
 	err = CreateOrUpdateService(cr.Namespace, additionalObjectMetaInfo, redisClusterAsOwner(cr), disableMetrics, false, additionalServiceType, *cr.Spec.Port)
 	if err != nil {
@@ -276,3 +335,32 @@
 	}
 	return nil
 }
+
+func (service RedisClusterService) createOrUpdateClusterNodePortService(cr *redisv1beta2.RedisCluster) error {
+	replicas := cr.Spec.GetReplicaCounts(service.RedisServiceRole)
+
+	for i := 0; i < int(replicas); i++ {
+		serviceName := cr.ObjectMeta.Name + "-" + service.RedisServiceRole + "-" + strconv.Itoa(i)
+		logger := serviceLogger(cr.Namespace, serviceName)
+		labels := getRedisLabels(cr.ObjectMeta.Name+"-"+service.RedisServiceRole, cluster, service.RedisServiceRole, map[string]string{
+			"statefulset.kubernetes.io/pod-name": serviceName,
+		})
+		annotations := generateServiceAnots(cr.ObjectMeta, nil, disableMetrics)
+		objectMetaInfo := generateObjectMetaInformation(serviceName, cr.Namespace, labels, annotations)
+		busPort := corev1.ServicePort{
+			Name:     "redis-bus",
+			Port:     int32(*cr.Spec.Port + 10000),
+			Protocol: corev1.ProtocolTCP,
+			TargetPort: intstr.IntOrString{
+				Type:   intstr.Int,
+				IntVal: int32(*cr.Spec.Port + 10000),
+			},
+		}
+		err := CreateOrUpdateService(cr.Namespace, objectMetaInfo, redisClusterAsOwner(cr), disableMetrics, false, "NodePort", *cr.Spec.Port, busPort)
+		if err != nil {
+			logger.Error(err, "Cannot create nodeport service for Redis", "Setup.Type", service.RedisServiceRole)
+			return err
+		}
+	}
+	return nil
+}

k8sutils/redis-cluster_test.go

@@ -427,10 +427,10 @@ func Test_generateRedisClusterContainerParams(t *testing.T) {
 		t.Fatalf("Failed to unmarshal file %s: %v", path, err)
 	}
 
-	actualLeaderContainer := generateRedisClusterContainerParams(input, input.Spec.RedisLeader.SecurityContext, input.Spec.RedisLeader.ReadinessProbe, input.Spec.RedisLeader.LivenessProbe)
+	actualLeaderContainer := generateRedisClusterContainerParams(input, input.Spec.RedisLeader.SecurityContext, input.Spec.RedisLeader.ReadinessProbe, input.Spec.RedisLeader.LivenessProbe, "leader")
 	assert.EqualValues(t, expectedLeaderContainer, actualLeaderContainer, "Expected %+v, got %+v", expectedLeaderContainer, actualLeaderContainer)
 
-	actualFollowerContainer := generateRedisClusterContainerParams(input, input.Spec.RedisFollower.SecurityContext, input.Spec.RedisFollower.ReadinessProbe, input.Spec.RedisFollower.LivenessProbe)
+	actualFollowerContainer := generateRedisClusterContainerParams(input, input.Spec.RedisFollower.SecurityContext, input.Spec.RedisFollower.ReadinessProbe, input.Spec.RedisFollower.LivenessProbe, "follower")
 	assert.EqualValues(t, expectedFollowerContainer, actualFollowerContainer, "Expected %+v, got %+v", expectedFollowerContainer, actualFollowerContainer)
 }
 

k8sutils/services.go

@@ -30,7 +30,7 @@
 }
 
 // generateServiceDef generates service definition for Redis
-func generateServiceDef(serviceMeta metav1.ObjectMeta, epp exporterPortProvider, ownerDef metav1.OwnerReference, headless bool, serviceType string, port int) *corev1.Service {
+func generateServiceDef(serviceMeta metav1.ObjectMeta, epp exporterPortProvider, ownerDef metav1.OwnerReference, headless bool, serviceType string, port int, extra ...corev1.ServicePort) *corev1.Service {
 	var PortName string
 	if serviceMeta.Labels["role"] == "sentinel" {
 		PortName = "sentinel-client"
@@ -61,6 +61,10 @@
 		redisExporterService := enableMetricsPort(exporterPort)
 		service.Spec.Ports = append(service.Spec.Ports, *redisExporterService)
 	}
+	if len(extra) > 0 {
+		service.Spec.Ports = append(service.Spec.Ports, extra...)
+	}
+
 	AddOwnerRefToObject(service, ownerDef)
 	return service
 }
@@ -150,9 +154,9 @@
 }
 
 // CreateOrUpdateService method will create or update Redis service
-func CreateOrUpdateService(namespace string, serviceMeta metav1.ObjectMeta, ownerDef metav1.OwnerReference, epp exporterPortProvider, headless bool, serviceType string, port int) error {
+func CreateOrUpdateService(namespace string, serviceMeta metav1.ObjectMeta, ownerDef metav1.OwnerReference, epp exporterPortProvider, headless bool, serviceType string, port int, extra ...corev1.ServicePort) error {
 	logger := serviceLogger(namespace, serviceMeta.Name)
-	serviceDef := generateServiceDef(serviceMeta, epp, ownerDef, headless, serviceType, port)
+	serviceDef := generateServiceDef(serviceMeta, epp, ownerDef, headless, serviceType, port, extra...)
 	storedService, err := getService(namespace, serviceMeta.Name)
 	if err != nil {
 		if errors.IsNotFound(err) {

k8sutils/statefulset.go

@@ -32,6 +32,7 @@
 type statefulSetParameters struct {
 	Replicas                      *int32
 	ClusterMode                   bool
+	ClusterVersion                *string
 	NodeConfVolume                bool
 	NodeSelector                  map[string]string
 	PodSecurityContext            *corev1.PodSecurityContext
@@ -240,6 +241,7 @@
 						params.NodeConfVolume,
 						params.EnableMetrics,
 						params.ExternalConfig,
+						params.ClusterVersion,
 						containerParams.AdditionalMountPath,
 						sidecars,
 					),
@@ -344,7 +346,7 @@
 }
 
 // generateContainerDef generates container definition for Redis
-func generateContainerDef(name string, containerParams containerParameters, clusterMode, nodeConfVolume, enableMetrics bool, externalConfig *string, mountpath []corev1.VolumeMount, sidecars []redisv1beta2.Sidecar) []corev1.Container {
+func generateContainerDef(name string, containerParams containerParameters, clusterMode, nodeConfVolume, enableMetrics bool, externalConfig, clusterVersion *string, mountpath []corev1.VolumeMount, sidecars []redisv1beta2.Sidecar) []corev1.Container {
 	containerDefinition := []corev1.Container{
 		{
 			Name:            name,
@@ -361,6 +363,7 @@
 				containerParams.ACLConfig,
 				containerParams.EnvVars,
 				containerParams.Port,
+				clusterVersion,
 			),
 			ReadinessProbe: getProbeInfo(containerParams.ReadinessProbe),
 			LivenessProbe:  getProbeInfo(containerParams.LivenessProbe),
@@ -588,12 +591,19 @@
 // getEnvironmentVariables returns all the required Environment Variables
 func getEnvironmentVariables(role string, enabledPassword *bool, secretName *string,
 	secretKey *string, persistenceEnabled *bool, tlsConfig *redisv1beta2.TLSConfig,
-	aclConfig *redisv1beta2.ACLConfig, envVar *[]corev1.EnvVar, port *int) []corev1.EnvVar {
+	aclConfig *redisv1beta2.ACLConfig, envVar *[]corev1.EnvVar, port *int, clusterVersion *string) []corev1.EnvVar {
 	envVars := []corev1.EnvVar{
 		{Name: "SERVER_MODE", Value: role},
 		{Name: "SETUP_MODE", Value: role},
 	}
 
+	if clusterVersion != nil {
+		envVars = append(envVars, corev1.EnvVar{
+			Name:  "REDIS_MAJOR_VERSION",
+			Value: *clusterVersion,
+		})
+	}
+
 	var redisHost string
 	if role == "sentinel" {
 		redisHost = "redis://localhost:" + strconv.Itoa(sentinelPort)

k8sutils/statefulset_test.go

@@ -230,6 +230,7 @@ func TestGetEnvironmentVariables(t *testing.T) {
 		aclConfig           *redisv1beta2.ACLConfig
 		envVar              *[]corev1.EnvVar
 		port                *int
+		clusterVersion      *string
 		expectedEnvironment []corev1.EnvVar
 	}{
 		{
@@ -257,6 +258,7 @@ func TestGetEnvironmentVariables(t *testing.T) {
 			envVar: &[]corev1.EnvVar{
 				{Name: "TEST_ENV", Value: "test-value"},
 			},
+			clusterVersion: pointer.String("v6"),
 			expectedEnvironment: []corev1.EnvVar{
 				{Name: "ACL_MODE", Value: "true"},
 				{Name: "PERSISTENCE_ENABLED", Value: "true"},
@@ -276,6 +278,7 @@ func TestGetEnvironmentVariables(t *testing.T) {
 				{Name: "SERVER_MODE", Value: "sentinel"},
 				{Name: "SETUP_MODE", Value: "sentinel"},
 				{Name: "TEST_ENV", Value: "test-value"},
+				{Name: "REDIS_MAJOR_VERSION", Value: "v6"},
 			},
 		},
 		{
@@ -289,6 +292,7 @@ func TestGetEnvironmentVariables(t *testing.T) {
 			aclConfig:          nil,
 			envVar:             nil,
 			port:               nil,
+			clusterVersion:     nil,
 			expectedEnvironment: []corev1.EnvVar{
 				{Name: "REDIS_ADDR", Value: "redis://localhost:6379"},
 				{Name: "SERVER_MODE", Value: "redis"},
@@ -363,7 +367,7 @@ func TestGetEnvironmentVariables(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			actualEnvironment := getEnvironmentVariables(tt.role, tt.enabledPassword, tt.secretName,
-				tt.secretKey, tt.persistenceEnabled, tt.tlsConfig, tt.aclConfig, tt.envVar, tt.port)
+				tt.secretKey, tt.persistenceEnabled, tt.tlsConfig, tt.aclConfig, tt.envVar, tt.port, tt.clusterVersion)
 
 			assert.ElementsMatch(t, tt.expectedEnvironment, actualEnvironment)
 		})

tests/e2e-chainsaw/v1beta2/hostnetwork/redis-cluster/ready-pod.yaml

@@ -13,6 +13,8 @@ spec:
           value: "true"
         - name: REDIS_ADDR
           value: "redis://localhost:6379"
+        - name: REDIS_MAJOR_VERSION
+          value: "v6"
         - name: REDIS_PORT
           value: "6380"
         - name: SERVER_MODE
@@ -39,6 +41,8 @@ spec:
           value: "true"
         - name: REDIS_ADDR
           value: "redis://localhost:6379"
+        - name: REDIS_MAJOR_VERSION
+          value: "v6"
         - name: REDIS_PORT
           value: "6380"
         - name: SERVER_MODE
@@ -65,6 +69,8 @@ spec:
           value: "true"
         - name: REDIS_ADDR
           value: "redis://localhost:6379"
+        - name: REDIS_MAJOR_VERSION
+          value: "v6"
         - name: REDIS_PORT
           value: "6380"
         - name: SERVER_MODE
@@ -91,6 +97,8 @@ spec:
           value: "true"
         - name: REDIS_ADDR
           value: "redis://localhost:6379"
+        - name: REDIS_MAJOR_VERSION
+          value: "v6"
         - name: REDIS_PORT
           value: "6380"
         - name: SERVER_MODE
@@ -117,6 +125,8 @@ spec:
           value: "true"
         - name: REDIS_ADDR
           value: "redis://localhost:6379"
+        - name: REDIS_MAJOR_VERSION
+          value: "v6"
         - name: REDIS_PORT
           value: "6380"
         - name: SERVER_MODE
@@ -143,6 +153,8 @@ spec:
           value: "true"
         - name: REDIS_ADDR
           value: "redis://localhost:6379"
+        - name: REDIS_MAJOR_VERSION
+          value: "v6"
         - name: REDIS_PORT
           value: "6380"
         - name: SERVER_MODE