upgrade to latest dependencies (#1046)

bumping knative.dev/eventing bd67450...1f9f4d3:
  > 1f9f4d3 Log stream from system namespace in upgrade tests (# 6699)
  > 466d123 Fix reconciler-tests Prow job (# 6694)

Signed-off-by: Knative Automation <[email protected]>

Signed-off-by: Knative Automation <[email protected]>

Author: knative-automation

go.mod

@@ -21,7 +21,7 @@ require (
 	k8s.io/client-go v0.25.4
 	k8s.io/code-generator v0.25.4
 	k8s.io/kube-openapi v0.0.0-20220803162953-67bda5d908f1
-	knative.dev/eventing v0.35.1-0.20230124081938-bd674502e6b8
+	knative.dev/eventing v0.35.1-0.20230125080738-1f9f4d37aa56
 	knative.dev/hack v0.0.0-20230113013652-c7cfcb062de9
 	knative.dev/pkg v0.0.0-20230117181655-247510c00e9d
 	knative.dev/reconciler-test v0.0.0-20230123181139-476a442e3644

go.sum

@@ -1054,8 +1054,8 @@ k8s.io/kube-openapi v0.0.0-20220803162953-67bda5d908f1 h1:MQ8BAZPZlWk3S9K4a9NCkI
 k8s.io/kube-openapi v0.0.0-20220803162953-67bda5d908f1/go.mod h1:C/N6wCaBHeBHkHUesQOQy2/MZqGgMAFPqGsGQLdbZBU=
 k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2 h1:GfD9OzL11kvZN5iArC6oTS7RTj7oJOIfnislxYlqTj8=
 k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
-knative.dev/eventing v0.35.1-0.20230124081938-bd674502e6b8 h1:Kkdh5EWgWPy4JLbIGpW6Y5aeKdTMX/kC6YSWXj1pOyw=
-knative.dev/eventing v0.35.1-0.20230124081938-bd674502e6b8/go.mod h1:PqYrXKXhZU7rQaS5TQuZDSOd9jPX7AegF8uNNUY4kcU=
+knative.dev/eventing v0.35.1-0.20230125080738-1f9f4d37aa56 h1:aBxfcRszMz6Ey6IgNVrEeDrkhApAeV8TV09qq95JtU4=
+knative.dev/eventing v0.35.1-0.20230125080738-1f9f4d37aa56/go.mod h1:Qka5Z6+LeMoHGL1QAznVdmq5LAu21b4F3rgxc2AMgRg=
 knative.dev/hack v0.0.0-20230113013652-c7cfcb062de9 h1:CDa7s9KspEZqPhk7cN68ZypRLuAvSgr+knoOaXSsrHk=
 knative.dev/hack v0.0.0-20230113013652-c7cfcb062de9/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
 knative.dev/pkg v0.0.0-20230117181655-247510c00e9d h1:pjKDcvHoMib8nRp56eISRmMj/pFMzJljnzvMvGCIReI=

vendor/knative.dev/eventing/test/rekt/features/broker/control_plane.go

@@ -370,8 +370,7 @@ func addControlPlaneDelivery(fs *feature.FeatureSet, brokerOpts ...manifest.CfgF
 
 		// Send an event into the matrix and hope for the best
 		prober.SenderFullEvents(1)
-		f.Setup("install source", prober.SenderInstall("source"))
-		f.Requirement("sender is finished", prober.SenderDone("source"))
+		f.Requirement("install source", prober.SenderInstall("source"))
 
 		// All events have been sent, time to look at the specs and confirm we got them.
 		expectedEvents := createExpectedEventPatterns(tt.brokerDS, cfg)

vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go

@@ -19,6 +19,7 @@ package broker
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/google/uuid"
 	"knative.dev/reconciler-test/pkg/environment"
@@ -56,13 +57,10 @@ func SourceToSink(brokerName string) *feature.Feature {
 
 	f.Setup("trigger goes ready", trigger.IsReady(via))
 
-	f.Setup("install source", func(ctx context.Context, t feature.T) {
-		u, err := broker.Address(ctx, brokerName)
-		if err != nil || u == nil {
-			t.Error("failed to get the address of the broker", brokerName, err)
-		}
-		eventshub.Install(source, eventshub.StartSenderURL(u.String()), eventshub.InputEvent(event))(ctx, t)
-	})
+	f.Requirement("install source", eventshub.Install(source,
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
+		eventshub.InputEvent(event),
+	))
 
 	f.Stable("broker as middleware").
 		Must("deliver an event",
@@ -97,9 +95,9 @@ func SourceToSinkWithDLQ() *feature.Feature {
 	ce := FullEvent()
 	ce.SetID(uuid.New().String())
 
-	// Install events after data plane is ready.
+	// Send events after data plane is ready.
 	f.Requirement("install source", eventshub.Install(source,
-		eventshub.StartSenderToResource(broker.GVR(), source),
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEvent(ce),
 	))
 
@@ -123,15 +121,7 @@ func SourceToSinkWithFlakyDLQ(brokerName string) *feature.Feature {
 	dlq := feature.MakeRandomK8sName("dlq")
 	via := feature.MakeRandomK8sName("via")
 
-	uuids := []string{
-		uuid.New().String(),
-		uuid.New().String(),
-		uuid.New().String(),
-	}
-
-	f := feature.NewFeature()
-
-	//
+	f := feature.NewFeatureNamed("Source to sink with flaky DLS")
 
 	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
 
@@ -142,39 +132,25 @@ func SourceToSinkWithFlakyDLQ(brokerName string) *feature.Feature {
 	})
 
 	f.Setup("install dlq", eventshub.Install(dlq, eventshub.StartReceiver))
-
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver, eventshub.DropFirstN(2)))
 	f.Setup("update broker with DLQ", broker.Install(brokerName, broker.WithDeadLetterSink(svc.AsKReference(dlq), "")))
-
-	// Point the Trigger subscriber to the sink svc.
-	cfg := []manifest.CfgFn{trigger.WithSubscriber(flaker.AsRef(flake), "")}
-
-	// Install the trigger
-	f.Setup("install trigger", trigger.Install(via, brokerName, cfg...))
-
+	f.Setup("install trigger", trigger.Install(via, brokerName, trigger.WithSubscriber(svc.AsKReference(sink), "")))
 	f.Setup("trigger goes ready", trigger.IsReady(via))
+	f.Setup("broker goes ready", broker.IsReady(via))
 
-	f.Setup("install source", func(ctx context.Context, t feature.T) {
-		u, err := broker.Address(ctx, brokerName)
-		if err != nil || u == nil {
-			t.Error("failed to get the address of the broker", brokerName, err)
-		}
-
-		opts := []eventshub.EventsHubOption{eventshub.StartSenderURL(u.String())}
-		for _, id := range uuids {
-			event := FullEvent()
-			event.SetID(id)
-			opts = append(opts, eventshub.InputEvent(event))
-		}
-		eventshub.Install(source, opts...)(ctx, t)
-	})
+	f.Requirement("install source", eventshub.Install(source,
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
+		eventshub.SendMultipleEvents(3, time.Millisecond),
+		eventshub.EnableIncrementalId,
+	))
 
-	f.Stable("broker with DQL").
+	f.Stable("broker with DLS").
 		Must("deliver event flaky sent to DLQ event[0]",
-			OnStore(dlq).MatchEvent(HasId(uuids[0])).Exact(1)).
+			OnStore(dlq).MatchEvent(HasId("1")).Exact(1)).
 		Must("deliver event flaky sent to DLQ event[1]",
-			OnStore(dlq).MatchEvent(HasId(uuids[1])).Exact(1)).
+			OnStore(dlq).MatchEvent(HasId("2")).Exact(1)).
 		Must("deliver event sink receiver got event[2]",
-			OnStore(sink).MatchEvent(HasId(uuids[2])).Exact(1))
+			OnStore(sink).MatchEvent(HasId("3")).Exact(1))
 
 	return f
 }

vendor/knative.dev/eventing/test/rekt/features/trigger/feature.go

@@ -20,14 +20,15 @@ import (
 	"context"
 
 	"github.com/cloudevents/sdk-go/v2/test"
-	"knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/eventing/test/rekt/resources/pingsource"
-	"knative.dev/eventing/test/rekt/resources/trigger"
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/eventshub/assert"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/manifest"
 	"knative.dev/reconciler-test/resources/svc"
+
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/pingsource"
+	"knative.dev/eventing/test/rekt/resources/trigger"
 )
 
 // This test is for avoiding regressions on the trigger dependency annotation functionality.
@@ -40,8 +41,8 @@ func TriggerDependencyAnnotation() *feature.Feature {
 	//Install the broker
 	brokerName := feature.MakeRandomK8sName("broker")
 	f.Setup("install broker", broker.Install(brokerName, broker.WithEnvConfig()...))
-	f.Requirement("broker is ready", broker.IsReady(brokerName))
-	f.Requirement("broker is addressable", broker.IsAddressable(brokerName))
+	f.Setup("broker is ready", broker.IsReady(brokerName))
+	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
 
 	psourcename := "test-ping-source-annotation"
 	dependencyAnnotation := `{"kind":"PingSource","name":"test-ping-source-annotation","apiVersion":"sources.knative.dev/v1"}`
@@ -61,7 +62,7 @@ func TriggerDependencyAnnotation() *feature.Feature {
 
 	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
 
-	f.Setup("install pingsource", func(ctx context.Context, t feature.T) {
+	f.Requirement("install pingsource", func(ctx context.Context, t feature.T) {
 		brokeruri, err := broker.Address(ctx, brokerName)
 		if err != nil {
 			t.Error("failed to get address of broker", err)
@@ -73,7 +74,7 @@ func TriggerDependencyAnnotation() *feature.Feature {
 		}
 		pingsource.Install(psourcename, cfg...)(ctx, t)
 	})
-	f.Setup("PingSource goes ready", pingsource.IsReady(psourcename))
+	f.Requirement("PingSource goes ready", pingsource.IsReady(psourcename))
 
 	f.Stable("pingsource as event source to test trigger with annotations").
 		Must("delivers events on broker with URI", assert.OnStore(sink).MatchEvent(

vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go

@@ -18,14 +18,12 @@ package trigger
 
 import (
 	"context"
-	"fmt"
 
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
 
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/delivery"
-	"knative.dev/eventing/test/rekt/resources/eventlibrary"
 	"knative.dev/eventing/test/rekt/resources/trigger"
 )
 
@@ -45,14 +43,7 @@ func SourceToTriggerSinkWithDLS() *feature.Feature {
 
 	prober := eventshub.NewProber()
 	prober.SetTargetResource(broker.GVR(), brokerName)
-
-	lib := feature.MakeRandomK8sName("lib")
-	f.Setup("install events", eventlibrary.Install(lib))
-	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
-	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
-	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
-		panic(fmt.Errorf("can not find event files: %s", err))
-	}
+	prober.SenderFullEvents(3)
 
 	f.Setup("install broker", broker.Install(brokerName, broker.WithEnvConfig()...))
 
@@ -70,10 +61,7 @@ func SourceToTriggerSinkWithDLS() *feature.Feature {
 	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
 
 	// Install sender.
-	f.Setup("install source", prober.SenderInstall("source"))
-
-	// After we have finished sending.
-	f.Requirement("sender is finished", prober.SenderDone("source"))
+	f.Requirement("install source", prober.SenderInstall("source"))
 
 	// Assert events ended up where we expected.
 	f.Stable("trigger with DLS").
@@ -100,14 +88,7 @@ func SourceToTriggerSinkWithDLSDontUseBrokers() *feature.Feature {
 
 	prober := eventshub.NewProber()
 	prober.SetTargetResource(broker.GVR(), brokerName)
-
-	lib := feature.MakeRandomK8sName("lib")
-	f.Setup("install events", eventlibrary.Install(lib))
-	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
-	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
-	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
-		panic(fmt.Errorf("can not find event files: %s", err))
-	}
+	prober.SenderFullEvents(3)
 
 	// Setup Probes
 	f.Setup("install trigger recorder", prober.ReceiverInstall(triggerSinkName))
@@ -132,10 +113,7 @@ func SourceToTriggerSinkWithDLSDontUseBrokers() *feature.Feature {
 	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
 
 	// Install events after topology is ready.
-	f.Setup("install source", prober.SenderInstall("source"))
-
-	// After we have finished sending.
-	f.Requirement("sender is finished", prober.SenderDone("source"))
+	f.Requirement("install source", prober.SenderInstall("source"))
 
 	// Assert events ended up where we expected.
 	f.Stable("trigger with a valid DLS ref").
@@ -163,13 +141,7 @@ func BadTriggerDoesNotAffectOkTrigger() *feature.Feature {
 	sink := feature.MakeRandomK8sName("sink")
 	source := feature.MakeRandomK8sName("source")
 
-	lib := feature.MakeRandomK8sName("lib")
-	f.Setup("install events", eventlibrary.Install(lib))
-	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
-	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
-	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
-		panic(fmt.Errorf("can not find event files: %s", err))
-	}
+	prober.SenderFullEvents(3)
 
 	// Setup Probes
 	f.Setup("install dlq", prober.ReceiverInstall(dlq))
@@ -192,11 +164,6 @@ func BadTriggerDoesNotAffectOkTrigger() *feature.Feature {
 	// Install events after data plane is ready.
 	f.Requirement("install source", prober.SenderInstall(source))
 
-	// After we have finished sending.
-	f.Requirement("sender is finished", prober.SenderDone(source))
-	f.Requirement("receiver 1 is finished", prober.ReceiverDone(source, dlq))
-	f.Requirement("receiver 2 is finished", prober.ReceiverDone(source, sink))
-
 	// Assert events ended up where we expected.
 	f.Stable("broker with DLQ").
 		Must("accepted all events", prober.AssertSentAll(source)).

vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.go

@@ -19,14 +19,17 @@ package channel_impl
 import (
 	"context"
 	"embed"
+	"encoding/json"
 	"log"
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/injection/clients/dynamicclient"
@@ -108,21 +111,35 @@ func IsAddressable(name string, timing ...time.Duration) feature.StepFn {
 func HasDeadLetterSinkURI(name string, gvr schema.GroupVersionResource) feature.StepFn {
 	return func(ctx context.Context, t feature.T) {
 		ns := environment.FromContext(ctx).Namespace()
-		ch, err := dynamicclient.Get(ctx).
-			Resource(gvr).
-			Namespace(ns).
-			Get(ctx, name, metav1.GetOptions{})
+		interval, timeout := environment.PollTimingsFromContext(ctx)
+		var lastState *eventingduck.Channelable
+		err := wait.Poll(interval, timeout, func() (done bool, err error) {
+			ch, err := dynamicclient.Get(ctx).
+				Resource(gvr).
+				Namespace(ns).
+				Get(ctx, name, metav1.GetOptions{})
+			if apierrors.IsNotFound(err) {
+				return false, nil
+			}
+			if err != nil {
+				t.Fatalf("failed to get %s/%s channel: %v", ns, name, err)
+			}
+
+			channelable := &eventingduck.Channelable{}
+			if err := runtime.DefaultUnstructuredConverter.FromUnstructured(ch.UnstructuredContent(), channelable); err != nil {
+				t.Fatal(err)
+			}
+			lastState = channelable
+
+			if channelable.Status.DeadLetterSinkURI.String() == "" {
+				return false, nil
+			}
+
+			return true, nil
+		})
 		if err != nil {
-			t.Fatalf("failed to get %s/%s channel: %v", ns, name, err)
-		}
-
-		channelable := &eventingduck.Channelable{}
-		if err := runtime.DefaultUnstructuredConverter.FromUnstructured(ch.UnstructuredContent(), channelable); err != nil {
-			t.Fatal(err)
-		}
-
-		if channelable.Status.DeadLetterSinkURI.String() == "" {
-			t.Fatalf("channel %s/%s has no dead letter sink uri in the status", ns, name)
+			bytes, _ := json.MarshalIndent(lastState, "", "  ")
+			t.Errorf("failed to verify channel has dead letter sink: %w, last state:\n%s", err, string(bytes))
 		}
 	}
 }