implement the ChannelSource

Author: lichuqiang

pkg/controller/manager.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/kubernetes-sigs/controller-runtime/pkg/client"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/client/config"
+	"github.com/kubernetes-sigs/controller-runtime/pkg/controller/event"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/controller/reconcile"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/internal/apiutil"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/internal/informer"
@@ -74,6 +75,9 @@ type controllerManager struct {
 	// informers are injected into Controllers (,and transitively EventHandlers, Sources and Predicates).
 	informers informer.Informers
 
+	// outerChan is a channel to receive events originating outside the cluster (e.g. GitHub Webhook callback)
+	outerChan chan event.GenericEvent
+
 	// TODO(directxman12): Provide an escape hatch to get individual indexers
 	// client is the client injected into Controllers (and EventHandlers, Sources and Predicates).
 	client client.Interface
@@ -139,6 +143,12 @@ func (cm *controllerManager) injectInto(i interface{}) error {
 	if _, err := inject.DoInformers(cm.informers, i); err != nil {
 		return err
 	}
+	if _, err := inject.DoSourceChannel(cm.outerChan, i); err != nil {
+		return err
+	}
+	if _, err := inject.DoStop(cm.stop, i); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -196,11 +206,14 @@ type ManagerArgs struct {
 	// Scheme is the scheme used to resolve runtime.Objects to GroupVersionKinds / Resources
 	// Defaults to the kubernetes/client-go scheme.Scheme
 	Scheme *runtime.Scheme
+
+	// OuterChan is a channel to receive events originating outside the cluster (e.g. GitHub Webhook callback)
+	OuterChan chan event.GenericEvent
 }
 
 // NewManager returns a new fully initialized Manager.
 func NewManager(args ManagerArgs) (Manager, error) {
-	cm := &controllerManager{config: args.Config, scheme: args.Scheme, errChan: make(chan error)}
+	cm := &controllerManager{config: args.Config, scheme: args.Scheme, outerChan: args.OuterChan, errChan: make(chan error)}
 
 	// Initialize a rest.config if none was specified
 	if cm.config == nil {

pkg/controller/source/example_test.go

@@ -18,7 +18,6 @@ package source_test
 
 import (
 	"github.com/kubernetes-sigs/controller-runtime/pkg/controller"
-	"github.com/kubernetes-sigs/controller-runtime/pkg/controller/event"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/controller/eventhandler"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/controller/source"
 	"k8s.io/api/core/v1"
@@ -38,10 +37,8 @@ func ExampleKindSource() {
 // This example reads GenericEvents from a channel and enqueues a reconcile.Request containing the Name and Namespace
 // provided by the event.
 func ExampleChannelSource() {
-	events := make(chan event.GenericEvent)
-
 	ctrl.Watch(
-		source.ChannelSource(events),
+		&source.ChannelSource{},
 		&eventhandler.EnqueueHandler{},
 	)
 }

pkg/controller/source/source.go

@@ -18,6 +18,7 @@ package source
 
 import (
 	"fmt"
+	"sync"
 
 	"github.com/kubernetes-sigs/controller-runtime/pkg/controller/event"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/controller/eventhandler"
@@ -31,6 +32,11 @@ import (
 	logf "github.com/kubernetes-sigs/controller-runtime/pkg/runtime/log"
 )
 
+const (
+	// initialBufferSize is the initial number of event notifications that can be buffered.
+	initialBufferSize = 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.
 //
@@ -46,18 +52,113 @@ type Source interface {
 	Start(eventhandler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
 }
 
+var _ Source = &ChannelSource{}
+
 // ChannelSource is used to provide a source of events originating outside the cluster
 // (eh.g. GitHub Webhook callback).  ChannelSource requires the user to wire the external
 // source (eh.g. http handler) to write GenericEvents to the underlying channel.
-type ChannelSource chan event.GenericEvent
+type ChannelSource 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
+	// stop is to end ongoing goroutine, and close the channels
+	stop <-chan struct{}
+	// dest is the destination channels of the added event handlers
+	dest []chan event.GenericEvent
+	// destLock is to ensure the destination channels are safely added/removed
+	destLock sync.Mutex
+}
+
+var _ inject.SourceChannel = &ChannelSource{}
+
+// InjectSourceChannel is internal should be called only by the Controller.
+// It is used to inject the source event channel initialized by the ControllerManager.
+func (cs *ChannelSource) InjectSourceChannel(c chan event.GenericEvent) error {
+	if cs.source == nil {
+		cs.source = c
+	}
+
+	return nil
+}
 
-var _ Source = ChannelSource(make(chan event.GenericEvent))
+var _ inject.Stop = &ChannelSource{}
+
+// InjectStop is internal should be called only by the Controller.
+// It is used to inject the stop channel initialized by the ControllerManager.
+func (cs *ChannelSource) InjectStop(stop <-chan struct{}) error {
+	if cs.stop == nil {
+		cs.stop = stop
+	}
+
+	return nil
+}
 
 // Start implements Source and should only be called by the Controller.
-func (ks ChannelSource) Start(
+func (cs *ChannelSource) Start(
 	handler eventhandler.EventHandler,
 	queue workqueue.RateLimitingInterface,
 	prct ...predicate.Predicate) error {
+	// source should have been injected before Start was called.
+	if cs.source == nil {
+		return fmt.Errorf("must call DoSourceChannel on ChannelSource before calling Start")
+	}
+
+	// stop should have been injected before Start was called
+	if cs.stop == nil {
+		return fmt.Errorf("must call DoStop on ChannelSource before calling Start")
+	}
+
+	cs.once.Do(func() {
+		// Distribute GenericEvents to all EventHandler / Queue pairs Watching this source
+		go func() {
+			for {
+				select {
+				case <-cs.stop:
+					// Close source channel.
+					// TODO: should we close the source channel here?
+					close(cs.source)
+
+					// Close destination channels
+					cs.destLock.Lock()
+					for _, dst := range cs.dest {
+						close(dst)
+					}
+					cs.destLock.Unlock()
+					return
+				case evt := <-cs.source:
+					cs.destLock.Lock()
+					for _, dst := range cs.dest {
+						dst <- evt
+					}
+					cs.destLock.Unlock()
+				}
+			}
+		}()
+	})
+
+	dst := make(chan event.GenericEvent, initialBufferSize)
+	go func() {
+		for evt := range dst {
+			shouldHandle := true
+			for _, p := range prct {
+				if !p.Generic(evt) {
+					shouldHandle = false
+					break
+				}
+			}
+
+			if shouldHandle {
+				handler.Generic(queue, evt)
+			}
+		}
+	}()
+
+	cs.destLock.Lock()
+	defer cs.destLock.Unlock()
+
+	cs.dest = append(cs.dest, dst)
+
 	return nil
 }
 

pkg/controller/source/source_test.go

@@ -26,9 +26,10 @@ import (
 	"github.com/kubernetes-sigs/controller-runtime/pkg/runtime/inject"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-	"k8s.io/client-go/util/workqueue"
 
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/util/workqueue"
 )
 
 var _ = Describe("Source", func() {
@@ -203,4 +204,57 @@ var _ = Describe("Source", func() {
 			})
 		})
 	})
+	Describe("ChannelSource", func() {
+		var c chan struct{}
+		var ch chan event.GenericEvent
+
+		BeforeEach(func() {
+			c = make(chan struct{})
+			ch = make(chan event.GenericEvent)
+		})
+
+		Context("for a GenericEvent", func() {
+			It("should provide a GenericEvent", func(done Done) {
+				p := &corev1.Pod{
+					ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "bar"},
+				}
+				evt := event.GenericEvent{
+					Object: p,
+					Meta:   p.GetObjectMeta(),
+				}
+
+				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
+				instance := &source.ChannelSource{}
+				inject.DoStop(c, instance)
+				inject.DoSourceChannel(ch, instance)
+				err := instance.Start(eventhandler.Funcs{
+					CreateFunc: func(workqueue.RateLimitingInterface, event.CreateEvent) {
+						defer GinkgoRecover()
+						Fail("Unexpected CreateEvent")
+					},
+					UpdateFunc: func(workqueue.RateLimitingInterface, event.UpdateEvent) {
+						defer GinkgoRecover()
+						Fail("Unexpected UpdateEvent")
+					},
+					DeleteFunc: func(workqueue.RateLimitingInterface, event.DeleteEvent) {
+						defer GinkgoRecover()
+						Fail("Unexpected DeleteEvent")
+					},
+					GenericFunc: func(q2 workqueue.RateLimitingInterface, evt event.GenericEvent) {
+						defer GinkgoRecover()
+						Expect(q2).To(Equal(q))
+						Expect(evt.Meta).To(Equal(p.GetObjectMeta()))
+						Expect(evt.Object).To(Equal(p))
+						close(c)
+					},
+				}, q)
+				Expect(err).NotTo(HaveOccurred())
+
+				ch <- evt
+				<-c
+				close(done)
+			})
+
+		})
+	})
 })

pkg/runtime/inject/inject.go

@@ -18,6 +18,7 @@ package inject
 
 import (
 	"github.com/kubernetes-sigs/controller-runtime/pkg/client"
+	"github.com/kubernetes-sigs/controller-runtime/pkg/controller/event"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/internal/informer"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
@@ -82,3 +83,33 @@ func DoScheme(scheme *runtime.Scheme, i interface{}) (bool, error) {
 	}
 	return false, nil
 }
+
+// SourceChannel is used by the ControllerManager to inject channel into Sources, EventHandlers, Predicates, and
+// Reconciles
+type SourceChannel interface {
+	InjectSourceChannel(chan event.GenericEvent) error
+}
+
+// DoSourceChannel will set source channel on i and return the result if it implements SourceChannel.
+// Returns false if i does not implement SourceChannel.
+func DoSourceChannel(ch chan event.GenericEvent, i interface{}) (bool, error) {
+	if c, ok := i.(SourceChannel); ok {
+		return true, c.InjectSourceChannel(ch)
+	}
+	return false, nil
+}
+
+// Stop is used by the ControllerManager to inject stop channel into Sources, EventHandlers, Predicates, and
+// Reconciles
+type Stop interface {
+	InjectStop(<-chan struct{}) error
+}
+
+// DoStop will set stop channel on i and return the result if it implements Stop.
+// Returns false if i does not implement Stop.
+func DoStop(stop <-chan struct{}, i interface{}) (bool, error) {
+	if s, ok := i.(Stop); ok {
+		return true, s.InjectStop(stop)
+	}
+	return false, nil
+}