WIP: add configmap for flow controller

config/400-flow-controller-config.yaml

@@ -0,0 +1,22 @@
+# 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
+#
+#     https://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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: flow-controller-config
+  namespace: knative-eventing
+data:
+  # Configuration for the flow controller
+  default-cluster-bus: stub

pkg/controller/flow/provider.go

@@ -17,9 +17,15 @@ limitations under the License.
 package flow
 
 import (
+	"sync"
+
 	channelsv1alpha1 "github.com/knative/eventing/pkg/apis/channels/v1alpha1"
 	feedsv1alpha1 "github.com/knative/eventing/pkg/apis/feeds/v1alpha1"
 	"github.com/knative/eventing/pkg/apis/flows/v1alpha1"
+	"github.com/knative/eventing/pkg/system"
+	"github.com/knative/pkg/configmap"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -32,10 +38,23 @@ import (
 
 const controllerAgentName = "flow-controller"
 
+// controllerConfigMapName is the name of the configmap in the eventing
+// namespace that holds the configuration for this controller.
+const controllerConfigMapName = "flow-controller-config"
+
 type reconciler struct {
 	client     client.Client
 	restConfig *rest.Config
 	recorder   record.EventRecorder
+
+	// configMutex guards the controller state that comes from the controller's configmap.
+	configMutex sync.RWMutex
+	// controllerConfigMapWatcher is used to watch the knative-system/flow-controller-config configMap.
+	controllerConfigMapWatcher configmap.Watcher
+	// defaultClusterBusName is the default bus name to use to create channels; it is
+	// updated if the knative-system/flow-controller-config ConfigMap exists and
+	// contains the 'default-cluster-bus-name' key.
+	defaultClusterBusName string
 }
 
 // Verify the struct implements reconcile.Reconciler
@@ -44,15 +63,25 @@ var _ reconcile.Reconciler = &reconciler{}
 // ProvideController returns a flow controller.
 func ProvideController(mrg manager.Manager) (controller.Controller, error) {
 	// Setup a new controller to Reconcile Flows.
+	k8sClient, err := kubernetes.NewForConfig(mrg.GetConfig())
+	if err != nil {
+		return nil, err
+	}
+
+	reconciler := &reconciler{
+		recorder:                   mrg.GetRecorder(controllerAgentName),
+		controllerConfigMapWatcher: configmap.NewDefaultWatcher(k8sClient, system.Namespace),
+	}
+
 	c, err := controller.New(controllerAgentName, mrg, controller.Options{
-		Reconciler: &reconciler{
-			recorder: mrg.GetRecorder(controllerAgentName),
-		},
+		Reconciler: reconciler,
 	})
 	if err != nil {
 		return nil, err
 	}
 
+	mrg.Add(reconciler.controllerConfigMapWatcher)
+
 	// Watch Flow events and enqueue Flow object key.
 	if err := c.Watch(&source.Kind{Type: &v1alpha1.Flow{}}, &handler.EnqueueRequestForObject{}); err != nil {
 		return nil, err
@@ -73,6 +102,8 @@ func ProvideController(mrg manager.Manager) (controller.Controller, error) {
 		return nil, err
 	}
 
+	reconciler.controllerConfigMapWatcher.Watch(controllerConfigMapName, reconciler.receiveControllerConfig)
+
 	return c, nil
 }
 
@@ -85,3 +116,24 @@ func (r *reconciler) InjectConfig(c *rest.Config) error {
 	r.restConfig = c
 	return nil
 }
+
+// defaultClusterBusConfigMapKey is the name of the key in this controller's
+// ConfigMap that contains the name of the default cluster bus for the flow
+// controller to use.
+const defaultClusterBusConfigMapKey = "default-cluster-bus"
+
+func (r *reconciler) receiveControllerConfig(configMap *corev1.ConfigMap) {
+	r.configMutex.Lock()
+	defer r.configMutex.Unlock()
+
+	if value, ok := configMap.Data[defaultClusterBusConfigMapKey]; ok {
+		r.defaultClusterBusName = value
+	} else {
+		r.defaultClusterBusName = "stub"
+	}
+}
+func (r *reconciler) getDefaultClusterBusName() string {
+	r.configMutex.RLock()
+	defer r.configMutex.RUnlock()
+	return r.defaultClusterBusName
+}

pkg/controller/flow/reconcile.go

@@ -34,9 +34,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
-// TODO: This should come from a configmap
-const defaultBusName = "stub"
-
 // What field do we assume Object Reference exports as a resolvable target
 const targetFieldName = "domainInternal"
 
@@ -232,7 +229,7 @@ func (r *reconciler) createChannel(flow *v1alpha1.Flow) (*channelsv1alpha1.Chann
 			},
 		},
 		Spec: channelsv1alpha1.ChannelSpec{
-			ClusterBus: defaultBusName,
+			ClusterBus: r.getDefaultClusterBusName(),
 		},
 	}
 	if err := r.client.Create(context.TODO(), channel); err != nil {

pkg/controller/flow/reconcile_test.go

@@ -79,8 +79,9 @@ func TestAllCases(t *testing.T) {
 
 	for _, tc := range testCases {
 		r := &reconciler{
-			client:   tc.GetClient(),
-			recorder: recorder,
+			client:                tc.GetClient(),
+			recorder:              recorder,
+			defaultClusterBusName: "stub",
 		}
 		t.Run(tc.Name, tc.Runner(t, r, r.client))
 	}

test/e2e-tests.sh

@@ -40,6 +40,14 @@ readonly E2E_TEST_FUNCTION=e2e-k8s-events-function
 
 function teardown() {
   ko delete --ignore-not-found=true -f config/
+  dump_controller_logs
+}
+
+function dump_controller_logs() {
+  echo -e "Dumping controller manager logs"
+  kubectl -n knative-eventing logs `kubectl -n knative-eventing get pods -oname | grep controller-manager` controller-manager
+  echo -e "Dumping controller logs"
+  kubectl -n knative-eventing logs `kubectl -n knative-eventing get pods -oname | grep eventing-controller`
 }
 
 function wait_until_flow_ready() {
@@ -72,10 +80,6 @@ function wait_until_flow_ready() {
   kubectl get -n $NAMESPACE flows $NAME -oyaml
   kubectl get -n $NAMESPACE jobs $NAME-start -oyaml
   kubectl get -n $NAMESPACE feeds $NAME -oyaml
-  echo -e "Dumping controller manager logs"
-  kubectl -n knative-eventing logs `kubectl -n knative-eventing get pods -oname | grep controller-manager` controller-manager
-  echo -e "Dumping controller logs"
-  kubectl -n knative-eventing logs `kubectl -n knative-eventing get pods -oname | grep eventing-controller`
   return 1
 }