Skip to content

Commit af8d903

Browse files
k8s-ci-robotk8s-ci-robot
authored
Merge pull request #2046 from FillZpp/add-return-registration-into-event-handler
⚠ Support registration and removal of event handler
2 parents 50077a4 + 6d2d247 commit af8d903

File tree

7 files changed

+66
-48
lines changed

7 files changed

+66
-48
lines changed

‎pkg/cache/cache.go

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -78,11 +78,19 @@ type Informer interface {
7878
// AddEventHandler adds an event handler to the shared informer using the shared informer's resync
7979
// period. Events to a single handler are delivered sequentially, but there is no coordination
8080
// between different handlers.
81-
AddEventHandler(handler toolscache.ResourceEventHandler)
81+
// It returns a registration handle for the handler that can be used to remove
82+
// the handler again.
83+
AddEventHandler(handler toolscache.ResourceEventHandler) (toolscache.ResourceEventHandlerRegistration, error)
8284
// AddEventHandlerWithResyncPeriod adds an event handler to the shared informer using the
8385
// specified resync period. Events to a single handler are delivered sequentially, but there is
8486
// no coordination between different handlers.
85-
AddEventHandlerWithResyncPeriod(handler toolscache.ResourceEventHandler, resyncPeriod time.Duration)
87+
// It returns a registration handle for the handler that can be used to remove
88+
// the handler again and an error if the handler cannot be added.
89+
AddEventHandlerWithResyncPeriod(handler toolscache.ResourceEventHandler, resyncPeriod time.Duration) (toolscache.ResourceEventHandlerRegistration, error)
90+
// RemoveEventHandler removes a formerly added event handler given by
91+
// its registration handle.
92+
// This function is guaranteed to be idempotent, and thread-safe.
93+
RemoveEventHandler(handle toolscache.ResourceEventHandlerRegistration) error
8694
// AddIndexers adds more indexers to this store. If you call this after you already have data
8795
// in the store, the results are undefined.
8896
AddIndexers(indexers toolscache.Indexers) error

‎pkg/cache/cache_test.go

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1345,7 +1345,7 @@ func CacheTest(createCacheFunc func(config *rest.Config, opts cache.Options) (ca
13451345
addFunc := func(obj interface{}) {
13461346
out <- obj
13471347
}
1348-
sii.AddEventHandler(kcache.ResourceEventHandlerFuncs{AddFunc: addFunc})
1348+
_, _ = sii.AddEventHandler(kcache.ResourceEventHandlerFuncs{AddFunc: addFunc})
13491349

13501350
By("adding an object")
13511351
cl, err := client.New(cfg, client.Options{})
@@ -1369,7 +1369,7 @@ func CacheTest(createCacheFunc func(config *rest.Config, opts cache.Options) (ca
13691369
addFunc := func(obj interface{}) {
13701370
out <- obj
13711371
}
1372-
sii.AddEventHandler(kcache.ResourceEventHandlerFuncs{AddFunc: addFunc})
1372+
_, _ = sii.AddEventHandler(kcache.ResourceEventHandlerFuncs{AddFunc: addFunc})
13731373

13741374
By("adding an object")
13751375
cl, err := client.New(cfg, client.Options{})
@@ -1528,7 +1528,7 @@ func CacheTest(createCacheFunc func(config *rest.Config, opts cache.Options) (ca
15281528
addFunc := func(obj interface{}) {
15291529
out <- obj
15301530
}
1531-
sii.AddEventHandler(kcache.ResourceEventHandlerFuncs{AddFunc: addFunc})
1531+
_, _ = sii.AddEventHandler(kcache.ResourceEventHandlerFuncs{AddFunc: addFunc})
15321532

15331533
By("adding an object")
15341534
cl, err := client.New(cfg, client.Options{})
@@ -1646,7 +1646,7 @@ func CacheTest(createCacheFunc func(config *rest.Config, opts cache.Options) (ca
16461646
addFunc := func(obj interface{}) {
16471647
out <- obj
16481648
}
1649-
sii.AddEventHandler(kcache.ResourceEventHandlerFuncs{AddFunc: addFunc})
1649+
_, _ = sii.AddEventHandler(kcache.ResourceEventHandlerFuncs{AddFunc: addFunc})
16501650

16511651
By("adding an object")
16521652
cl, err := client.New(cfg, client.Options{})

‎pkg/cache/informer_cache.go

Lines changed: 2 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@ import (
2121
"fmt"
2222
"reflect"
2323
"strings"
24-
"time"
2524

2625
apimeta "k8s.io/apimachinery/pkg/api/meta"
2726
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -142,7 +141,7 @@ func (ip *informerCache) GetInformerForKind(ctx context.Context, gvk schema.Grou
142141
if err != nil {
143142
return nil, err
144143
}
145-
return WrapInformer(i.Informer), err
144+
return i.Informer, err
146145
}
147146

148147
// GetInformer returns the informer for the obj.
@@ -156,7 +155,7 @@ func (ip *informerCache) GetInformer(ctx context.Context, obj client.Object) (In
156155
if err != nil {
157156
return nil, err
158157
}
159-
return WrapInformer(i.Informer), err
158+
return i.Informer, err
160159
}
161160

162161
// NeedLeaderElection implements the LeaderElectionRunnable interface
@@ -216,20 +215,3 @@ func indexByField(indexer Informer, field string, extractor client.IndexerFunc)
216215

217216
return indexer.AddIndexers(cache.Indexers{internal.FieldIndexName(field): indexFunc})
218217
}
219-
220-
type informerWrapper struct {
221-
cache.SharedIndexInformer
222-
}
223-
224-
func (iw *informerWrapper) AddEventHandler(handler cache.ResourceEventHandler) {
225-
_, _ = iw.SharedIndexInformer.AddEventHandler(handler)
226-
}
227-
228-
func (iw *informerWrapper) AddEventHandlerWithResyncPeriod(handler cache.ResourceEventHandler, resyncPeriod time.Duration) {
229-
_, _ = iw.SharedIndexInformer.AddEventHandlerWithResyncPeriod(handler, resyncPeriod)
230-
}
231-
232-
// WrapInformer is a temporary wrapper to make Informer compatible with the SharedIndexInformer in client-go v0.26.0
233-
func WrapInformer(i cache.SharedIndexInformer) Informer {
234-
return &informerWrapper{SharedIndexInformer: i}
235-
}

‎pkg/cache/informertest/fake_cache.go

Lines changed: 2 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -47,11 +47,7 @@ func (c *FakeInformers) GetInformerForKind(ctx context.Context, gvk schema.Group
4747
if err != nil {
4848
return nil, err
4949
}
50-
i, err := c.informerFor(gvk, obj)
51-
if err != nil {
52-
return nil, err
53-
}
54-
return cache.WrapInformer(i), nil
50+
return c.informerFor(gvk, obj)
5551
}
5652

5753
// FakeInformerForKind implements Informers.
@@ -80,11 +76,7 @@ func (c *FakeInformers) GetInformer(ctx context.Context, obj client.Object) (cac
8076
return nil, err
8177
}
8278
gvk := gvks[0]
83-
i, err := c.informerFor(gvk, obj)
84-
if err != nil {
85-
return nil, err
86-
}
87-
return cache.WrapInformer(i), nil
79+
return c.informerFor(gvk, obj)
8880
}
8981

9082
// WaitForCacheSync implements Informers.

‎pkg/cache/multi_namespace_cache.go

Lines changed: 36 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -296,17 +296,47 @@ type multiNamespaceInformer struct {
296296
var _ Informer = &multiNamespaceInformer{}
297297

298298
// AddEventHandler adds the handler to each namespaced informer.
299-
func (i *multiNamespaceInformer) AddEventHandler(handler toolscache.ResourceEventHandler) {
300-
for _, informer := range i.namespaceToInformer {
301-
informer.AddEventHandler(handler)
299+
func (i *multiNamespaceInformer) AddEventHandler(handler toolscache.ResourceEventHandler) (toolscache.ResourceEventHandlerRegistration, error) {
300+
handles := make(map[string]toolscache.ResourceEventHandlerRegistration, len(i.namespaceToInformer))
301+
for ns, informer := range i.namespaceToInformer {
302+
registration, err := informer.AddEventHandler(handler)
303+
if err != nil {
304+
return nil, err
305+
}
306+
handles[ns] = registration
302307
}
308+
return handles, nil
303309
}
304310

305311
// AddEventHandlerWithResyncPeriod adds the handler with a resync period to each namespaced informer.
306-
func (i *multiNamespaceInformer) AddEventHandlerWithResyncPeriod(handler toolscache.ResourceEventHandler, resyncPeriod time.Duration) {
307-
for _, informer := range i.namespaceToInformer {
308-
informer.AddEventHandlerWithResyncPeriod(handler, resyncPeriod)
312+
func (i *multiNamespaceInformer) AddEventHandlerWithResyncPeriod(handler toolscache.ResourceEventHandler, resyncPeriod time.Duration) (toolscache.ResourceEventHandlerRegistration, error) {
313+
handles := make(map[string]toolscache.ResourceEventHandlerRegistration, len(i.namespaceToInformer))
314+
for ns, informer := range i.namespaceToInformer {
315+
registration, err := informer.AddEventHandlerWithResyncPeriod(handler, resyncPeriod)
316+
if err != nil {
317+
return nil, err
318+
}
319+
handles[ns] = registration
320+
}
321+
return handles, nil
322+
}
323+
324+
// RemoveEventHandler removes a formerly added event handler given by its registration handle.
325+
func (i *multiNamespaceInformer) RemoveEventHandler(h toolscache.ResourceEventHandlerRegistration) error {
326+
handles, ok := h.(map[string]toolscache.ResourceEventHandlerRegistration)
327+
if !ok {
328+
return fmt.Errorf("it is not the registration returned by multiNamespaceInformer")
309329
}
330+
for ns, informer := range i.namespaceToInformer {
331+
registration, ok := handles[ns]
332+
if !ok {
333+
continue
334+
}
335+
if err := informer.RemoveEventHandler(registration); err != nil {
336+
return err
337+
}
338+
}
339+
return nil
310340
}
311341

312342
// AddIndexers adds the indexer for each namespaced informer.

‎pkg/source/source.go

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -155,7 +155,11 @@ func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue w
155155
return
156156
}
157157

158-
i.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
158+
_, err := i.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
159+
if err != nil {
160+
ks.started <- err
161+
return
162+
}
159163
if !ks.cache.WaitForCacheSync(ctx) {
160164
// Would be great to return something more informative here
161165
ks.started <- errors.New("cache did not sync")
@@ -351,7 +355,10 @@ func (is *Informer) Start(ctx context.Context, handler handler.EventHandler, que
351355
return fmt.Errorf("must specify Informer.Informer")
352356
}
353357

354-
is.Informer.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
358+
_, err := is.Informer.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
359+
if err != nil {
360+
return err
361+
}
355362
return nil
356363
}
357364

‎pkg/source/source_integration_test.go

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@ import (
2020
"fmt"
2121
"time"
2222

23-
"sigs.k8s.io/controller-runtime/pkg/cache"
2423
"sigs.k8s.io/controller-runtime/pkg/client"
2524
"sigs.k8s.io/controller-runtime/pkg/event"
2625
"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -245,7 +244,7 @@ var _ = Describe("Source", func() {
245244
c := make(chan struct{})
246245

247246
q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
248-
instance := &source.Informer{Informer: cache.WrapInformer(depInformer)}
247+
instance := &source.Informer{Informer: depInformer}
249248
err := instance.Start(ctx, handler.Funcs{
250249
CreateFunc: func(evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
251250
defer GinkgoRecover()
@@ -286,7 +285,7 @@ var _ = Describe("Source", func() {
286285
rs2.SetLabels(map[string]string{"biz": "baz"})
287286

288287
q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
289-
instance := &source.Informer{Informer: cache.WrapInformer(depInformer)}
288+
instance := &source.Informer{Informer: depInformer}
290289
err = instance.Start(ctx, handler.Funcs{
291290
CreateFunc: func(evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
292291
},
@@ -323,7 +322,7 @@ var _ = Describe("Source", func() {
323322
c := make(chan struct{})
324323

325324
q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
326-
instance := &source.Informer{Informer: cache.WrapInformer(depInformer)}
325+
instance := &source.Informer{Informer: depInformer}
327326
err := instance.Start(ctx, handler.Funcs{
328327
CreateFunc: func(event.CreateEvent, workqueue.RateLimitingInterface) {
329328
},

0 commit comments

Comments
 (0)