From 2951ef5bddbd7ec04c30d97ba08fa5fd56f52024 Mon Sep 17 00:00:00 2001 From: Tim Ramlot <42113979+inteon@users.noreply.github.com> Date: Sat, 6 Jan 2024 13:59:19 +0100 Subject: [PATCH] move arguments from Start() to Source objects Signed-off-by: Tim Ramlot <42113979+inteon@users.noreply.github.com> --- examples/builtins/main.go | 9 +- pkg/builder/controller.go | 104 +++--- pkg/builder/options.go | 24 +- pkg/controller/controller.go | 12 +- pkg/controller/controller_integration_test.go | 9 +- pkg/controller/controller_test.go | 4 +- pkg/controller/example_test.go | 6 +- pkg/handler/example_test.go | 24 +- pkg/internal/controller/controller.go | 15 +- pkg/internal/controller/controller_test.go | 70 ++-- .../recorder/recorder_integration_test.go | 2 +- pkg/internal/source/channel_broadcast.go | 196 ++++++++++ pkg/internal/source/kind.go | 28 +- pkg/predicate/predicate.go | 23 +- pkg/source/example_test.go | 9 +- pkg/source/source.go | 221 +++++------ pkg/source/source_integration_test.go | 68 ++-- pkg/source/source_test.go | 345 +++++++++--------- 18 files changed, 713 insertions(+), 456 deletions(-) create mode 100644 pkg/internal/source/channel_broadcast.go diff --git a/examples/builtins/main.go b/examples/builtins/main.go index 8ea173b248..c791b0bf10 100644 --- a/examples/builtins/main.go +++ b/examples/builtins/main.go @@ -59,14 +59,17 @@ func main() { } // Watch ReplicaSets and enqueue ReplicaSet object key - if err := c.Watch(source.Kind(mgr.GetCache(), &appsv1.ReplicaSet{}), &handler.EnqueueRequestForObject{}); err != nil { + if err := c.Watch(source.Kind(mgr.GetCache(), &appsv1.ReplicaSet{}, &handler.EnqueueRequestForObject{})); err != nil { entryLog.Error(err, "unable to watch ReplicaSets") os.Exit(1) } // Watch Pods and enqueue owning ReplicaSet key - if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}), - handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &appsv1.ReplicaSet{}, handler.OnlyControllerOwner())); err != nil { + if err := c.Watch(source.Kind( + mgr.GetCache(), + &corev1.Pod{}, + handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &appsv1.ReplicaSet{}, handler.OnlyControllerOwner()), + )); err != nil { entryLog.Error(err, "unable to watch Pods") os.Exit(1) } diff --git a/pkg/builder/controller.go b/pkg/builder/controller.go index 1a115f2f7b..2f0efaea89 100644 --- a/pkg/builder/controller.go +++ b/pkg/builder/controller.go @@ -30,7 +30,6 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client/apiutil" "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/handler" - internalsource "sigs.k8s.io/controller-runtime/pkg/internal/source" "sigs.k8s.io/controller-runtime/pkg/manager" "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" @@ -54,14 +53,15 @@ const ( // Builder builds a Controller. type Builder struct { - forInput ForInput - ownsInput []OwnsInput - watchesInput []WatchesInput - mgr manager.Manager - globalPredicates []predicate.Predicate - ctrl controller.Controller - ctrlOptions controller.Options - name string + forInput ForInput + ownsInput []OwnsInput + watchesObjectInput []WatchesObjectInput + watchesSourceInput []WatchesSourceInput + mgr manager.Manager + globalPredicates []predicate.Predicate + ctrl controller.Controller + ctrlOptions controller.Options + name string } // ControllerManagedBy returns a new controller builder that will be started by the provided Manager. @@ -80,7 +80,7 @@ type ForInput struct { // 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{}). +// Watches(source.Kind(mgr.GetCache(), apiType, &handler.EnqueueRequestForObject{})). func (blder *Builder) For(object client.Object, opts ...ForOption) *Builder { if blder.forInput.object != nil { blder.forInput.err = fmt.Errorf("For(...) should only be called once, could not assign multiple objects for reconciliation") @@ -121,10 +121,10 @@ func (blder *Builder) Owns(object client.Object, opts ...OwnsOption) *Builder { return blder } -// WatchesInput represents the information set by Watches method. -type WatchesInput struct { - src source.Source - eventHandler handler.EventHandler +// WatchesObjectInput represents the information set by Watches method. +type WatchesObjectInput struct { + object client.Object + eventhandler handler.EventHandler predicates []predicate.Predicate objectProjection objectProjection } @@ -133,10 +133,15 @@ type WatchesInput struct { // update events by *reconciling the object* with the given EventHandler. // // 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.Kind(blder.mgr.GetCache(), object) - return blder.WatchesRawSource(src, eventHandler, opts...) +// WatchesRawSource(source.Kind(scheme, object, eventhandler, opts...)). +func (blder *Builder) Watches(object client.Object, eventhandler handler.EventHandler, opts ...WatchesObjectOption) *Builder { + input := WatchesObjectInput{object: object, eventhandler: eventhandler} + for _, opt := range opts { + opt.ApplyToWatchesObject(&input) + } + + blder.watchesObjectInput = append(blder.watchesObjectInput, input) + return blder } // WatchesMetadata is the same as Watches, but forces the internal cache to only watch PartialObjectMetadata. @@ -166,29 +171,30 @@ 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 ...WatchesObjectOption) *Builder { opts = append(opts, OnlyMetadata) - return blder.Watches(object, eventHandler, opts...) + return blder.Watches(object, eventhandler, opts...) +} + +// WatchesSourceInput represents the information set by Watches method. +type WatchesSourceInput struct { + src source.Source } // WatchesRawSource exposes the lower-level ControllerManagedBy Watches functions through the builder. -// Specified predicates are registered only for given source. // // STOP! Consider using For(...), Owns(...), Watches(...), WatchesMetadata(...) instead. -// This method is only exposed for more advanced use cases, most users should use one of the higher level functions. -func (blder *Builder) WatchesRawSource(src source.Source, eventHandler handler.EventHandler, opts ...WatchesOption) *Builder { - input := WatchesInput{src: src, eventHandler: eventHandler} - for _, opt := range opts { - opt.ApplyToWatches(&input) - } - - blder.watchesInput = append(blder.watchesInput, input) +// This method is only exposed for more advanced use cases, most users should use higher level functions. +// This method does generally disregard all the global configuration set by the builder. +func (blder *Builder) WatchesRawSource(src source.Source) *Builder { + blder.watchesSourceInput = append(blder.watchesSourceInput, WatchesSourceInput{src: src}) return blder } // 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. +// The predicates are not applied to sources watched with WatchesRawSource(...). // Defaults to the empty list. func (blder *Builder) WithEventFilter(p predicate.Predicate) *Builder { blder.globalPredicates = append(blder.globalPredicates, p) @@ -272,11 +278,15 @@ func (blder *Builder) doWatch() error { if err != nil { return err } - src := source.Kind(blder.mgr.GetCache(), obj) - hdler := &handler.EnqueueRequestForObject{} allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...) allPredicates = append(allPredicates, blder.forInput.predicates...) - if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil { + src := source.Kind( + blder.mgr.GetCache(), + obj, + &handler.EnqueueRequestForObject{}, + allPredicates..., + ) + if err := blder.ctrl.Watch(src); err != nil { return err } } @@ -290,7 +300,6 @@ func (blder *Builder) doWatch() error { if err != nil { return err } - src := source.Kind(blder.mgr.GetCache(), obj) opts := []handler.OwnerOption{} if !own.matchEveryOwner { opts = append(opts, handler.OnlyControllerOwner()) @@ -302,30 +311,37 @@ func (blder *Builder) doWatch() error { ) allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...) allPredicates = append(allPredicates, own.predicates...) - if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil { + src := source.Kind(blder.mgr.GetCache(), obj, hdler, allPredicates...) + if err := blder.ctrl.Watch(src); err != nil { return err } } // Do the watch requests - if len(blder.watchesInput) == 0 && blder.forInput.object == nil { + if len(blder.watchesObjectInput) == 0 && len(blder.watchesSourceInput) == 0 && blder.forInput.object == nil { return errors.New("there are no watches configured, controller will never get triggered. Use For(), Owns() or Watches() to set them up") } - for _, w := range blder.watchesInput { - // If the source of this watch is of type Kind, project it. - if srcKind, ok := w.src.(*internalsource.Kind); ok { - typeForSrc, err := blder.project(srcKind.Type, w.objectProjection) - if err != nil { - return err - } - srcKind.Type = typeForSrc + + for _, w := range blder.watchesObjectInput { + obj, err := blder.project(w.object, w.objectProjection) + if err != nil { + return err } + allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...) allPredicates = append(allPredicates, w.predicates...) - if err := blder.ctrl.Watch(w.src, w.eventHandler, allPredicates...); err != nil { + src := source.Kind(blder.mgr.GetCache(), obj, w.eventhandler, allPredicates...) + if err := blder.ctrl.Watch(src); err != nil { return err } } + + for _, w := range blder.watchesSourceInput { + if err := blder.ctrl.Watch(w.src); err != nil { + return err + } + } + return nil } diff --git a/pkg/builder/options.go b/pkg/builder/options.go index 15f66b2a82..6824932695 100644 --- a/pkg/builder/options.go +++ b/pkg/builder/options.go @@ -34,10 +34,10 @@ type OwnsOption interface { ApplyToOwns(*OwnsInput) } -// WatchesOption is some configuration that modifies options for a watches request. -type WatchesOption interface { - // ApplyToWatches applies this configuration to the given watches options. - ApplyToWatches(*WatchesInput) +// WatchesObjectOption is some configuration that modifies options for a watches request. +type WatchesObjectOption interface { + // ApplyToWatchesObject applies this configuration to the given watches options. + ApplyToWatchesObject(*WatchesObjectInput) } // }}} @@ -66,14 +66,14 @@ func (w Predicates) ApplyToOwns(opts *OwnsInput) { opts.predicates = w.predicates } -// ApplyToWatches applies this configuration to the given WatchesInput options. -func (w Predicates) ApplyToWatches(opts *WatchesInput) { +// ApplyToObjectWatches applies this configuration to the given WatchesInput options. +func (w Predicates) ApplyToWatchesObject(opts *WatchesObjectInput) { opts.predicates = w.predicates } var _ ForOption = &Predicates{} var _ OwnsOption = &Predicates{} -var _ WatchesOption = &Predicates{} +var _ WatchesObjectOption = &Predicates{} // }}} @@ -94,8 +94,8 @@ func (p projectAs) ApplyToOwns(opts *OwnsInput) { opts.objectProjection = objectProjection(p) } -// ApplyToWatches applies this configuration to the given WatchesInput options. -func (p projectAs) ApplyToWatches(opts *WatchesInput) { +// ApplyToWatches applies this configuration to the given WatchesObjectInput options. +func (p projectAs) ApplyToWatchesObject(opts *WatchesObjectInput) { opts.objectProjection = objectProjection(p) } @@ -132,9 +132,9 @@ var ( // consumption and leads to race conditions as caches are not in sync. OnlyMetadata = projectAs(projectAsMetadata) - _ ForOption = OnlyMetadata - _ OwnsOption = OnlyMetadata - _ WatchesOption = OnlyMetadata + _ ForOption = OnlyMetadata + _ OwnsOption = OnlyMetadata + _ WatchesObjectOption = OnlyMetadata ) // }}} diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go index e48db41f94..4edf01ed85 100644 --- a/pkg/controller/controller.go +++ b/pkg/controller/controller.go @@ -25,10 +25,8 @@ import ( "k8s.io/client-go/util/workqueue" "k8s.io/klog/v2" - "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/internal/controller" "sigs.k8s.io/controller-runtime/pkg/manager" - "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/ratelimiter" "sigs.k8s.io/controller-runtime/pkg/reconcile" "sigs.k8s.io/controller-runtime/pkg/source" @@ -72,13 +70,9 @@ type Controller interface { // Reconciler is called to reconcile an object by Namespace/Name reconcile.Reconciler - // Watch takes events provided by a Source and uses the EventHandler to - // enqueue reconcile.Requests in response to the events. - // - // Watch may be provided one or more Predicates to filter events before - // they are given to the EventHandler. Events will be passed to the - // EventHandler if all provided Predicates evaluate to true. - Watch(src source.Source, eventhandler handler.EventHandler, predicates ...predicate.Predicate) error + // Watch takes events provided by a Source and enqueues reconcile.Requests + // in response to the events. + Watch(src source.Source) error // Start starts the controller. Start blocks until the context is closed or a // controller has an error starting. diff --git a/pkg/controller/controller_integration_test.go b/pkg/controller/controller_integration_test.go index 48facf1e94..73e5e807e4 100644 --- a/pkg/controller/controller_integration_test.go +++ b/pkg/controller/controller_integration_test.go @@ -64,13 +64,14 @@ var _ = Describe("controller", func() { Expect(err).NotTo(HaveOccurred()) By("Watching Resources") - err = instance.Watch( - source.Kind(cm.GetCache(), &appsv1.ReplicaSet{}), + err = instance.Watch(source.Kind( + cm.GetCache(), + &appsv1.ReplicaSet{}, handler.EnqueueRequestForOwner(cm.GetScheme(), cm.GetRESTMapper(), &appsv1.Deployment{}), - ) + )) Expect(err).NotTo(HaveOccurred()) - err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}), &handler.EnqueueRequestForObject{}) + err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}, &handler.EnqueueRequestForObject{})) Expect(err).NotTo(HaveOccurred()) err = cm.GetClient().Get(ctx, types.NamespacedName{Name: "foo"}, &corev1.Namespace{}) diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go index e49a2c5774..a595a05f05 100644 --- a/pkg/controller/controller_test.go +++ b/pkg/controller/controller_test.go @@ -77,7 +77,7 @@ var _ = Describe("controller.Controller", func() { ctx, cancel := context.WithCancel(context.Background()) watchChan := make(chan event.GenericEvent, 1) - watch := &source.Channel{Source: watchChan} + watch := source.Channel(source.NewChannelBroadcaster(watchChan), nil, &handler.EnqueueRequestForObject{}) watchChan <- event.GenericEvent{Object: &corev1.Pod{}} reconcileStarted := make(chan struct{}) @@ -99,7 +99,7 @@ var _ = Describe("controller.Controller", func() { Expect(err).NotTo(HaveOccurred()) c, err := controller.New("new-controller", m, controller.Options{Reconciler: rec}) - Expect(c.Watch(watch, &handler.EnqueueRequestForObject{})).To(Succeed()) + Expect(c.Watch(watch)).To(Succeed()) Expect(err).NotTo(HaveOccurred()) go func() { diff --git a/pkg/controller/example_test.go b/pkg/controller/example_test.go index d4fa1aef0b..85e9b9737b 100644 --- a/pkg/controller/example_test.go +++ b/pkg/controller/example_test.go @@ -71,7 +71,7 @@ func ExampleController() { } // Watch for Pod create / update / delete events and call Reconcile - err = c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}), &handler.EnqueueRequestForObject{}) + err = c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.EnqueueRequestForObject{})) if err != nil { log.Error(err, "unable to watch pods") os.Exit(1) @@ -108,7 +108,7 @@ func ExampleController_unstructured() { Version: "v1", }) // Watch for Pod create / update / delete events and call Reconcile - err = c.Watch(source.Kind(mgr.GetCache(), u), &handler.EnqueueRequestForObject{}) + err = c.Watch(source.Kind(mgr.GetCache(), u, &handler.EnqueueRequestForObject{})) if err != nil { log.Error(err, "unable to watch pods") os.Exit(1) @@ -139,7 +139,7 @@ func ExampleNewUnmanaged() { os.Exit(1) } - if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}), &handler.EnqueueRequestForObject{}); err != nil { + if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.EnqueueRequestForObject{})); err != nil { log.Error(err, "unable to watch pods") os.Exit(1) } diff --git a/pkg/handler/example_test.go b/pkg/handler/example_test.go index 575ea05fca..cd73fb53fc 100644 --- a/pkg/handler/example_test.go +++ b/pkg/handler/example_test.go @@ -42,8 +42,7 @@ var ( func ExampleEnqueueRequestForObject() { // controller is a controller.controller err := c.Watch( - source.Kind(mgr.GetCache(), &corev1.Pod{}), - &handler.EnqueueRequestForObject{}, + source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.EnqueueRequestForObject{}), ) if err != nil { // handle it @@ -54,10 +53,11 @@ func ExampleEnqueueRequestForObject() { // owning (direct) Deployment responsible for the creation of the ReplicaSet. func ExampleEnqueueRequestForOwner() { // controller is a controller.controller - err := c.Watch( - source.Kind(mgr.GetCache(), &appsv1.ReplicaSet{}), + err := c.Watch(source.Kind( + mgr.GetCache(), + &appsv1.ReplicaSet{}, handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &appsv1.Deployment{}, handler.OnlyControllerOwner()), - ) + )) if err != nil { // handle it } @@ -67,8 +67,9 @@ func ExampleEnqueueRequestForOwner() { // objects (of Type: MyKind) using a mapping function defined by the user. func ExampleEnqueueRequestsFromMapFunc() { // controller is a controller.controller - err := c.Watch( - source.Kind(mgr.GetCache(), &appsv1.Deployment{}), + err := c.Watch(source.Kind( + mgr.GetCache(), + &appsv1.Deployment{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, a client.Object) []reconcile.Request { return []reconcile.Request{ {NamespacedName: types.NamespacedName{ @@ -81,7 +82,7 @@ func ExampleEnqueueRequestsFromMapFunc() { }}, } }), - ) + )) if err != nil { // handle it } @@ -90,8 +91,9 @@ func ExampleEnqueueRequestsFromMapFunc() { // This example implements handler.EnqueueRequestForObject. func ExampleFuncs() { // controller is a controller.controller - err := c.Watch( - source.Kind(mgr.GetCache(), &corev1.Pod{}), + err := c.Watch(source.Kind( + mgr.GetCache(), + &corev1.Pod{}, handler.Funcs{ CreateFunc: func(ctx context.Context, e event.CreateEvent, q workqueue.RateLimitingInterface) { q.Add(reconcile.Request{NamespacedName: types.NamespacedName{ @@ -118,7 +120,7 @@ func ExampleFuncs() { }}) }, }, - ) + )) if err != nil { // handle it } diff --git a/pkg/internal/controller/controller.go b/pkg/internal/controller/controller.go index 33883647b9..270951dbad 100644 --- a/pkg/internal/controller/controller.go +++ b/pkg/internal/controller/controller.go @@ -28,11 +28,8 @@ import ( utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/uuid" "k8s.io/client-go/util/workqueue" - - "sigs.k8s.io/controller-runtime/pkg/handler" ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics" logf "sigs.k8s.io/controller-runtime/pkg/log" - "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" "sigs.k8s.io/controller-runtime/pkg/source" ) @@ -94,9 +91,7 @@ type Controller struct { // watchDescription contains all the information necessary to start a watch. type watchDescription struct { - src source.Source - handler handler.EventHandler - predicates []predicate.Predicate + src source.Source } // Reconcile implements reconcile.Reconciler. @@ -120,7 +115,7 @@ func (c *Controller) Reconcile(ctx context.Context, req reconcile.Request) (_ re } // Watch implements controller.Controller. -func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prct ...predicate.Predicate) error { +func (c *Controller) Watch(src source.Source) error { c.mu.Lock() defer c.mu.Unlock() @@ -128,12 +123,12 @@ func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prc // // These watches are going to be held on the controller struct until the manager or user calls Start(...). if !c.Started { - c.startWatches = append(c.startWatches, watchDescription{src: src, handler: evthdler, predicates: prct}) + c.startWatches = append(c.startWatches, watchDescription{src: src}) return nil } c.LogConstructor(nil).Info("Starting EventSource", "source", src) - return src.Start(c.ctx, evthdler, c.Queue, prct...) + return src.Start(c.ctx, c.Queue) } // NeedLeaderElection implements the manager.LeaderElectionRunnable interface. @@ -177,7 +172,7 @@ func (c *Controller) Start(ctx context.Context) error { for _, watch := range c.startWatches { c.LogConstructor(nil).Info("Starting EventSource", "source", fmt.Sprintf("%s", watch.src)) - if err := watch.src.Start(ctx, watch.handler, c.Queue, watch.predicates...); err != nil { + if err := watch.src.Start(ctx, c.Queue); err != nil { return err } } diff --git a/pkg/internal/controller/controller_test.go b/pkg/internal/controller/controller_test.go index ce2245e60f..4610efc3ee 100644 --- a/pkg/internal/controller/controller_test.go +++ b/pkg/internal/controller/controller_test.go @@ -42,7 +42,6 @@ import ( "sigs.k8s.io/controller-runtime/pkg/handler" ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics" "sigs.k8s.io/controller-runtime/pkg/internal/log" - "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" "sigs.k8s.io/controller-runtime/pkg/source" ) @@ -127,7 +126,7 @@ var _ = Describe("controller", func() { It("should return an error if there is an error waiting for the informers", func() { f := false ctrl.startWatches = []watchDescription{{ - src: source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{}), + src: source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{}, &handler.EnqueueRequestForObject{}), }} ctrl.Name = "foo" ctx, cancel := context.WithCancel(context.Background()) @@ -145,7 +144,7 @@ var _ = Describe("controller", func() { c = &cacheWithIndefinitelyBlockingGetInformer{c} ctrl.startWatches = []watchDescription{{ - src: source.Kind(c, &appsv1.Deployment{}), + src: source.Kind(c, &appsv1.Deployment{}, &handler.EnqueueRequestForObject{}), }} ctrl.Name = "testcontroller" @@ -163,7 +162,7 @@ var _ = Describe("controller", func() { c = &cacheWithIndefinitelyBlockingGetInformer{c} ctrl.startWatches = []watchDescription{{ src: &singnallingSourceWrapper{ - SyncingSource: source.Kind(c, &appsv1.Deployment{}), + SyncingSource: source.Kind(c, &appsv1.Deployment{}, &handler.EnqueueRequestForObject{}), cacheSyncDone: sourceSynced, }, }} @@ -191,7 +190,7 @@ var _ = Describe("controller", func() { Expect(err).NotTo(HaveOccurred()) ctrl.startWatches = []watchDescription{{ src: &singnallingSourceWrapper{ - SyncingSource: source.Kind(c, &appsv1.Deployment{}), + SyncingSource: source.Kind(c, &appsv1.Deployment{}, &handler.EnqueueRequestForObject{}), cacheSyncDone: sourceSynced, }, }} @@ -226,20 +225,22 @@ var _ = Describe("controller", func() { Object: p, } - ins := &source.Channel{Source: ch} - ins.DestBufferSize = 1 + ins := source.Channel( + source.NewChannelBroadcaster(ch), + nil, + handler.Funcs{ + GenericFunc: func(ctx context.Context, evt event.GenericEvent, q workqueue.RateLimitingInterface) { + defer GinkgoRecover() + close(processed) + }, + }, + ) // send the event to the channel ch <- evt ctrl.startWatches = []watchDescription{{ src: ins, - handler: handler.Funcs{ - GenericFunc: func(ctx context.Context, evt event.GenericEvent, q workqueue.RateLimitingInterface) { - defer GinkgoRecover() - close(processed) - }, - }, }} go func() { @@ -249,35 +250,52 @@ var _ = Describe("controller", func() { <-processed }) - It("should error when channel source is not specified", func() { + It("should error when ChannelBroadcaster is not specified", func() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ins := source.Channel( + nil, + nil, + &handler.EnqueueRequestForObject{}, + ) + ctrl.startWatches = []watchDescription{{ + src: ins, + }} + + e := ctrl.Start(ctx) + Expect(e).To(HaveOccurred()) + Expect(e.Error()).To(ContainSubstring("must create Channel with a non-nil Broadcaster")) + }) + + It("should error when channel eventhandler is not specified", func() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - ins := &source.Channel{} + ins := source.Channel( + source.NewChannelBroadcaster(make(<-chan event.GenericEvent)), + nil, + nil, + ) ctrl.startWatches = []watchDescription{{ src: ins, }} e := ctrl.Start(ctx) Expect(e).To(HaveOccurred()) - Expect(e.Error()).To(ContainSubstring("must specify Channel.Source")) + Expect(e.Error()).To(ContainSubstring("must create Channel with a non-nil EventHandler")) }) It("should call Start on sources with the appropriate EventHandler, Queue, and Predicates", func() { - pr1 := &predicate.Funcs{} - pr2 := &predicate.Funcs{} - evthdl := &handler.EnqueueRequestForObject{} started := false - src := source.Func(func(ctx context.Context, e handler.EventHandler, q workqueue.RateLimitingInterface, p ...predicate.Predicate) error { + src := source.Func(func(ctx context.Context, q workqueue.RateLimitingInterface) error { defer GinkgoRecover() - Expect(e).To(Equal(evthdl)) Expect(q).To(Equal(ctrl.Queue)) - Expect(p).To(ConsistOf(pr1, pr2)) started = true return nil }) - Expect(ctrl.Watch(src, evthdl, pr1, pr2)).NotTo(HaveOccurred()) + Expect(ctrl.Watch(src)).NotTo(HaveOccurred()) // Use a cancelled context so Start doesn't block ctx, cancel := context.WithCancel(context.Background()) @@ -288,13 +306,11 @@ var _ = Describe("controller", func() { It("should return an error if there is an error starting sources", func() { err := fmt.Errorf("Expected Error: could not start source") - src := source.Func(func(context.Context, handler.EventHandler, - workqueue.RateLimitingInterface, - ...predicate.Predicate) error { + src := source.Func(func(context.Context, workqueue.RateLimitingInterface) error { defer GinkgoRecover() return err }) - Expect(ctrl.Watch(src, &handler.EnqueueRequestForObject{})).To(Succeed()) + Expect(ctrl.Watch(src)).To(Succeed()) ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/pkg/internal/recorder/recorder_integration_test.go b/pkg/internal/recorder/recorder_integration_test.go index 130a306053..f9a20737fc 100644 --- a/pkg/internal/recorder/recorder_integration_test.go +++ b/pkg/internal/recorder/recorder_integration_test.go @@ -56,7 +56,7 @@ var _ = Describe("recorder", func() { Expect(err).NotTo(HaveOccurred()) By("Watching Resources") - err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}), &handler.EnqueueRequestForObject{}) + err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}, &handler.EnqueueRequestForObject{})) Expect(err).NotTo(HaveOccurred()) By("Starting the Manager") diff --git a/pkg/internal/source/channel_broadcast.go b/pkg/internal/source/channel_broadcast.go new file mode 100644 index 0000000000..cb7c9431c6 --- /dev/null +++ b/pkg/internal/source/channel_broadcast.go @@ -0,0 +1,196 @@ +/* +Copyright 2023 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package internal + +import ( + "sync" + + "sigs.k8s.io/controller-runtime/pkg/event" +) + +// ChannelBroadcaster is a wrapper around a channel that allows multiple listeners to all +// receive the events from the channel. +type ChannelBroadcaster struct { + source <-chan event.GenericEvent + + mu sync.Mutex + rcCount uint + managementCh chan managementMsg + doneCh chan struct{} +} + +type managementOperation bool + +const ( + addChannel managementOperation = true + removeChannel managementOperation = false +) + +type managementMsg struct { + operation managementOperation + ch chan event.GenericEvent +} + +// NewChannelBroadcaster creates a new ChannelBroadcaster for the given channel. +func NewChannelBroadcaster(source <-chan event.GenericEvent) *ChannelBroadcaster { + return &ChannelBroadcaster{ + source: source, + } +} + +// AddListener adds a new listener to the ChannelBroadcaster. Each listener +// will receive all events from the source channel. All listeners have to be +// removed using RemoveListener before the ChannelBroadcaster can be garbage +// collected. +func (sc *ChannelBroadcaster) AddListener(ch chan event.GenericEvent) { + var managementCh chan managementMsg + var doneCh chan struct{} + isFirst := false + func() { + sc.mu.Lock() + defer sc.mu.Unlock() + + isFirst = sc.rcCount == 0 + sc.rcCount++ + + if isFirst { + sc.managementCh = make(chan managementMsg) + sc.doneCh = make(chan struct{}) + } + + managementCh = sc.managementCh + doneCh = sc.doneCh + }() + + if isFirst { + go startLoop(sc.source, managementCh, doneCh) + } + + // If the goroutine is not yet stopped, send a message to add the + // destination channel. The routine might be stopped already because + // the source channel was closed. + select { + case <-doneCh: + default: + managementCh <- managementMsg{ + operation: addChannel, + ch: ch, + } + } +} + +func startLoop( + source <-chan event.GenericEvent, + managementCh chan managementMsg, + doneCh chan struct{}, +) { + defer close(doneCh) + + var destinations []chan event.GenericEvent + + // Close all remaining destinations in case the Source channel is closed. + defer func() { + for _, dst := range destinations { + close(dst) + } + }() + + // Wait for the first destination to be added before starting the loop. + for len(destinations) == 0 { + managementMsg := <-managementCh + if managementMsg.operation == addChannel { + destinations = append(destinations, managementMsg.ch) + } + } + + for { + select { + case msg := <-managementCh: + + switch msg.operation { + case addChannel: + destinations = append(destinations, msg.ch) + case removeChannel: + SearchLoop: + for i, dst := range destinations { + if dst == msg.ch { + destinations = append(destinations[:i], destinations[i+1:]...) + close(dst) + break SearchLoop + } + } + + if len(destinations) == 0 { + return + } + } + + case evt, stillOpen := <-source: + if !stillOpen { + return + } + + for _, dst := range destinations { + // We cannot make it under goroutine here, or we'll meet the + // race condition of writing message to closed channels. + // To avoid blocking, the dest channels are expected to be of + // proper buffer size. If we still see it blocked, then + // the controller is thought to be in an abnormal state. + dst <- evt + } + } + } +} + +// RemoveListener removes a listener from the ChannelBroadcaster. The listener +// will no longer receive events from the source channel. If this is the last +// listener, this function will block until the ChannelBroadcaster's is stopped. +func (sc *ChannelBroadcaster) RemoveListener(ch chan event.GenericEvent) { + var managementCh chan managementMsg + var doneCh chan struct{} + isLast := false + func() { + sc.mu.Lock() + defer sc.mu.Unlock() + + sc.rcCount-- + isLast = sc.rcCount == 0 + + managementCh = sc.managementCh + doneCh = sc.doneCh + }() + + // If the goroutine is not yet stopped, send a message to remove the + // destination channel. The routine might be stopped already because + // the source channel was closed. + select { + case <-doneCh: + default: + managementCh <- managementMsg{ + operation: removeChannel, + ch: ch, + } + } + + // Wait for the doneCh to be closed (in case we are the last one) + if isLast { + <-doneCh + } + + // Wait for the destination channel to be closed. + <-ch +} diff --git a/pkg/internal/source/kind.go b/pkg/internal/source/kind.go index b3a8227125..f3ce9da10d 100644 --- a/pkg/internal/source/kind.go +++ b/pkg/internal/source/kind.go @@ -18,11 +18,17 @@ import ( // Kind is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create). type Kind struct { + // Cache used to watch APIs + Cache cache.Cache + // Type is the type of object to watch. e.g. &v1.Pod{} Type client.Object - // Cache used to watch APIs - Cache cache.Cache + // EventHandler is the handler to call when events are received. + EventHandler handler.EventHandler + + // Predicates are the predicates to evaluate whether to handle the event. + Predicates []predicate.Predicate // started may contain an error if one was encountered during startup. If its closed and does not // contain an error, startup and syncing finished. @@ -32,13 +38,19 @@ type Kind struct { // Start is internal and should be called only by the Controller to register an EventHandler with the Informer // to enqueue reconcile.Requests. -func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue workqueue.RateLimitingInterface, - prct ...predicate.Predicate) error { +func (ks *Kind) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error { + if ks.Cache == nil { + return fmt.Errorf("must create Kind with a non-nil Cache") + } if ks.Type == nil { - return fmt.Errorf("must create Kind with a non-nil object") + return fmt.Errorf("must create Kind with a non-nil Type") } - if ks.Cache == nil { - return fmt.Errorf("must create Kind with a non-nil cache") + if ks.EventHandler == nil { + return fmt.Errorf("must create Kind with a non-nil EventHandler") + } + + if ks.started != nil { + return fmt.Errorf("cannot start an already started Kind source") } // cache.GetInformer will block until its context is cancelled if the cache was already started and it can not @@ -79,7 +91,7 @@ func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue w return } - _, err := i.AddEventHandler(NewEventHandler(ctx, queue, handler, prct).HandlerFuncs()) + _, err := i.AddEventHandler(NewEventHandler(ctx, queue, ks.EventHandler, ks.Predicates).HandlerFuncs()) if err != nil { ks.started <- err return diff --git a/pkg/predicate/predicate.go b/pkg/predicate/predicate.go index 314635875e..b50c0c529e 100644 --- a/pkg/predicate/predicate.go +++ b/pkg/predicate/predicate.go @@ -177,9 +177,13 @@ func (GenerationChangedPredicate) Update(e event.UpdateEvent) bool { // It is intended to be used in conjunction with the GenerationChangedPredicate, as in the following example: // // Controller.Watch( -// &source.Kind{Type: v1.MyCustomKind}, -// &handler.EnqueueRequestForObject{}, -// predicate.Or(predicate.GenerationChangedPredicate{}, predicate.AnnotationChangedPredicate{})) +// source.Kind( +// mgr.GetCache(), +// v1.MyCustomKind, +// &handler.EnqueueRequestForObject{}, +// predicate.Or(predicate.GenerationChangedPredicate{}, predicate.AnnotationChangedPredicate{})) +// ), +// ) // // This is mostly useful for controllers that needs to trigger both when the resource's generation is incremented // (i.e., when the resource' .spec changes), or an annotation changes (e.g., for a staging/alpha API). @@ -206,11 +210,14 @@ func (AnnotationChangedPredicate) Update(e event.UpdateEvent) bool { // This predicate will skip update events that have no change in the object's label. // It is intended to be used in conjunction with the GenerationChangedPredicate, as in the following example: // -// Controller.Watch( -// -// &source.Kind{Type: v1.MyCustomKind}, -// &handler.EnqueueRequestForObject{}, -// predicate.Or(predicate.GenerationChangedPredicate{}, predicate.LabelChangedPredicate{})) +// Controller.Watch( +// source.Kind( +// mgr.GetCache(), +// v1.MyCustomKind, +// &handler.EnqueueRequestForObject{}, +// predicate.Or(predicate.GenerationChangedPredicate{}, predicate.LabelChangedPredicate{})) +// ), +// ) // // This will be helpful when object's labels is carrying some extra specification information beyond object's spec, // and the controller will be triggered if any valid spec change (not only in spec, but also in labels) happens. diff --git a/pkg/source/example_test.go b/pkg/source/example_test.go index 77857729de..7f177f1d26 100644 --- a/pkg/source/example_test.go +++ b/pkg/source/example_test.go @@ -31,7 +31,7 @@ var ctrl controller.Controller // This example Watches for Pod Events (e.g. Create / Update / Delete) and enqueues a reconcile.Request // with the Name and Namespace of the Pod. func ExampleKind() { - err := ctrl.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}), &handler.EnqueueRequestForObject{}) + err := ctrl.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.EnqueueRequestForObject{})) if err != nil { // handle it } @@ -43,8 +43,11 @@ func ExampleChannel() { events := make(chan event.GenericEvent) err := ctrl.Watch( - &source.Channel{Source: events}, - &handler.EnqueueRequestForObject{}, + source.Channel( + source.NewChannelBroadcaster(events), + nil, + &handler.EnqueueRequestForObject{}, + ), ) if err != nil { // handle it diff --git a/pkg/source/source.go b/pkg/source/source.go index 099c8d68fa..7ad1aedccd 100644 --- a/pkg/source/source.go +++ b/pkg/source/source.go @@ -22,13 +22,14 @@ import ( "sync" "k8s.io/client-go/util/workqueue" + "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" internal "sigs.k8s.io/controller-runtime/pkg/internal/source" + "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/cache" - "sigs.k8s.io/controller-runtime/pkg/predicate" ) const ( @@ -47,7 +48,7 @@ const ( type Source interface { // Start is internal and should be called only by the Controller to register an EventHandler with the Informer // to enqueue reconcile.Requests. - Start(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error + Start(context.Context, workqueue.RateLimitingInterface) error } // SyncingSource is a source that needs syncing prior to being usable. The controller @@ -58,168 +59,180 @@ type SyncingSource interface { } // Kind creates a KindSource with the given cache provider. -func Kind(cache cache.Cache, object client.Object) SyncingSource { - return &internal.Kind{Type: object, Cache: cache} +func Kind(cache cache.Cache, object client.Object, eventhandler handler.EventHandler, prds ...predicate.Predicate) SyncingSource { + return &internal.Kind{Cache: cache, Type: object, EventHandler: eventhandler, Predicates: prds} } -var _ Source = &Channel{} - -// Channel is used to provide a source of events originating outside the cluster -// (e.g. GitHub Webhook callback). Channel requires the user to wire the external -// source (eh.g. http handler) to write GenericEvents to the underlying channel. -type Channel struct { - // once ensures the event distribution goroutine will be performed only once - once sync.Once +// NewChannelBroadcaster creates a new ChannelBroadcaster for the given channel. +// A ChannelBroadcaster is a wrapper around a channel that allows multiple listeners to all +// receive the events from the channel. +var NewChannelBroadcaster = internal.NewChannelBroadcaster - // Source is the source channel to fetch GenericEvents - Source <-chan event.GenericEvent +var _ Source = &ChannelSource{} - // dest is the destination channels of the added event handlers - dest []chan event.GenericEvent +// ChannelSource is used to provide a source of events originating outside the cluster +// (e.g. GitHub Webhook callback). ChannelSource requires the user to wire the external +// source (eh.g. http handler) to write GenericEvents to the underlying channel. +type ChannelSource struct { + // Broadcaster contains the source channel for events. + Broadcaster *internal.ChannelBroadcaster // DestBufferSize is the specified buffer size of dest channels. // Default to 1024 if not specified. DestBufferSize int - // destLock is to ensure the destination channels are safely added/removed - destLock sync.Mutex + // EventHandler is the handler to call when events are received. + EventHandler handler.EventHandler + + // Predicates are the predicates to evaluate whether to handle the event. + Predicates []predicate.Predicate + + mu sync.Mutex + // isStarted is true if the source has been started. A source can only be started once. + isStarted bool } -func (cs *Channel) String() string { +func (cs *ChannelSource) String() string { return fmt.Sprintf("channel source: %p", cs) } // Start implements Source and should only be called by the Controller. -func (cs *Channel) Start( - ctx context.Context, - handler handler.EventHandler, - queue workqueue.RateLimitingInterface, - prct ...predicate.Predicate) error { - // Source should have been specified by the user. - if cs.Source == nil { - return fmt.Errorf("must specify Channel.Source") +func (cs *ChannelSource) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error { + // Broadcaster should have been specified by the user. + if cs.Broadcaster == nil { + return fmt.Errorf("must create Channel with a non-nil Broadcaster") } - - // use default value if DestBufferSize not specified - if cs.DestBufferSize == 0 { - cs.DestBufferSize = defaultBufferSize + if cs.EventHandler == nil { + return fmt.Errorf("must create Channel with a non-nil EventHandler") } - dst := make(chan event.GenericEvent, cs.DestBufferSize) + cs.mu.Lock() + defer cs.mu.Unlock() + if cs.isStarted { + return fmt.Errorf("cannot start an already started Channel source") + } + cs.isStarted = true - cs.destLock.Lock() - cs.dest = append(cs.dest, dst) - cs.destLock.Unlock() + if cs.DestBufferSize <= 0 { + cs.DestBufferSize = defaultBufferSize + } - cs.once.Do(func() { - // Distribute GenericEvents to all EventHandler / Queue pairs Watching this source - go cs.syncLoop(ctx) - }) + // Create a destination channel for the event handler + // and add it to the list of destinations + destination := make(chan event.GenericEvent, cs.DestBufferSize) + cs.Broadcaster.AddListener(destination) go func() { - for evt := range dst { - shouldHandle := true - for _, p := range prct { - if !p.Generic(evt) { - shouldHandle = false - break - } - } - - if shouldHandle { - func() { - ctx, cancel := context.WithCancel(ctx) - defer cancel() - handler.Generic(ctx, evt, queue) - }() - } - } + // Remove the listener and wait for the broadcaster + // to stop sending events to the destination channel. + defer cs.Broadcaster.RemoveListener(destination) + + cs.processReceivedEvents( + ctx, + destination, + queue, + ) }() return nil } -func (cs *Channel) doStop() { - cs.destLock.Lock() - defer cs.destLock.Unlock() - - for _, dst := range cs.dest { - close(dst) - } -} - -func (cs *Channel) distribute(evt event.GenericEvent) { - cs.destLock.Lock() - defer cs.destLock.Unlock() - - for _, dst := range cs.dest { - // We cannot make it under goroutine here, or we'll meet the - // race condition of writing message to closed channels. - // To avoid blocking, the dest channels are expected to be of - // proper buffer size. If we still see it blocked, then - // the controller is thought to be in an abnormal state. - dst <- evt - } -} - -func (cs *Channel) syncLoop(ctx context.Context) { +func (cs *ChannelSource) processReceivedEvents( + ctx context.Context, + destination <-chan event.GenericEvent, + queue workqueue.RateLimitingInterface, +) { +eventloop: for { select { case <-ctx.Done(): - // Close destination channels - cs.doStop() return - case evt, stillOpen := <-cs.Source: + case event, stillOpen := <-destination: if !stillOpen { - // if the source channel is closed, we're never gonna get - // anything more on it, so stop & bail - cs.doStop() return } - cs.distribute(evt) + + // Check predicates against the event first + // and continue the outer loop if any of them fail. + for _, p := range cs.Predicates { + if !p.Generic(event) { + continue eventloop + } + } + + // Call the event handler with the event. + cs.EventHandler.Generic(ctx, event, queue) } } } -// Informer is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create). -type Informer struct { +// Channel creates a ChannelSource with the given buffer size. +func Channel(broadcaster *internal.ChannelBroadcaster, destBufferSize *int, eventhandler handler.EventHandler, prds ...predicate.Predicate) Source { + return &ChannelSource{Broadcaster: broadcaster, DestBufferSize: ptr.Deref(destBufferSize, defaultBufferSize), EventHandler: eventhandler, Predicates: prds} +} + +// InformerSource is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create). +type InformerSource struct { // Informer is the controller-runtime Informer Informer cache.Informer -} -var _ Source = &Informer{} + // EventHandler is the handler to call when events are received. + EventHandler handler.EventHandler + + // Predicates are the predicates to evaluate whether to handle the event. + Predicates []predicate.Predicate + + mu sync.Mutex + // isStarted is true if the source has been started. A source can only be started once. + isStarted bool +} // Start is internal and should be called only by the Controller to register an EventHandler with the Informer // to enqueue reconcile.Requests. -func (is *Informer) Start(ctx context.Context, handler handler.EventHandler, queue workqueue.RateLimitingInterface, - prct ...predicate.Predicate) error { - // Informer should have been specified by the user. +func (is *InformerSource) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error { if is.Informer == nil { - return fmt.Errorf("must specify Informer.Informer") + return fmt.Errorf("must create Informer with a non-nil Informer") + } + if is.EventHandler == nil { + return fmt.Errorf("must create Informer with a non-nil EventHandler") } - _, err := is.Informer.AddEventHandler(internal.NewEventHandler(ctx, queue, handler, prct).HandlerFuncs()) + is.mu.Lock() + defer is.mu.Unlock() + if is.isStarted { + return fmt.Errorf("cannot start an already started Informer source") + } + is.isStarted = true + + _, err := is.Informer.AddEventHandler(internal.NewEventHandler(ctx, queue, is.EventHandler, is.Predicates).HandlerFuncs()) if err != nil { return err } return nil } -func (is *Informer) String() string { +func (is *InformerSource) String() string { return fmt.Sprintf("informer source: %p", is.Informer) } -var _ Source = Func(nil) +// Informer creates an InformerSource with the given cache provider. +func Informer(informer cache.Informer, eventhandler handler.EventHandler, prds ...predicate.Predicate) Source { + return &InformerSource{Informer: informer, EventHandler: eventhandler, Predicates: prds} +} -// Func is a function that implements Source. -type Func func(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error +// FuncSource is a function that implements Source. +type FuncSource func(context.Context, workqueue.RateLimitingInterface) error // Start implements Source. -func (f Func) Start(ctx context.Context, evt handler.EventHandler, queue workqueue.RateLimitingInterface, - pr ...predicate.Predicate) error { - return f(ctx, evt, queue, pr...) +func (f FuncSource) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error { + return f(ctx, queue) } -func (f Func) String() string { +func (f FuncSource) String() string { return fmt.Sprintf("func source: %p", f) } + +// Func creates a FuncSource with the given function. +func Func(f FuncSource) Source { + return f +} diff --git a/pkg/source/source_integration_test.go b/pkg/source/source_integration_test.go index 594d3c9a9c..6d6515d8aa 100644 --- a/pkg/source/source_integration_test.go +++ b/pkg/source/source_integration_test.go @@ -37,7 +37,6 @@ import ( ) var _ = Describe("Source", func() { - var instance1, instance2 source.Source var obj client.Object var q workqueue.RateLimitingInterface var c1, c2 chan interface{} @@ -58,11 +57,6 @@ var _ = Describe("Source", func() { c2 = make(chan interface{}) }) - JustBeforeEach(func() { - instance1 = source.Kind(icache, obj) - instance2 = source.Kind(icache, obj) - }) - AfterEach(func() { err := clientset.CoreV1().Namespaces().Delete(ctx, ns, metav1.DeleteOptions{}) Expect(err).NotTo(HaveOccurred()) @@ -101,31 +95,33 @@ var _ = Describe("Source", func() { } // Create an event handler to verify the events - newHandler := func(c chan interface{}) handler.Funcs { - return handler.Funcs{ - CreateFunc: func(ctx context.Context, evt event.CreateEvent, rli workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Expect(rli).To(Equal(q)) - c <- evt - }, - UpdateFunc: func(ctx context.Context, evt event.UpdateEvent, rli workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Expect(rli).To(Equal(q)) - c <- evt - }, - DeleteFunc: func(ctx context.Context, evt event.DeleteEvent, rli workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Expect(rli).To(Equal(q)) - c <- evt + newSource := func(c chan interface{}) source.Source { + return source.Kind( + icache, + obj, + handler.Funcs{ + CreateFunc: func(ctx context.Context, evt event.CreateEvent, rli workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Expect(rli).To(Equal(q)) + c <- evt + }, + UpdateFunc: func(ctx context.Context, evt event.UpdateEvent, rli workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Expect(rli).To(Equal(q)) + c <- evt + }, + DeleteFunc: func(ctx context.Context, evt event.DeleteEvent, rli workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Expect(rli).To(Equal(q)) + c <- evt + }, }, - } + ) } - handler1 := newHandler(c1) - handler2 := newHandler(c2) // Create 2 instances - Expect(instance1.Start(ctx, handler1, q)).To(Succeed()) - Expect(instance2.Start(ctx, handler2, q)).To(Succeed()) + Expect(newSource(c1).Start(ctx, q)).To(Succeed()) + Expect(newSource(c2).Start(ctx, q)).To(Succeed()) By("Creating a Deployment and expecting the CreateEvent.") created, err = client.Create(ctx, deployment, metav1.CreateOptions{}) @@ -241,8 +237,7 @@ var _ = Describe("Source", func() { c := make(chan struct{}) q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := &source.Informer{Informer: depInformer} - err := instance.Start(ctx, handler.Funcs{ + instance := source.Informer(depInformer, handler.Funcs{ CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) { defer GinkgoRecover() var err error @@ -265,7 +260,8 @@ var _ = Describe("Source", func() { defer GinkgoRecover() Fail("Unexpected GenericEvent") }, - }, q) + }) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) _, err = clientset.AppsV1().ReplicaSets("default").Create(ctx, rs, metav1.CreateOptions{}) @@ -282,8 +278,7 @@ var _ = Describe("Source", func() { rs2.SetLabels(map[string]string{"biz": "baz"}) q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := &source.Informer{Informer: depInformer} - err = instance.Start(ctx, handler.Funcs{ + instance := source.Informer(depInformer, handler.Funcs{ CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) { }, UpdateFunc: func(ctx context.Context, evt event.UpdateEvent, q2 workqueue.RateLimitingInterface) { @@ -307,7 +302,8 @@ var _ = Describe("Source", func() { defer GinkgoRecover() Fail("Unexpected GenericEvent") }, - }, q) + }) + err = instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) _, err = clientset.AppsV1().ReplicaSets("default").Update(ctx, rs2, metav1.UpdateOptions{}) @@ -319,8 +315,7 @@ var _ = Describe("Source", func() { c := make(chan struct{}) q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := &source.Informer{Informer: depInformer} - err := instance.Start(ctx, handler.Funcs{ + instance := source.Informer(depInformer, handler.Funcs{ CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { }, UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { @@ -335,7 +330,8 @@ var _ = Describe("Source", func() { defer GinkgoRecover() Fail("Unexpected GenericEvent") }, - }, q) + }) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) err = clientset.AppsV1().ReplicaSets("default").Delete(ctx, rs.Name, metav1.DeleteOptions{}) diff --git a/pkg/source/source_test.go b/pkg/source/source_test.go index 16c365e8a2..ab617ce181 100644 --- a/pkg/source/source_test.go +++ b/pkg/source/source_test.go @@ -65,8 +65,7 @@ var _ = Describe("Source", func() { } q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := source.Kind(ic, &corev1.Pod{}) - err := instance.Start(ctx, handler.Funcs{ + instance := source.Kind(ic, &corev1.Pod{}, handler.Funcs{ CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) { defer GinkgoRecover() Expect(q2).To(Equal(q)) @@ -85,7 +84,8 @@ var _ = Describe("Source", func() { defer GinkgoRecover() Fail("Unexpected GenericEvent") }, - }, q) + }) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) Expect(instance.WaitForSync(context.Background())).NotTo(HaveOccurred()) @@ -102,8 +102,7 @@ var _ = Describe("Source", func() { ic := &informertest.FakeInformers{} q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := source.Kind(ic, &corev1.Pod{}) - err := instance.Start(ctx, handler.Funcs{ + instance := source.Kind(ic, &corev1.Pod{}, handler.Funcs{ CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) { defer GinkgoRecover() Fail("Unexpected CreateEvent") @@ -125,7 +124,8 @@ var _ = Describe("Source", func() { defer GinkgoRecover() Fail("Unexpected GenericEvent") }, - }, q) + }) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) Expect(instance.WaitForSync(context.Background())).NotTo(HaveOccurred()) @@ -147,8 +147,7 @@ var _ = Describe("Source", func() { } q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := source.Kind(ic, &corev1.Pod{}) - err := instance.Start(ctx, handler.Funcs{ + instance := source.Kind(ic, &corev1.Pod{}, handler.Funcs{ CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { defer GinkgoRecover() Fail("Unexpected DeleteEvent") @@ -167,7 +166,8 @@ var _ = Describe("Source", func() { defer GinkgoRecover() Fail("Unexpected GenericEvent") }, - }, q) + }) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) Expect(instance.WaitForSync(context.Background())).NotTo(HaveOccurred()) @@ -179,24 +179,31 @@ var _ = Describe("Source", func() { }) }) + It("should return an error from Start eventhandler was not provided", func() { + instance := source.Kind(ic, &corev1.Pod{}, nil) + err := instance.Start(ctx, nil) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("must create Kind with a non-nil EventHandler")) + }) + It("should return an error from Start cache was not provided", func() { - instance := source.Kind(nil, &corev1.Pod{}) - err := instance.Start(ctx, nil, nil) + instance := source.Kind(nil, &corev1.Pod{}, &handler.EnqueueRequestForObject{}) + err := instance.Start(ctx, nil) Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(ContainSubstring("must create Kind with a non-nil cache")) + Expect(err.Error()).To(ContainSubstring("must create Kind with a non-nil Cache")) }) It("should return an error from Start if a type was not provided", func() { - instance := source.Kind(ic, nil) - err := instance.Start(ctx, nil, nil) + instance := source.Kind(ic, nil, &handler.EnqueueRequestForObject{}) + err := instance.Start(ctx, nil) Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(ContainSubstring("must create Kind with a non-nil object")) + Expect(err.Error()).To(ContainSubstring("must create Kind with a non-nil Type")) }) It("should return an error if syncing fails", func() { f := false - instance := source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{}) - Expect(instance.Start(context.Background(), nil, nil)).NotTo(HaveOccurred()) + instance := source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{}, &handler.EnqueueRequestForObject{}) + Expect(instance.Start(context.Background(), nil)).NotTo(HaveOccurred()) err := instance.WaitForSync(context.Background()) Expect(err).To(HaveOccurred()) Expect(err.Error()).To(Equal("cache did not sync")) @@ -211,8 +218,8 @@ var _ = Describe("Source", func() { ctx, cancel := context.WithTimeout(ctx, 2*time.Second) defer cancel() - instance := source.Kind(ic, &corev1.Pod{}) - err := instance.Start(ctx, handler.Funcs{}, q) + instance := source.Kind(ic, &corev1.Pod{}, handler.Funcs{}) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) Eventually(instance.WaitForSync).WithArguments(context.Background()).Should(HaveOccurred()) }) @@ -220,8 +227,8 @@ var _ = Describe("Source", func() { It("should return an error if syncing fails", func() { f := false - instance := source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{}) - Expect(instance.Start(context.Background(), nil, nil)).NotTo(HaveOccurred()) + instance := source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{}, &handler.EnqueueRequestForObject{}) + Expect(instance.Start(context.Background(), nil)).NotTo(HaveOccurred()) err := instance.WaitForSync(context.Background()) Expect(err).To(HaveOccurred()) Expect(err.Error()).To(Equal("cache did not sync")) @@ -232,24 +239,18 @@ var _ = Describe("Source", func() { Describe("Func", func() { It("should be called from Start", func() { run := false - instance := source.Func(func( - context.Context, - handler.EventHandler, - workqueue.RateLimitingInterface, ...predicate.Predicate) error { + instance := source.Func(func(context.Context, workqueue.RateLimitingInterface) error { run = true return nil }) - Expect(instance.Start(ctx, nil, nil)).NotTo(HaveOccurred()) + Expect(instance.Start(ctx, nil)).NotTo(HaveOccurred()) Expect(run).To(BeTrue()) expected := fmt.Errorf("expected error: Func") - instance = source.Func(func( - context.Context, - handler.EventHandler, - workqueue.RateLimitingInterface, ...predicate.Predicate) error { + instance = source.Func(func(context.Context, workqueue.RateLimitingInterface) error { return expected }) - Expect(instance.Start(ctx, nil, nil)).To(Equal(expected)) + Expect(instance.Start(ctx, nil)).To(Equal(expected)) }) }) @@ -289,29 +290,34 @@ var _ = Describe("Source", func() { } q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := &source.Channel{Source: ch} - err := instance.Start(ctx, handler.Funcs{ - CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected CreateEvent") - }, - UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected UpdateEvent") - }, - DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected DeleteEvent") - }, - GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { - defer GinkgoRecover() - // The empty event should have been filtered out by the predicates, - // and will not be passed to the handler. - Expect(q2).To(BeIdenticalTo(q)) - Expect(evt.Object).To(Equal(p)) - close(c) + instance := source.Channel( + source.NewChannelBroadcaster(ch), + nil, + handler.Funcs{ + CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected CreateEvent") + }, + UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected UpdateEvent") + }, + DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected DeleteEvent") + }, + GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { + defer GinkgoRecover() + // The empty event should have been filtered out by the predicates, + // and will not be passed to the handler. + Expect(q2).To(BeIdenticalTo(q)) + Expect(evt.Object).To(Equal(p)) + close(c) + }, }, - }, q, prct) + prct, + ) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) ch <- invalidEvt @@ -327,34 +333,37 @@ var _ = Describe("Source", func() { q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") // Add a handler to get distribution blocked - instance := &source.Channel{Source: ch} - instance.DestBufferSize = 1 - err := instance.Start(ctx, handler.Funcs{ - CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected CreateEvent") - }, - UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected UpdateEvent") - }, - DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected DeleteEvent") - }, - GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { - defer GinkgoRecover() - // Block for the first time - if eventCount == 0 { - <-unblock - } - eventCount++ - - if eventCount == 3 { - close(processed) - } + instance := source.Channel( + source.NewChannelBroadcaster(ch), + nil, + handler.Funcs{ + CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected CreateEvent") + }, + UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected UpdateEvent") + }, + DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected DeleteEvent") + }, + GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { + defer GinkgoRecover() + // Block for the first time + if eventCount == 0 { + <-unblock + } + eventCount++ + + if eventCount == 3 { + close(processed) + } + }, }, - }, q) + ) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) // Write 3 events into the source channel. @@ -383,28 +392,31 @@ var _ = Describe("Source", func() { q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") // Add a handler to get distribution blocked - instance := &source.Channel{Source: ch} - instance.DestBufferSize = 1 + instance := source.Channel( + source.NewChannelBroadcaster(ch), + nil, + handler.Funcs{ + CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected CreateEvent") + }, + UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected UpdateEvent") + }, + DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected DeleteEvent") + }, + GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { + defer GinkgoRecover() - err := instance.Start(ctx, handler.Funcs{ - CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected CreateEvent") - }, - UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected UpdateEvent") - }, - DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected DeleteEvent") + close(processed) + }, }, - GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { - defer GinkgoRecover() + ) - close(processed) - }, - }, q) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) <-processed @@ -421,31 +433,35 @@ var _ = Describe("Source", func() { ch <- evt close(ch) - By("feeding that channel to a channel source") - src := &source.Channel{Source: ch} - processed := make(chan struct{}) defer close(processed) - err := src.Start(ctx, handler.Funcs{ - CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected CreateEvent") - }, - UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected UpdateEvent") - }, - DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected DeleteEvent") - }, - GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { - defer GinkgoRecover() + By("feeding that channel to a channel source") + src := source.Channel( + source.NewChannelBroadcaster(ch), + nil, + handler.Funcs{ + CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected CreateEvent") + }, + UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected UpdateEvent") + }, + DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected DeleteEvent") + }, + GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { + defer GinkgoRecover() - processed <- struct{}{} + processed <- struct{}{} + }, }, - }, q) + ) + + err := src.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) By("expecting to only get one event") @@ -454,9 +470,9 @@ var _ = Describe("Source", func() { }) It("should get error if no source specified", func() { q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := &source.Channel{ /*no source specified*/ } - err := instance.Start(ctx, handler.Funcs{}, q) - Expect(err).To(Equal(fmt.Errorf("must specify Channel.Source"))) + instance := source.Channel(nil, nil, handler.Funcs{}) + err := instance.Start(ctx, q) + Expect(err).To(Equal(fmt.Errorf("must create Channel with a non-nil Broadcaster"))) }) }) Context("for multi sources (handlers)", func() { @@ -469,61 +485,48 @@ var _ = Describe("Source", func() { Object: p, } - var resEvent1, resEvent2 event.GenericEvent - c1 := make(chan struct{}) - c2 := make(chan struct{}) - q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := &source.Channel{Source: ch} - err := instance.Start(ctx, handler.Funcs{ - CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected CreateEvent") - }, - UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected UpdateEvent") - }, - DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected DeleteEvent") - }, - GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Expect(q2).To(BeIdenticalTo(q)) - Expect(evt.Object).To(Equal(p)) - resEvent1 = evt - close(c1) - }, - }, q) + channelSource := source.NewChannelBroadcaster(ch) + + createInstance := func(c chan event.GenericEvent) source.Source { + return source.Channel( + channelSource, + nil, + handler.Funcs{ + CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected CreateEvent") + }, + UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected UpdateEvent") + }, + DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Fail("Unexpected DeleteEvent") + }, + GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { + defer GinkgoRecover() + Expect(q2).To(BeIdenticalTo(q)) + Expect(evt.Object).To(Equal(p)) + c <- evt + }, + }, + ) + } + + c1 := make(chan event.GenericEvent) + c2 := make(chan event.GenericEvent) + + err := createInstance(c1).Start(ctx, q) Expect(err).NotTo(HaveOccurred()) - err = instance.Start(ctx, handler.Funcs{ - CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected CreateEvent") - }, - UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected UpdateEvent") - }, - DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Fail("Unexpected DeleteEvent") - }, - GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) { - defer GinkgoRecover() - Expect(q2).To(BeIdenticalTo(q)) - Expect(evt.Object).To(Equal(p)) - resEvent2 = evt - close(c2) - }, - }, q) + err = createInstance(c2).Start(ctx, q) Expect(err).NotTo(HaveOccurred()) ch <- evt - <-c1 - <-c2 + resEvent1 := <-c1 + resEvent2 := <-c2 // Validate the two handlers received same event Expect(resEvent1).To(Equal(resEvent2))