Add builder handlers for generic objects

Signed-off-by: Danil Grigorev <[email protected]>

Author: Danil-Grigorev

alias.go

@@ -20,6 +20,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	cfg "sigs.k8s.io/controller-runtime/pkg/config"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -33,6 +34,9 @@ import (
 // Builder builds an Application ControllerManagedBy (e.g. Operator) and returns a manager.Manager to start it.
 type Builder = builder.Builder
 
+// WatchObject defines an interface on the watch object with inherited type info.
+type WatchObject = builder.WatchObject
+
 // Request contains the information necessary to reconcile a Kubernetes object.  This includes the
 // information to uniquely identify the object - its Name and Namespace.  It does NOT contain information about
 // any specific Event or the object contents itself.
@@ -157,3 +161,7 @@ var (
 	// SetLogger sets a concrete logging implementation for all deferred Loggers.
 	SetLogger = log.SetLogger
 )
+
+func Object[T client.Object](obj T) WatchObject {
+	return builder.Object(obj)
+}

example_test.go

@@ -69,6 +69,37 @@ func Example() {
 	}
 }
 
+// This example creates a generic application Controller that is configured for ReplicaSets and Pods.
+//
+// * Create a new application for ReplicaSets that manages Pods owned by the ReplicaSet and calls into
+// ReplicaSetReconciler.
+//
+// * Start the application.
+func GenericExample() {
+	log := ctrl.Log.WithName("builder-examples")
+
+	manager, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{})
+	if err != nil {
+		log.Error(err, "could not create manager")
+		os.Exit(1)
+	}
+
+	err = ctrl.
+		NewControllerManagedBy(manager).         // Create the Controller
+		With(ctrl.Object(&appsv1.ReplicaSet{})). // ReplicaSet is the Application API
+		Own(ctrl.Object(&corev1.Pod{})).         // ReplicaSet owns Pods created by it
+		Complete(&ReplicaSetReconciler{Client: manager.GetClient()})
+	if err != nil {
+		log.Error(err, "could not create controller")
+		os.Exit(1)
+	}
+
+	if err := manager.Start(ctrl.SetupSignalHandler()); err != nil {
+		log.Error(err, "could not start manager")
+		os.Exit(1)
+	}
+}
+
 type ExampleCRDWithConfigMapRef struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`

pkg/builder/controller.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/klog/v2"
 
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -52,14 +53,58 @@ const (
 	projectAsMetadata
 )
 
+// WatchObject is an interface on an object wrapper with preserved type information
+type WatchObject interface {
+	GetObject() client.Object
+	SetSource(cache.Cache) source.SyncingSource
+	// SetPredicates()
+}
+
+type watchObject[T client.Object] struct {
+	object T
+	source source.SyncingSource
+}
+
+// // SetPredicates implements WatchObject.
+// func (w watchObject[T]) SetPredicates() {
+// 	w.source.Start()
+// }
+
+// SetSource returns inner client.Object
+func (w watchObject[T]) GetObject() client.Object {
+	return w.object
+}
+
+// SetSource sets and returns the source.SyncingSource on the object
+func (w watchObject[T]) SetSource(cache cache.Cache) source.SyncingSource {
+	if w.source == nil {
+		w.source = source.Object(cache, w.object)
+	}
+
+	return w.source
+}
+
+// Object constructs a wrapper on a generic object with stored type information
+func Object[T client.Object](obj T) WatchObject {
+	return watchObject[T]{object: obj}
+}
+
+type State struct {
+	options []Option
+	mgr     manager.Manager
+}
+
 // Builder builds a Controller.
 type Builder struct {
 	forInput         ForInput
 	ownsInput        []OwnsInput
 	watchesInput     []WatchesInput
+	blocks           []func(State)
+	options          []Option
 	mgr              manager.Manager
 	globalPredicates []predicate.Predicate
 	ctrl             controller.Controller
+	ctrls            []controller.ControllerConstraint
 	ctrlOptions      controller.Options
 	name             string
 }
@@ -71,7 +116,7 @@ func ControllerManagedBy(m manager.Manager) *Builder {
 
 // ForInput represents the information set by the For method.
 type ForInput struct {
-	object           client.Object
+	object           WatchObject
 	predicates       []predicate.Predicate
 	objectProjection objectProjection
 	err              error
@@ -81,12 +126,22 @@ type ForInput struct {
 // update events by *reconciling the object*.
 // This is the equivalent of calling
 // Watches(&source.Kind{Type: apiType}, &handler.EnqueueRequestForObject{}).
-func (blder *Builder) For(object client.Object, opts ...ForOption) *Builder {
+func (blder *Builder) For(object client.Object, opts ...Option) *Builder {
+	return blder.With(Object(object), opts...)
+}
+
+// With 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{}).
+//
+// Unlike For, With ensures the type of client.Object returned by predicates or event handlers.
+func (blder *Builder) With(watch WatchObject, opts ...Option) *Builder {
 	if blder.forInput.object != nil {
 		blder.forInput.err = fmt.Errorf("For(...) should only be called once, could not assign multiple objects for reconciliation")
 		return blder
 	}
-	input := ForInput{object: object}
+	input := ForInput{object: watch}
 	for _, opt := range opts {
 		opt.ApplyToFor(&input)
 	}
@@ -97,10 +152,11 @@ func (blder *Builder) For(object client.Object, opts ...ForOption) *Builder {
 
 // OwnsInput represents the information set by Owns method.
 type OwnsInput struct {
-	matchEveryOwner  bool
-	object           client.Object
-	predicates       []predicate.Predicate
-	objectProjection objectProjection
+	matchEveryOwner   bool
+	object            WatchObject
+	predicates        []predicate.Predicate
+	genericPredicates []predicate.PredicateConstraint
+	objectProjection  objectProjection
 }
 
 // Owns defines types of Objects being *generated* by the ControllerManagedBy, and configures the ControllerManagedBy to respond to
@@ -111,7 +167,21 @@ type OwnsInput struct {
 //
 // By default, this is the equivalent of calling
 // Watches(object, handler.EnqueueRequestForOwner([...], ownerType, OnlyControllerOwner())).
-func (blder *Builder) Owns(object client.Object, opts ...OwnsOption) *Builder {
+func (blder *Builder) Owns(object client.Object, opts ...Option) *Builder {
+	return blder.Own(Object(object), opts...)
+}
+
+// Own 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*.
+//
+// The default behavior reconciles only the first controller-type OwnerReference of the given type.
+// Use Owns(object, builder.MatchEveryOwner) to reconcile all owners.
+//
+// By default, this is the equivalent of calling
+// Watches(object, handler.EnqueueRequestForOwner([...], ownerType, OnlyControllerOwner())).
+//
+// Unlike Owns, Own ensures the type of client.Object returned by predicates or event handlers.
+func (blder *Builder) Own(object WatchObject, opts ...Option) *Builder {
 	input := OwnsInput{object: object}
 	for _, opt := range opts {
 		opt.ApplyToOwns(&input)
@@ -134,11 +204,72 @@ type WatchesInput struct {
 //
 // This is the equivalent of calling
 // WatchesRawSource(source.Kind(cache, object), eventHandler, opts...).
-func (blder *Builder) Watches(object client.Object, eventHandler handler.EventHandler, opts ...WatchesOption) *Builder {
-	src := source.ObjectKind(blder.mgr.GetCache(), object)
+func (blder *Builder) Watches(object client.Object, eventHandler handler.EventHandler, opts ...Option) *Builder {
+	return blder.Watch(Object(object), eventHandler, opts...)
+}
+
+// Watch defines the type of Object to watch, and configures the ControllerManagedBy to respond to create / delete /
+// update events by *reconciling the object* with the given EventHandler.
+//
+// This is the equivalent of calling
+// WatchesRawSource(source.Kind(cache, object), eventHandler, opts...).
+// Unlike Watches, Watch ensures the type of client.Object returned by predicates or event handlers.
+func (blder *Builder) Watch(object WatchObject, eventHandler handler.EventHandler, opts ...Option) *Builder {
+	src := object.SetSource(blder.mgr.GetCache())
 	return blder.WatchesRawSource(src, eventHandler, opts...)
 }
 
+func (blder *Builder) Add(w AddWatch) *Builder {
+	w.AddTo(blder)
+	return blder
+}
+
+func (blder *Builder) AddBlock(f func(State)) {
+	blder.blocks = append(blder.blocks, f)
+}
+
+type AddWatch interface {
+	SetSource(*Builder)
+	SetEventHandler(*Builder)
+	GetOptions() []Option
+	AddTo(*Builder)
+}
+
+type Adder interface {
+	AddTo(*Builder) *Builder
+}
+
+var _ AddWatch = &RawAdder[any]{}
+
+type RawAdder[T any] struct {
+	*Raw[T]
+}
+
+func (r *RawAdder[T]) AddTo(b *Builder) {
+	b.options = append(b.options, r.GetOptions()...) // state
+	// Establishing watches, event handlers
+	r.SetSource(b)
+	r.SetEventHandler(b)
+}
+
+func (r *RawAdder[T]) SetSource(b *Builder) {}
+
+func (r *RawAdder[T]) SetEventHandler(b *Builder) {
+	b.AddBlock(func(s State) {
+		allPredicates := append([]predicate.Predicate(nil), s.options)
+		allPredicates = append(allPredicates, own.predicates...)
+		if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {
+			return err
+		}
+	})
+}
+
+func (blder *RawAdder[T]) GetOptions() []Option {
+	return []Option{}
+}
+
+var _ controller.ControllerConstraint
+
 // WatchesMetadata is the same as Watches, but forces the internal cache to only watch PartialObjectMetadata.
 //
 // This is useful when watching lots of objects, really big objects, or objects for which you only know
@@ -166,11 +297,18 @@ func (blder *Builder) Watches(object client.Object, eventHandler handler.EventHa
 // In the first case, controller-runtime will create another cache for the
 // concrete type on top of the metadata cache; this increases memory
 // consumption and leads to race conditions as caches are not in sync.
-func (blder *Builder) WatchesMetadata(object client.Object, eventHandler handler.EventHandler, opts ...WatchesOption) *Builder {
+func (blder *Builder) WatchesMetadata(object client.Object, eventHandler handler.EventHandler, opts ...Option) *Builder {
 	opts = append(opts, OnlyMetadata)
+	_ = RawAdder[any]{}
 	return blder.Watches(object, eventHandler, opts...)
 }
 
+type Raw[T any] struct {
+	src          source.ObjectSource[T]
+	eventHandler handler.ObjectHandler[T]
+	predicates   []predicate.ObjectPredicate[T]
+}
+
 // WatchesRawSource exposes the lower-level ControllerManagedBy Watches functions through the builder.
 // Specified predicates are registered only for given source.
 //
@@ -179,7 +317,7 @@ func (blder *Builder) WatchesMetadata(object client.Object, eventHandler handler
 //
 // Example:
 // WatchesRawSource(source.Kind(cache, &corev1.Pod{}), eventHandler, opts...) // ensure that source propagates only valid Pod objects.
-func (blder *Builder) WatchesRawSource(src source.Source, eventHandler handler.EventHandler, opts ...WatchesOption) *Builder {
+func (blder *Builder) WatchesRawSource(src source.Source, eventHandler handler.EventHandler, opts ...Option) *Builder {
 	input := WatchesInput{src: src, eventHandler: eventHandler}
 	for _, opt := range opts {
 		opt.ApplyToWatches(&input)
@@ -271,14 +409,22 @@ func (blder *Builder) project(obj client.Object, proj objectProjection) (client.
 func (blder *Builder) doWatch() error {
 	// Reconcile type
 	if blder.forInput.object != nil {
-		obj, err := blder.project(blder.forInput.object, blder.forInput.objectProjection)
-		if err != nil {
-			return err
+		src := blder.forInput.object.SetSource(blder.mgr.GetCache())
+		if _, ok := blder.forInput.object.(watchObject[client.Object]); ok {
+			obj, err := blder.project(blder.forInput.object.GetObject(), blder.forInput.objectProjection)
+			if err != nil {
+				return err
+			}
+			src = source.Object(blder.mgr.GetCache(), obj)
 		}
-		src := source.ObjectKind(blder.mgr.GetCache(), obj)
 		hdler := &handler.EnqueueRequestForObject{}
 		allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
 		allPredicates = append(allPredicates, blder.forInput.predicates...)
+		for _, c := range blder.ctrls {
+			if err := c.DoWatch(); err != nil {
+				return err
+			}
+		}
 		if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {
 			return err
 		}
@@ -289,18 +435,21 @@ func (blder *Builder) doWatch() error {
 		return errors.New("Owns() can only be used together with For()")
 	}
 	for _, own := range blder.ownsInput {
-		obj, err := blder.project(own.object, own.objectProjection)
-		if err != nil {
-			return err
+		src := own.object.SetSource(blder.mgr.GetCache())
+		if _, ok := own.object.(watchObject[client.Object]); ok {
+			obj, err := blder.project(own.object.GetObject(), own.objectProjection)
+			if err != nil {
+				return err
+			}
+			src = source.Object(blder.mgr.GetCache(), obj)
 		}
-		src := source.ObjectKind(blder.mgr.GetCache(), obj)
 		opts := []handler.OwnerOption{}
 		if !own.matchEveryOwner {
 			opts = append(opts, handler.OnlyControllerOwner())
 		}
 		hdler := handler.EnqueueRequestForOwner(
 			blder.mgr.GetScheme(), blder.mgr.GetRESTMapper(),
-			blder.forInput.object,
+			blder.forInput.object.GetObject(),
 			opts...,
 		)
 		allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
@@ -359,7 +508,7 @@ func (blder *Builder) doController(r reconcile.Reconciler) error {
 	hasGVK := blder.forInput.object != nil
 	if hasGVK {
 		var err error
-		gvk, err = getGvk(blder.forInput.object, blder.mgr.GetScheme())
+		gvk, err = getGvk(blder.forInput.object.GetObject(), blder.mgr.GetScheme())
 		if err != nil {
 			return err
 		}