Refactor the way our Controllers watch ConfigMaps

cmd/controller/main.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/knative/serving/pkg"
+	"github.com/knative/serving/pkg/configmap"
 
 	"github.com/josephburnett/k8sflag/pkg/k8sflag"
 	"github.com/knative/serving/pkg/controller"
@@ -151,10 +152,13 @@ func main() {
 		QueueProxyLoggingLevel:  queueProxyLoggingLevel.Get(),
 	}
 
+	configMapWatcher := configmap.NewDefaultWatcher(kubeClient, pkg.GetServingSystemNamespace())
+
 	opt := controller.Options{
 		KubeClientSet:    kubeClient,
 		ServingClientSet: elaClient,
 		BuildClientSet:   buildClient,
+		ConfigMapWatcher: configMapWatcher,
 		Logger:           logger,
 	}
 
@@ -163,7 +167,6 @@ func main() {
 	configurationInformer := elaInformerFactory.Serving().V1alpha1().Configurations()
 	revisionInformer := elaInformerFactory.Serving().V1alpha1().Revisions()
 	buildInformer := buildInformerFactory.Build().V1alpha1().Builds()
-	configMapInformer := servingSystemInformerFactory.Core().V1().ConfigMaps()
 	deploymentInformer := kubeInformerFactory.Apps().V1().Deployments()
 	endpointsInformer := kubeInformerFactory.Core().V1().Endpoints()
 	coreServiceInformer := kubeInformerFactory.Core().V1().Services()
@@ -174,19 +177,23 @@ func main() {
 	// Add new controllers to this array.
 	controllers := []controller.Interface{
 		configuration.NewController(opt, configurationInformer, revisionInformer, cfg),
-		revision.NewController(opt, vpaClient, revisionInformer, buildInformer, configMapInformer,
+		revision.NewController(opt, vpaClient, revisionInformer, buildInformer,
 			deploymentInformer, coreServiceInformer, endpointsInformer, vpaInformer,
 			cfg, &revControllerConfig),
 		route.NewController(opt, routeInformer, configurationInformer, ingressInformer,
-			configMapInformer, cfg, autoscaleEnableScaleToZero),
+			cfg, autoscaleEnableScaleToZero),
 		service.NewController(opt, serviceInformer, configurationInformer, routeInformer, cfg),
 	}
 
-	go kubeInformerFactory.Start(stopCh)
-	go elaInformerFactory.Start(stopCh)
-	go buildInformerFactory.Start(stopCh)
-	go servingSystemInformerFactory.Start(stopCh)
-	go vpaInformerFactory.Start(stopCh)
+	// These are non-blocking.
+	kubeInformerFactory.Start(stopCh)
+	elaInformerFactory.Start(stopCh)
+	buildInformerFactory.Start(stopCh)
+	servingSystemInformerFactory.Start(stopCh)
+	vpaInformerFactory.Start(stopCh)
+	if err := configMapWatcher.Start(stopCh); err != nil {
+		logger.Fatalf("failed to start configuration manager: %v", err)
+	}
 
 	// Wait for the caches to be synced before starting controllers.
 	logger.Info("Waiting for informer caches to sync")
@@ -196,7 +203,6 @@ func main() {
 		configurationInformer.Informer().HasSynced,
 		revisionInformer.Informer().HasSynced,
 		buildInformer.Informer().HasSynced,
-		configMapInformer.Informer().HasSynced,
 		deploymentInformer.Informer().HasSynced,
 		coreServiceInformer.Informer().HasSynced,
 		endpointsInformer.Informer().HasSynced,

pkg/configmap/OWNERS

@@ -0,0 +1,5 @@
+# The OWNERS file is used by prow to automatically merge approved PRs.
+
+approvers:
+- mattmoor
+- mdemirhan

pkg/configmap/default.go

@@ -0,0 +1,128 @@
+/*
+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
+distributed under the License is distributed 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"
+	"sync"
+
+	corev1 "k8s.io/api/core/v1"
+	informers "k8s.io/client-go/informers"
+	corev1informers "k8s.io/client-go/informers/core/v1"
+	"k8s.io/client-go/tools/cache"
+)
+
+// defaultImpl provides a default informer-based implementation of Watcher.
+type defaultImpl struct {
+	sif      informers.SharedInformerFactory
+	informer corev1informers.ConfigMapInformer
+	ns       string
+
+	// Guards mutations to defaultImpl fields
+	m sync.Mutex
+
+	observers map[string][]Observer
+	started   bool
+}
+
+// Asserts that defaultImpl implements Watcher.
+var _ Watcher = (*defaultImpl)(nil)
+
+// Watch implements Watcher
+func (di *defaultImpl) Watch(name string, w Observer) {
+	di.m.Lock()
+	defer di.m.Unlock()
+
+	if di.observers == nil {
+		di.observers = make(map[string][]Observer)
+	}
+
+	wl, _ := di.observers[name]
+	di.observers[name] = append(wl, w)
+}
+
+// Start implements Watcher
+func (di *defaultImpl) Start(stopCh <-chan struct{}) error {
+	if err := di.registerCallbackAndStartInformer(stopCh); err != nil {
+		return err
+	}
+
+	// Wait until it has been synced (WITHOUT holding the mutex, so callbacks happen)
+	if ok := cache.WaitForCacheSync(stopCh, di.informer.Informer().HasSynced); !ok {
+		return errors.New("Error waiting for ConfigMap informer to sync.")
+	}
+
+	return di.checkObservedResourcesExist()
+}
+
+func (di *defaultImpl) registerCallbackAndStartInformer(stopCh <-chan struct{}) error {
+	di.m.Lock()
+	defer di.m.Unlock()
+	if di.started {
+		return errors.New("Watcher already started!")
+	}
+	di.started = true
+
+	di.informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    di.addConfigMapEvent,
+		UpdateFunc: di.updateConfigMapEvent,
+	})
+
+	// Start the shared informer factory (non-blocking)
+	di.sif.Start(stopCh)
+	return nil
+}
+
+func (di *defaultImpl) checkObservedResourcesExist() error {
+	di.m.Lock()
+	defer di.m.Unlock()
+	// Check that all objects with Observers exist in our informers.
+	for k := range di.observers {
+		_, err := di.informer.Lister().ConfigMaps(di.ns).Get(k)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (di *defaultImpl) addConfigMapEvent(obj interface{}) {
+	// If the ConfigMap update is outside of our namespace, then quickly disregard it.
+	configMap := obj.(*corev1.ConfigMap)
+	if configMap.Namespace != di.ns {
+		// Outside of our namespace.
+		// This shouldn't happen due to our filtered informer.
+		return
+	}
+
+	// Within our namespace, take the lock and see if there are any registered observers.
+	di.m.Lock()
+	defer di.m.Unlock()
+	wl, ok := di.observers[configMap.Name]
+	if !ok {
+		return // No observers.
+	}
+
+	// Iterate over the observers and invoke their callbacks.
+	for _, w := range wl {
+		w(configMap)
+	}
+}
+
+func (di *defaultImpl) updateConfigMapEvent(old, new interface{}) {
+	di.addConfigMapEvent(new)
+}

pkg/configmap/default_test.go

@@ -0,0 +1,177 @@
+/*
+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
+distributed under the License is distributed 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 (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+)
+
+type counter struct {
+	name  string
+	count int
+}
+
+func (c *counter) callback(*corev1.ConfigMap) {
+	c.count++
+}
+
+func TestWatch(t *testing.T) {
+	fooCM := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "knative-system",
+			Name:      "foo",
+		},
+	}
+	barCM := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "knative-system",
+			Name:      "bar",
+		},
+	}
+	kc := fakekubeclientset.NewSimpleClientset(fooCM, barCM)
+	cm := NewDefaultWatcher(kc, "knative-system").(*defaultImpl)
+
+	foo1 := &counter{name: "foo1"}
+	cm.Watch("foo", foo1.callback)
+	foo2 := &counter{name: "foo2"}
+	cm.Watch("foo", foo2.callback)
+	bar := &counter{name: "bar"}
+	cm.Watch("bar", bar.callback)
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	err := cm.Start(stopCh)
+	if err != nil {
+		t.Fatalf("cm.Start() = %v", err)
+	}
+
+	// When Start returns the callbacks should have been called with the
+	// version of the objects that is available.
+	for _, obj := range []*counter{foo1, foo2, bar} {
+		if got, want := obj.count, 1; got != want {
+			t.Errorf("%v.count = %v, want %v", obj, got, want)
+		}
+	}
+
+	// After a "foo" event, the "foo" watchers should have 2,
+	// and the "bar" watchers should still have 1
+	cm.updateConfigMapEvent(nil, fooCM)
+	for _, obj := range []*counter{foo1, foo2} {
+		if got, want := obj.count, 2; got != want {
+			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)
+		}
+	}
+
+	// After a "foo" and "bar" event, the "foo" watchers should have 3,
+	// and the "bar" watchers should still have 2
+	cm.updateConfigMapEvent(nil, fooCM)
+	cm.updateConfigMapEvent(nil, barCM)
+	for _, obj := range []*counter{foo1, foo2} {
+		if got, want := obj.count, 3; got != want {
+			t.Errorf("%v.count = %v, want %v", obj, got, want)
+		}
+	}
+	for _, obj := range []*counter{bar} {
+		if got, want := obj.count, 2; got != want {
+			t.Errorf("%v.count = %v, want %v", obj, got, want)
+		}
+	}
+
+	// After a "bar" event, all watchers should have 3
+	cm.updateConfigMapEvent(nil, barCM)
+	for _, obj := range []*counter{foo1, foo2, bar} {
+		if got, want := obj.count, 3; got != want {
+			t.Errorf("%v.count = %v, want %v", obj, got, want)
+		}
+	}
+
+	// After an unwatched ConfigMap update, no change.
+	cm.updateConfigMapEvent(nil, &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "knative-system",
+			Name:      "not-watched",
+		},
+	})
+	for _, obj := range []*counter{foo1, foo2, bar} {
+		if got, want := obj.count, 3; got != want {
+			t.Errorf("%v.count = %v, want %v", obj, got, want)
+		}
+	}
+
+	// After a change in an unrelated namespace, no change.
+	cm.updateConfigMapEvent(nil, &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "different-system",
+			Name:      "foo",
+		},
+	})
+	for _, obj := range []*counter{foo1, foo2, bar} {
+		if got, want := obj.count, 3; got != want {
+			t.Errorf("%v.count = %v, want %v", obj, got, want)
+		}
+	}
+}
+
+func TestWatchMissingFailsOnStart(t *testing.T) {
+	kc := fakekubeclientset.NewSimpleClientset()
+	cm := NewDefaultWatcher(kc, "knative-system").(*defaultImpl)
+
+	foo1 := &counter{name: "foo1"}
+	cm.Watch("foo", foo1.callback)
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+
+	// This should error because we don't have a ConfigMap named "foo".
+	err := cm.Start(stopCh)
+	if err == nil {
+		t.Fatal("cm.Start() succeeded, wanted error")
+	}
+}
+
+func TestErrorOnMultipleStarts(t *testing.T) {
+	fooCM := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "knative-system",
+			Name:      "foo",
+		},
+	}
+	kc := fakekubeclientset.NewSimpleClientset(fooCM)
+	cm := NewDefaultWatcher(kc, "knative-system").(*defaultImpl)
+
+	foo1 := &counter{name: "foo1"}
+	cm.Watch("foo", foo1.callback)
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+
+	// This should succeed because the watched resource exists.
+	if err := cm.Start(stopCh); err != nil {
+		t.Fatalf("cm.Start() = %v", err)
+	}
+
+	// This should error because we already called Start()
+	if err := cm.Start(stopCh); err == nil {
+		t.Fatal("cm.Start() succeeded, wanted error")
+	}
+}