add redis cluster support

Author: odacremolbap

config/300-redisbroker.yaml

@@ -56,8 +56,13 @@ spec:
                     type: object
                     properties:
                       url:
-                        description: URL of the Redis instance.
+                        description: URL of the Redis standalone instance.
                         type: string
+                      clusterURLs:
+                        description: URLs for the Redis cluster instances.
+                        type: array
+                        items:
+                          type: string
                       username:
                         description: Redis username.
                         type: object
@@ -88,8 +93,9 @@ spec:
                       tlsSkipVerify:
                         description: Skip TLS certificate verification.
                         type: boolean
-                    required:
-                    - url
+                    oneOf:
+                    - required: [url]
+                    - required: [clusterURLs]
 
                   stream:
                     description: Redis stream to be used by the broker.

pkg/apis/eventing/v1alpha1/redisbroker_types.go

@@ -43,8 +43,11 @@ var (
 )
 
 type RedisConnection struct {
-	// Redis URL.
-	URL string `json:"url"`
+	// Redis URL for standalone instances
+	URL *string `json:"url,omitempty"`
+
+	// Redis URLs for cluster instances
+	ClusterURLs []string `json:"clusterURLs,omitempty"`
 
 	// Redis username.
 	Username *SecretValueFromSource `json:"username,omitempty"`

pkg/reconciler/redisbroker/controller.go

@@ -20,7 +20,6 @@ import (
 	"knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 	"knative.dev/pkg/client/injection/kube/informers/core/v1/service"
 	"knative.dev/pkg/client/injection/kube/informers/core/v1/serviceaccount"
-	"knative.dev/pkg/client/injection/kube/informers/rbac/v1/rolebinding"
 	rolebindingsinformer "knative.dev/pkg/client/injection/kube/informers/rbac/v1/rolebinding"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -64,7 +63,7 @@ func NewController(
 	serviceAccountInformer := serviceaccount.Get(ctx)
 	roleBindingsInformer := rolebindingsinformer.Get(ctx)
 
-	_ = rolebinding.Get(ctx)
+	_ = rolebindingsinformer.Get(ctx)
 
 	r := &reconciler{
 		secretReconciler: common.NewSecretReconciler(ctx, secretInformer.Lister(), trgInformer.Lister()),

pkg/reconciler/redisbroker/reconciler.go

@@ -7,11 +7,11 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"strings"
 
 	"go.uber.org/zap"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
-	v1 "k8s.io/api/core/v1"
 
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/logging"
@@ -61,7 +61,16 @@ func redisDeploymentOption(rb *eventingv1alpha1.RedisBroker, redisSvc *corev1.Se
 		resources.ContainerAddEnvFromValue("REDIS_STREAM_MAX_LEN", maxLen)(c)
 
 		if rb.IsUserProvidedRedis() {
-			resources.ContainerAddEnvFromValue("REDIS_ADDRESS", rb.Spec.Redis.Connection.URL)(c)
+
+			// Standalone connections require an address, while cluster connections require an
+			// address list of each endpoint available for the initial connection.
+			if rb.Spec.Redis.Connection.ClusterURLs != nil &&
+				len(rb.Spec.Redis.Connection.ClusterURLs) != 0 {
+				resources.ContainerAddEnvFromValue("REDIS_CLUSTER_ADDRESSES",
+					strings.Join(rb.Spec.Redis.Connection.ClusterURLs, ","))(c)
+			} else {
+				resources.ContainerAddEnvFromValue("REDIS_ADDRESS", *rb.Spec.Redis.Connection.URL)(c)
+			}
 
 			if rb.Spec.Redis.Connection.Username != nil {
 				resources.ContainerAddEnvVarFromSecret("REDIS_USERNAME",
@@ -123,7 +132,7 @@ func (r *reconciler) ReconcileKind(ctx context.Context, rb *eventingv1alpha1.Red
 	return nil
 }
 
-func getServiceAddress(svc *v1.Service) *apis.URL {
+func getServiceAddress(svc *corev1.Service) *apis.URL {
 	var port string
 	if svc.Spec.Ports[0].Port != 80 {
 		port = ":" + strconv.Itoa(int(svc.Spec.Ports[0].Port))