reconcile redis service endpoints

Author: odacremolbap

pkg/apis/eventing/v1alpha1/redisbroker_lifecycle.go

@@ -18,26 +18,24 @@ import (
 // RedisBrokerBroker refers to the TriggerMesh Broker that manages events on top of Redis.
 
 const (
-	RedisBrokerConditionReady                                         = apis.ConditionReady
-	RedisBrokerRedisDeployment                     apis.ConditionType = "RedisDeploymentReady"
-	RedisBrokerRedisService                        apis.ConditionType = "RedisServiceReady"
-	RedisBrokerRedisServiceConditionEndpointsReady apis.ConditionType = "RedisEndpointsReady"
-	RedisBrokerBrokerDeployment                    apis.ConditionType = "BrokerDeploymentReady"
-	RedisBrokerBrokerService                       apis.ConditionType = "BrokerServiceReady"
-
-	// RedisBrokerBrokerServiceConditionEndpointsReady has status True when a k8s Service Endpoints
-	// are backed by at least one endpoint.
-	RedisBrokerBrokerServiceConditionEndpointsReady apis.ConditionType = "BrokerEndpointsReady"
+	RedisBrokerConditionReady                                          = apis.ConditionReady
+	RedisBrokerRedisDeployment                      apis.ConditionType = "RedisDeploymentReady"
+	RedisBrokerRedisService                         apis.ConditionType = "RedisServiceReady"
+	RedisBrokerRedisServiceEndpointsConditionReady  apis.ConditionType = "RedisEndpointsReady"
+	RedisBrokerBrokerDeployment                     apis.ConditionType = "BrokerDeploymentReady"
+	RedisBrokerBrokerService                        apis.ConditionType = "BrokerServiceReady"
+	RedisBrokerBrokerServiceEndpointsConditionReady apis.ConditionType = "BrokerEndpointsReady"
 	RedisBrokerConfigSecret                         apis.ConditionType = "BrokerConfigSecretReady"
 	RedisBrokerConditionAddressable                 apis.ConditionType = "Addressable"
 )
 
 var redisBrokerCondSet = apis.NewLivingConditionSet(
 	RedisBrokerRedisDeployment,
 	RedisBrokerRedisService,
+	RedisBrokerRedisServiceEndpointsConditionReady,
 	RedisBrokerBrokerDeployment,
 	RedisBrokerBrokerService,
-	RedisBrokerBrokerServiceConditionEndpointsReady,
+	RedisBrokerBrokerServiceEndpointsConditionReady,
 	RedisBrokerConfigSecret,
 
 // TODO RedisBrokerConditionAddressable,
@@ -155,6 +153,18 @@ func (bs *RedisBrokerStatus) MarkRedisServiceReady() {
 	redisBrokerCondSet.Manage(bs).MarkTrue(RedisBrokerRedisService)
 }
 
+func (bs *RedisBrokerStatus) MarkRedisEndpointsFailed(reason, messageFormat string, messageA ...interface{}) {
+	redisBrokerCondSet.Manage(bs).MarkFalse(RedisBrokerRedisServiceEndpointsConditionReady, reason, messageFormat, messageA...)
+}
+
+func (bs *RedisBrokerStatus) MarkRedisEndpointsUnknown(reason, messageFormat string, messageA ...interface{}) {
+	redisBrokerCondSet.Manage(bs).MarkUnknown(RedisBrokerRedisServiceEndpointsConditionReady, reason, messageFormat, messageA...)
+}
+
+func (bs *RedisBrokerStatus) MarkRedisEndpointsTrue() {
+	redisBrokerCondSet.Manage(bs).MarkTrue(RedisBrokerRedisServiceEndpointsConditionReady)
+}
+
 // Manage Redis broker state for
 // Deployment, Service and Endpoint
 
@@ -196,13 +206,13 @@ func (bs *RedisBrokerStatus) MarkBrokerServiceReady() {
 }
 
 func (bs *RedisBrokerStatus) MarkBrokerEndpointsFailed(reason, messageFormat string, messageA ...interface{}) {
-	redisBrokerCondSet.Manage(bs).MarkFalse(RedisBrokerBrokerServiceConditionEndpointsReady, reason, messageFormat, messageA...)
+	redisBrokerCondSet.Manage(bs).MarkFalse(RedisBrokerBrokerServiceEndpointsConditionReady, reason, messageFormat, messageA...)
 }
 
 func (bs *RedisBrokerStatus) MarkBrokerEndpointsUnknown(reason, messageFormat string, messageA ...interface{}) {
-	redisBrokerCondSet.Manage(bs).MarkUnknown(RedisBrokerBrokerServiceConditionEndpointsReady, reason, messageFormat, messageA...)
+	redisBrokerCondSet.Manage(bs).MarkUnknown(RedisBrokerBrokerServiceEndpointsConditionReady, reason, messageFormat, messageA...)
 }
 
 func (bs *RedisBrokerStatus) MarkBrokerEndpointsTrue() {
-	redisBrokerCondSet.Manage(bs).MarkTrue(RedisBrokerBrokerServiceConditionEndpointsReady)
+	redisBrokerCondSet.Manage(bs).MarkTrue(RedisBrokerBrokerServiceEndpointsConditionReady)
 }

pkg/reconciler/redisbroker/controller.go

@@ -63,7 +63,14 @@ func NewController(
 	r := &Reconciler{
 		kubeClientSet:    kubeclient.Get(ctx),
 		secretReconciler: newSecretReconciler(ctx, secretInformer.Lister(), trgInformer.Lister()),
-		redisReconciler:  newRedisReconciler(ctx, deploymentInformer.Lister(), serviceInformer.Lister(), env.RedisImage),
+		redisReconciler: redisReconciler{
+			client:           kubeclient.Get(ctx),
+			deploymentLister: deploymentInformer.Lister(),
+			serviceLister:    serviceInformer.Lister(),
+			endpointsLister:  endpointsInformer.Lister(),
+			image:            env.RedisImage,
+		},
+
 		brokerReconciler: brokerReconciler{
 			client:           kubeclient.Get(ctx),
 			deploymentLister: deploymentInformer.Lister(),

pkg/reconciler/redisbroker/reconcile_broker.go

@@ -55,7 +55,6 @@ func (r *brokerReconciler) reconcile(ctx context.Context, rb *eventingv1alpha1.R
 		return d, nil, err
 	}
 
-	// update endpoint statuses
 	_, err = r.reconcileEndpoints(ctx, svc, rb)
 	if err != nil {
 		return d, nil, err
@@ -221,12 +220,12 @@ func (r *brokerReconciler) reconcileEndpoints(ctx context.Context, service *core
 		rb.Status.MarkBrokerEndpointsFailed(reconciler.ReasonUnavailableEndpoints, "Endpoints for broker service do not exist")
 		return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonUnavailableEndpoints,
 			"Endpoints for broker service do not exist %s",
-			types.NamespacedName{Namespace: ep.Namespace, Name: ep.Name})
+			types.NamespacedName{Namespace: service.Namespace, Name: service.Name})
 	}
 
-	fullname := types.NamespacedName{Namespace: ep.Namespace, Name: ep.Name}
-	rb.Status.MarkBrokerEndpointsFailed(reconciler.ReasonFailedEndpointsGet, "Could not retrieve endpoints for broker service")
-	logging.FromContext(ctx).Error("Unable to get the service endpoints", zap.String("endpoint", fullname.String()), zap.Error(err))
+	fullname := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
+	rb.Status.MarkBrokerEndpointsUnknown(reconciler.ReasonFailedEndpointsGet, "Could not retrieve endpoints for broker service")
+	logging.FromContext(ctx).Error("Unable to get the broker service endpoints", zap.String("endpoint", fullname.String()), zap.Error(err))
 	return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonFailedEndpointsGet,
 		"Failed to get broker service ednpoints %s: %w", fullname, err)
 }

pkg/reconciler/redisbroker/reconcile_redis.go

@@ -12,7 +12,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	appsv1listers "k8s.io/client-go/listers/apps/v1"
 	corev1listers "k8s.io/client-go/listers/core/v1"
-	k8sclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/eventing/pkg/apis/duck"
 	"knative.dev/pkg/logging"
 	pkgreconciler "knative.dev/pkg/reconciler"
 
@@ -34,15 +34,6 @@ type redisReconciler struct {
 	image            string
 }
 
-func newRedisReconciler(ctx context.Context, deploymentLister appsv1listers.DeploymentLister, serviceLister corev1listers.ServiceLister, image string) redisReconciler {
-	return redisReconciler{
-		client:           k8sclient.Get(ctx),
-		deploymentLister: deploymentLister,
-		serviceLister:    serviceLister,
-		image:            image,
-	}
-}
-
 func (r *redisReconciler) reconcile(ctx context.Context, rb *eventingv1alpha1.RedisBroker) (*appsv1.Deployment, *corev1.Service, error) {
 	d, err := r.reconcileDeployment(ctx, rb)
 	if err != nil {
@@ -54,7 +45,10 @@ func (r *redisReconciler) reconcile(ctx context.Context, rb *eventingv1alpha1.Re
 		return d, nil, err
 	}
 
-	// update endpoint statuses
+	_, err = r.reconcileEndpoints(ctx, svc, rb)
+	if err != nil {
+		return d, nil, err
+	}
 
 	return d, svc, nil
 }
@@ -184,3 +178,31 @@ func (r *redisReconciler) reconcileService(ctx context.Context, rb *eventingv1al
 
 	return current, nil
 }
+
+func (r *redisReconciler) reconcileEndpoints(ctx context.Context, service *corev1.Service, rb *eventingv1alpha1.RedisBroker) (*corev1.Endpoints, error) {
+	ep, err := r.endpointsLister.Endpoints(service.Namespace).Get(service.Name)
+	switch {
+	case err == nil:
+		if duck.EndpointsAreAvailable(ep) {
+			rb.Status.MarkRedisEndpointsTrue()
+			return ep, nil
+		}
+
+		rb.Status.MarkRedisEndpointsFailed(reconciler.ReasonUnavailableEndpoints, "Endpoints for redis service are not available")
+		return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonUnavailableEndpoints,
+			"Endpoints for redis service are not available %s",
+			types.NamespacedName{Namespace: ep.Namespace, Name: ep.Name})
+
+	case apierrs.IsNotFound(err):
+		rb.Status.MarkRedisEndpointsFailed(reconciler.ReasonUnavailableEndpoints, "Endpoints for redis service do not exist")
+		return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonUnavailableEndpoints,
+			"Endpoints for redis service do not exist %s",
+			types.NamespacedName{Namespace: service.Namespace, Name: service.Name})
+	}
+
+	fullname := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
+	rb.Status.MarkRedisEndpointsUnknown(reconciler.ReasonFailedEndpointsGet, "Could not retrieve endpoints for redis service")
+	logging.FromContext(ctx).Error("Unable to get the redis service endpoints", zap.String("endpoint", fullname.String()), zap.Error(err))
+	return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonFailedEndpointsGet,
+		"Failed to get redis service ednpoints %s: %w", fullname, err)
+}