Glue Delivery.Timeout to RetryConfig.RequestTimeout

config/channels/in-memory-channel/resources/in-memory-channel.yaml

@@ -118,6 +118,7 @@ spec:
                           description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                           type: integer
                           format: int32
+                      x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature
                     generation:
                       description: Generation of the origin of the subscriber with uid:UID.
                       type: integer

config/core/resources/channel.yaml

@@ -146,6 +146,7 @@ spec:
                           description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                           type: integer
                           format: int32
+                      x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature
                     generation:
                       description: Generation of the origin of the subscriber with uid:UID.
                       type: integer

pkg/channel/fanout/fanout_message_handler.go

@@ -33,7 +33,6 @@ import (
 	"github.com/cloudevents/sdk-go/v2/binding/buffering"
 	"go.opencensus.io/trace"
 	"go.uber.org/zap"
-
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/channel"
 	"knative.dev/eventing/pkg/kncloudevents"

pkg/kncloudevents/message_sender_test.go

@@ -165,8 +165,8 @@ func TestHTTPMessageSenderSendWithRetriesWithSingleRequestTimeout(t *testing.T)
 		if newVal >= 5 {
 			writer.WriteHeader(http.StatusOK)
 		} else {
-			// Let's add one more second
-			time.Sleep(timeout)
+			// Let's add a bit more time
+			time.Sleep(timeout + (200 * time.Millisecond))
 			writer.WriteHeader(http.StatusAccepted)
 		}
 	}))

pkg/kncloudevents/retries.go

@@ -104,6 +104,14 @@ func RetryConfigFromDeliverySpec(spec v1.DeliverySpec) (RetryConfig, error) {
 		}
 	}
 
+	if spec.Timeout != nil {
+		timeout, err := period.Parse(*spec.Timeout)
+		if err != nil {
+			return retryConfig, fmt.Errorf("failed to parse Spec.Timeout: %w", err)
+		}
+		retryConfig.RequestTimeout, _ = timeout.Duration()
+	}
+
 	return retryConfig, nil
 }
 

pkg/kncloudevents/retries_test.go

@@ -339,6 +339,7 @@ func TestRetryConfigFromDeliverySpecCheckRetry(t *testing.T) {
 			Retry:         pointer.Int32Ptr(10),
 			BackoffPolicy: &linear,
 			BackoffDelay:  pointer.StringPtr("PT1S"),
+			Timeout:       pointer.StringPtr("PT10S"),
 		},
 	}, {
 		name: "only retry",
@@ -347,15 +348,21 @@ func TestRetryConfigFromDeliverySpecCheckRetry(t *testing.T) {
 			BackoffPolicy: &linear,
 		},
 	}, {
-		name: "not ISO8601",
+		name: "delay not ISO8601",
 		spec: v1.DeliverySpec{
 			Retry:         pointer.Int32Ptr(10),
 			BackoffDelay:  pointer.StringPtr("PP1"),
 			BackoffPolicy: &linear,
 		},
 		wantErr: true,
-	},
-	}
+	}, {
+		name: "timeout not ISO8601",
+		spec: v1.DeliverySpec{
+			Retry:   pointer.Int32Ptr(10),
+			Timeout: pointer.StringPtr("PP1"),
+		},
+		wantErr: true,
+	}}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

pkg/reconciler/subscription/subscription.go

@@ -510,13 +510,14 @@ func deliverySpec(sub *v1.Subscription, channel *eventingduckv1.Channelable) (de
 				},
 			}
 		}
-		if channel.Spec.Delivery.BackoffDelay != nil || channel.Spec.Delivery.Retry != nil || channel.Spec.Delivery.BackoffPolicy != nil {
+		if channel.Spec.Delivery.BackoffDelay != nil || channel.Spec.Delivery.Retry != nil || channel.Spec.Delivery.BackoffPolicy != nil || channel.Spec.Delivery.Timeout != nil {
 			if delivery == nil {
 				delivery = &eventingduckv1.DeliverySpec{}
 			}
 			delivery.BackoffPolicy = channel.Spec.Delivery.BackoffPolicy
 			delivery.Retry = channel.Spec.Delivery.Retry
 			delivery.BackoffDelay = channel.Spec.Delivery.BackoffDelay
+			delivery.Timeout = channel.Spec.Delivery.Timeout
 		}
 		return
 	}
@@ -530,13 +531,14 @@ func deliverySpec(sub *v1.Subscription, channel *eventingduckv1.Channelable) (de
 			},
 		}
 	}
-	if sub.Spec.Delivery != nil && (sub.Spec.Delivery.BackoffDelay != nil || sub.Spec.Delivery.Retry != nil || sub.Spec.Delivery.BackoffPolicy != nil) {
+	if sub.Spec.Delivery != nil && (sub.Spec.Delivery.BackoffDelay != nil || sub.Spec.Delivery.Retry != nil || sub.Spec.Delivery.BackoffPolicy != nil || sub.Spec.Delivery.Timeout != nil) {
 		if delivery == nil {
 			delivery = &eventingduckv1.DeliverySpec{}
 		}
 		delivery.BackoffPolicy = sub.Spec.Delivery.BackoffPolicy
 		delivery.Retry = sub.Spec.Delivery.Retry
 		delivery.BackoffDelay = sub.Spec.Delivery.BackoffDelay
+		delivery.Timeout = sub.Spec.Delivery.Timeout
 	}
 	return
 }

test/experimental/config/features.yaml

@@ -23,3 +23,4 @@ metadata:
     knative.dev/config-category: eventing
 data:
   kreference-group: "enabled"
+  delivery-timeout: "enabled"

test/experimental/delivery_timeout_test.go

@@ -0,0 +1,43 @@
+// +build e2e
+
+/*
+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 experimental
+
+import (
+	"testing"
+
+	"knative.dev/eventing/test/experimental/features/delivery_timeout"
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+)
+
+func TestDeliveryTimeout(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, delivery_timeout.ChannelToSink())
+}

test/experimental/features/delivery_timeout/channel.go

@@ -0,0 +1,119 @@
+/*
+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 delivery_timeout
+
+import (
+	"context"
+	"time"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/rickb777/date/period"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
+	"knative.dev/eventing/test/rekt/resources/channel"
+	"knative.dev/eventing/test/rekt/resources/subscription"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/eventshub/assert"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+// ChannelToSink tests a scenario where the flow is source -> channel -> sink (timeout) -- fallback to -> dead letter sink
+func ChannelToSink() *feature.Feature {
+	f := feature.NewFeature()
+
+	timeout := 6 * time.Second
+	// Clocks are funny, let's add 1 second to take in account eventual clock skews
+	timeoutPeriod, _ := period.NewOf(timeout - time.Second)
+	timeoutString := timeoutPeriod.String()
+
+	channelAPIVersion, imcKind := channel.GVK().ToAPIVersionAndKind()
+
+	channelName := feature.MakeRandomK8sName("channel")
+	subName := feature.MakeRandomK8sName("sub-sink")
+	sinkName := feature.MakeRandomK8sName("sink")
+	deadLetterSinkName := feature.MakeRandomK8sName("dead-letter")
+	sourceName := feature.MakeRandomK8sName("source")
+
+	ev := cetest.FullEvent()
+
+	f.Setup("install sink", eventshub.Install(
+		sinkName,
+		eventshub.StartReceiver,
+		eventshub.ResponseWaitTime(timeout),
+	))
+	f.Setup("install dead letter sink", eventshub.Install(
+		deadLetterSinkName,
+		eventshub.StartReceiver,
+	))
+
+	f.Setup("Install channel", channel.Install(channelName))
+	f.Setup("Channel is ready", channel.IsReady(channelName))
+
+	f.Setup("Install channel -> sink subscription", func(ctx context.Context, t feature.T) {
+		namespace := environment.FromContext(ctx).Namespace()
+		_, err := eventingclient.Get(ctx).MessagingV1().Subscriptions(namespace).Create(ctx,
+			&messagingv1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      subName,
+					Namespace: namespace,
+				},
+				Spec: messagingv1.SubscriptionSpec{
+					Channel: duckv1.KReference{
+						APIVersion: channelAPIVersion,
+						Kind:       imcKind,
+						Name:       channelName,
+					},
+					Subscriber: &duckv1.Destination{
+						Ref: &duckv1.KReference{
+							APIVersion: "v1",
+							Kind:       "Service",
+							Name:       sinkName,
+						},
+					},
+					Delivery: &eventingduckv1.DeliverySpec{
+						DeadLetterSink: &duckv1.Destination{
+							Ref: &duckv1.KReference{
+								APIVersion: "v1",
+								Kind:       "Service",
+								Name:       deadLetterSinkName,
+							},
+						},
+						Timeout: &timeoutString,
+					},
+				},
+			}, metav1.CreateOptions{})
+		require.NoError(t, err)
+	})
+
+	f.Setup("subscription channel -> Sink is ready", subscription.IsReady(subName))
+
+	f.Setup("install source", eventshub.Install(
+		sourceName,
+		eventshub.StartSenderToResource(channel.GVR(), channelName),
+		eventshub.InputEvent(ev),
+	))
+
+	f.Assert("receive event on sink", assert.OnStore(sinkName).MatchEvent(cetest.HasId(ev.ID())).Exact(1))
+	f.Assert("receive event on dead letter sink", assert.OnStore(deadLetterSinkName).MatchEvent(cetest.HasId(ev.ID())).Exact(1))
+
+	return f
+}