Skip to content

Commit

Permalink
Make event handler type aware
Browse files Browse the repository at this point in the history
Signed-off-by: Danil Grigorev <danil.grigorev@suse.com>
  • Loading branch information
Danil-Grigorev committed Feb 29, 2024
1 parent 5615941 commit d6d1d93
Show file tree
Hide file tree
Showing 4 changed files with 458 additions and 9 deletions.
17 changes: 9 additions & 8 deletions pkg/internal/source/kind.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"reflect"
"time"

"k8s.io/apimachinery/pkg/api/meta"
Expand All @@ -17,9 +18,9 @@ import (
)

// Kind is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create).
type Kind struct {
type Kind[T client.Object] struct {
// Type is the type of object to watch. e.g. &v1.Pod{}
Type client.Object
Type T

// Cache used to watch APIs
Cache cache.Cache
Expand All @@ -32,9 +33,9 @@ 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,
func (ks *Kind[T]) Start(ctx context.Context, handler handler.EventHandler, queue workqueue.RateLimitingInterface,
prct ...predicate.Predicate) error {
if ks.Type == nil {
if reflect.DeepEqual(ks.Type, *new(T)) {
return fmt.Errorf("must create Kind with a non-nil object")
}
if ks.Cache == nil {
Expand Down Expand Up @@ -79,7 +80,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(NewObjectEventHandler[T](ctx, queue, handler, prct).HandlerFuncs())
if err != nil {
ks.started <- err
return
Expand All @@ -94,16 +95,16 @@ func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue w
return nil
}

func (ks *Kind) String() string {
if ks.Type != nil {
func (ks *Kind[T]) String() string {
if !reflect.DeepEqual(ks.Type, *new(T)) {
return fmt.Sprintf("kind source: %T", ks.Type)
}
return "kind source: unknown type"
}

// WaitForSync implements SyncingSource to allow controllers to wait with starting
// workers until the cache is synced.
func (ks *Kind) WaitForSync(ctx context.Context) error {
func (ks *Kind[T]) WaitForSync(ctx context.Context) error {
select {
case err := <-ks.started:
return err
Expand Down
167 changes: 167 additions & 0 deletions pkg/internal/source/object_event_handler.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,167 @@
/*
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/tools/cache"
"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"

"sigs.k8s.io/controller-runtime/pkg/predicate"
)

// NewObjectEventHandler creates a new TypedEventHandler.
func NewObjectEventHandler[T client.Object](ctx context.Context, queue workqueue.RateLimitingInterface, handler handler.EventHandler, predicates []predicate.Predicate) *ObjectEventHandler[T] {
return &ObjectEventHandler[T]{
ctx: ctx,
handler: handler,
queue: queue,
predicates: predicates,
}
}

// ObjectEventHandler adapts a handler.EventHandler interface to a cache.ResourceEventHandler interface.
type ObjectEventHandler[T client.Object] struct {
// ctx stores the context that created the event handler
// that is used to propagate cancellation signals to each handler function.
ctx context.Context

handler handler.EventHandler
queue workqueue.RateLimitingInterface
predicates []predicate.Predicate
}

// HandlerFuncs converts EventHandler to a ResourceEventHandlerFuncs
// TODO: switch to ResourceEventHandlerDetailedFuncs with client-go 1.27
func (e *ObjectEventHandler[T]) HandlerFuncs() cache.ResourceEventHandlerFuncs {
return cache.ResourceEventHandlerFuncs{
AddFunc: e.OnAdd,
UpdateFunc: e.OnUpdate,
DeleteFunc: e.OnDelete,
}
}

// OnAdd creates CreateEvent and calls Create on EventHandler.
func (e *ObjectEventHandler[T]) OnAdd(obj interface{}) {
c := event.CreateEvent{}

// Pull Object out of the object
if o, ok := obj.(T); ok {
c.Object = o
} else {
log.Error(nil, "OnAdd missing Object",
"object", obj, "type", fmt.Sprintf("%T", obj))
return
}

for _, p := range e.predicates {
if !p.Create(c) {
return
}
}

// Invoke create handler
ctx, cancel := context.WithCancel(e.ctx)
defer cancel()
e.handler.Create(ctx, c, e.queue)
}

// OnUpdate creates UpdateEvent and calls Update on EventHandler.
func (e *ObjectEventHandler[T]) OnUpdate(oldObj, newObj interface{}) {
u := event.UpdateEvent{}

if o, ok := oldObj.(T); ok {
u.ObjectOld = o
} else {
log.Error(nil, "OnUpdate missing ObjectOld",
"object", oldObj, "type", fmt.Sprintf("%T", oldObj))
return
}

// Pull Object out of the object
if o, ok := newObj.(T); ok {
u.ObjectNew = o
} else {
log.Error(nil, "OnUpdate missing ObjectNew",
"object", newObj, "type", fmt.Sprintf("%T", newObj))
return
}

for _, p := range e.predicates {
if !p.Update(u) {
return
}
}

// Invoke update handler
ctx, cancel := context.WithCancel(e.ctx)
defer cancel()
e.handler.Update(ctx, u, e.queue)
}

// OnDelete creates DeleteEvent and calls Delete on EventHandler.
func (e *ObjectEventHandler[T]) OnDelete(obj interface{}) {
d := event.DeleteEvent{}

// Deal with tombstone events by pulling the object out. Tombstone events wrap the object in a
// DeleteFinalStateUnknown struct, so the object needs to be pulled out.
// Copied from sample-controller
// This should never happen if we aren't missing events, which we have concluded that we are not
// and made decisions off of this belief. Maybe this shouldn't be here?
var ok bool
if _, ok = obj.(T); !ok {
// If the object doesn't have Metadata, assume it is a tombstone object of type DeletedFinalStateUnknown
tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
if !ok {
log.Error(nil, "Error decoding objects. Expected cache.DeletedFinalStateUnknown",
"type", fmt.Sprintf("%T", obj),
"object", obj)
return
}

// Set DeleteStateUnknown to true
d.DeleteStateUnknown = true

// Set obj to the tombstone obj
obj = tombstone.Obj
}

// Pull Object out of the object
if o, ok := obj.(T); ok {
d.Object = o
} else {
log.Error(nil, "OnDelete missing Object",
"object", obj, "type", fmt.Sprintf("%T", obj))
return
}

for _, p := range e.predicates {
if !p.Delete(d) {
return
}
}

// Invoke delete handler
ctx, cancel := context.WithCancel(e.ctx)
defer cancel()
e.handler.Delete(ctx, d, e.queue)
}
Loading

0 comments on commit d6d1d93

Please sign in to comment.