From a4e8aca85c917413ad406fff1281bd1578c7ebc9 Mon Sep 17 00:00:00 2001 From: Tim Ramlot <42113979+inteon@users.noreply.github.com> Date: Sun, 11 Feb 2024 10:59:00 +0100 Subject: [PATCH 1/3] move Source definitions to a single folder without breaking the API Signed-off-by: Tim Ramlot <42113979+inteon@users.noreply.github.com> --- pkg/builder/controller.go | 11 +- pkg/source/internal/channel.go | 150 +++++++++++++++ .../internal}/event_handler.go | 0 pkg/source/internal/func.go | 39 ++++ pkg/source/internal/informer.go | 53 ++++++ .../internal}/internal_suite_test.go | 0 .../internal}/internal_test.go | 2 +- .../source => source/internal}/kind.go | 33 ++++ pkg/source/source.go | 173 +----------------- 9 files changed, 292 insertions(+), 169 deletions(-) create mode 100644 pkg/source/internal/channel.go rename pkg/{internal/source => source/internal}/event_handler.go (100%) create mode 100644 pkg/source/internal/func.go create mode 100644 pkg/source/internal/informer.go rename pkg/{internal/source => source/internal}/internal_suite_test.go (100%) rename pkg/{internal/source => source/internal}/internal_test.go (99%) rename pkg/{internal/source => source/internal}/kind.go (77%) diff --git a/pkg/builder/controller.go b/pkg/builder/controller.go index 1a115f2f7b..a649f47071 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" @@ -313,12 +312,14 @@ func (blder *Builder) doWatch() error { } 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 { + if srcKind, ok := w.src.(interface { + ProjectObject(func(client.Object) (client.Object, error)) error + }); ok { + if err := srcKind.ProjectObject(func(o client.Object) (client.Object, error) { + return blder.project(o, w.objectProjection) + }); err != nil { return err } - srcKind.Type = typeForSrc } allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...) allPredicates = append(allPredicates, w.predicates...) diff --git a/pkg/source/internal/channel.go b/pkg/source/internal/channel.go new file mode 100644 index 0000000000..9471be24e4 --- /dev/null +++ b/pkg/source/internal/channel.go @@ -0,0 +1,150 @@ +/* +Copyright 2018 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 ( + "context" + "fmt" + "sync" + + "k8s.io/client-go/util/workqueue" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/predicate" +) + +const ( + // defaultBufferSize is the default number of event notifications that can be buffered. + defaultBufferSize = 1024 +) + +// 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 + + // Source is the source channel to fetch GenericEvents + Source <-chan event.GenericEvent + + // dest is the destination channels of the added event handlers + dest []chan event.GenericEvent + + // 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 +} + +func (cs *Channel) 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") + } + + // use default value if DestBufferSize not specified + if cs.DestBufferSize == 0 { + cs.DestBufferSize = defaultBufferSize + } + + dst := make(chan event.GenericEvent, cs.DestBufferSize) + + cs.destLock.Lock() + cs.dest = append(cs.dest, dst) + cs.destLock.Unlock() + + cs.once.Do(func() { + // Distribute GenericEvents to all EventHandler / Queue pairs Watching this source + go cs.syncLoop(ctx) + }) + + 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) + }() + } + } + }() + + 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) { + for { + select { + case <-ctx.Done(): + // Close destination channels + cs.doStop() + return + case evt, stillOpen := <-cs.Source: + 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) + } + } +} diff --git a/pkg/internal/source/event_handler.go b/pkg/source/internal/event_handler.go similarity index 100% rename from pkg/internal/source/event_handler.go rename to pkg/source/internal/event_handler.go diff --git a/pkg/source/internal/func.go b/pkg/source/internal/func.go new file mode 100644 index 0000000000..9a0e43bb09 --- /dev/null +++ b/pkg/source/internal/func.go @@ -0,0 +1,39 @@ +/* +Copyright 2018 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 ( + "context" + "fmt" + + "k8s.io/client-go/util/workqueue" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/predicate" +) + +// Func is a function that implements Source. +type Func func(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) 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 Func) String() string { + return fmt.Sprintf("func source: %p", f) +} diff --git a/pkg/source/internal/informer.go b/pkg/source/internal/informer.go new file mode 100644 index 0000000000..63023c583e --- /dev/null +++ b/pkg/source/internal/informer.go @@ -0,0 +1,53 @@ +/* +Copyright 2018 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 ( + "context" + "fmt" + + "k8s.io/client-go/util/workqueue" + "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/predicate" +) + +// Informer is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create). +type Informer struct { + // Informer is the controller-runtime Informer + Informer cache.Informer +} + +// 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. + if is.Informer == nil { + return fmt.Errorf("must specify Informer.Informer") + } + + _, err := is.Informer.AddEventHandler(NewEventHandler(ctx, queue, handler, prct).HandlerFuncs()) + if err != nil { + return err + } + return nil +} + +func (is *Informer) String() string { + return fmt.Sprintf("informer source: %p", is.Informer) +} diff --git a/pkg/internal/source/internal_suite_test.go b/pkg/source/internal/internal_suite_test.go similarity index 100% rename from pkg/internal/source/internal_suite_test.go rename to pkg/source/internal/internal_suite_test.go diff --git a/pkg/internal/source/internal_test.go b/pkg/source/internal/internal_test.go similarity index 99% rename from pkg/internal/source/internal_test.go rename to pkg/source/internal/internal_test.go index 0574f7180e..c7b7aa4847 100644 --- a/pkg/internal/source/internal_test.go +++ b/pkg/source/internal/internal_test.go @@ -25,7 +25,7 @@ import ( "k8s.io/client-go/util/workqueue" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" - internal "sigs.k8s.io/controller-runtime/pkg/internal/source" + internal "sigs.k8s.io/controller-runtime/pkg/source/internal" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" diff --git a/pkg/internal/source/kind.go b/pkg/source/internal/kind.go similarity index 77% rename from pkg/internal/source/kind.go rename to pkg/source/internal/kind.go index b3a8227125..14483dcd08 100644 --- a/pkg/internal/source/kind.go +++ b/pkg/source/internal/kind.go @@ -1,3 +1,19 @@ +/* +Copyright 2018 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 ( @@ -94,6 +110,23 @@ func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue w return nil } +// ProjectObject sets the Kind's object to the given object. +// This function should only be called by the Controller builder. +// NOTE: make sure to update pkg/builder/controller.go if you change this function. +func (ks *Kind) ProjectObject(fn func(client.Object) (client.Object, error)) error { + if ks.startCancel != nil { + return fmt.Errorf("cannot project object after Start has been called") + } + + newType, err := fn(ks.Type) + if err != nil { + return err + } + + ks.Type = newType + return nil +} + func (ks *Kind) String() string { if ks.Type != nil { return fmt.Sprintf("kind source: %T", ks.Type) diff --git a/pkg/source/source.go b/pkg/source/source.go index 099c8d68fa..cf359d355c 100644 --- a/pkg/source/source.go +++ b/pkg/source/source.go @@ -18,24 +18,16 @@ package source import ( "context" - "fmt" - "sync" "k8s.io/client-go/util/workqueue" "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" + internal "sigs.k8s.io/controller-runtime/pkg/source/internal" "sigs.k8s.io/controller-runtime/pkg/cache" "sigs.k8s.io/controller-runtime/pkg/predicate" ) -const ( - // defaultBufferSize is the default number of event notifications that can be buffered. - defaultBufferSize = 1024 -) - // Source is a source of events (eh.g. Create, Update, Delete operations on Kubernetes Objects, Webhook callbacks, etc) // which should be processed by event.EventHandlers to enqueue reconcile.Requests. // @@ -62,164 +54,19 @@ func Kind(cache cache.Cache, object client.Object) SyncingSource { return &internal.Kind{Type: object, Cache: cache} } -var _ Source = &Channel{} +// Informer is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create). +type Informer = internal.Informer // 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 - - // Source is the source channel to fetch GenericEvents - Source <-chan event.GenericEvent - - // dest is the destination channels of the added event handlers - dest []chan event.GenericEvent - - // 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 -} - -func (cs *Channel) 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") - } - - // use default value if DestBufferSize not specified - if cs.DestBufferSize == 0 { - cs.DestBufferSize = defaultBufferSize - } - - dst := make(chan event.GenericEvent, cs.DestBufferSize) - - cs.destLock.Lock() - cs.dest = append(cs.dest, dst) - cs.destLock.Unlock() - - cs.once.Do(func() { - // Distribute GenericEvents to all EventHandler / Queue pairs Watching this source - go cs.syncLoop(ctx) - }) - - 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) - }() - } - } - }() - - 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) { - for { - select { - case <-ctx.Done(): - // Close destination channels - cs.doStop() - return - case evt, stillOpen := <-cs.Source: - 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) - } - } -} - -// Informer is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create). -type Informer struct { - // Informer is the controller-runtime Informer - Informer cache.Informer -} - -var _ Source = &Informer{} - -// 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. - if is.Informer == nil { - return fmt.Errorf("must specify Informer.Informer") - } - - _, err := is.Informer.AddEventHandler(internal.NewEventHandler(ctx, queue, handler, prct).HandlerFuncs()) - if err != nil { - return err - } - return nil -} - -func (is *Informer) String() string { - return fmt.Sprintf("informer source: %p", is.Informer) -} - -var _ Source = Func(nil) +type Channel = internal.Channel // Func is a function that implements Source. -type Func func(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error +type Func = internal.Func -// 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 Func) String() string { - return fmt.Sprintf("func source: %p", f) -} +var _ Source = &internal.Kind{} +var _ SyncingSource = &internal.Kind{} +var _ Source = &internal.Informer{} +var _ Source = &internal.Channel{} +var _ Source = internal.Func(nil) From bde0a5e2d9d01c39fc4b714ac4d7712470f19646 Mon Sep 17 00:00:00 2001 From: Tim Ramlot <42113979+inteon@users.noreply.github.com> Date: Tue, 13 Feb 2024 22:57:52 +0100 Subject: [PATCH 2/3] only allow starting a source once Signed-off-by: Tim Ramlot <42113979+inteon@users.noreply.github.com> --- pkg/controller/controller_test.go | 2 +- pkg/internal/controller/controller_test.go | 9 +- pkg/source/example_test.go | 4 +- pkg/source/internal/channel.go | 136 ++++++-------- pkg/source/internal/channel_broadcast.go | 196 +++++++++++++++++++++ pkg/source/internal/informer.go | 15 +- pkg/source/internal/kind.go | 9 +- pkg/source/source.go | 64 +++++-- pkg/source/source_integration_test.go | 6 +- pkg/source/source_test.go | 103 +++++------ 10 files changed, 372 insertions(+), 172 deletions(-) create mode 100644 pkg/source/internal/channel_broadcast.go diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go index e49a2c5774..3568ab5d89 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)) watchChan <- event.GenericEvent{Object: &corev1.Pod{}} reconcileStarted := make(chan struct{}) diff --git a/pkg/internal/controller/controller_test.go b/pkg/internal/controller/controller_test.go index ce2245e60f..f551598cbd 100644 --- a/pkg/internal/controller/controller_test.go +++ b/pkg/internal/controller/controller_test.go @@ -226,8 +226,7 @@ var _ = Describe("controller", func() { Object: p, } - ins := &source.Channel{Source: ch} - ins.DestBufferSize = 1 + ins := source.Channel(source.NewChannelBroadcaster(ch), source.WithDestBufferSize(1)) // send the event to the channel ch <- evt @@ -249,18 +248,18 @@ 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{} + ins := source.Channel(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 Broadcaster")) }) It("should call Start on sources with the appropriate EventHandler, Queue, and Predicates", func() { diff --git a/pkg/source/example_test.go b/pkg/source/example_test.go index 77857729de..fc96dc4d8d 100644 --- a/pkg/source/example_test.go +++ b/pkg/source/example_test.go @@ -43,7 +43,9 @@ func ExampleChannel() { events := make(chan event.GenericEvent) err := ctrl.Watch( - &source.Channel{Source: events}, + source.Channel( + source.NewChannelBroadcaster(events), + ), &handler.EnqueueRequestForObject{}, ) if err != nil { diff --git a/pkg/source/internal/channel.go b/pkg/source/internal/channel.go index 9471be24e4..3272e9a6fd 100644 --- a/pkg/source/internal/channel.go +++ b/pkg/source/internal/channel.go @@ -27,30 +27,25 @@ import ( "sigs.k8s.io/controller-runtime/pkg/predicate" ) -const ( - // defaultBufferSize is the default number of event notifications that can be buffered. - defaultBufferSize = 1024 -) +// ChannelOptions contains the options for the Channel source. +type ChannelOptions struct { + // DestBufferSize is the specified buffer size of dest channels. + // Default to 1024 if not specified. + DestBufferSize int +} // 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 - - // Source is the source channel to fetch GenericEvents - Source <-chan event.GenericEvent + Options ChannelOptions - // dest is the destination channels of the added event handlers - dest []chan event.GenericEvent + // Broadcaster contains the source channel for events. + Broadcaster *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 + 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 { @@ -63,88 +58,67 @@ func (cs *Channel) Start( 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") + // 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 + cs.mu.Lock() + defer cs.mu.Unlock() + if cs.isStarted { + return fmt.Errorf("cannot start an already started Channel source") } + cs.isStarted = true - dst := make(chan event.GenericEvent, cs.DestBufferSize) - - cs.destLock.Lock() - cs.dest = append(cs.dest, dst) - cs.destLock.Unlock() - - 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.Options.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, + handler, + prct, + ) }() 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 *Channel) processReceivedEvents( + ctx context.Context, + destination <-chan event.GenericEvent, + queue workqueue.RateLimitingInterface, + eventHandler handler.EventHandler, + predicates []predicate.Predicate, +) { +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 predicates { + if !p.Generic(event) { + continue eventloop + } + } + + // Call the event handler with the event. + eventHandler.Generic(ctx, event, queue) } } } diff --git a/pkg/source/internal/channel_broadcast.go b/pkg/source/internal/channel_broadcast.go new file mode 100644 index 0000000000..cb7c9431c6 --- /dev/null +++ b/pkg/source/internal/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/source/internal/informer.go b/pkg/source/internal/informer.go index 63023c583e..d7a8f8ba49 100644 --- a/pkg/source/internal/informer.go +++ b/pkg/source/internal/informer.go @@ -19,6 +19,7 @@ package internal import ( "context" "fmt" + "sync" "k8s.io/client-go/util/workqueue" "sigs.k8s.io/controller-runtime/pkg/cache" @@ -30,16 +31,26 @@ import ( type Informer struct { // Informer is the controller-runtime Informer Informer cache.Informer + + 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. if is.Informer == nil { - return fmt.Errorf("must specify Informer.Informer") + return fmt.Errorf("must create Informer with a non-nil Informer") + } + + 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(NewEventHandler(ctx, queue, handler, prct).HandlerFuncs()) if err != nil { diff --git a/pkg/source/internal/kind.go b/pkg/source/internal/kind.go index 14483dcd08..37f0487ede 100644 --- a/pkg/source/internal/kind.go +++ b/pkg/source/internal/kind.go @@ -36,7 +36,6 @@ import ( type Kind struct { // Type is the type of object to watch. e.g. &v1.Pod{} Type client.Object - // Cache used to watch APIs Cache cache.Cache @@ -51,10 +50,14 @@ type Kind struct { func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue workqueue.RateLimitingInterface, prct ...predicate.Predicate) error { 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") + return fmt.Errorf("must create Kind with a non-nil Cache") + } + + 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 diff --git a/pkg/source/source.go b/pkg/source/source.go index cf359d355c..74796e315c 100644 --- a/pkg/source/source.go +++ b/pkg/source/source.go @@ -22,10 +22,10 @@ import ( "k8s.io/client-go/util/workqueue" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/predicate" internal "sigs.k8s.io/controller-runtime/pkg/source/internal" "sigs.k8s.io/controller-runtime/pkg/cache" - "sigs.k8s.io/controller-runtime/pkg/predicate" ) // Source is a source of events (eh.g. Create, Update, Delete operations on Kubernetes Objects, Webhook callbacks, etc) @@ -49,24 +49,58 @@ type SyncingSource interface { WaitForSync(ctx context.Context) error } +var _ Source = &internal.Kind{} +var _ SyncingSource = &internal.Kind{} +var _ Source = &internal.Informer{} +var _ Source = &internal.Channel{} +var _ Source = internal.Func(nil) + // 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} + return &internal.Kind{Cache: cache, Type: object} } -// Informer is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create). -type Informer = internal.Informer +// 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 -// 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 = internal.Channel +// ChannelOption is a functional option for configuring a Channel source. +type ChannelOption func(*internal.ChannelOptions) -// Func is a function that implements Source. -type Func = internal.Func +// WithDestBufferSize specifies the buffer size of dest channels. +func WithDestBufferSize(destBufferSize int) ChannelOption { + return func(o *internal.ChannelOptions) { + if destBufferSize <= 0 { + return // ignore invalid buffer size + } -var _ Source = &internal.Kind{} -var _ SyncingSource = &internal.Kind{} -var _ Source = &internal.Informer{} -var _ Source = &internal.Channel{} -var _ Source = internal.Func(nil) + o.DestBufferSize = destBufferSize + } +} + +// Channel creates a ChannelSource with the given buffer size. +func Channel(broadcaster *internal.ChannelBroadcaster, options ...ChannelOption) Source { + opts := internal.ChannelOptions{ + // 1024 is the default number of event notifications that can be buffered. + DestBufferSize: 1024, + } + for _, o := range options { + if o == nil { + continue // ignore nil options + } + o(&opts) + } + + return &internal.Channel{Options: opts, Broadcaster: broadcaster} +} + +// Informer creates an InformerSource with the given cache provider. +func Informer(informer cache.Informer) Source { + return &internal.Informer{Informer: informer} +} + +// Func creates a FuncSource with the given function. +func Func(f internal.Func) Source { + return f +} diff --git a/pkg/source/source_integration_test.go b/pkg/source/source_integration_test.go index 594d3c9a9c..cfb92cb88f 100644 --- a/pkg/source/source_integration_test.go +++ b/pkg/source/source_integration_test.go @@ -241,7 +241,7 @@ var _ = Describe("Source", func() { c := make(chan struct{}) q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := &source.Informer{Informer: depInformer} + instance := source.Informer(depInformer) err := instance.Start(ctx, handler.Funcs{ CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) { defer GinkgoRecover() @@ -282,7 +282,7 @@ var _ = Describe("Source", func() { rs2.SetLabels(map[string]string{"biz": "baz"}) q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := &source.Informer{Informer: depInformer} + instance := source.Informer(depInformer) err = instance.Start(ctx, handler.Funcs{ CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) { }, @@ -319,7 +319,7 @@ var _ = Describe("Source", func() { c := make(chan struct{}) q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := &source.Informer{Informer: depInformer} + instance := source.Informer(depInformer) err := instance.Start(ctx, handler.Funcs{ CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { }, diff --git a/pkg/source/source_test.go b/pkg/source/source_test.go index 16c365e8a2..39777278c3 100644 --- a/pkg/source/source_test.go +++ b/pkg/source/source_test.go @@ -183,20 +183,20 @@ var _ = Describe("Source", func() { instance := source.Kind(nil, &corev1.Pod{}) err := instance.Start(ctx, nil, 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) 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()) + Expect(instance.Start(context.Background(), &handler.EnqueueRequestForObject{}, nil)).NotTo(HaveOccurred()) err := instance.WaitForSync(context.Background()) Expect(err).To(HaveOccurred()) Expect(err.Error()).To(Equal("cache did not sync")) @@ -221,7 +221,7 @@ 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()) + Expect(instance.Start(context.Background(), &handler.EnqueueRequestForObject{}, nil)).NotTo(HaveOccurred()) err := instance.WaitForSync(context.Background()) Expect(err).To(HaveOccurred()) Expect(err.Error()).To(Equal("cache did not sync")) @@ -289,7 +289,7 @@ var _ = Describe("Source", func() { } q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := &source.Channel{Source: ch} + instance := source.Channel(source.NewChannelBroadcaster(ch)) err := instance.Start(ctx, handler.Funcs{ CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { defer GinkgoRecover() @@ -327,8 +327,7 @@ 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), source.WithDestBufferSize(1)) err := instance.Start(ctx, handler.Funcs{ CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { defer GinkgoRecover() @@ -383,8 +382,7 @@ 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), source.WithDestBufferSize(1)) err := instance.Start(ctx, handler.Funcs{ CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) { @@ -422,7 +420,7 @@ var _ = Describe("Source", func() { close(ch) By("feeding that channel to a channel source") - src := &source.Channel{Source: ch} + src := source.Channel(source.NewChannelBroadcaster(ch)) processed := make(chan struct{}) defer close(processed) @@ -454,9 +452,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*/ } + instance := source.Channel(nil) err := instance.Start(ctx, handler.Funcs{}, q) - Expect(err).To(Equal(fmt.Errorf("must specify Channel.Source"))) + Expect(err).To(Equal(fmt.Errorf("must create Channel with a non-nil Broadcaster"))) }) }) Context("for multi sources (handlers)", func() { @@ -469,61 +467,44 @@ 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) + + createHandler := func(c chan event.GenericEvent) handler.Funcs { + return 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 := source.Channel(channelSource).Start(ctx, createHandler(c1), 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 = source.Channel(channelSource).Start(ctx, createHandler(c2), 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)) From 98944019ace34b2d4f3a54d2abad5d6cb85ee258 Mon Sep 17 00:00:00 2001 From: Tim Ramlot <42113979+inteon@users.noreply.github.com> Date: Tue, 13 Feb 2024 23:02:04 +0100 Subject: [PATCH 3/3] 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 | 110 ++++--- 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/handler/predicates.go | 47 +++ pkg/internal/controller/controller.go | 15 +- pkg/internal/controller/controller_test.go | 62 ++-- .../recorder/recorder_integration_test.go | 2 +- pkg/predicate/predicate.go | 23 +- pkg/source/example_test.go | 4 +- pkg/source/internal/channel.go | 29 +- pkg/source/internal/event_handler.go | 34 +- pkg/source/internal/func.go | 9 +- pkg/source/internal/informer.go | 12 +- pkg/source/internal/internal_test.go | 84 ++--- pkg/source/internal/kind.go | 29 +- pkg/source/source.go | 15 +- pkg/source/source_integration_test.go | 68 ++-- pkg/source/source_test.go | 301 ++++++++++-------- 23 files changed, 495 insertions(+), 437 deletions(-) create mode 100644 pkg/handler/predicates.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 a649f47071..8c51fa5ca7 100644 --- a/pkg/builder/controller.go +++ b/pkg/builder/controller.go @@ -53,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. @@ -79,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") @@ -120,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 } @@ -132,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. @@ -165,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) @@ -271,11 +278,14 @@ 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.WithPredicates(&handler.EnqueueRequestForObject{}, allPredicates...), + ) + if err := blder.ctrl.Watch(src); err != nil { return err } } @@ -289,7 +299,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()) @@ -301,32 +310,43 @@ 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, + handler.WithPredicates(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.(interface { - ProjectObject(func(client.Object) (client.Object, error)) error - }); ok { - if err := srcKind.ProjectObject(func(o client.Object) (client.Object, error) { - return blder.project(o, w.objectProjection) - }); err != nil { - return err - } + 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, + handler.WithPredicates(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..1ffc7c5287 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) { +// ApplyToWatchesObject 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) { +// ApplyToWatchesObject 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 3568ab5d89..5ad4a3de8b 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.NewChannelBroadcaster(watchChan)) + watch := source.Channel(source.NewChannelBroadcaster(watchChan), &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/handler/predicates.go b/pkg/handler/predicates.go new file mode 100644 index 0000000000..81175e755b --- /dev/null +++ b/pkg/handler/predicates.go @@ -0,0 +1,47 @@ +package handler + +import ( + "context" + + "k8s.io/client-go/util/workqueue" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/predicate" +) + +// WithPredicates returns an EventHandler that only calls the underlying handler if all the given predicates return true. +func WithPredicates(handler EventHandler, predicates ...predicate.Predicate) EventHandler { + return &Funcs{ + CreateFunc: func(ctx context.Context, createEvent event.CreateEvent, queue workqueue.RateLimitingInterface) { + for _, predicate := range predicates { + if !predicate.Create(createEvent) { + return + } + } + handler.Create(ctx, createEvent, queue) + }, + UpdateFunc: func(ctx context.Context, updateEvent event.UpdateEvent, queue workqueue.RateLimitingInterface) { + for _, predicate := range predicates { + if !predicate.Update(updateEvent) { + return + } + } + handler.Update(ctx, updateEvent, queue) + }, + DeleteFunc: func(ctx context.Context, deleteEvent event.DeleteEvent, queue workqueue.RateLimitingInterface) { + for _, predicate := range predicates { + if !predicate.Delete(deleteEvent) { + return + } + } + handler.Delete(ctx, deleteEvent, queue) + }, + GenericFunc: func(ctx context.Context, genericEvent event.GenericEvent, queue workqueue.RateLimitingInterface) { + for _, predicate := range predicates { + if !predicate.Generic(genericEvent) { + return + } + } + handler.Generic(ctx, genericEvent, queue) + }, + } +} 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 f551598cbd..f1a27852e2 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,19 +225,21 @@ var _ = Describe("controller", func() { Object: p, } - ins := source.Channel(source.NewChannelBroadcaster(ch), source.WithDestBufferSize(1)) + ins := source.Channel( + source.NewChannelBroadcaster(ch), + 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() { @@ -252,7 +253,10 @@ var _ = Describe("controller", func() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - ins := source.Channel(nil) + ins := source.Channel( + nil, + &handler.EnqueueRequestForObject{}, + ) ctrl.startWatches = []watchDescription{{ src: ins, }} @@ -262,21 +266,33 @@ var _ = Describe("controller", func() { 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( + source.NewChannelBroadcaster(make(<-chan event.GenericEvent)), + nil, + ) + 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 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()) @@ -287,13 +303,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/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 fc96dc4d8d..4a668f7158 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 } @@ -45,8 +45,8 @@ func ExampleChannel() { err := ctrl.Watch( source.Channel( source.NewChannelBroadcaster(events), + &handler.EnqueueRequestForObject{}, ), - &handler.EnqueueRequestForObject{}, ) if err != nil { // handle it diff --git a/pkg/source/internal/channel.go b/pkg/source/internal/channel.go index 3272e9a6fd..d1ccae51e7 100644 --- a/pkg/source/internal/channel.go +++ b/pkg/source/internal/channel.go @@ -24,7 +24,6 @@ import ( "k8s.io/client-go/util/workqueue" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" - "sigs.k8s.io/controller-runtime/pkg/predicate" ) // ChannelOptions contains the options for the Channel source. @@ -43,6 +42,9 @@ type Channel struct { // Broadcaster contains the source channel for events. Broadcaster *ChannelBroadcaster + // EventHandler is the handler to call when events are received. + EventHandler handler.EventHandler + mu sync.Mutex // isStarted is true if the source has been started. A source can only be started once. isStarted bool @@ -53,15 +55,13 @@ func (cs *Channel) String() string { } // 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 { - // Broadcaster should have been specified by the user. +func (cs *Channel) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error { if cs.Broadcaster == nil { return fmt.Errorf("must create Channel with a non-nil Broadcaster") } + if cs.EventHandler == nil { + return fmt.Errorf("must create Channel with a non-nil EventHandler") + } cs.mu.Lock() defer cs.mu.Unlock() @@ -84,8 +84,6 @@ func (cs *Channel) Start( ctx, destination, queue, - handler, - prct, ) }() @@ -96,10 +94,7 @@ func (cs *Channel) processReceivedEvents( ctx context.Context, destination <-chan event.GenericEvent, queue workqueue.RateLimitingInterface, - eventHandler handler.EventHandler, - predicates []predicate.Predicate, ) { -eventloop: for { select { case <-ctx.Done(): @@ -109,16 +104,8 @@ eventloop: return } - // Check predicates against the event first - // and continue the outer loop if any of them fail. - for _, p := range predicates { - if !p.Generic(event) { - continue eventloop - } - } - // Call the event handler with the event. - eventHandler.Generic(ctx, event, queue) + cs.EventHandler.Generic(ctx, event, queue) } } } diff --git a/pkg/source/internal/event_handler.go b/pkg/source/internal/event_handler.go index ae8404a1fa..43edadb977 100644 --- a/pkg/source/internal/event_handler.go +++ b/pkg/source/internal/event_handler.go @@ -26,19 +26,16 @@ import ( "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" logf "sigs.k8s.io/controller-runtime/pkg/internal/log" - - "sigs.k8s.io/controller-runtime/pkg/predicate" ) var log = logf.RuntimeLog.WithName("source").WithName("EventHandler") // NewEventHandler creates a new EventHandler. -func NewEventHandler(ctx context.Context, queue workqueue.RateLimitingInterface, handler handler.EventHandler, predicates []predicate.Predicate) *EventHandler { +func NewEventHandler(ctx context.Context, queue workqueue.RateLimitingInterface, handler handler.EventHandler) *EventHandler { return &EventHandler{ - ctx: ctx, - handler: handler, - queue: queue, - predicates: predicates, + ctx: ctx, + handler: handler, + queue: queue, } } @@ -48,9 +45,8 @@ type EventHandler struct { // that is used to propagate cancellation signals to each handler function. ctx context.Context - handler handler.EventHandler - queue workqueue.RateLimitingInterface - predicates []predicate.Predicate + handler handler.EventHandler + queue workqueue.RateLimitingInterface } // HandlerFuncs converts EventHandler to a ResourceEventHandlerFuncs @@ -76,12 +72,6 @@ func (e *EventHandler) OnAdd(obj interface{}) { return } - for _, p := range e.predicates { - if !p.Create(c) { - return - } - } - // Invoke create handler ctx, cancel := context.WithCancel(e.ctx) defer cancel() @@ -109,12 +99,6 @@ func (e *EventHandler) OnUpdate(oldObj, newObj interface{}) { return } - for _, p := range e.predicates { - if !p.Update(u) { - return - } - } - // Invoke update handler ctx, cancel := context.WithCancel(e.ctx) defer cancel() @@ -157,12 +141,6 @@ func (e *EventHandler) OnDelete(obj interface{}) { return } - for _, p := range e.predicates { - if !p.Delete(d) { - return - } - } - // Invoke delete handler ctx, cancel := context.WithCancel(e.ctx) defer cancel() diff --git a/pkg/source/internal/func.go b/pkg/source/internal/func.go index 9a0e43bb09..3db10b54fc 100644 --- a/pkg/source/internal/func.go +++ b/pkg/source/internal/func.go @@ -21,17 +21,14 @@ import ( "fmt" "k8s.io/client-go/util/workqueue" - "sigs.k8s.io/controller-runtime/pkg/handler" - "sigs.k8s.io/controller-runtime/pkg/predicate" ) // Func is a function that implements Source. -type Func func(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error +type Func 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 Func) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error { + return f(ctx, queue) } func (f Func) String() string { diff --git a/pkg/source/internal/informer.go b/pkg/source/internal/informer.go index d7a8f8ba49..65b321e9d9 100644 --- a/pkg/source/internal/informer.go +++ b/pkg/source/internal/informer.go @@ -24,7 +24,6 @@ import ( "k8s.io/client-go/util/workqueue" "sigs.k8s.io/controller-runtime/pkg/cache" "sigs.k8s.io/controller-runtime/pkg/handler" - "sigs.k8s.io/controller-runtime/pkg/predicate" ) // Informer is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create). @@ -32,6 +31,9 @@ type Informer struct { // Informer is the controller-runtime Informer Informer cache.Informer + // EventHandler is the handler to call when events are received. + EventHandler handler.EventHandler + mu sync.Mutex // isStarted is true if the source has been started. A source can only be started once. isStarted bool @@ -39,11 +41,13 @@ type Informer struct { // 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 { +func (is *Informer) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error { if is.Informer == nil { 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") + } is.mu.Lock() defer is.mu.Unlock() @@ -52,7 +56,7 @@ func (is *Informer) Start(ctx context.Context, handler handler.EventHandler, que } is.isStarted = true - _, err := is.Informer.AddEventHandler(NewEventHandler(ctx, queue, handler, prct).HandlerFuncs()) + _, err := is.Informer.AddEventHandler(NewEventHandler(ctx, queue, is.EventHandler).HandlerFuncs()) if err != nil { return err } diff --git a/pkg/source/internal/internal_test.go b/pkg/source/internal/internal_test.go index c7b7aa4847..6446fc3d86 100644 --- a/pkg/source/internal/internal_test.go +++ b/pkg/source/internal/internal_test.go @@ -74,7 +74,7 @@ var _ = Describe("Internal", func() { set = true }, } - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, funcs, nil) + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, funcs) }) Describe("EventHandler", func() { @@ -99,41 +99,47 @@ var _ = Describe("Internal", func() { }) It("should used Predicates to filter CreateEvents", func() { - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ - predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }}, - }) + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, + handler.WithPredicates(setfuncs, predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }}), + ) set = false instance.OnAdd(pod) Expect(set).To(BeFalse()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ - predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, - }) + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, + handler.WithPredicates(setfuncs, predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}), + ) instance.OnAdd(pod) Expect(set).To(BeTrue()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ - predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, - predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }}, - }) + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, + handler.WithPredicates(setfuncs, + predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, + predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }}, + ), + ) instance.OnAdd(pod) Expect(set).To(BeFalse()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ - predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }}, - predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, - }) + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, + handler.WithPredicates(setfuncs, + predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }}, + predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, + ), + ) instance.OnAdd(pod) Expect(set).To(BeFalse()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ - predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, - predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, - }) + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, + handler.WithPredicates(setfuncs, + predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, + predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, + ), + ) instance.OnAdd(pod) Expect(set).To(BeTrue()) }) @@ -157,40 +163,40 @@ var _ = Describe("Internal", func() { It("should used Predicates to filter UpdateEvents", func() { set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs, predicate.Funcs{UpdateFunc: func(updateEvent event.UpdateEvent) bool { return false }}, - }) + )) instance.OnUpdate(pod, newPod) Expect(set).To(BeFalse()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs, predicate.Funcs{UpdateFunc: func(event.UpdateEvent) bool { return true }}, - }) + )) instance.OnUpdate(pod, newPod) Expect(set).To(BeTrue()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs, predicate.Funcs{UpdateFunc: func(event.UpdateEvent) bool { return true }}, predicate.Funcs{UpdateFunc: func(event.UpdateEvent) bool { return false }}, - }) + )) instance.OnUpdate(pod, newPod) Expect(set).To(BeFalse()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs, predicate.Funcs{UpdateFunc: func(event.UpdateEvent) bool { return false }}, predicate.Funcs{UpdateFunc: func(event.UpdateEvent) bool { return true }}, - }) + )) instance.OnUpdate(pod, newPod) Expect(set).To(BeFalse()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs, predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}, - }) + )) instance.OnUpdate(pod, newPod) Expect(set).To(BeTrue()) }) @@ -215,40 +221,40 @@ var _ = Describe("Internal", func() { It("should used Predicates to filter DeleteEvents", func() { set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs, predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return false }}, - }) + )) instance.OnDelete(pod) Expect(set).To(BeFalse()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs, predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return true }}, - }) + )) instance.OnDelete(pod) Expect(set).To(BeTrue()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs, predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return true }}, predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return false }}, - }) + )) instance.OnDelete(pod) Expect(set).To(BeFalse()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs, predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return false }}, predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return true }}, - }) + )) instance.OnDelete(pod) Expect(set).To(BeFalse()) set = false - instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{ + instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs, predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return true }}, predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return true }}, - }) + )) instance.OnDelete(pod) Expect(set).To(BeTrue()) }) diff --git a/pkg/source/internal/kind.go b/pkg/source/internal/kind.go index 37f0487ede..ba3931ae99 100644 --- a/pkg/source/internal/kind.go +++ b/pkg/source/internal/kind.go @@ -29,7 +29,6 @@ import ( "sigs.k8s.io/controller-runtime/pkg/cache" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/handler" - "sigs.k8s.io/controller-runtime/pkg/predicate" ) // Kind is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create). @@ -39,6 +38,9 @@ type Kind struct { // Cache used to watch APIs Cache cache.Cache + // EventHandler is the handler to call when events are received. + EventHandler handler.EventHandler + // started may contain an error if one was encountered during startup. If its closed and does not // contain an error, startup and syncing finished. started chan error @@ -47,14 +49,16 @@ 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.Type == nil { 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") @@ -98,7 +102,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).HandlerFuncs()) if err != nil { ks.started <- err return @@ -113,23 +117,6 @@ func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue w return nil } -// ProjectObject sets the Kind's object to the given object. -// This function should only be called by the Controller builder. -// NOTE: make sure to update pkg/builder/controller.go if you change this function. -func (ks *Kind) ProjectObject(fn func(client.Object) (client.Object, error)) error { - if ks.startCancel != nil { - return fmt.Errorf("cannot project object after Start has been called") - } - - newType, err := fn(ks.Type) - if err != nil { - return err - } - - ks.Type = newType - return nil -} - func (ks *Kind) String() string { if ks.Type != nil { return fmt.Sprintf("kind source: %T", ks.Type) diff --git a/pkg/source/source.go b/pkg/source/source.go index 74796e315c..2b2fcb8d43 100644 --- a/pkg/source/source.go +++ b/pkg/source/source.go @@ -22,7 +22,6 @@ import ( "k8s.io/client-go/util/workqueue" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/handler" - "sigs.k8s.io/controller-runtime/pkg/predicate" internal "sigs.k8s.io/controller-runtime/pkg/source/internal" "sigs.k8s.io/controller-runtime/pkg/cache" @@ -39,7 +38,7 @@ import ( 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 @@ -56,8 +55,8 @@ var _ Source = &internal.Channel{} var _ Source = internal.Func(nil) // Kind creates a KindSource with the given cache provider. -func Kind(cache cache.Cache, object client.Object) SyncingSource { - return &internal.Kind{Cache: cache, Type: object} +func Kind(cache cache.Cache, object client.Object, eventhandler handler.EventHandler) SyncingSource { + return &internal.Kind{Cache: cache, Type: object, EventHandler: eventhandler} } // NewChannelBroadcaster creates a new ChannelBroadcaster for the given channel. @@ -80,7 +79,7 @@ func WithDestBufferSize(destBufferSize int) ChannelOption { } // Channel creates a ChannelSource with the given buffer size. -func Channel(broadcaster *internal.ChannelBroadcaster, options ...ChannelOption) Source { +func Channel(broadcaster *internal.ChannelBroadcaster, eventhandler handler.EventHandler, options ...ChannelOption) Source { opts := internal.ChannelOptions{ // 1024 is the default number of event notifications that can be buffered. DestBufferSize: 1024, @@ -92,12 +91,12 @@ func Channel(broadcaster *internal.ChannelBroadcaster, options ...ChannelOption) o(&opts) } - return &internal.Channel{Options: opts, Broadcaster: broadcaster} + return &internal.Channel{Options: opts, Broadcaster: broadcaster, EventHandler: eventhandler} } // Informer creates an InformerSource with the given cache provider. -func Informer(informer cache.Informer) Source { - return &internal.Informer{Informer: informer} +func Informer(informer cache.Informer, eventhandler handler.EventHandler) Source { + return &internal.Informer{Informer: informer, EventHandler: eventhandler} } // Func creates a FuncSource with the given function. diff --git a/pkg/source/source_integration_test.go b/pkg/source/source_integration_test.go index cfb92cb88f..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(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(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(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 39777278c3..6e2d723cb8 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")) }) 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 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(), &handler.EnqueueRequestForObject{}, 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(), &handler.EnqueueRequestForObject{}, 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,35 @@ var _ = Describe("Source", func() { } q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := source.Channel(source.NewChannelBroadcaster(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) - }, - }, q, prct) + instance := source.Channel( + source.NewChannelBroadcaster(ch), + handler.WithPredicates( + 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) + }, + }, + prct, + ), + ) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) ch <- invalidEvt @@ -327,33 +334,37 @@ var _ = Describe("Source", func() { q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") // Add a handler to get distribution blocked - instance := source.Channel(source.NewChannelBroadcaster(ch), source.WithDestBufferSize(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), + 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) + source.WithDestBufferSize(1), + ) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) // Write 3 events into the source channel. @@ -382,27 +393,31 @@ var _ = Describe("Source", func() { q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") // Add a handler to get distribution blocked - instance := source.Channel(source.NewChannelBroadcaster(ch), source.WithDestBufferSize(1)) + instance := source.Channel( + source.NewChannelBroadcaster(ch), + 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() + source.WithDestBufferSize(1), + ) - close(processed) - }, - }, q) + err := instance.Start(ctx, q) Expect(err).NotTo(HaveOccurred()) <-processed @@ -419,31 +434,34 @@ var _ = Describe("Source", func() { ch <- evt close(ch) - By("feeding that channel to a channel source") - src := source.Channel(source.NewChannelBroadcaster(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), + 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") @@ -452,8 +470,8 @@ var _ = Describe("Source", func() { }) It("should get error if no source specified", func() { q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") - instance := source.Channel(nil) - err := instance.Start(ctx, handler.Funcs{}, q) + instance := source.Channel(nil, handler.Funcs{}) + err := instance.Start(ctx, q) Expect(err).To(Equal(fmt.Errorf("must create Channel with a non-nil Broadcaster"))) }) }) @@ -470,36 +488,39 @@ var _ = Describe("Source", func() { q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test") channelSource := source.NewChannelBroadcaster(ch) - createHandler := func(c chan event.GenericEvent) handler.Funcs { - return 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 + createInstance := func(c chan event.GenericEvent) source.Source { + return source.Channel( + channelSource, + 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 := source.Channel(channelSource).Start(ctx, createHandler(c1), q) + err := createInstance(c1).Start(ctx, q) Expect(err).NotTo(HaveOccurred()) - err = source.Channel(channelSource).Start(ctx, createHandler(c2), q) + err = createInstance(c2).Start(ctx, q) Expect(err).NotTo(HaveOccurred()) ch <- evt