Dedicated notifier object to mediate events

This will probably help with unit tests.

Author: bpineau

pkg/controller/controller.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/bpineau/katafygio/config"
+	"github.com/bpineau/katafygio/pkg/event"
 
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -22,38 +23,19 @@ import (
 
 const maxProcessRetry = 6
 
-// Action represents the kind of object change we're notifying
-type Action int
-
-const (
-	// Delete is the object deletion Action
-	Delete Action = iota
-
-	// Upsert is the update or create Action
-	Upsert
-)
-
-// Event conveys an object delete/upsert notification
-type Event struct {
-	Action Action
-	Key    string
-	Kind   string
-	Obj    string
-}
-
 // Controller is a generic kubernetes controller
 type Controller struct {
 	stopCh   chan struct{}
 	doneCh   chan struct{}
-	evchan   chan Event
-	name     string
+	notifier *event.Notifier
 	config   *config.KfConfig
+	name     string
 	queue    workqueue.RateLimitingInterface
 	informer cache.SharedIndexInformer
 }
 
 // New return an untyped, generic Kubernetes controller
-func New(lw cache.ListerWatcher, evchan chan Event, name string, config *config.KfConfig) *Controller {
+func New(lw cache.ListerWatcher, notifier *event.Notifier, name string, config *config.KfConfig) *Controller {
 	queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
 
 	informer := cache.NewSharedIndexInformer(
@@ -87,7 +69,7 @@ func New(lw cache.ListerWatcher, evchan chan Event, name string, config *config.
 	return &Controller{
 		stopCh:   make(chan struct{}),
 		doneCh:   make(chan struct{}),
-		evchan:   evchan,
+		notifier: notifier,
 		name:     name,
 		config:   config,
 		queue:    queue,
@@ -158,7 +140,7 @@ func (c *Controller) processItem(key string) error {
 
 	if !exists {
 		// deleted object
-		c.enqueue(Event{Action: Delete, Key: key, Kind: c.name, Obj: ""})
+		c.enqueue(&event.Notification{Action: event.Delete, Key: key, Kind: c.name, Object: ""})
 		return nil
 	}
 
@@ -180,10 +162,10 @@ func (c *Controller) processItem(key string) error {
 		return fmt.Errorf("failed to marshal %s: %v", key, err)
 	}
 
-	c.enqueue(Event{Action: Upsert, Key: key, Kind: c.name, Obj: string(yml)})
+	c.enqueue(&event.Notification{Action: event.Upsert, Key: key, Kind: c.name, Object: string(yml)})
 	return nil
 }
 
-func (c *Controller) enqueue(ev Event) {
-	c.evchan <- ev
+func (c *Controller) enqueue(notif *event.Notification) {
+	c.notifier.Send(notif)
 }

pkg/event/event.go

@@ -0,0 +1,38 @@
+// Package event mediates notification between controllers and recorder
+package event
+
+// Action represents the kind of object change we're notifying
+type Action int
+
+const (
+	// Delete is the object deletion Action
+	Delete Action = iota
+
+	// Upsert is the update or create Action
+	Upsert
+)
+
+// Notification conveys an object delete/upsert notification
+type Notification struct {
+	Action Action
+	Key    string
+	Kind   string
+	Object string
+}
+
+// Notifier mediates notifications between controllers and recorder
+type Notifier struct {
+	C chan Notification
+}
+
+// New creates a new event.Notifier
+func New() *Notifier {
+	return &Notifier{
+		C: make(chan Notification),
+	}
+}
+
+// Send sends a notification
+func (n *Notifier) Send(notif *Notification) {
+	n.C <- *notif
+}

pkg/observer/observer.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/bpineau/katafygio/config"
 	"github.com/bpineau/katafygio/pkg/controller"
+	"github.com/bpineau/katafygio/pkg/event"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -27,7 +28,7 @@ const discoveryInterval = 60 * time.Second
 type Observer struct {
 	stop   chan struct{}
 	done   chan struct{}
-	evch   chan controller.Event
+	notif  *event.Notifier
 	disc   *discovery.DiscoveryClient
 	cpool  dynamic.ClientPool
 	ctrls  map[string]*controller.Controller
@@ -46,10 +47,10 @@ type resources map[string]*gvk
 
 // New returns a new observer, that will watch for api resource kinds
 // and create new controllers for each one.
-func New(config *config.KfConfig, evch chan controller.Event) *Observer {
+func New(config *config.KfConfig, notif *event.Notifier) *Observer {
 	return &Observer{
 		config: config,
-		evch:   evch,
+		notif:  notif,
 		disc:   discovery.NewDiscoveryClientForConfigOrDie(config.Client),
 		cpool:  dynamic.NewDynamicClientPool(config.Client),
 		ctrls:  make(map[string]*controller.Controller),
@@ -126,7 +127,7 @@ func (c *Observer) refresh() error {
 			},
 		}
 
-		c.ctrls[name] = controller.New(lw, c.evch, strings.ToLower(res.ar.Kind), c.config)
+		c.ctrls[name] = controller.New(lw, c.notif, strings.ToLower(res.ar.Kind), c.config)
 		go c.ctrls[name].Start()
 	}
 

pkg/recorder/recorder.go

@@ -11,7 +11,7 @@ import (
 	"time"
 
 	"github.com/bpineau/katafygio/config"
-	"github.com/bpineau/katafygio/pkg/controller"
+	"github.com/bpineau/katafygio/pkg/event"
 )
 
 // activeFiles will contain a list of active (present in cluster) objets; we'll
@@ -22,18 +22,18 @@ type activeFiles map[string]bool
 // Listener receive events from controllers and save them to disk as yaml files
 type Listener struct {
 	config      *config.KfConfig
-	evchan      chan controller.Event
+	events      *event.Notifier
 	actives     activeFiles
 	activesLock sync.RWMutex
 	stopch      chan struct{}
 	donech      chan struct{}
 }
 
 // New creates a new Listener
-func New(config *config.KfConfig, evchan chan controller.Event) *Listener {
+func New(config *config.KfConfig, events *event.Notifier) *Listener {
 	return &Listener{
 		config:  config,
-		evchan:  evchan,
+		events:  events,
 		actives: activeFiles{},
 	}
 }
@@ -57,8 +57,8 @@ func (w *Listener) Start() *Listener {
 			select {
 			case <-w.stopch:
 				return
-			case ev := <-w.evchan:
-				w.processNextEvent(ev)
+			case ev := <-w.events.C:
+				w.processNextEvent(&ev)
 			case <-gcTick.C:
 				w.deleteObsoleteFiles()
 			}
@@ -75,7 +75,7 @@ func (w *Listener) Stop() {
 	<-w.donech
 }
 
-func (w *Listener) processNextEvent(ev controller.Event) {
+func (w *Listener) processNextEvent(ev *event.Notification) {
 	if w.shouldIgnore(ev) {
 		return
 	}
@@ -88,9 +88,9 @@ func (w *Listener) processNextEvent(ev controller.Event) {
 	}
 
 	switch ev.Action {
-	case controller.Upsert:
-		err = w.save(path, ev.Obj)
-	case controller.Delete:
+	case event.Upsert:
+		err = w.save(path, ev.Object)
+	case event.Delete:
 		err = w.remove(path)
 	}
 
@@ -99,7 +99,7 @@ func (w *Listener) processNextEvent(ev controller.Event) {
 	}
 }
 
-func (w *Listener) shouldIgnore(ev controller.Event) bool {
+func (w *Listener) shouldIgnore(ev *event.Notification) bool {
 	for _, kind := range w.config.ExcludeKind {
 		if strings.Compare(strings.ToLower(kind), ev.Kind) == 0 {
 			return true
@@ -115,7 +115,7 @@ func (w *Listener) shouldIgnore(ev controller.Event) bool {
 	return w.config.DryRun
 }
 
-func getPath(root string, ev controller.Event) (string, error) {
+func getPath(root string, ev *event.Notification) (string, error) {
 	filename := ev.Kind + "-" + filepath.Base(ev.Key) + ".yaml"
 
 	dir, err := filepath.Abs(filepath.Dir(root + "/" + ev.Key))

pkg/run/run.go

@@ -8,7 +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"
 	"github.com/bpineau/katafygio/pkg/recorder"
@@ -22,10 +22,9 @@ func Run(config *config.KfConfig) {
 		config.Logger.Fatalf("failed to start git repo handler: %v", err)
 	}
 
-	evchan := make(chan controller.Event)
-
-	reco := recorder.New(config, evchan).Start()
-	ctrl := observer.New(config, evchan).Start()
+	evts := event.New()
+	reco := recorder.New(config, evts).Start()
+	ctrl := observer.New(config, evts).Start()
 
 	http, err := health.New(config).Start()
 	if err != nil {