Glue code + e2e test with channel

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

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) {

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 TestChannelToChannel(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,116 @@
+/*
+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/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 := 5 * time.Second
+	timeoutString := "PT5S"
+
+	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: channelAPIVersion,
+							Kind:       imcKind,
+							Name:       sinkName,
+						},
+					},
+					Delivery: &eventingduckv1.DeliverySpec{
+						DeadLetterSink: &duckv1.Destination{
+							Ref: &duckv1.KReference{
+								APIVersion: channelAPIVersion,
+								Kind:       imcKind,
+								Name:       channelName,
+							},
+						},
+						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
+}