Merge 9d6583a into 47ee8fa

config/brokers/mt-channel-broker/300-config-mtbroker-delivery.yaml

@@ -0,0 +1,26 @@
+# Copyright 2021 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: mt-broker-internal-delivery
+  namespace: knative-eventing
+  labels:
+    eventing.knative.dev/release: devel
+data:
+  _example: |
+    # Configure the internal channels subscription delivery spec
+    delivery: |
+      retry: 10

pkg/reconciler/mtbroker/config.go → pkg/reconciler/mtbroker/channel_config.go

@@ -23,11 +23,12 @@ import (
 	"fmt"
 
 	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/yaml"
 
-	corev1 "k8s.io/api/core/v1"
-	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
 	"knative.dev/pkg/logging"
+
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
 )
 
 type Config struct {

pkg/reconciler/mtbroker/delivery_config.go

@@ -0,0 +1,75 @@
+/*
+Copyright 2021 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 mtbroker
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/pkg/configmap"
+	"sigs.k8s.io/yaml"
+
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+)
+
+const (
+	InternalDeliveryConfigMapName        = "mt-broker-internal-delivery"
+	internalDeliveryConfigMapDeliveryKey = "delivery"
+)
+
+// NewInternalDeliveryConfigFromConfigMap parses the config map into a DeliverySpec
+func NewInternalDeliveryConfigFromConfigMap(cm *corev1.ConfigMap) (*eventingduck.DeliverySpec, error) {
+	if cm == nil || len(cm.Data) == 0 {
+		return nil, nil
+	}
+	d, ok := cm.Data[internalDeliveryConfigMapDeliveryKey]
+	if !ok {
+		return nil, nil
+	}
+
+	var delivery eventingduck.DeliverySpec
+	err := yaml.Unmarshal([]byte(d), &delivery)
+	if err != nil {
+		return nil, err
+	}
+	return &delivery, nil
+}
+
+// InternalDeliveryConfigStore is a typed wrapper around configmap.Untyped store to handle our configmaps.
+// +k8s:deepcopy-gen=false
+type InternalDeliveryConfigStore struct {
+	*configmap.UntypedStore
+}
+
+// NewInternalDeliveryConfigStore creates a new store of Configs and optionally calls functions when ConfigMaps are updated.
+func NewInternalDeliveryConfigStore(logger configmap.Logger, onAfterStore ...func(name string, value interface{})) *InternalDeliveryConfigStore {
+	store := &InternalDeliveryConfigStore{
+		UntypedStore: configmap.NewUntypedStore(
+			"channeldefaults",
+			logger,
+			configmap.Constructors{
+				InternalDeliveryConfigMapName: NewInternalDeliveryConfigFromConfigMap,
+			},
+			onAfterStore...,
+		),
+	}
+
+	return store
+}
+
+// Load creates a InternalDeliveryConfigStore from the current config state of the InternalDeliveryConfigStore.
+func (s *InternalDeliveryConfigStore) Load() *eventingduck.DeliverySpec {
+	return s.UntypedLoad(InternalDeliveryConfigMapName).(*eventingduck.DeliverySpec)
+}

pkg/reconciler/mtbroker/delivery_config_test.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2021 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 mtbroker
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/utils/pointer"
+
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+
+	. "knative.dev/pkg/configmap/testing"
+)
+
+func TestMtBrokerInternalConfig(t *testing.T) {
+	_, example := ConfigMapsFromTestFile(t, "config-mtbroker-delivery")
+
+	for _, tt := range []struct {
+		name string
+		fail bool
+		want *eventingduck.DeliverySpec
+		data *corev1.ConfigMap
+	}{{
+		name: "Nil config",
+		fail: false,
+		want: nil,
+		data: nil,
+	}, {
+		name: "Empty config",
+		fail: false,
+		want: nil,
+		data: &corev1.ConfigMap{},
+	}, {
+		name: "With values",
+		fail: false,
+		want: &eventingduck.DeliverySpec{Retry: pointer.Int32Ptr(10)},
+		data: example,
+	}} {
+		t.Run(tt.name, func(t *testing.T) {
+			testConfig, err := NewInternalDeliveryConfigFromConfigMap(tt.data)
+			if tt.fail != (err != nil) {
+				t.Fatal("Unexpected error value:", err)
+			}
+
+			require.Equal(t, tt.want, testConfig)
+		})
+	}
+}

pkg/reconciler/mtbroker/testdata/config-mtbroker-delivery.yaml

@@ -0,0 +1,24 @@
+# Copyright 2021 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: mt-broker-internal-delivery
+  namespace: default
+
+data:
+  _example: |
+    delivery: |
+      retry: 10

pkg/reconciler/mtbroker/trigger/controller.go

@@ -31,6 +31,8 @@ import (
 	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
 	triggerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger"
 	"knative.dev/eventing/pkg/duck"
+	"knative.dev/eventing/pkg/reconciler/mtbroker"
+
 	"knative.dev/pkg/client/injection/ducks/duck/v1/source"
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
 	"knative.dev/pkg/configmap"
@@ -68,6 +70,8 @@ func NewController(
 
 	r.sourceTracker = duck.NewListableTracker(ctx, source.Get, impl.EnqueueKey, controller.GetTrackerLease(ctx))
 	r.uriResolver = resolver.NewURIResolver(ctx, impl.EnqueueKey)
+	r.internalDeliveryConfigStore = mtbroker.NewInternalDeliveryConfigStore(logger)
+	r.internalDeliveryConfigStore.WatchConfigs(cmw)
 
 	triggerInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 

pkg/reconciler/mtbroker/trigger/controller_test.go

@@ -19,21 +19,33 @@ package mttrigger
 import (
 	"testing"
 
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"knative.dev/pkg/configmap"
 	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/system"
 
 	// Fake injection informers
 	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake"
+	"knative.dev/eventing/pkg/reconciler/mtbroker"
+
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/source/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
 )
 
 func TestNew(t *testing.T) {
 	ctx, _ := SetupFakeContext(t)
 
-	c := NewController(ctx, configmap.NewStaticWatcher())
+	cmw := configmap.NewStaticWatcher(&corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      mtbroker.InternalDeliveryConfigMapName,
+			Namespace: system.Namespace(),
+		},
+	})
+
+	c := NewController(ctx, cmw)
 
 	if c == nil {
 		t.Fatal("Expected NewController to return a non-nil value")

pkg/reconciler/mtbroker/trigger/trigger.go

@@ -29,23 +29,25 @@ import (
 	"k8s.io/client-go/dynamic"
 	corev1listers "k8s.io/client-go/listers/core/v1"
 
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/network"
+	pkgreconciler "knative.dev/pkg/reconciler"
+	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/system"
+
 	"knative.dev/eventing/pkg/apis/eventing"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
 	clientset "knative.dev/eventing/pkg/client/clientset/versioned"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
 	"knative.dev/eventing/pkg/duck"
+	"knative.dev/eventing/pkg/reconciler/mtbroker"
 	"knative.dev/eventing/pkg/reconciler/mtbroker/resources"
 	"knative.dev/eventing/pkg/reconciler/sugar/trigger/path"
-	"knative.dev/pkg/apis"
-	duckv1 "knative.dev/pkg/apis/duck/v1"
-	"knative.dev/pkg/controller"
-	"knative.dev/pkg/logging"
-	"knative.dev/pkg/network"
-	pkgreconciler "knative.dev/pkg/reconciler"
-	"knative.dev/pkg/resolver"
-	"knative.dev/pkg/system"
 )
 
 var brokerGVK = eventingv1.SchemeGroupVersion.WithKind("Broker")
@@ -67,6 +69,9 @@ type Reconciler struct {
 	triggerLister      eventinglisters.TriggerLister
 	configmapLister    corev1listers.ConfigMapLister
 
+	// Internal Delivery store
+	internalDeliveryConfigStore *mtbroker.InternalDeliveryConfigStore
+
 	// Dynamic tracker to track Sources. In particular, it tracks the dependency between Triggers and Sources.
 	sourceTracker duck.ListableTracker
 
@@ -163,7 +168,7 @@ func (r *Reconciler) subscribeToBrokerChannel(ctx context.Context, b *eventingv1
 		Name:       b.Name,
 		Namespace:  b.Namespace,
 	}
-	expected := resources.NewSubscription(t, brokerTrigger, brokerObjRef, uri, b.Spec.Delivery)
+	expected := resources.NewSubscription(t, brokerTrigger, brokerObjRef, uri, r.internalDeliveryConfigStore.Load())
 
 	sub, err := r.subscriptionLister.Subscriptions(t.Namespace).Get(expected.Name)
 	// If the resource doesn't exist, we'll create it.