Add ability to watch secondary resources (#117)

Author: fabianlindfors

pkg/controller/watch/watch.go

@@ -0,0 +1,72 @@
+package watch
+
+import (
+	"github.com/mongodb/mongodb-kubernetes-operator/pkg/util/contains"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+// ResourceWatcher implements handler.EventHandler and is used to trigger reconciliation when
+// a watched object changes. It's designed to only be used for a single type of object.
+// If multiple types should be watched, one ResourceWatcher for each type should be used.
+type ResourceWatcher struct {
+	watched map[types.NamespacedName][]types.NamespacedName
+}
+
+// New will create a new ResourceWatcher with no watched objects.
+func New() ResourceWatcher {
+	return ResourceWatcher{
+		watched: make(map[types.NamespacedName][]types.NamespacedName),
+	}
+}
+
+// Watch will add a new object to watch.
+func (w ResourceWatcher) Watch(watchedName, dependentName types.NamespacedName) {
+	existing, hasExisting := w.watched[watchedName]
+	if !hasExisting {
+		existing = []types.NamespacedName{}
+	}
+
+	// Check if resource is already being watched.
+	if contains.NamespacedName(existing, dependentName) {
+		return
+	}
+
+	w.watched[watchedName] = append(existing, dependentName)
+}
+
+func (w ResourceWatcher) Create(event event.CreateEvent, queue workqueue.RateLimitingInterface) {
+	w.handleEvent(event.Meta, queue)
+}
+
+func (w ResourceWatcher) Update(event event.UpdateEvent, queue workqueue.RateLimitingInterface) {
+	w.handleEvent(event.MetaOld, queue)
+}
+
+func (w ResourceWatcher) Delete(event event.DeleteEvent, queue workqueue.RateLimitingInterface) {
+	w.handleEvent(event.Meta, queue)
+}
+
+func (w ResourceWatcher) Generic(event event.GenericEvent, queue workqueue.RateLimitingInterface) {
+	w.handleEvent(event.Meta, queue)
+}
+
+// handleEvent is called when an event is received for an object.
+// It will check if the object is being watched and trigger a reconciliation for
+// the dependent object.
+func (w ResourceWatcher) handleEvent(meta metav1.Object, queue workqueue.RateLimitingInterface) {
+	changedObjectName := types.NamespacedName{
+		Name:      meta.GetName(),
+		Namespace: meta.GetNamespace(),
+	}
+
+	// Enqueue reconciliation for each dependent object.
+	for _, reconciledObjectName := range w.watched[changedObjectName] {
+		queue.Add(reconcile.Request{
+			NamespacedName: reconciledObjectName,
+		})
+	}
+}

pkg/controller/watch/watch_test.go

@@ -0,0 +1,163 @@
+package watch
+
+import (
+	"testing"
+
+	"k8s.io/apimachinery/pkg/types"
+
+	mdbv1 "github.com/mongodb/mongodb-kubernetes-operator/pkg/apis/mongodb/v1"
+
+	"github.com/stretchr/testify/assert"
+
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllertest"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/util/workqueue"
+
+	"sigs.k8s.io/controller-runtime/pkg/event"
+)
+
+func TestWatcher(t *testing.T) {
+	obj := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pod",
+			Namespace: "namespace",
+		},
+	}
+	objNsName := types.NamespacedName{Name: obj.Name, Namespace: obj.Namespace}
+
+	mdb1 := mdbv1.MongoDB{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "mdb1",
+			Namespace: "namespace",
+		},
+	}
+
+	mdb2 := mdbv1.MongoDB{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "mdb2",
+			Namespace: "namespace",
+		},
+	}
+
+	t.Run("Non-watched object", func(t *testing.T) {
+		watcher := New()
+		queue := controllertest.Queue{Interface: workqueue.New()}
+
+		watcher.Create(event.CreateEvent{
+			Meta:   obj.GetObjectMeta(),
+			Object: obj,
+		}, queue)
+
+		// Ensure no reconciliation is queued if object is not watched.
+		assert.Equal(t, 0, queue.Len())
+	})
+
+	t.Run("Multiple objects to reconile", func(t *testing.T) {
+		watcher := New()
+		queue := controllertest.Queue{Interface: workqueue.New()}
+		watcher.Watch(objNsName, mdb1.NamespacedName())
+		watcher.Watch(objNsName, mdb2.NamespacedName())
+
+		watcher.Create(event.CreateEvent{
+			Meta:   obj.GetObjectMeta(),
+			Object: obj,
+		}, queue)
+
+		// Ensure multiple reconciliations are enqueued.
+		assert.Equal(t, 2, queue.Len())
+	})
+
+	t.Run("Create event", func(t *testing.T) {
+		watcher := New()
+		queue := controllertest.Queue{Interface: workqueue.New()}
+		watcher.Watch(objNsName, mdb1.NamespacedName())
+
+		watcher.Create(event.CreateEvent{
+			Meta:   obj.GetObjectMeta(),
+			Object: obj,
+		}, queue)
+
+		assert.Equal(t, 1, queue.Len())
+	})
+
+	t.Run("Update event", func(t *testing.T) {
+		watcher := New()
+		queue := controllertest.Queue{Interface: workqueue.New()}
+		watcher.Watch(objNsName, mdb1.NamespacedName())
+
+		watcher.Update(event.UpdateEvent{
+			MetaOld:   obj.GetObjectMeta(),
+			ObjectOld: obj,
+			MetaNew:   obj.GetObjectMeta(),
+			ObjectNew: obj,
+		}, queue)
+
+		assert.Equal(t, 1, queue.Len())
+	})
+
+	t.Run("Delete event", func(t *testing.T) {
+		watcher := New()
+		queue := controllertest.Queue{Interface: workqueue.New()}
+		watcher.Watch(objNsName, mdb1.NamespacedName())
+
+		watcher.Delete(event.DeleteEvent{
+			Meta:   obj.GetObjectMeta(),
+			Object: obj,
+		}, queue)
+
+		assert.Equal(t, 1, queue.Len())
+	})
+
+	t.Run("Generic event", func(t *testing.T) {
+		watcher := New()
+		queue := controllertest.Queue{Interface: workqueue.New()}
+		watcher.Watch(objNsName, mdb1.NamespacedName())
+
+		watcher.Generic(event.GenericEvent{
+			Meta:   obj.GetObjectMeta(),
+			Object: obj,
+		}, queue)
+
+		assert.Equal(t, 1, queue.Len())
+	})
+}
+
+func TestWatcherAdd(t *testing.T) {
+	watcher := New()
+	assert.Empty(t, watcher.watched)
+
+	watchedName := types.NamespacedName{Name: "object", Namespace: "namespace"}
+
+	mdb1 := mdbv1.MongoDB{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "mdb1",
+			Namespace: "namespace",
+		},
+	}
+	mdb2 := mdbv1.MongoDB{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "mdb2",
+			Namespace: "namespace",
+		},
+	}
+
+	// Ensure single object can be added to empty watchlist.
+	watcher.Watch(watchedName, mdb1.NamespacedName())
+	assert.Len(t, watcher.watched, 1)
+	assert.Equal(t, []types.NamespacedName{mdb1.NamespacedName()}, watcher.watched[watchedName])
+
+	// Ensure object can only be watched once.
+	watcher.Watch(watchedName, mdb1.NamespacedName())
+	assert.Len(t, watcher.watched, 1)
+	assert.Equal(t, []types.NamespacedName{mdb1.NamespacedName()}, watcher.watched[watchedName])
+
+	// Ensure a single object can be watched for multiple reconciliations.
+	watcher.Watch(watchedName, mdb2.NamespacedName())
+	assert.Len(t, watcher.watched, 1)
+	assert.Equal(t, []types.NamespacedName{
+		mdb1.NamespacedName(),
+		mdb2.NamespacedName(),
+	}, watcher.watched[watchedName])
+}

pkg/util/contains/contains.go

@@ -1,6 +1,9 @@
 package contains
 
-import mdbv1 "github.com/mongodb/mongodb-kubernetes-operator/pkg/apis/mongodb/v1"
+import (
+	mdbv1 "github.com/mongodb/mongodb-kubernetes-operator/pkg/apis/mongodb/v1"
+	"k8s.io/apimachinery/pkg/types"
+)
 
 func String(slice []string, s string) bool {
 	for _, elem := range slice {
@@ -19,3 +22,12 @@ func AuthMode(slice []mdbv1.AuthMode, s mdbv1.AuthMode) bool {
 	}
 	return false
 }
+
+func NamespacedName(nsNames []types.NamespacedName, nsName types.NamespacedName) bool {
+	for _, elem := range nsNames {
+		if elem == nsName {
+			return true
+		}
+	}
+	return false
+}