Add per-watch predicates for Owns/For/Watches

This allows per-watch predicates for For, Owns, and Watches in the
builder.  These are in addition to the global predicate set, which apply
to all watches.

Author: mszostok

pkg/builder/controller.go

@@ -37,15 +37,15 @@ var getGvk = apiutil.GVKForObject
 
 // Builder builds a Controller.
 type Builder struct {
-	apiType        runtime.Object
-	mgr            manager.Manager
-	predicates     []predicate.Predicate
-	managedObjects []runtime.Object
-	watchRequest   []watchRequest
-	config         *rest.Config
-	ctrl           controller.Controller
-	ctrlOptions    controller.Options
-	name           string
+	apiType          simpleWatch
+	mgr              manager.Manager
+	globalPredicates []predicate.Predicate
+	managedObjects   []simpleWatch
+	watchRequest     []watchRequest
+	config           *rest.Config
+	ctrl             controller.Controller
+	ctrlOptions      controller.Options
+	name             string
 }
 
 // ControllerManagedBy returns a new controller builder that will be started by the provided Manager
@@ -63,32 +63,46 @@ func (blder *Builder) ForType(apiType runtime.Object) *Builder {
 	return blder.For(apiType)
 }
 
+// simpleWatch represents the information required to construct a "simple" For
+// or Owns watch.
+type simpleWatch struct {
+	item       runtime.Object
+	predicates []predicate.Predicate
+}
+
 // For defines the type of Object being *reconciled*, and configures the ControllerManagedBy to respond to create / delete /
 // update events by *reconciling the object*.
 // This is the equivalent of calling
 // Watches(&source.Kind{Type: apiType}, &handler.EnqueueRequestForObject{})
-func (blder *Builder) For(apiType runtime.Object) *Builder {
-	blder.apiType = apiType
+func (blder *Builder) For(reconciledAPIType runtime.Object, prct ...predicate.Predicate) *Builder {
+	blder.apiType = simpleWatch{
+		item:       reconciledAPIType,
+		predicates: prct,
+	}
 	return blder
 }
 
 // Owns defines types of Objects being *generated* by the ControllerManagedBy, and configures the ControllerManagedBy to respond to
 // create / delete / update events by *reconciling the owner object*.  This is the equivalent of calling
 // Watches(&source.Kind{Type: <ForType-apiType>}, &handler.EnqueueRequestForOwner{OwnerType: apiType, IsController: true})
-func (blder *Builder) Owns(apiType runtime.Object) *Builder {
-	blder.managedObjects = append(blder.managedObjects, apiType)
+func (blder *Builder) Owns(apiType runtime.Object, prct ...predicate.Predicate) *Builder {
+	blder.managedObjects = append(blder.managedObjects, simpleWatch{item: apiType, predicates: prct})
 	return blder
 }
 
+// watchRequest represents the information needed to construct a
+// manually-defined (non-"simple") watch.
 type watchRequest struct {
 	src          source.Source
 	eventhandler handler.EventHandler
+	predicates   []predicate.Predicate
 }
 
 // Watches exposes the lower-level ControllerManagedBy Watches functions through the builder.  Consider using
 // Owns or For instead of Watches directly.
-func (blder *Builder) Watches(src source.Source, eventhandler handler.EventHandler) *Builder {
-	blder.watchRequest = append(blder.watchRequest, watchRequest{src: src, eventhandler: eventhandler})
+// Specified predicates are registered only for given source.
+func (blder *Builder) Watches(src source.Source, eventhandler handler.EventHandler, prct ...predicate.Predicate) *Builder {
+	blder.watchRequest = append(blder.watchRequest, watchRequest{src: src, eventhandler: eventhandler, predicates: prct})
 	return blder
 }
 
@@ -102,9 +116,10 @@ func (blder *Builder) WithConfig(config *rest.Config) *Builder {
 
 // WithEventFilter sets the event filters, to filter which create/update/delete/generic events eventually
 // trigger reconciliations.  For example, filtering on whether the resource version has changed.
+// Given predicate is added for all watched objects.
 // Defaults to the empty list.
 func (blder *Builder) WithEventFilter(p predicate.Predicate) *Builder {
-	blder.predicates = append(blder.predicates, p)
+	blder.globalPredicates = append(blder.globalPredicates, p)
 	return blder
 }
 
@@ -157,28 +172,33 @@ func (blder *Builder) Build(r reconcile.Reconciler) (controller.Controller, erro
 
 func (blder *Builder) doWatch() error {
 	// Reconcile type
-	src := &source.Kind{Type: blder.apiType}
+	src := &source.Kind{Type: blder.apiType.item}
 	hdler := &handler.EnqueueRequestForObject{}
-	err := blder.ctrl.Watch(src, hdler, blder.predicates...)
+	allPredicates := append(blder.globalPredicates, blder.apiType.predicates...)
+	err := blder.ctrl.Watch(src, hdler, allPredicates...)
 	if err != nil {
 		return err
 	}
 
 	// Watches the managed types
 	for _, obj := range blder.managedObjects {
-		src := &source.Kind{Type: obj}
+		src := &source.Kind{Type: obj.item}
 		hdler := &handler.EnqueueRequestForOwner{
-			OwnerType:    blder.apiType,
+			OwnerType:    blder.apiType.item,
 			IsController: true,
 		}
-		if err := blder.ctrl.Watch(src, hdler, blder.predicates...); err != nil {
+		allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
+		allPredicates = append(allPredicates, obj.predicates...)
+		if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {
 			return err
 		}
 	}
 
 	// Do the watch requests
 	for _, w := range blder.watchRequest {
-		if err := blder.ctrl.Watch(w.src, w.eventhandler, blder.predicates...); err != nil {
+		allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
+		allPredicates = append(allPredicates, w.predicates...)
+		if err := blder.ctrl.Watch(w.src, w.eventhandler, allPredicates...); err != nil {
 			return err
 		}
 
@@ -196,7 +216,7 @@ func (blder *Builder) getControllerName() (string, error) {
 	if blder.name != "" {
 		return blder.name, nil
 	}
-	gvk, err := getGvk(blder.apiType, blder.mgr.GetScheme())
+	gvk, err := getGvk(blder.apiType.item, blder.mgr.GetScheme())
 	if err != nil {
 		return "", err
 	}

pkg/builder/controller_test.go

@@ -31,8 +31,10 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/source"
@@ -180,6 +182,67 @@ var _ = Describe("application", func() {
 			close(done)
 		}, 10)
 	})
+
+	Describe("Set custom predicates", func() {
+		It("should execute registered predicates only for assigned kind", func(done Done) {
+			m, err := manager.New(cfg, manager.Options{})
+			Expect(err).NotTo(HaveOccurred())
+
+			var (
+				deployPrctExecuted     = false
+				replicaSetPrctExecuted = false
+				allPrctExecuted        = 0
+			)
+
+			deployPrct := predicate.Funcs{
+				CreateFunc: func(e event.CreateEvent) bool {
+					defer GinkgoRecover()
+					// check that it was called only for deployment
+					Expect(e.Meta).To(BeAssignableToTypeOf(&appsv1.Deployment{}))
+					deployPrctExecuted = true
+					return true
+				},
+			}
+
+			replicaSetPrct := predicate.Funcs{
+				CreateFunc: func(e event.CreateEvent) bool {
+					defer GinkgoRecover()
+					// check that it was called only for replicaset
+					Expect(e.Meta).To(BeAssignableToTypeOf(&appsv1.ReplicaSet{}))
+					replicaSetPrctExecuted = true
+					return true
+				},
+			}
+
+			allPrct := predicate.Funcs{
+				CreateFunc: func(e event.CreateEvent) bool {
+					defer GinkgoRecover()
+					//check that it was called for all registered kinds
+					Expect(e.Meta).Should(Or(
+						BeAssignableToTypeOf(&appsv1.Deployment{}),
+						BeAssignableToTypeOf(&appsv1.ReplicaSet{}),
+					))
+
+					allPrctExecuted++
+					return true
+				},
+			}
+
+			bldr := ControllerManagedBy(m).
+				For(&appsv1.Deployment{}, deployPrct).
+				Owns(&appsv1.ReplicaSet{}, replicaSetPrct).
+				WithEventFilter(allPrct)
+
+			doReconcileTest("5", stop, bldr, m, true)
+
+			Expect(deployPrctExecuted).To(BeTrue(), "Deploy predicated should be called at least once")
+			Expect(replicaSetPrctExecuted).To(BeTrue(), "ReplicaSet predicated should be called at least once")
+			Expect(allPrctExecuted).To(BeNumerically(">=", 2), "Global Predicated should be called at least twice")
+
+			close(done)
+		})
+	})
+
 })
 
 func doReconcileTest(nameSuffix string, stop chan struct{}, blder *Builder, mgr manager.Manager, complete bool) {