Updated broker and channel controllers to watch kafka feature config …

…maps

Signed-off-by: Calum Murray <cmurray@redhat.com>

control-plane/cmd/kafka-controller/main.go

@@ -96,7 +96,7 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "channel-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return channel.NewController(ctx, channelEnv)
+				return channel.NewController(ctx, watcher, channelEnv)
 			},
 		},
 

control-plane/pkg/reconciler/broker/controller.go

@@ -98,6 +98,12 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 
 	brokerInformer := brokerinformer.Get(ctx)
 
+	kafkaConfigStore := apisconfig.NewStore(ctx, func(name string, value *apisconfig.KafkaFeatureFlags) {
+		reconciler.KafkaFeatureFlags.Reset(value)
+		impl.GlobalResync(brokerInformer.Informer())
+	})
+	kafkaConfigStore.WatchConfigs(watcher)
+
 	brokerInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: kafka.BrokerClassFilter(),
 		Handler:    controller.HandleAll(impl.Enqueue),

control-plane/pkg/reconciler/broker/controller_test.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	apisconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake"
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
@@ -84,6 +85,10 @@ func TestNewController(t *testing.T) {
 			ObjectMeta: metav1.ObjectMeta{
 				Name: "cm",
 			},
+		}, &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: apisconfig.FlagsConfigName,
+			},
 		}),
 		env,
 	)

control-plane/pkg/reconciler/channel/controller.go

@@ -37,6 +37,7 @@ import (
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 	serviceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/service"
+	"knative.dev/pkg/configmap"
 
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
@@ -49,7 +50,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
 
-func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
+func NewController(ctx context.Context, watcher configmap.Watcher, configs *config.Env) *controller.Impl {
 
 	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
 
@@ -95,6 +96,12 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 
 	channelInformer := kafkachannelinformer.Get(ctx)
 
+	kafkaConfigStore := apisconfig.NewStore(ctx, func(name string, value *apisconfig.KafkaFeatureFlags) {
+		reconciler.KafkaFeatureFlags.Reset(value)
+		impl.GlobalResync(channelInformer.Informer())
+	})
+	kafkaConfigStore.WatchConfigs(watcher)
+
 	channelInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
 	reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)

control-plane/pkg/reconciler/channel/controller_test.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	apisconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake"
 	"knative.dev/eventing/pkg/eventingtls/eventingtlstesting"
@@ -34,6 +35,7 @@ import (
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake"
+	"knative.dev/pkg/configmap"
 	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
@@ -81,6 +83,11 @@ func TestNewController(t *testing.T) {
 
 	controller := NewController(
 		ctx,
+		configmap.NewStaticWatcher(&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: apisconfig.FlagsConfigName,
+			},
+		}),
 		configs,
 	)
 	if controller == nil {

control-plane/pkg/reconciler/channel/v2/controllerv2.go

@@ -25,6 +25,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
 	serviceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/service"
+	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/resolver"
 
@@ -45,12 +46,13 @@ import (
 
 	"knative.dev/pkg/controller"
 
+	apisconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup"
 )
 
-func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
+func NewController(ctx context.Context, watcher configmap.Watcher, configs *config.Env) *controller.Impl {
 
 	configmapInformer := configmapinformer.Get(ctx)
 	channelInformer := kafkachannelinformer.Get(ctx)
@@ -75,6 +77,7 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 		SubscriptionLister:         subscriptioninformer.Get(ctx).Lister(),
 		ConsumerGroupLister:        consumerGroupInformer.Lister(),
 		InternalsClient:            consumergroupclient.Get(ctx),
+		KafkaFeatureFlags:          apisconfig.DefaultFeaturesConfig(),
 	}
 
 	logger := logging.FromContext(ctx)
@@ -88,6 +91,13 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 	}
 
 	impl := kafkachannelreconciler.NewImpl(ctx, reconciler)
+
+	kafkaConfigStore := apisconfig.NewStore(ctx, func(name string, value *apisconfig.KafkaFeatureFlags) {
+		reconciler.KafkaFeatureFlags.Reset(value)
+		impl.GlobalResync(channelInformer.Informer())
+	})
+	kafkaConfigStore.WatchConfigs(watcher)
+
 	IPsLister := prober.IdentityIPsLister()
 	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, "", IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)

control-plane/pkg/reconciler/channel/v2/controllerv2_test.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 
+	apisconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake"
 	"knative.dev/eventing/pkg/eventingtls/eventingtlstesting"
@@ -35,6 +36,7 @@ import (
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake"
+	"knative.dev/pkg/configmap"
 	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
@@ -84,6 +86,11 @@ func TestNewController(t *testing.T) {
 
 	controller := NewController(
 		ctx,
+		configmap.NewStaticWatcher(&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: apisconfig.FlagsConfigName,
+			},
+		}),
 		configs,
 	)
 	if controller == nil {