Merge pull request #159 from pwittrock/master

Support for client-go code generated controllers as Controller Watch …

Author: k8s-ci-robot

pkg/source/source.go

@@ -27,6 +27,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/runtime/inject"
 	"sigs.k8s.io/controller-runtime/pkg/source/internal"
 
+	toolscache "k8s.io/client-go/tools/cache"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
@@ -39,7 +40,7 @@ const (
 // 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.
 //
-// * Use Kind for events originating in the cluster (eh.g. Pod Create, Pod Update, Deployment Update).
+// * Use Kind for events originating in the cluster (e.g. Pod Create, Pod Update, Deployment Update).
 //
 // * Use Channel for events originating outside the cluster (eh.g. GitHub Webhook callback, Polling external urls).
 //
@@ -51,7 +52,7 @@ type Source interface {
 	Start(handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
 }
 
-// Kind is used to provide a source of events originating inside the cluster from Watches (eh.g. Pod Create)
+// Kind is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create)
 type Kind struct {
 	// Type is the type of object to watch.  e.g. &v1.Pod{}
 	Type runtime.Object
@@ -100,7 +101,7 @@ func (ks *Kind) InjectCache(c cache.Cache) error {
 var _ Source = &Channel{}
 
 // Channel is used to provide a source of events originating outside the cluster
-// (eh.g. GitHub Webhook callback).  Channel requires the user to wire the external
+// (e.g. GitHub Webhook callback).  Channel requires the user to wire the external
 // source (eh.g. http handler) to write GenericEvents to the underlying channel.
 type Channel struct {
 	// once ensures the event distribution goroutine will be performed only once
@@ -221,6 +222,28 @@ func (cs *Channel) syncLoop() {
 	}
 }
 
+// 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 generated client-go Informer
+	Informer toolscache.SharedIndexInformer
+}
+
+var _ Source = &Informer{}
+
+// Start is internal and should be called only by the Controller to register an EventHandler with the Informer
+// to enqueue reconcile.Requests.
+func (ks *Informer) Start(handler handler.EventHandler, queue workqueue.RateLimitingInterface,
+	prct ...predicate.Predicate) error {
+
+	// Informer should have been specified by the user.
+	if ks.Informer == nil {
+		return fmt.Errorf("must specify Informer.Informer")
+	}
+
+	ks.Informer.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
+	return nil
+}
+
 // Func is a function that implements Source
 type Func func(handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
 

pkg/source/source_integration_test.go

@@ -18,6 +18,7 @@ package source_test
 
 import (
 	"fmt"
+	"time"
 
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -30,6 +31,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	kubeinformers "k8s.io/client-go/informers"
+	toolscache "k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 )
 
@@ -209,4 +212,161 @@ var _ = Describe("Source", func() {
 			})
 		})
 	})
+
+	Describe("Informer", func() {
+		var c chan struct{}
+		var rs *appsv1.ReplicaSet
+		var depInformer toolscache.SharedIndexInformer
+		var informerFactory kubeinformers.SharedInformerFactory
+		var stopTest chan struct{}
+
+		BeforeEach(func(done Done) {
+			stopTest = make(chan struct{})
+			informerFactory = kubeinformers.NewSharedInformerFactory(clientset, time.Second*30)
+			depInformer = informerFactory.Apps().V1().ReplicaSets().Informer()
+			informerFactory.Start(stopTest)
+			Eventually(depInformer.HasSynced).Should(BeTrue())
+
+			c = make(chan struct{})
+			rs = &appsv1.ReplicaSet{
+				ObjectMeta: metav1.ObjectMeta{Name: "informer-rs-name"},
+				Spec: appsv1.ReplicaSetSpec{
+					Selector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{"foo": "bar"},
+					},
+					Template: corev1.PodTemplateSpec{
+						ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"foo": "bar"}},
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name:  "nginx",
+									Image: "nginx",
+								},
+							},
+						},
+					},
+				},
+			}
+			close(done)
+		})
+
+		AfterEach(func(done Done) {
+			close(stopTest)
+			close(done)
+		})
+
+		Context("for a ReplicaSet resource", func() {
+			It("should provide a ReplicaSet CreateEvent", func(done Done) {
+				c := make(chan struct{})
+
+				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
+				instance := &source.Informer{Informer: depInformer}
+				err := instance.Start(handler.Funcs{
+					CreateFunc: func(evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						var err error
+						rs, err = clientset.AppsV1().ReplicaSets("default").Get(rs.Name, metav1.GetOptions{})
+						Expect(err).NotTo(HaveOccurred())
+
+						Expect(q2).To(BeIdenticalTo(q))
+						Expect(evt.Meta).To(Equal(rs))
+						Expect(evt.Object).To(Equal(rs))
+						close(c)
+					},
+					UpdateFunc: func(event.UpdateEvent, workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						Fail("Unexpected UpdateEvent")
+					},
+					DeleteFunc: func(event.DeleteEvent, workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						Fail("Unexpected DeleteEvent")
+					},
+					GenericFunc: func(event.GenericEvent, workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						Fail("Unexpected GenericEvent")
+					},
+				}, q)
+				Expect(err).NotTo(HaveOccurred())
+
+				rs, err = clientset.AppsV1().ReplicaSets("default").Create(rs)
+				Expect(err).NotTo(HaveOccurred())
+				<-c
+				close(done)
+			}, 30)
+
+			It("should provide a ReplicaSet UpdateEvent", func(done Done) {
+				var err error
+				rs, err = clientset.AppsV1().ReplicaSets("default").Get(rs.Name, metav1.GetOptions{})
+				Expect(err).NotTo(HaveOccurred())
+
+				rs2 := rs.DeepCopy()
+				rs2.SetLabels(map[string]string{"biz": "baz"})
+
+				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
+				instance := &source.Informer{Informer: depInformer}
+				err = instance.Start(handler.Funcs{
+					CreateFunc: func(evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
+					},
+					UpdateFunc: func(evt event.UpdateEvent, q2 workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						var err error
+						rs2, err = clientset.AppsV1().ReplicaSets("default").Get(rs.Name, metav1.GetOptions{})
+						Expect(err).NotTo(HaveOccurred())
+
+						Expect(q2).To(Equal(q))
+						Expect(evt.MetaOld).To(Equal(rs))
+						Expect(evt.ObjectOld).To(Equal(rs))
+
+						Expect(evt.MetaNew).To(Equal(rs2))
+						Expect(evt.ObjectNew).To(Equal(rs2))
+
+						close(c)
+					},
+					DeleteFunc: func(event.DeleteEvent, workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						Fail("Unexpected DeleteEvent")
+					},
+					GenericFunc: func(event.GenericEvent, workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						Fail("Unexpected GenericEvent")
+					},
+				}, q)
+				Expect(err).NotTo(HaveOccurred())
+
+				rs2, err = clientset.AppsV1().ReplicaSets("default").Update(rs2)
+				Expect(err).NotTo(HaveOccurred())
+				<-c
+				close(done)
+			})
+
+			It("should provide a ReplicaSet DeletedEvent", func(done Done) {
+				c := make(chan struct{})
+
+				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
+				instance := &source.Informer{Informer: depInformer}
+				err := instance.Start(handler.Funcs{
+					CreateFunc: func(event.CreateEvent, workqueue.RateLimitingInterface) {
+					},
+					UpdateFunc: func(event.UpdateEvent, workqueue.RateLimitingInterface) {
+					},
+					DeleteFunc: func(evt event.DeleteEvent, q2 workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						Expect(q2).To(Equal(q))
+						Expect(evt.Meta.GetName()).To(Equal(rs.Name))
+						close(c)
+					},
+					GenericFunc: func(event.GenericEvent, workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						Fail("Unexpected GenericEvent")
+					},
+				}, q)
+				Expect(err).NotTo(HaveOccurred())
+
+				err = clientset.AppsV1().ReplicaSets("default").Delete(rs.Name, &metav1.DeleteOptions{})
+				Expect(err).NotTo(HaveOccurred())
+				<-c
+				close(done)
+			})
+		})
+	})
 })

pkg/source/source_test.go

@@ -116,7 +116,7 @@ var _ = Describe("Source", func() {
 					},
 					UpdateFunc: func(evt event.UpdateEvent, q2 workqueue.RateLimitingInterface) {
 						defer GinkgoRecover()
-						Expect(q2).To(Equal(q))
+						Expect(q2).To(BeIdenticalTo(q))
 						Expect(evt.MetaOld).To(Equal(p))
 						Expect(evt.ObjectOld).To(Equal(p))
 
@@ -170,7 +170,7 @@ var _ = Describe("Source", func() {
 					},
 					DeleteFunc: func(evt event.DeleteEvent, q2 workqueue.RateLimitingInterface) {
 						defer GinkgoRecover()
-						Expect(q2).To(Equal(q))
+						Expect(q2).To(BeIdenticalTo(q))
 						Expect(evt.Meta).To(Equal(p))
 						Expect(evt.Object).To(Equal(p))
 						close(c)
@@ -306,7 +306,7 @@ var _ = Describe("Source", func() {
 						defer GinkgoRecover()
 						// The empty event should have been filtered out by the predicates,
 						// and will not be passed to the handler.
-						Expect(q2).To(Equal(q))
+						Expect(q2).To(BeIdenticalTo(q))
 						Expect(evt.Meta).To(Equal(p))
 						Expect(evt.Object).To(Equal(p))
 						close(c)
@@ -429,7 +429,7 @@ var _ = Describe("Source", func() {
 					},
 					GenericFunc: func(evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
 						defer GinkgoRecover()
-						Expect(q2).To(Equal(q))
+						Expect(q2).To(BeIdenticalTo(q))
 						Expect(evt.Meta).To(Equal(p))
 						Expect(evt.Object).To(Equal(p))
 						resEvent1 = evt
@@ -453,7 +453,7 @@ var _ = Describe("Source", func() {
 					},
 					GenericFunc: func(evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
 						defer GinkgoRecover()
-						Expect(q2).To(Equal(q))
+						Expect(q2).To(BeIdenticalTo(q))
 						Expect(evt.Meta).To(Equal(p))
 						Expect(evt.Object).To(Equal(p))
 						resEvent2 = evt