add broker service endpoint reconcile

Author: odacremolbap

config/200-clusterroles.yaml

@@ -34,11 +34,12 @@ rules:
   - delete
   - patch
 
-# Manage broker services and configuration as secrets
+# Manage broker services, endpoints and secrets (for configuration)
 - apiGroups:
   - ''
   resources:
   - services
+  - endpoints
   - secrets
   verbs:
   - get

pkg/apis/eventing/v1alpha1/redisbroker_lifecycle.go

@@ -18,20 +18,26 @@ 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"
-	RedisBrokerBrokerDeployment     apis.ConditionType = "BrokerDeploymentReady"
-	RedisBrokerBrokerService        apis.ConditionType = "BrokerServiceReady"
-	RedisBrokerConfigSecret         apis.ConditionType = "BrokerConfigSecretReady"
-	RedisBrokerConditionAddressable apis.ConditionType = "Addressable"
+	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"
+	RedisBrokerConfigSecret                         apis.ConditionType = "BrokerConfigSecretReady"
+	RedisBrokerConditionAddressable                 apis.ConditionType = "Addressable"
 )
 
 var redisBrokerCondSet = apis.NewLivingConditionSet(
 	RedisBrokerRedisDeployment,
 	RedisBrokerRedisService,
 	RedisBrokerBrokerDeployment,
 	RedisBrokerBrokerService,
+	RedisBrokerBrokerServiceConditionEndpointsReady,
 	RedisBrokerConfigSecret,
 
 // TODO RedisBrokerConditionAddressable,
@@ -149,8 +155,8 @@ func (bs *RedisBrokerStatus) MarkRedisServiceReady() {
 	redisBrokerCondSet.Manage(bs).MarkTrue(RedisBrokerRedisService)
 }
 
-// Manage Redis broker state for both
-// Service and Deployment
+// Manage Redis broker state for
+// Deployment, Service and Endpoint
 
 func (bs *RedisBrokerStatus) MarkBrokerDeploymentFailed(reason, messageFormat string, messageA ...interface{}) {
 	redisBrokerCondSet.Manage(bs).MarkFalse(RedisBrokerBrokerDeployment, reason, messageFormat, messageA...)
@@ -188,3 +194,15 @@ func (bs *RedisBrokerStatus) MarkBrokerServiceUnknown(reason, messageFormat stri
 func (bs *RedisBrokerStatus) MarkBrokerServiceReady() {
 	redisBrokerCondSet.Manage(bs).MarkTrue(RedisBrokerBrokerService)
 }
+
+func (bs *RedisBrokerStatus) MarkBrokerEndpointsFailed(reason, messageFormat string, messageA ...interface{}) {
+	redisBrokerCondSet.Manage(bs).MarkFalse(RedisBrokerBrokerServiceConditionEndpointsReady, reason, messageFormat, messageA...)
+}
+
+func (bs *RedisBrokerStatus) MarkBrokerEndpointsUnknown(reason, messageFormat string, messageA ...interface{}) {
+	redisBrokerCondSet.Manage(bs).MarkUnknown(RedisBrokerBrokerServiceConditionEndpointsReady, reason, messageFormat, messageA...)
+}
+
+func (bs *RedisBrokerStatus) MarkBrokerEndpointsTrue() {
+	redisBrokerCondSet.Manage(bs).MarkTrue(RedisBrokerBrokerServiceConditionEndpointsReady)
+}

pkg/reconciler/events.go

@@ -28,6 +28,9 @@ const (
 
 	ReasonFailedTriggerList     = "FailedTriggerList"
 	ReasonFailedConfigSerialize = "FailedConfigSerialize"
+
+	ReasonUnavailableEndpoints = "UnavailableEndpoints"
+	ReasonFailedEndpointsGet   = "FailedEndpointsGet"
 	// ReasonServiceUpdate       = "UpdateService"
 	// ReasonFailedServiceGet    = "FailedServiceGet"
 	// ReasonFailedServiceCreate = "FailedServiceCreate"

pkg/reconciler/redisbroker/controller.go

@@ -8,11 +8,14 @@ import (
 
 	"github.com/kelseyhightower/envconfig"
 	"go.uber.org/zap"
+
+	corev1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
+	endpointsinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/endpoints"
 	"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"
@@ -52,6 +55,7 @@ func NewController(
 	secretInformer := secret.Get(ctx)
 	deploymentInformer := deployment.Get(ctx)
 	serviceInformer := service.Get(ctx)
+	endpointsInformer := endpointsinformer.Get(ctx)
 	serviceAccountInformer := serviceaccount.Get(ctx)
 	// TODO rolebinding
 	_ = rolebinding.Get(ctx)
@@ -60,7 +64,13 @@ func NewController(
 		kubeClientSet:    kubeclient.Get(ctx),
 		secretReconciler: newSecretReconciler(ctx, secretInformer.Lister(), trgInformer.Lister()),
 		redisReconciler:  newRedisReconciler(ctx, deploymentInformer.Lister(), serviceInformer.Lister(), env.RedisImage),
-		brokerReconciler: newBrokerReconciler(ctx, deploymentInformer.Lister(), serviceInformer.Lister(), env.BrokerImage),
+		brokerReconciler: brokerReconciler{
+			client:           kubeclient.Get(ctx),
+			deploymentLister: deploymentInformer.Lister(),
+			serviceLister:    serviceInformer.Lister(),
+			endpointsLister:  endpointsInformer.Lister(),
+			image:            env.BrokerImage,
+		},
 	}
 
 	impl := rbreconciler.NewImpl(ctx, r)
@@ -81,6 +91,32 @@ func NewController(
 		FilterFunc: controller.FilterController(rb),
 		Handler:    controller.HandleAll(impl.EnqueueControllerOf),
 	})
+	endpointsInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: func(obj interface{}) bool {
+			ep, ok := obj.(*corev1.Endpoints)
+			if !ok || ep.Labels != nil || ep.Labels[appAnnotation] == appAnnotationValue {
+				return false
+			}
+
+			return true
+		},
+		Handler: controller.HandleAll(func(obj interface{}) {
+			ep, ok := obj.(*corev1.Endpoints)
+			if !ok {
+				return
+			}
+
+			svc, err := serviceInformer.Lister().Services(ep.Namespace).Get(ep.Name)
+			if err != nil {
+				// no matter the error, if we cannot retrieve the service we cannot
+				// read the owner and enqueue the key.
+				return
+			}
+
+			impl.EnqueueControllerOf(svc)
+		}),
+	})
+
 	serviceAccountInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: controller.FilterController(rb),
 		Handler:    controller.HandleAll(impl.EnqueueControllerOf),

pkg/reconciler/redisbroker/reconcile_broker.go

@@ -6,6 +6,7 @@ import (
 	"path"
 
 	"go.uber.org/zap"
+
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
@@ -14,7 +15,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"
 
@@ -39,18 +40,10 @@ type brokerReconciler struct {
 	client           kubernetes.Interface
 	deploymentLister appsv1listers.DeploymentLister
 	serviceLister    corev1listers.ServiceLister
+	endpointsLister  corev1listers.EndpointsLister
 	image            string
 }
 
-func newBrokerReconciler(ctx context.Context, deploymentLister appsv1listers.DeploymentLister, serviceLister corev1listers.ServiceLister, image string) brokerReconciler {
-	return brokerReconciler{
-		client:           k8sclient.Get(ctx),
-		deploymentLister: deploymentLister,
-		serviceLister:    serviceLister,
-		image:            image,
-	}
-}
-
 func (r *brokerReconciler) reconcile(ctx context.Context, rb *eventingv1alpha1.RedisBroker, redis *corev1.Service, secret *corev1.Secret) (*appsv1.Deployment, *corev1.Service, error) {
 	d, err := r.reconcileDeployment(ctx, rb, redis, secret)
 	if err != nil {
@@ -62,6 +55,12 @@ 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
+	}
+
 	return d, svc, nil
 }
 
@@ -76,7 +75,7 @@ func buildBrokerDeployment(rb *eventingv1alpha1.RedisBroker, redis *corev1.Servi
 
 	return resources.NewDeployment(rb.Namespace, rb.Name+"-"+brokerResourceSuffix,
 		resources.DeploymentWithMetaOptions(
-			resources.MetaAddLabel("app", appAnnotationValue),
+			resources.MetaAddLabel(appAnnotation, appAnnotationValue),
 			resources.MetaAddLabel("component", brokerResourceSuffix),
 			resources.MetaAddLabel(resourceNameAnnotation, rb.Name+"-"+brokerResourceSuffix),
 			resources.MetaAddOwner(rb, rb.GetGroupVersionKind())),
@@ -145,7 +144,7 @@ func (r *brokerReconciler) reconcileDeployment(ctx context.Context, rb *eventing
 func buildBrokerService(rb *eventingv1alpha1.RedisBroker) *corev1.Service {
 	return resources.NewService(rb.Namespace, rb.Name+"-"+brokerResourceSuffix,
 		resources.ServiceWithMetaOptions(
-			resources.MetaAddLabel("app", appAnnotationValue),
+			resources.MetaAddLabel(appAnnotation, appAnnotationValue),
 			resources.MetaAddLabel("component", brokerResourceSuffix),
 			resources.MetaAddLabel(resourceNameAnnotation, rb.Name+"-"+brokerResourceSuffix),
 			resources.MetaAddOwner(rb, rb.GetGroupVersionKind())),
@@ -159,6 +158,7 @@ func (r *brokerReconciler) reconcileService(ctx context.Context, rb *eventingv1a
 	current, err := r.serviceLister.Services(desired.Namespace).Get(desired.Name)
 	switch {
 	case err == nil:
+		// Set Status
 		// Compare current object with desired, update if needed.
 		if !semantic.Semantic.DeepEqual(desired, current) {
 			desired.Status = current.Status
@@ -202,3 +202,31 @@ func (r *brokerReconciler) reconcileService(ctx context.Context, rb *eventingv1a
 
 	return current, nil
 }
+
+func (r *brokerReconciler) 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.MarkBrokerEndpointsTrue()
+			return ep, nil
+		}
+
+		rb.Status.MarkBrokerEndpointsFailed(reconciler.ReasonUnavailableEndpoints, "Endpoints for broker service are not available")
+		return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonUnavailableEndpoints,
+			"Endpoints for broker service are not available %s",
+			types.NamespacedName{Namespace: ep.Namespace, Name: ep.Name})
+
+	case apierrs.IsNotFound(err):
+		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})
+	}
+
+	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))
+	return nil, pkgreconciler.NewEvent(corev1.EventTypeWarning, reconciler.ReasonFailedEndpointsGet,
+		"Failed to get broker service ednpoints %s: %w", fullname, err)
+}

pkg/reconciler/redisbroker/reconcile_redis.go

@@ -30,6 +30,7 @@ type redisReconciler struct {
 	client           kubernetes.Interface
 	deploymentLister appsv1listers.DeploymentLister
 	serviceLister    corev1listers.ServiceLister
+	endpointsLister  corev1listers.EndpointsLister
 	image            string
 }
 
@@ -53,13 +54,15 @@ func (r *redisReconciler) reconcile(ctx context.Context, rb *eventingv1alpha1.Re
 		return d, nil, err
 	}
 
+	// update endpoint statuses
+
 	return d, svc, nil
 }
 
 func buildRedisDeployment(rb *eventingv1alpha1.RedisBroker, image string) *appsv1.Deployment {
 	return resources.NewDeployment(rb.Namespace, rb.Name+"-"+redisResourceSuffix,
 		resources.DeploymentWithMetaOptions(
-			resources.MetaAddLabel("app", appAnnotationValue),
+			resources.MetaAddLabel(appAnnotation, appAnnotationValue),
 			resources.MetaAddLabel("component", redisResourceSuffix),
 			resources.MetaAddLabel(resourceNameAnnotation, rb.Name+"-"+redisResourceSuffix),
 			resources.MetaAddOwner(rb, rb.GetGroupVersionKind())),
@@ -124,7 +127,7 @@ func (r *redisReconciler) reconcileDeployment(ctx context.Context, rb *eventingv
 func buildRedisService(rb *eventingv1alpha1.RedisBroker) *corev1.Service {
 	return resources.NewService(rb.Namespace, rb.Name+"-"+redisResourceSuffix,
 		resources.ServiceWithMetaOptions(
-			resources.MetaAddLabel("app", appAnnotationValue),
+			resources.MetaAddLabel(appAnnotation, appAnnotationValue),
 			resources.MetaAddLabel("component", redisResourceSuffix),
 			resources.MetaAddLabel(resourceNameAnnotation, rb.Name+"-"+redisResourceSuffix),
 			resources.MetaAddOwner(rb, rb.GetGroupVersionKind())),

pkg/reconciler/redisbroker/reconcile_secret.go

@@ -165,7 +165,7 @@ func (r *secretReconciler) buildConfigSecret(ctx context.Context, rb *eventingv1
 
 	return resources.NewSecret(rb.Namespace, rb.Name,
 		resources.SecretWithMetaOptions(
-			resources.MetaAddLabel("app", redisResourceSuffix),
+			resources.MetaAddLabel(appAnnotation, redisResourceSuffix),
 			resources.MetaAddLabel(resourceNameAnnotation, rb.Name+"-"+"redisbroker-config"),
 			resources.MetaAddOwner(rb, rb.GetGroupVersionKind())),
 		resources.SecretSetData(configSecretKey, b)), nil

pkg/reconciler/redisbroker/reconciler.go

@@ -16,6 +16,7 @@ import (
 )
 
 const (
+	appAnnotation          = "app"
 	appAnnotationValue     = "redisbroker"
 	resourceNameAnnotation = "eventing.triggermesh.io/name"
 )