Skip to content

Commit ae0f6ab

Browse files
authored
Merge pull request #2783 from alvaroaleman/compatible-generics
⚠️ Source, Event, Predicate, Handler: Add generics support
2 parents a92b961 + 2add01e commit ae0f6ab

24 files changed

+773
-614
lines changed

examples/builtins/main.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -59,14 +59,14 @@ func main() {
5959
}
6060

6161
// Watch ReplicaSets and enqueue ReplicaSet object key
62-
if err := c.Watch(source.Kind(mgr.GetCache(), &appsv1.ReplicaSet{}), &handler.EnqueueRequestForObject{}); err != nil {
62+
if err := c.Watch(source.Kind(mgr.GetCache(), &appsv1.ReplicaSet{}, &handler.TypedEnqueueRequestForObject[*appsv1.ReplicaSet]{})); err != nil {
6363
entryLog.Error(err, "unable to watch ReplicaSets")
6464
os.Exit(1)
6565
}
6666

6767
// Watch Pods and enqueue owning ReplicaSet key
68-
if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}),
69-
handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &appsv1.ReplicaSet{}, handler.OnlyControllerOwner())); err != nil {
68+
if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{},
69+
handler.TypedEnqueueRequestForOwner[*corev1.Pod](mgr.GetScheme(), mgr.GetRESTMapper(), &appsv1.ReplicaSet{}, handler.OnlyControllerOwner()))); err != nil {
7070
entryLog.Error(err, "unable to watch Pods")
7171
os.Exit(1)
7272
}

pkg/builder/controller.go

Lines changed: 34 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@ import (
3030
"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
3131
"sigs.k8s.io/controller-runtime/pkg/controller"
3232
"sigs.k8s.io/controller-runtime/pkg/handler"
33-
internalsource "sigs.k8s.io/controller-runtime/pkg/internal/source"
3433
"sigs.k8s.io/controller-runtime/pkg/manager"
3534
"sigs.k8s.io/controller-runtime/pkg/predicate"
3635
"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -56,6 +55,7 @@ const (
5655
type Builder struct {
5756
forInput ForInput
5857
ownsInput []OwnsInput
58+
rawSources []source.Source
5959
watchesInput []WatchesInput
6060
mgr manager.Manager
6161
globalPredicates []predicate.Predicate
@@ -123,8 +123,8 @@ func (blder *Builder) Owns(object client.Object, opts ...OwnsOption) *Builder {
123123

124124
// WatchesInput represents the information set by Watches method.
125125
type WatchesInput struct {
126-
src source.Source
127-
eventHandler handler.EventHandler
126+
obj client.Object
127+
handler handler.EventHandler
128128
predicates []predicate.Predicate
129129
objectProjection objectProjection
130130
}
@@ -133,10 +133,19 @@ type WatchesInput struct {
133133
// update events by *reconciling the object* with the given EventHandler.
134134
//
135135
// This is the equivalent of calling
136-
// WatchesRawSource(source.Kind(cache, object), eventHandler, opts...).
136+
// WatchesRawSource(source.Kind(cache, object, eventHandler, predicates...)).
137137
func (blder *Builder) Watches(object client.Object, eventHandler handler.EventHandler, opts ...WatchesOption) *Builder {
138-
src := source.Kind(blder.mgr.GetCache(), object)
139-
return blder.WatchesRawSource(src, eventHandler, opts...)
138+
input := WatchesInput{
139+
obj: object,
140+
handler: eventHandler,
141+
}
142+
for _, opt := range opts {
143+
opt.ApplyToWatches(&input)
144+
}
145+
146+
blder.watchesInput = append(blder.watchesInput, input)
147+
148+
return blder
140149
}
141150

142151
// WatchesMetadata is the same as Watches, but forces the internal cache to only watch PartialObjectMetadata.
@@ -176,13 +185,11 @@ func (blder *Builder) WatchesMetadata(object client.Object, eventHandler handler
176185
//
177186
// STOP! Consider using For(...), Owns(...), Watches(...), WatchesMetadata(...) instead.
178187
// This method is only exposed for more advanced use cases, most users should use one of the higher level functions.
179-
func (blder *Builder) WatchesRawSource(src source.Source, eventHandler handler.EventHandler, opts ...WatchesOption) *Builder {
180-
input := WatchesInput{src: src, eventHandler: eventHandler}
181-
for _, opt := range opts {
182-
opt.ApplyToWatches(&input)
183-
}
188+
//
189+
// WatchesRawSource does not respect predicates configured through WithEventFilter.
190+
func (blder *Builder) WatchesRawSource(src source.Source) *Builder {
191+
blder.rawSources = append(blder.rawSources, src)
184192

185-
blder.watchesInput = append(blder.watchesInput, input)
186193
return blder
187194
}
188195

@@ -272,11 +279,11 @@ func (blder *Builder) doWatch() error {
272279
if err != nil {
273280
return err
274281
}
275-
src := source.Kind(blder.mgr.GetCache(), obj)
276282
hdler := &handler.EnqueueRequestForObject{}
277283
allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
278284
allPredicates = append(allPredicates, blder.forInput.predicates...)
279-
if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {
285+
src := source.Kind(blder.mgr.GetCache(), obj, hdler, allPredicates...)
286+
if err := blder.ctrl.Watch(src); err != nil {
280287
return err
281288
}
282289
}
@@ -290,7 +297,6 @@ func (blder *Builder) doWatch() error {
290297
if err != nil {
291298
return err
292299
}
293-
src := source.Kind(blder.mgr.GetCache(), obj)
294300
opts := []handler.OwnerOption{}
295301
if !own.matchEveryOwner {
296302
opts = append(opts, handler.OnlyControllerOwner())
@@ -302,27 +308,29 @@ func (blder *Builder) doWatch() error {
302308
)
303309
allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
304310
allPredicates = append(allPredicates, own.predicates...)
305-
if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {
311+
src := source.Kind(blder.mgr.GetCache(), obj, hdler, allPredicates...)
312+
if err := blder.ctrl.Watch(src); err != nil {
306313
return err
307314
}
308315
}
309316

310317
// Do the watch requests
311-
if len(blder.watchesInput) == 0 && blder.forInput.object == nil {
312-
return errors.New("there are no watches configured, controller will never get triggered. Use For(), Owns() or Watches() to set them up")
318+
if len(blder.watchesInput) == 0 && blder.forInput.object == nil && len(blder.rawSources) == 0 {
319+
return errors.New("there are no watches configured, controller will never get triggered. Use For(), Owns(), Watches() or WatchesRawSource() to set them up")
313320
}
314321
for _, w := range blder.watchesInput {
315-
// If the source of this watch is of type Kind, project it.
316-
if srcKind, ok := w.src.(*internalsource.Kind); ok {
317-
typeForSrc, err := blder.project(srcKind.Type, w.objectProjection)
318-
if err != nil {
319-
return err
320-
}
321-
srcKind.Type = typeForSrc
322+
projected, err := blder.project(w.obj, w.objectProjection)
323+
if err != nil {
324+
return fmt.Errorf("failed to project for %T: %w", w.obj, err)
322325
}
323326
allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
324327
allPredicates = append(allPredicates, w.predicates...)
325-
if err := blder.ctrl.Watch(w.src, w.eventHandler, allPredicates...); err != nil {
328+
if err := blder.ctrl.Watch(source.Kind(blder.mgr.GetCache(), projected, w.handler, allPredicates...)); err != nil {
329+
return err
330+
}
331+
}
332+
for _, src := range blder.rawSources {
333+
if err := blder.ctrl.Watch(src); err != nil {
326334
return err
327335
}
328336
}

pkg/builder/controller_test.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -146,7 +146,7 @@ var _ = Describe("application", func() {
146146
instance, err := ControllerManagedBy(m).
147147
Named("my_controller").
148148
Build(noop)
149-
Expect(err).To(MatchError(ContainSubstring("there are no watches configured, controller will never get triggered. Use For(), Owns() or Watches() to set them up")))
149+
Expect(err).To(MatchError(ContainSubstring("there are no watches configured, controller will never get triggered. Use For(), Owns(), Watches() or WatchesRawSource() to set them up")))
150150
Expect(instance).To(BeNil())
151151
})
152152

pkg/controller/controller.go

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -25,10 +25,8 @@ import (
2525
"k8s.io/client-go/util/workqueue"
2626
"k8s.io/klog/v2"
2727

28-
"sigs.k8s.io/controller-runtime/pkg/handler"
2928
"sigs.k8s.io/controller-runtime/pkg/internal/controller"
3029
"sigs.k8s.io/controller-runtime/pkg/manager"
31-
"sigs.k8s.io/controller-runtime/pkg/predicate"
3230
"sigs.k8s.io/controller-runtime/pkg/ratelimiter"
3331
"sigs.k8s.io/controller-runtime/pkg/reconcile"
3432
"sigs.k8s.io/controller-runtime/pkg/source"
@@ -84,13 +82,8 @@ type Controller interface {
8482
// Reconciler is called to reconcile an object by Namespace/Name
8583
reconcile.Reconciler
8684

87-
// Watch takes events provided by a Source and uses the EventHandler to
88-
// enqueue reconcile.Requests in response to the events.
89-
//
90-
// Watch may be provided one or more Predicates to filter events before
91-
// they are given to the EventHandler. Events will be passed to the
92-
// EventHandler if all provided Predicates evaluate to true.
93-
Watch(src source.Source, eventhandler handler.EventHandler, predicates ...predicate.Predicate) error
85+
// Watch watches the provided Source.
86+
Watch(src source.Source) error
9487

9588
// Start starts the controller. Start blocks until the context is closed or a
9689
// controller has an error starting.

pkg/controller/controller_integration_test.go

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -65,12 +65,13 @@ var _ = Describe("controller", func() {
6565

6666
By("Watching Resources")
6767
err = instance.Watch(
68-
source.Kind(cm.GetCache(), &appsv1.ReplicaSet{}),
69-
handler.EnqueueRequestForOwner(cm.GetScheme(), cm.GetRESTMapper(), &appsv1.Deployment{}),
68+
source.Kind(cm.GetCache(), &appsv1.ReplicaSet{},
69+
handler.TypedEnqueueRequestForOwner[*appsv1.ReplicaSet](cm.GetScheme(), cm.GetRESTMapper(), &appsv1.Deployment{}),
70+
),
7071
)
7172
Expect(err).NotTo(HaveOccurred())
7273

73-
err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}), &handler.EnqueueRequestForObject{})
74+
err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}, &handler.TypedEnqueueRequestForObject[*appsv1.Deployment]{}))
7475
Expect(err).NotTo(HaveOccurred())
7576

7677
err = cm.GetClient().Get(ctx, types.NamespacedName{Name: "foo"}, &corev1.Namespace{})

pkg/controller/controller_test.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ var _ = Describe("controller.Controller", func() {
7979

8080
ctx, cancel := context.WithCancel(context.Background())
8181
watchChan := make(chan event.GenericEvent, 1)
82-
watch := &source.Channel{Source: watchChan}
82+
watch := source.Channel(watchChan, &handler.EnqueueRequestForObject{})
8383
watchChan <- event.GenericEvent{Object: &corev1.Pod{}}
8484

8585
reconcileStarted := make(chan struct{})
@@ -101,7 +101,7 @@ var _ = Describe("controller.Controller", func() {
101101
Expect(err).NotTo(HaveOccurred())
102102

103103
c, err := controller.New("new-controller", m, controller.Options{Reconciler: rec})
104-
Expect(c.Watch(watch, &handler.EnqueueRequestForObject{})).To(Succeed())
104+
Expect(c.Watch(watch)).To(Succeed())
105105
Expect(err).NotTo(HaveOccurred())
106106

107107
go func() {

pkg/controller/example_test.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -71,7 +71,7 @@ func ExampleController() {
7171
}
7272

7373
// Watch for Pod create / update / delete events and call Reconcile
74-
err = c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}), &handler.EnqueueRequestForObject{})
74+
err = c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.TypedEnqueueRequestForObject[*corev1.Pod]{}))
7575
if err != nil {
7676
log.Error(err, "unable to watch pods")
7777
os.Exit(1)
@@ -108,7 +108,7 @@ func ExampleController_unstructured() {
108108
Version: "v1",
109109
})
110110
// Watch for Pod create / update / delete events and call Reconcile
111-
err = c.Watch(source.Kind(mgr.GetCache(), u), &handler.EnqueueRequestForObject{})
111+
err = c.Watch(source.Kind(mgr.GetCache(), u, &handler.TypedEnqueueRequestForObject[*unstructured.Unstructured]{}))
112112
if err != nil {
113113
log.Error(err, "unable to watch pods")
114114
os.Exit(1)
@@ -139,7 +139,7 @@ func ExampleNewUnmanaged() {
139139
os.Exit(1)
140140
}
141141

142-
if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}), &handler.EnqueueRequestForObject{}); err != nil {
142+
if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.TypedEnqueueRequestForObject[*corev1.Pod]{})); err != nil {
143143
log.Error(err, "unable to watch pods")
144144
os.Exit(1)
145145
}

pkg/event/event.go

Lines changed: 34 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -18,38 +18,55 @@ package event
1818

1919
import "sigs.k8s.io/controller-runtime/pkg/client"
2020

21-
// CreateEvent is an event where a Kubernetes object was created. CreateEvent should be generated
21+
// CreateEvent is an event where a Kubernetes object was created. CreateEvent should be generated
22+
// by a source.Source and transformed into a reconcile.Request by a handler.EventHandler.
23+
type CreateEvent = TypedCreateEvent[client.Object]
24+
25+
// UpdateEvent is an event where a Kubernetes object was updated. UpdateEvent should be generated
26+
// by a source.Source and transformed into a reconcile.Request by an handler.EventHandler.
27+
type UpdateEvent = TypedUpdateEvent[client.Object]
28+
29+
// DeleteEvent is an event where a Kubernetes object was deleted. DeleteEvent should be generated
2230
// by a source.Source and transformed into a reconcile.Request by an handler.EventHandler.
23-
type CreateEvent struct {
31+
type DeleteEvent = TypedDeleteEvent[client.Object]
32+
33+
// GenericEvent is an event where the operation type is unknown (e.g. polling or event originating outside the cluster).
34+
// GenericEvent should be generated by a source.Source and transformed into a reconcile.Request by an
35+
// handler.EventHandler.
36+
type GenericEvent = TypedGenericEvent[client.Object]
37+
38+
// TypedCreateEvent is an event where a Kubernetes object was created. TypedCreateEvent should be generated
39+
// by a source.Source and transformed into a reconcile.Request by an handler.TypedEventHandler.
40+
type TypedCreateEvent[T any] struct {
2441
// Object is the object from the event
25-
Object client.Object
42+
Object T
2643
}
2744

28-
// UpdateEvent is an event where a Kubernetes object was updated. UpdateEvent should be generated
29-
// by a source.Source and transformed into a reconcile.Request by an handler.EventHandler.
30-
type UpdateEvent struct {
45+
// TypedUpdateEvent is an event where a Kubernetes object was updated. TypedUpdateEvent should be generated
46+
// by a source.Source and transformed into a reconcile.Request by an handler.TypedEventHandler.
47+
type TypedUpdateEvent[T any] struct {
3148
// ObjectOld is the object from the event
32-
ObjectOld client.Object
49+
ObjectOld T
3350

3451
// ObjectNew is the object from the event
35-
ObjectNew client.Object
52+
ObjectNew T
3653
}
3754

38-
// DeleteEvent is an event where a Kubernetes object was deleted. DeleteEvent should be generated
39-
// by a source.Source and transformed into a reconcile.Request by an handler.EventHandler.
40-
type DeleteEvent struct {
55+
// TypedDeleteEvent is an event where a Kubernetes object was deleted. TypedDeleteEvent should be generated
56+
// by a source.Source and transformed into a reconcile.Request by an handler.TypedEventHandler.
57+
type TypedDeleteEvent[T any] struct {
4158
// Object is the object from the event
42-
Object client.Object
59+
Object T
4360

4461
// DeleteStateUnknown is true if the Delete event was missed but we identified the object
4562
// as having been deleted.
4663
DeleteStateUnknown bool
4764
}
4865

49-
// GenericEvent is an event where the operation type is unknown (e.g. polling or event originating outside the cluster).
50-
// GenericEvent should be generated by a source.Source and transformed into a reconcile.Request by an
51-
// handler.EventHandler.
52-
type GenericEvent struct {
66+
// TypedGenericEvent is an event where the operation type is unknown (e.g. polling or event originating outside the cluster).
67+
// TypedGenericEvent should be generated by a source.Source and transformed into a reconcile.Request by an
68+
// handler.TypedEventHandler.
69+
type TypedGenericEvent[T any] struct {
5370
// Object is the object from the event
54-
Object client.Object
71+
Object T
5572
}

0 commit comments

Comments
 (0)