Improve UX for using client-go generated Informers

Author: pwittrock

pkg/manager/manager.go

@@ -135,6 +135,14 @@ func (r RunnableFunc) Start(s <-chan struct{}) error {
 	return r(s)
 }
 
+// InformerFactoryRunnable wraps a SharedInformerFactory to make it implement Runnable
+func StartAdapter(s func(<-chan struct{})) Runnable {
+	return RunnableFunc(func(c <-chan struct{}) error {
+		s(c)
+		return nil
+	})
+}
+
 // New returns a new Manager for creating Controllers.
 func New(config *rest.Config, options Options) (Manager, error) {
 	// Initialize a rest.config if none was specified

pkg/source/example_test.go

@@ -17,14 +17,22 @@ limitations under the License.
 package source_test
 
 import (
+	"time"
+
+	"github.com/golang/glog"
 	"k8s.io/api/core/v1"
+	kubeinformers "k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/runtime/signals"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
 var ctrl controller.Controller
+var mgr manager.Manager
 
 // This example Watches for Pod Events (e.g. Create / Update / Delete) and enqueues a reconcile.Request
 // with the Name and Namespace of the Pod.
@@ -42,3 +50,27 @@ func ExampleChannel() {
 		&handler.EnqueueRequestForObject{},
 	)
 }
+
+// This example Watches for Service Events (e.g. Create / Update / Delete) and enqueues a reconcile.Request
+// with the Name and Namespace of the Service.  It uses the client-go generated Service Informer instead of the
+// Generic Informer.
+func ExampleInformer() {
+	generatedClient := kubernetes.NewForConfigOrDie(mgr.GetConfig())
+	generatedInformers := kubeinformers.NewSharedInformerFactory(generatedClient, time.Minute*30)
+
+	// Add it to the Manager
+	if err := mgr.Add(manager.StartAdapter(generatedInformers.Start)); err != nil {
+		glog.Fatalf("error Adding InformerFactory to the Manager: %v", err)
+	}
+
+	// Setup Watch using the client-go generated Informer
+	if err := ctrl.Watch(
+		&source.Informer{InformerProvider: generatedInformers.Core().V1().Services()},
+		&handler.EnqueueRequestForObject{},
+	); err != nil {
+		glog.Fatalf("error Watching Services: %v", err)
+	}
+
+	// Start the Manager
+	mgr.Start(signals.SetupSignalHandler())
+}

pkg/source/source.go

@@ -241,10 +241,17 @@ func (cs *Channel) syncLoop() {
 	}
 }
 
+type InformerProvider interface {
+	Informer() toolscache.SharedIndexInformer
+}
+
 // 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 is the generated client-go Informer.  Mutually exclusive with InformerProvider.
 	Informer toolscache.SharedIndexInformer
+
+	// InformerProvider provides a generated client-go Informer.  Mutually exclusive with Informer.
+	InformerProvider InformerProvider
 }
 
 var _ Source = &Informer{}
@@ -255,11 +262,23 @@ func (is *Informer) Start(handler handler.EventHandler, queue workqueue.RateLimi
 	prct ...predicate.Predicate) error {
 
 	// Informer should have been specified by the user.
-	if is.Informer == nil {
-		return fmt.Errorf("must specify Informer.Informer")
+	if is.Informer == nil && is.InformerProvider == nil {
+		return fmt.Errorf("must specify Informer.Informer or Informer.InformerProvider")
+	}
+
+	if is.Informer != nil && is.InformerProvider != nil {
+		return fmt.Errorf("must specify only one of Informer.Informer and Informer.InformerProvider")
+	}
+
+	if is.Informer != nil {
+		is.Informer.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
+	}
+
+	if is.InformerProvider != nil {
+		is.InformerProvider.Informer().AddEventHandler(
+			internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
 	}
 
-	is.Informer.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
 	return nil
 }
 

pkg/source/source_suite_test.go

@@ -34,7 +34,7 @@ func TestSource(t *testing.T) {
 }
 
 var testenv *envtest.Environment
-var config *rest.Config
+var cfg *rest.Config
 var clientset *kubernetes.Clientset
 var icache cache.Cache
 var stop chan struct{}
@@ -46,13 +46,13 @@ var _ = BeforeSuite(func(done Done) {
 	testenv = &envtest.Environment{}
 
 	var err error
-	config, err = testenv.Start()
+	cfg, err = testenv.Start()
 	Expect(err).NotTo(HaveOccurred())
 
-	clientset, err = kubernetes.NewForConfig(config)
+	clientset, err = kubernetes.NewForConfig(cfg)
 	Expect(err).NotTo(HaveOccurred())
 
-	icache, err = cache.New(config, cache.Options{})
+	icache, err = cache.New(cfg, cache.Options{})
 	Expect(err).NotTo(HaveOccurred())
 
 	go func() {