Separate controllers factory

Use a dedicated controllers factory interface, rather than
creating controllers inline as needed. This will make unit
tests much easier to plug in, and move all controllers
creation code in the correct package.

Author: bpineau

pkg/controller/controller.go

@@ -26,13 +26,22 @@ import (
 
 const maxProcessRetry = 6
 
+// Interface describe a standard kubernetes controller
+type Interface interface {
+	Start()
+	Stop()
+}
+
+// Factory generate controllers
+type Factory struct{}
+
 // Controller is a generic kubernetes controller
 type Controller struct {
+	name     string
 	stopCh   chan struct{}
 	doneCh   chan struct{}
 	notifier event.Notifier
 	config   *config.KfConfig
-	name     string
 	queue    workqueue.RateLimitingInterface
 	informer cache.SharedIndexInformer
 }
@@ -99,7 +108,7 @@ func (c *Controller) Start() {
 	go c.informer.Run(c.stopCh)
 
 	if !cache.WaitForCacheSync(c.stopCh, c.informer.HasSynced) {
-		utilruntime.HandleError(fmt.Errorf("Timed out waiting for caches to sync"))
+		utilruntime.HandleError(fmt.Errorf("Timed out waiting for cache sync"))
 		return
 	}
 
@@ -149,7 +158,7 @@ func (c *Controller) processItem(key string) error {
 	rawobj, exists, err := c.informer.GetIndexer().GetByKey(key)
 
 	if err != nil {
-		return fmt.Errorf("error fetching object with key %s from store: %v", key, err)
+		return fmt.Errorf("error fetching %s from store: %v", key, err)
 	}
 
 	if !exists {
@@ -183,3 +192,8 @@ func (c *Controller) processItem(key string) error {
 func (c *Controller) enqueue(notif *event.Notification) {
 	c.notifier.Send(notif)
 }
+
+// NewController create a controller.Controller
+func (f *Factory) NewController(client cache.ListerWatcher, notifier event.Notifier, name string, config *config.KfConfig) Interface {
+	return New(client, notifier, name, config)
+}

pkg/observer/observer.go

@@ -17,19 +17,29 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/tools/cache"
 )
 
 const discoveryInterval = 60 * time.Second
 
-// Observer watch api-server and manage kubernetes controllers
+// ControllerFactory make controllers generation interchangeable
+type ControllerFactory interface {
+	NewController(client cache.ListerWatcher, notifier event.Notifier,
+		name string, config *config.KfConfig) controller.Interface
+}
+
+type controllerCollection map[string]controller.Interface
+
+// Observer watch api-server and manage kubernetes controllers lifecyles
 type Observer struct {
-	stop   chan struct{}
-	done   chan struct{}
-	notif  event.Notifier
-	disc   discovery.DiscoveryInterface
-	cpool  dynamic.ClientPool
-	ctrls  map[string]*controller.Controller
-	config *config.KfConfig
+	config    *config.KfConfig
+	stopCh    chan struct{}
+	doneCh    chan struct{}
+	notifier  event.Notifier
+	discovery discovery.DiscoveryInterface
+	cpool     dynamic.ClientPool
+	ctrls     controllerCollection
+	factory   ControllerFactory
 }
 
 type gvk struct {
@@ -40,27 +50,28 @@ type gvk struct {
 type resources map[string]*gvk
 
 // New returns a new observer, that will watch API resources and create controllers
-func New(config *config.KfConfig, notif event.Notifier) *Observer {
+func New(config *config.KfConfig, notif event.Notifier, factory ControllerFactory) *Observer {
 	return &Observer{
-		config: config,
-		notif:  notif,
-		disc:   discovery.NewDiscoveryClientForConfigOrDie(config.Client),
-		cpool:  dynamic.NewDynamicClientPool(config.Client),
-		ctrls:  make(map[string]*controller.Controller),
+		config:    config,
+		notifier:  notif,
+		discovery: discovery.NewDiscoveryClientForConfigOrDie(config.Client),
+		cpool:     dynamic.NewDynamicClientPool(config.Client),
+		ctrls:     make(controllerCollection),
+		factory:   factory,
 	}
 }
 
 // Start starts the observer in a detached goroutine
 func (c *Observer) Start() *Observer {
 	c.config.Logger.Info("Starting all kubernetes controllers")
 
-	c.stop = make(chan struct{})
-	c.done = make(chan struct{})
+	c.stopCh = make(chan struct{})
+	c.doneCh = make(chan struct{})
 
 	go func() {
 		ticker := time.NewTicker(discoveryInterval)
 		defer ticker.Stop()
-		defer close(c.done)
+		defer close(c.doneCh)
 
 		for {
 			err := c.refresh()
@@ -69,7 +80,7 @@ func (c *Observer) Start() *Observer {
 			}
 
 			select {
-			case <-c.stop:
+			case <-c.stopCh:
 				return
 			case <-ticker.C:
 			}
@@ -83,17 +94,17 @@ func (c *Observer) Start() *Observer {
 func (c *Observer) Stop() {
 	c.config.Logger.Info("Stopping all kubernetes controllers")
 
-	close(c.stop)
+	close(c.stopCh)
 
 	for _, ct := range c.ctrls {
 		ct.Stop()
 	}
 
-	<-c.done
+	<-c.doneCh
 }
 
 func (c *Observer) refresh() error {
-	groups, err := c.disc.ServerResources()
+	groups, err := c.discovery.ServerResources()
 	if err != nil {
 		return fmt.Errorf("failed to collect server resources: %v", err)
 	}
@@ -114,7 +125,7 @@ func (c *Observer) refresh() error {
 
 		client := cl.Resource(res.apiResource.DeepCopy(), metav1.NamespaceAll)
 
-		c.ctrls[name] = controller.New(client, c.notif, cname, c.config)
+		c.ctrls[name] = c.factory.NewController(client, c.notifier, cname, c.config)
 		go c.ctrls[name].Start()
 	}
 

pkg/run/run.go

@@ -8,6 +8,7 @@ import (
 	"syscall"
 
 	"github.com/bpineau/katafygio/config"
+	"github.com/bpineau/katafygio/pkg/controller"
 	"github.com/bpineau/katafygio/pkg/event"
 	"github.com/bpineau/katafygio/pkg/health"
 	"github.com/bpineau/katafygio/pkg/observer"
@@ -24,7 +25,7 @@ func Run(config *config.KfConfig) {
 
 	evts := event.New()
 	reco := recorder.New(config, evts).Start()
-	obsv := observer.New(config, evts).Start()
+	obsv := observer.New(config, evts, &controller.Factory{}).Start()
 
 	http, err := health.New(config).Start()
 	if err != nil {