move all Source implementations to a single folder

Signed-off-by: Tim Ramlot <[email protected]>

pkg/source/internal/channel.go

@@ -0,0 +1,147 @@
+/*
+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"
+	"errors"
+	"fmt"
+	"sync"
+
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+)
+
+type Channel[T any] 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.TypedGenericEvent[T]
+
+	Handler handler.TypedEventHandler[T]
+
+	Predicates []predicate.TypedPredicate[T]
+
+	BufferSize *int
+
+	// dest is the destination channels of the added event handlers
+	dest []chan event.TypedGenericEvent[T]
+
+	// destLock is to ensure the destination channels are safely added/removed
+	destLock sync.Mutex
+}
+
+func (cs *Channel[T]) String() string {
+	return fmt.Sprintf("channel source: %p", cs)
+}
+
+// Start implements Source and should only be called by the Controller.
+func (cs *Channel[T]) Start(
+	ctx context.Context,
+	queue workqueue.RateLimitingInterface,
+) error {
+	// Source should have been specified by the user.
+	if cs.Source == nil {
+		return fmt.Errorf("must specify Channel.Source")
+	}
+	if cs.Handler == nil {
+		return errors.New("must specify Channel.Handler")
+	}
+
+	if cs.BufferSize == nil {
+		cs.BufferSize = ptr.To(1024)
+	}
+
+	dst := make(chan event.TypedGenericEvent[T], *cs.BufferSize)
+
+	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 cs.Predicates {
+				if !p.Generic(evt) {
+					shouldHandle = false
+					break
+				}
+			}
+
+			if shouldHandle {
+				func() {
+					ctx, cancel := context.WithCancel(ctx)
+					defer cancel()
+					cs.Handler.Generic(ctx, evt, queue)
+				}()
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (cs *Channel[T]) doStop() {
+	cs.destLock.Lock()
+	defer cs.destLock.Unlock()
+
+	for _, dst := range cs.dest {
+		close(dst)
+	}
+}
+
+func (cs *Channel[T]) distribute(evt event.TypedGenericEvent[T]) {
+	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[T]) 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)
+		}
+	}
+}

pkg/source/internal/func.go

@@ -0,0 +1,36 @@
+/*
+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"
+)
+
+// Func is a function that implements Source.
+type Func func(context.Context, workqueue.RateLimitingInterface) error
+
+// Start implements Source.
+func (f Func) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error {
+	return f(ctx, queue)
+}
+
+func (f Func) String() string {
+	return fmt.Sprintf("func source: %p", f)
+}

pkg/source/internal/informer.go

@@ -0,0 +1,58 @@
+/*
+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"
+	"errors"
+	"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
+	Handler    handler.EventHandler
+	Predicates []predicate.Predicate
+}
+
+// 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, queue workqueue.RateLimitingInterface) error {
+	// Informer should have been specified by the user.
+	if is.Informer == nil {
+		return fmt.Errorf("must specify Informer.Informer")
+	}
+	if is.Handler == nil {
+		return errors.New("must specify Informer.Handler")
+	}
+
+	_, err := is.Informer.AddEventHandler(NewEventHandler(ctx, queue, is.Handler, is.Predicates).HandlerFuncs())
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (is *Informer) String() string {
+	return fmt.Sprintf("informer source: %p", is.Informer)
+}

pkg/internal/source/internal_test.go → pkg/source/internal/internal_test.go

@@ -26,7 +26,7 @@ import (
 	"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"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

pkg/internal/source/kind.go → 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 (