support memorybrokers at triggers

Author: odacremolbap

pkg/reconciler/trigger/controller.go

@@ -12,10 +12,12 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
+	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/resolver"
 
 	eventingv1alpha1 "github.com/triggermesh/triggermesh-core/pkg/apis/eventing/v1alpha1"
+	mbinformer "github.com/triggermesh/triggermesh-core/pkg/client/generated/injection/informers/eventing/v1alpha1/memorybroker"
 	rbinformer "github.com/triggermesh/triggermesh-core/pkg/client/generated/injection/informers/eventing/v1alpha1/redisbroker"
 	tginformer "github.com/triggermesh/triggermesh-core/pkg/client/generated/injection/informers/eventing/v1alpha1/trigger"
 	tgreconciler "github.com/triggermesh/triggermesh-core/pkg/client/generated/injection/reconciler/eventing/v1alpha1/trigger"
@@ -29,9 +31,11 @@ func NewController(
 ) *controller.Impl {
 	tgInformer := tginformer.Get(ctx)
 	rbInformer := rbinformer.Get(ctx)
+	mbInformer := mbinformer.Get(ctx)
 
 	r := &Reconciler{
 		rbLister: rbInformer.Lister(),
+		mbLister: mbInformer.Lister(),
 	}
 
 	impl := tgreconciler.NewImpl(ctx, r)
@@ -40,22 +44,29 @@ func NewController(
 
 	tgInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
-	// Filter Redis brokers that are referenced by triggers.
+	// Filter brokers that are referenced by triggers.
 	filterBroker := func(obj interface{}) bool {
-		// TODO other brokers should be supported.
+		// TODO duck
+		var accessor kmeta.OwnerRefableAccessor
 		rb, ok := obj.(*eventingv1alpha1.RedisBroker)
 		if !ok {
-			return false
+			mb, ok := obj.(*eventingv1alpha1.MemoryBroker)
+			if !ok {
+				return false
+			}
+			accessor = kmeta.OwnerRefableAccessor(mb)
+		} else {
+			accessor = kmeta.OwnerRefableAccessor(rb)
 		}
 
-		tgl, err := tgInformer.Lister().Triggers(rb.Namespace).List(labels.Everything())
+		tgl, err := tgInformer.Lister().Triggers(accessor.GetNamespace()).List(labels.Everything())
 		if err != nil {
 			logging.FromContext(ctx).Error("Unable to list Triggers", zap.Error(err))
 			return false
 		}
 
 		for _, tg := range tgl {
-			if tg.ReferencesBroker(rb) {
+			if tg.ReferencesBroker(accessor) {
 				return true
 			}
 		}
@@ -64,20 +75,27 @@ func NewController(
 	}
 
 	enqueueFromBroker := func(obj interface{}) {
-		// TODO check GVK if other brokers are  supported.
+		// TODO duck
+		var accessor kmeta.OwnerRefableAccessor
 		rb, ok := obj.(*eventingv1alpha1.RedisBroker)
 		if !ok {
-			return
+			mb, ok := obj.(*eventingv1alpha1.MemoryBroker)
+			if !ok {
+				return
+			}
+			accessor = kmeta.OwnerRefableAccessor(mb)
+		} else {
+			accessor = kmeta.OwnerRefableAccessor(rb)
 		}
 
-		tgl, err := tgInformer.Lister().Triggers(rb.Namespace).List(labels.Everything())
+		tgl, err := tgInformer.Lister().Triggers(accessor.GetNamespace()).List(labels.Everything())
 		if err != nil {
 			logging.FromContext(ctx).Error("Unable to list Triggers", zap.Error(err))
 			return
 		}
 
 		for _, tg := range tgl {
-			if tg.ReferencesBroker(rb) {
+			if tg.ReferencesBroker(accessor) {
 				impl.EnqueueKey(types.NamespacedName{
 					Name:      tg.Name,
 					Namespace: tg.Namespace,
@@ -91,5 +109,10 @@ func NewController(
 		Handler:    controller.HandleAll(enqueueFromBroker),
 	})
 
+	mbInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: filterBroker,
+		Handler:    controller.HandleAll(enqueueFromBroker),
+	})
+
 	return impl
 }

pkg/reconciler/trigger/reconciler.go

@@ -21,7 +21,9 @@ import (
 )
 
 type Reconciler struct {
+	// TODO duck brokers
 	rbLister    eventingv1alpha1listers.RedisBrokerLister
+	mbLister    eventingv1alpha1listers.MemoryBrokerLister
 	uriResolver *resolver.URIResolver
 }
 
@@ -40,7 +42,26 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, t *eventingv1alpha1.Trig
 }
 
 func (r *Reconciler) resolveBroker(ctx context.Context, t *eventingv1alpha1.Trigger) pkgreconciler.Event {
-	// TODO, use any broker, not RedisBrokers
+	// TODO duck
+	// TODO move to webhook
+	if t.Spec.Broker.Group != "" && t.Spec.Broker.Group != eventingv1alpha1.SchemeGroupVersion.Group {
+		return controller.NewPermanentError(fmt.Errorf("not supported Broker Group %q", t.Spec.Broker.Group))
+	}
+
+	var rb *eventingv1alpha1.RedisBroker
+	if t.Spec.Broker.Kind == rb.GetGroupVersionKind().Kind {
+		return r.resolveRedisBroker(ctx, t)
+	}
+
+	var mb *eventingv1alpha1.MemoryBroker
+	if t.Spec.Broker.Kind != mb.GetGroupVersionKind().Kind {
+		return controller.NewPermanentError(fmt.Errorf("not supported Broker Kind %q", t.Spec.Broker.Kind))
+	}
+
+	return r.resolveMemoryBroker(ctx, t)
+}
+
+func (r *Reconciler) resolveRedisBroker(ctx context.Context, t *eventingv1alpha1.Trigger) pkgreconciler.Event {
 	rb, err := r.rbLister.RedisBrokers(t.Namespace).Get(t.Spec.Broker.Name)
 	if err != nil {
 		if apierrs.IsNotFound(err) {
@@ -65,6 +86,31 @@ func (r *Reconciler) resolveBroker(ctx context.Context, t *eventingv1alpha1.Trig
 	return nil
 }
 
+func (r *Reconciler) resolveMemoryBroker(ctx context.Context, t *eventingv1alpha1.Trigger) pkgreconciler.Event {
+	mb, err := r.mbLister.MemoryBrokers(t.Namespace).Get(t.Spec.Broker.Name)
+	if err != nil {
+		if apierrs.IsNotFound(err) {
+			logging.FromContext(ctx).Errorw(fmt.Sprintf("Trigger %s/%s references non existing broker %q", t.Namespace, t.Name, t.Spec.Broker.Name))
+			t.Status.MarkBrokerFailed(common.ReasonBrokerDoesNotExist, "Broker %q does not exist", t.Spec.Broker.Name)
+			// No need to requeue, we will be notified when if broker is created.
+			return controller.NewPermanentError(err)
+		}
+
+		t.Status.MarkBrokerFailed(common.ReasonFailedBrokerGet, "Failed to get broker %q : %s", t.Spec.Broker, err)
+		return pkgreconciler.NewEvent(corev1.EventTypeWarning, common.ReasonFailedBrokerGet,
+			"Failed to get broker for trigger %s/%s: %w", t.Namespace, t.Name, err)
+	}
+
+	t.Status.PropagateBrokerCondition(mb.Status.GetTopLevelCondition())
+
+	// No need to requeue, we'll get requeued when broker changes status.
+	if !mb.IsReady() {
+		logging.FromContext(ctx).Errorw(fmt.Sprintf("Trigger %s/%s references non ready broker %q", t.Namespace, t.Name, t.Spec.Broker.Name))
+	}
+
+	return nil
+}
+
 func (r *Reconciler) resolveTarget(ctx context.Context, t *eventingv1alpha1.Trigger) pkgreconciler.Event {
 	if t.Spec.Target.Ref != nil && t.Spec.Target.Ref.Namespace == "" {
 		// To call URIFromDestinationV1(ctx context.Context, dest v1.Destination, parent interface{}), dest.Ref must have a Namespace