Few changes to the configmap package

configmap/informed_watcher.go

@@ -0,0 +1,122 @@
+/*
+Copyright 2018 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+istributed under the License is istributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package configmap
+
+import (
+	"errors"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	informers "k8s.io/client-go/informers"
+	corev1informers "k8s.io/client-go/informers/core/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+// NewDefaultWatcher creates a new default configmap.Watcher instance.
+//
+// Deprecated: Use NewInformedWatcher
+func NewDefaultWatcher(kc kubernetes.Interface, namespace string) *InformedWatcher {
+	return NewInformedWatcher(kc, namespace)
+}
+
+// NewInformedWatcher watchers a Kubernetes namespace for configmap changs
+func NewInformedWatcher(kc kubernetes.Interface, namespace string) *InformedWatcher {
+	sif := informers.NewSharedInformerFactoryWithOptions(
+		kc,
+		5*time.Minute,
+		informers.WithNamespace(namespace),
+	)
+
+	return &InformedWatcher{
+		sif:      sif,
+		informer: sif.Core().V1().ConfigMaps(),
+		ManualWatcher: ManualWatcher{
+			Namespace: namespace,
+		},
+	}
+}
+
+// InformedWatcher provides an informer-based implementation of Watcher.
+type InformedWatcher struct {
+	sif      informers.SharedInformerFactory
+	informer corev1informers.ConfigMapInformer
+	started  bool
+
+	// Embedding this struct allows us to reuse the logic
+	// of registering and notifying observers. This simplifies the
+	// InformedWatcher to just setting up the Kubernetes informer
+	ManualWatcher
+}
+
+// Asserts that InformedWatcher implements Watcher.
+var _ Watcher = (*InformedWatcher)(nil)
+
+// Start implements Watcher
+func (i *InformedWatcher) Start(stopCh <-chan struct{}) error {
+	if err := i.registerCallbackAndStartInformer(stopCh); err != nil {
+		return err
+	}
+
+	// Wait until it has been synced (WITHOUT holing the mutex, so callbacks happen)
+	if ok := cache.WaitForCacheSync(stopCh, i.informer.Informer().HasSynced); !ok {
+		return errors.New("Error waiting for ConfigMap informer to sync.")
+	}
+
+	return i.checkObservedResourcesExist()
+}
+
+func (i *InformedWatcher) registerCallbackAndStartInformer(stopCh <-chan struct{}) error {
+	i.m.Lock()
+	defer i.m.Unlock()
+	if i.started {
+		return errors.New("Watcher already started!")
+	}
+	i.started = true
+
+	i.informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    i.addConfigMapEvent,
+		UpdateFunc: i.updateConfigMapEvent,
+	})
+
+	// Start the shared informer factory (non-blocking)
+	i.sif.Start(stopCh)
+	return nil
+}
+
+func (i *InformedWatcher) checkObservedResourcesExist() error {
+	i.m.Lock()
+	defer i.m.Unlock()
+	// Check that all objects with Observers exist in our informers.
+	for k := range i.observers {
+		_, err := i.informer.Lister().ConfigMaps(i.Namespace).Get(k)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (i *InformedWatcher) addConfigMapEvent(obj interface{}) {
+	configMap := obj.(*corev1.ConfigMap)
+	i.OnChange(configMap)
+}
+
+func (i *InformedWatcher) updateConfigMapEvent(old, new interface{}) {
+	configMap := new.(*corev1.ConfigMap)
+	i.OnChange(configMap)
+}

configmap/default_test.go → configmap/informed_watcher_test.go

@@ -33,27 +33,27 @@ func (c *counter) callback(*corev1.ConfigMap) {
 	c.count++
 }
 
-func TestWatch(t *testing.T) {
+func TestInformedWatcher(t *testing.T) {
 	fooCM := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "knative-system",
+			Namespace: "default",
 			Name:      "foo",
 		},
 	}
 	barCM := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "knative-system",
+			Namespace: "default",
 			Name:      "bar",
 		},
 	}
 	kc := fakekubeclientset.NewSimpleClientset(fooCM, barCM)
-	cm := NewDefaultWatcher(kc, "knative-system").(*defaultImpl)
+	cm := NewInformedWatcher(kc, "default")
 
 	foo1 := &counter{name: "foo1"}
-	cm.Watch("foo", foo1.callback)
 	foo2 := &counter{name: "foo2"}
-	cm.Watch("foo", foo2.callback)
 	bar := &counter{name: "bar"}
+	cm.Watch("foo", foo1.callback)
+	cm.Watch("foo", foo2.callback)
 	cm.Watch("bar", bar.callback)
 
 	stopCh := make(chan struct{})
@@ -79,6 +79,7 @@ func TestWatch(t *testing.T) {
 			t.Errorf("%v.count = %v, want %v", obj, got, want)
 		}
 	}
+
 	for _, obj := range []*counter{bar} {
 		if got, want := obj.count, 1; got != want {
 			t.Errorf("%v.count = %v, want %v", obj, got, want)
@@ -109,9 +110,10 @@ func TestWatch(t *testing.T) {
 	}
 
 	// After an unwatched ConfigMap update, no change.
+
 	cm.updateConfigMapEvent(nil, &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "knative-system",
+			Namespace: "default",
 			Name:      "not-watched",
 		},
 	})
@@ -124,7 +126,7 @@ func TestWatch(t *testing.T) {
 	// After a change in an unrelated namespace, no change.
 	cm.updateConfigMapEvent(nil, &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "different-system",
+			Namespace: "not-default",
 			Name:      "foo",
 		},
 	})
@@ -137,7 +139,7 @@ func TestWatch(t *testing.T) {
 
 func TestWatchMissingFailsOnStart(t *testing.T) {
 	kc := fakekubeclientset.NewSimpleClientset()
-	cm := NewDefaultWatcher(kc, "knative-system").(*defaultImpl)
+	cm := NewInformedWatcher(kc, "default")
 
 	foo1 := &counter{name: "foo1"}
 	cm.Watch("foo", foo1.callback)
@@ -155,12 +157,12 @@ func TestWatchMissingFailsOnStart(t *testing.T) {
 func TestErrorOnMultipleStarts(t *testing.T) {
 	fooCM := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "knative-system",
+			Namespace: "default",
 			Name:      "foo",
 		},
 	}
 	kc := fakekubeclientset.NewSimpleClientset(fooCM)
-	cm := NewDefaultWatcher(kc, "knative-system").(*defaultImpl)
+	cm := NewInformedWatcher(kc, "default")
 
 	foo1 := &counter{name: "foo1"}
 	cm.Watch("foo", foo1.callback)