Upgrade to latest dependencies and fix incompatible API call (#3891)

* Upgrade to latest dependencies

* Define InstallKafkaBrokerStepFn for newfilters.NewFiltersFeatureSet

The NewFiltersFeatureSet now requires a different type after upgrading
dependencies.

Author: mgencur

go.mod

@@ -35,10 +35,10 @@ require (
 	k8s.io/apiserver v0.29.2
 	k8s.io/client-go v0.29.2
 	k8s.io/utils v0.0.0-20240102154912-e7106e64919e
-	knative.dev/eventing v0.41.1-0.20240507013633-96e5c1853b7c
+	knative.dev/eventing v0.41.1-0.20240510065209-4f3282328ea6
 	knative.dev/hack v0.0.0-20240507013718-68e3bfb39d11
-	knative.dev/pkg v0.0.0-20240507013158-1d1616aa15db
-	knative.dev/reconciler-test v0.0.0-20240507013704-7b06e50b743e
+	knative.dev/pkg v0.0.0-20240507092124-360b72e4948e
+	knative.dev/reconciler-test v0.0.0-20240507120221-c76096ce6188
 	sigs.k8s.io/controller-runtime v0.12.3
 	sigs.k8s.io/yaml v1.4.0
 )

go.sum

@@ -1274,14 +1274,14 @@ k8s.io/utils v0.0.0-20200912215256-4140de9c8800/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20240102154912-e7106e64919e h1:eQ/4ljkx21sObifjzXwlPKpdGLrCfRziVtos3ofG/sQ=
 k8s.io/utils v0.0.0-20240102154912-e7106e64919e/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
-knative.dev/eventing v0.41.1-0.20240507013633-96e5c1853b7c h1:ZJ9+rdbp0/oFOD9en3FTxlHK7ZnZFbxrQKdAulWmTSY=
-knative.dev/eventing v0.41.1-0.20240507013633-96e5c1853b7c/go.mod h1:PLgoapXwPcHAr79U/cZldghvawUKrpZzw202zF9S7q8=
+knative.dev/eventing v0.41.1-0.20240510065209-4f3282328ea6 h1:QphhMEyvWiqAQ858HW2HuMa4/BsOcWwfV2RRSrqlYGo=
+knative.dev/eventing v0.41.1-0.20240510065209-4f3282328ea6/go.mod h1:Sstqyyuh67wF2tnbIR9gQ+lkTQmQpcLUw6tCNz/24gU=
 knative.dev/hack v0.0.0-20240507013718-68e3bfb39d11 h1:CYoD72R8/R35REjeY2nnWfBak+Q3f+NxXwEfwcID1eU=
 knative.dev/hack v0.0.0-20240507013718-68e3bfb39d11/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
-knative.dev/pkg v0.0.0-20240507013158-1d1616aa15db h1:wc8y5CoKhGWsIUxdkTqILXr2i1rJUxtVKQRc3P0tEKI=
-knative.dev/pkg v0.0.0-20240507013158-1d1616aa15db/go.mod h1:PQiq+p9Gr++TnU+w5P+ZpFS5MOwcMF6Y2oWFU8Ou7bw=
-knative.dev/reconciler-test v0.0.0-20240507013704-7b06e50b743e h1:v1MZlnB7hYVSDIPeLBiatmrC2NzJuV2hCo+trOeV+e0=
-knative.dev/reconciler-test v0.0.0-20240507013704-7b06e50b743e/go.mod h1:ctOcB7Yj+TAJ3f+jp+qoyAHnmT0GqzaTABZWSoE5Jrs=
+knative.dev/pkg v0.0.0-20240507092124-360b72e4948e h1:By0c/FKLZlisAerWiMj9crtaEhcl/sp8gN8Rp0s8fKM=
+knative.dev/pkg v0.0.0-20240507092124-360b72e4948e/go.mod h1:PQiq+p9Gr++TnU+w5P+ZpFS5MOwcMF6Y2oWFU8Ou7bw=
+knative.dev/reconciler-test v0.0.0-20240507120221-c76096ce6188 h1:uOzt7ZVFpHoMSfjut8H9d1pafNPF2Luat/w5QMV+CIY=
+knative.dev/reconciler-test v0.0.0-20240507120221-c76096ce6188/go.mod h1:kZEZ0/oQWnS1wBUgQqer/N9k6IzI4jwLLY2xCblEit4=
 pgregory.net/rapid v1.1.0 h1:CMa0sjHSru3puNx+J0MIAuiiEV4N0qj8/cMWGBBCsjw=
 pgregory.net/rapid v1.1.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=

test/e2e_new/new_trigger_filters_test.go

@@ -20,6 +20,7 @@
 package e2e_new
 
 import (
+	"context"
 	"testing"
 
 	"knative.dev/pkg/system"
@@ -45,22 +46,17 @@ func TestNewTriggerFilters(t *testing.T) {
 		environment.Managed(t),
 	)
 
-	env.ParallelTestSet(ctx, t, newfilters.NewFiltersFeatureSet(InstallKafkaBroker))
+	env.ParallelTestSet(ctx, t, newfilters.NewFiltersFeatureSet(InstallKafkaBrokerStepFn))
 }
 
-func InstallKafkaBroker(f *feature.Feature) string {
-	brokerName := feature.MakeRandomK8sName("kafka-broker")
-
-	install, cmName := single_partition_config.MakeInstall()
-
-	f.Setup("install one partition configuration", install)
-	f.Setup("install kafka broker", broker.Install(
-		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
-		broker.WithConfig(cmName),
-	))
-	f.Setup("kafka broker is ready", broker.IsReady(brokerName))
-	f.Setup("kafka broker is addressable", broker.IsAddressable(brokerName))
-
-	return brokerName
+func InstallKafkaBrokerStepFn(brokerName string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		install, cmName := single_partition_config.MakeInstall()
+		install(ctx, t)
+		broker.Install(
+			brokerName,
+			broker.WithBrokerClass(kafka.BrokerClass),
+			broker.WithConfig(cmName),
+		)(ctx, t)
+	}
 }

vendor/knative.dev/eventing/pkg/eventingtls/trust_bundle.go

@@ -135,13 +135,32 @@ func PropagateTrustBundles(ctx context.Context, k8s kubernetes.Interface, trustB
 		if p.userCm == nil {
 			// Update owner references
 			expected.OwnerReferences = withOwnerReferences(obj, gvk, []metav1.OwnerReference{})
-
-			if err := createConfigMap(ctx, k8s, expected); err != nil {
+			err := createConfigMap(ctx, k8s, expected)
+			if err != nil && !apierrors.IsAlreadyExists(err) {
 				return err
 			}
+			if apierrors.IsAlreadyExists(err) {
+				// Update existing ConfigMap
+				cm, getErr := k8s.CoreV1().
+					ConfigMaps(obj.GetNamespace()).
+					Get(ctx, userCMName(p.sysCM.Name), metav1.GetOptions{})
+				if getErr != nil {
+					return err // return original error
+				}
+
+				cm.ObjectMeta.DeepCopyInto(&expected.ObjectMeta)
+				expected.OwnerReferences = withOwnerReferences(obj, gvk, cm.OwnerReferences)
+
+				if !equality.Semantic.DeepDerivative(expected, cm) {
+					if err := updateConfigMap(ctx, k8s, expected); err != nil {
+						return err
+					}
+				}
+			}
 			continue
 		}
 
+		p.userCm.ObjectMeta.DeepCopyInto(&expected.ObjectMeta)
 		// Update owner references
 		expected.OwnerReferences = withOwnerReferences(obj, gvk, p.userCm.OwnerReferences)
 

vendor/knative.dev/eventing/pkg/reconciler/testing/v1/factory.go

@@ -19,16 +19,20 @@ package testing
 import (
 	"context"
 	"encoding/json"
+	"slices"
 	"testing"
 
+	authv1 "k8s.io/api/authorization/v1"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/logging"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	clientgotesting "k8s.io/client-go/testing"
 
 	"k8s.io/client-go/tools/record"
 
@@ -107,6 +111,47 @@ func MakeFactory(ctor Ctor, unstructured bool, logger *zap.SugaredLogger) Factor
 			return ValidateUpdates(ctx, action)
 		})
 
+		kubeClient.PrependReactor("create", "subjectaccessreviews", func(action ktesting.Action) (handled bool, ret runtime.Object, err error) {
+			obj := action.(clientgotesting.CreateAction).GetObject()
+			sar, ok := obj.(*authv1.SubjectAccessReview)
+			if !ok {
+				return false, nil, nil
+			}
+
+			// need a new kubeclient because otherwise we will deadlock
+			ctx, kubeClient := fakekubeclient.With(ctx, ls.GetKubeObjects()...)
+
+			roleBindings, err := kubeClient.RbacV1().RoleBindings(sar.Spec.ResourceAttributes.Namespace).List(ctx, metav1.ListOptions{})
+			logger.Infof("rolebindings: %+v", roleBindings)
+			if err != nil {
+				return false, nil, nil
+			}
+
+			for _, rb := range roleBindings.Items {
+				for _, s := range rb.Subjects {
+					if (s.Name == sar.Spec.User && (s.Kind == "User" || s.Kind == "ServiceAccount")) || (slices.Contains(sar.Spec.Groups, s.Name) && s.Kind == "Group") {
+						role, err := kubeClient.RbacV1().Roles(sar.Spec.ResourceAttributes.Namespace).Get(ctx, rb.RoleRef.Name, metav1.GetOptions{})
+						if err != nil {
+							// let's keep trying for other roles, no reason to fail here
+							continue
+						}
+						for _, rule := range role.Rules {
+							if slices.Contains(rule.APIGroups, sar.Spec.ResourceAttributes.Group) &&
+								(slices.Contains(rule.Resources, "*") || slices.Contains(rule.Resources, sar.Spec.ResourceAttributes.Resource)) &&
+								slices.Contains(rule.Verbs, sar.Spec.ResourceAttributes.Verb) {
+								res := sar.DeepCopy()
+								res.Status.Allowed = true
+								return true, res, nil
+							}
+
+						}
+					}
+				}
+			}
+
+			return true, sar, nil
+		})
+
 		for _, reactor := range r.WithReactors {
 			kubeClient.PrependReactor("*", "*", reactor)
 			client.PrependReactor("*", "*", reactor)

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

@@ -27,6 +27,7 @@ import (
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/delivery"
 	"knative.dev/eventing/test/rekt/resources/trigger"
+	triggerresources "knative.dev/eventing/test/rekt/resources/trigger"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/reconciler-test/pkg/eventshub"
 	eventassert "knative.dev/reconciler-test/pkg/eventshub/assert"
@@ -88,7 +89,10 @@ func BrokerSendEventWithOIDCTokenToSubscriber() *feature.Feature {
 	))
 
 	f.Alpha("Broker").
-		Must("handles event with valid OIDC token", eventassert.OnStore(sink).MatchReceivedEvent(test.HasId(event.ID())).Exact(1))
+		Must("handles event with valid OIDC token", eventassert.OnStore(sink).MatchReceivedEvent(test.HasId(event.ID())).Exact(1)).
+		Must("uses triggers identity for OIDC", eventassert.OnStore(sink).MatchWithContext(
+			eventassert.MatchKind(eventshub.EventReceived).WithContext(),
+			eventassert.MatchOIDCUserFromResource(triggerresources.GVR(), triggerName)).AtLeast(1))
 
 	return f
 }
@@ -145,7 +149,10 @@ func BrokerSendEventWithOIDCTokenToDLS() *feature.Feature {
 
 	// Assert events ended up where we expected.
 	f.Stable("broker with DLS").
-		Must("deliver event to DLQ", eventassert.OnStore(dls).MatchReceivedEvent(test.HasId(event.ID())).AtLeast(1))
+		Must("deliver event to DLQ", eventassert.OnStore(dls).MatchReceivedEvent(test.HasId(event.ID())).AtLeast(1)).
+		Must("uses triggers identity for OIDC", eventassert.OnStore(dls).MatchWithContext(
+			eventassert.MatchKind(eventshub.EventReceived).WithContext(),
+			eventassert.MatchOIDCUserFromResource(triggerresources.GVR(), triggerName)).AtLeast(1))
 
 	return f
 }

vendor/knative.dev/eventing/test/rekt/features/channel/oidc_feature.go

@@ -60,7 +60,11 @@ func DispatcherAuthenticatesRequestsWithOIDC() *feature.Feature {
 	event := test.FullEvent()
 	f.Requirement("install source", eventshub.Install(source, eventshub.InputEvent(event), eventshub.StartSenderToResourceTLS(channel_impl.GVR(), channelName, nil)))
 
-	f.Alpha("channel dispatcher").Must("authenticate requests with OIDC", assert.OnStore(sink).MatchReceivedEvent(test.HasId(event.ID())).AtLeast(1))
+	f.Alpha("channel dispatcher").
+		Must("authenticate requests with OIDC", assert.OnStore(sink).MatchReceivedEvent(test.HasId(event.ID())).AtLeast(1)).
+		Must("uses subscriptions identity for OIDC", assert.OnStore(sink).MatchWithContext(
+			assert.MatchKind(eventshub.EventReceived).WithContext(),
+			assert.MatchOIDCUserFromResource(subscription.GVR(), subscriptionName)).AtLeast(1))
 
 	return f
 }

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

@@ -17,15 +17,21 @@ limitations under the License.
 package new_trigger_filters
 
 import (
+	"context"
 	"fmt"
 
 	. "github.com/cloudevents/sdk-go/v2/test"
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/trigger"
+	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+	"knative.dev/reconciler-test/pkg/resources/service"
 
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 )
 
-type InstallBrokerFunc func(f *feature.Feature) string
+type InstallBrokerFunc func(brokerName string) feature.StepFn
 
 type CloudEventsContext struct {
 	eventType            string
@@ -329,14 +335,40 @@ func MultipleTriggersAndSinksFeature(installBroker InstallBrokerFunc) *feature.F
 		},
 	}
 
-	// We need to create the broker here and mock it later so that the test uses the same broker for both filters
-	brokerName := installBroker(f)
-	fakeInstallBroker := func(_ *feature.Feature) string {
-		return brokerName
-	}
-
-	createNewFiltersFeature(f, eventContextsFirstSink, filtersFirstTrigger, eventingv1.TriggerFilter{}, fakeInstallBroker)
-	createNewFiltersFeature(f, eventContextsSecondSink, filtersSecondTrigger, eventingv1.TriggerFilter{}, fakeInstallBroker)
+	subscriberName1 := feature.MakeRandomK8sName("subscriber1")
+	subscriberName2 := feature.MakeRandomK8sName("subscriber2")
+	triggerName1 := feature.MakeRandomK8sName("trigger1")
+	triggerName2 := feature.MakeRandomK8sName("trigger2")
+	brokerName := feature.MakeRandomK8sName("broker")
+
+	f.Setup("Install Broker, Sinks, Triggers", func(ctx context.Context, t feature.T) {
+		installBroker(brokerName)(ctx, t)
+
+		eventshub.Install(subscriberName1, eventshub.StartReceiver)(ctx, t)
+		eventshub.Install(subscriberName2, eventshub.StartReceiver)(ctx, t)
+
+		triggerCfg1 := []manifest.CfgFn{
+			trigger.WithSubscriber(service.AsKReference(subscriberName1), ""),
+			trigger.WithNewFilters(filtersFirstTrigger),
+			trigger.WithFilter(eventingv1.TriggerFilter{}.Attributes),
+		}
+		trigger.Install(triggerName1, brokerName, triggerCfg1...)(ctx, t)
+
+		triggerCfg2 := []manifest.CfgFn{
+			trigger.WithSubscriber(service.AsKReference(subscriberName2), ""),
+			trigger.WithNewFilters(filtersSecondTrigger),
+			trigger.WithFilter(eventingv1.TriggerFilter{}.Attributes),
+		}
+		trigger.Install(triggerName2, brokerName, triggerCfg2...)(ctx, t)
+
+		broker.IsReady(brokerName)(ctx, t)
+		broker.IsAddressable(brokerName)(ctx, t)
+		trigger.IsReady(triggerName1)(ctx, t)
+		trigger.IsReady(triggerName2)(ctx, t)
+	})
+
+	assertDelivery(f, brokerName, subscriberName1, eventContextsFirstSink)
+	assertDelivery(f, brokerName, subscriberName2, eventContextsSecondSink)
 
 	return f
 }

vendor/knative.dev/eventing/test/rekt/features/new_trigger_filters/filters.go

@@ -17,39 +17,49 @@ limitations under the License.
 package new_trigger_filters
 
 import (
+	"context"
 	"fmt"
 
+	"github.com/cloudevents/sdk-go/v2/event"
 	. "github.com/cloudevents/sdk-go/v2/test"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"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/pkg/resources/service"
-
-	"github.com/cloudevents/sdk-go/v2/event"
-	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
-	"knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/eventing/test/rekt/resources/trigger"
 )
 
 func createNewFiltersFeature(f *feature.Feature, eventContexts []CloudEventsContext, filters []eventingv1.SubscriptionsAPIFilter, filter eventingv1.TriggerFilter, installBroker InstallBrokerFunc) {
 	subscriberName := feature.MakeRandomK8sName("subscriber")
 	triggerName := feature.MakeRandomK8sName("trigger")
-	brokerName := installBroker(f)
+	brokerName := feature.MakeRandomK8sName("broker")
 
-	f.Setup("Install trigger subscriber", eventshub.Install(subscriberName, eventshub.StartReceiver))
+	f.Setup("Install Sink, Broker, Trigger", func(ctx context.Context, t feature.T) {
+		installBroker(brokerName)(ctx, t)
 
-	cfg := []manifest.CfgFn{
-		trigger.WithSubscriber(service.AsKReference(subscriberName), ""),
-		trigger.WithNewFilters(filters),
-		trigger.WithFilter(filter.Attributes),
-	}
+		eventshub.Install(subscriberName, eventshub.StartReceiver)(ctx, t)
 
-	f.Setup("Install trigger", trigger.Install(triggerName, brokerName, cfg...))
-	f.Setup("Wait for trigger to become ready", trigger.IsReady(triggerName))
+		triggerCfg := []manifest.CfgFn{
+			trigger.WithSubscriber(service.AsKReference(subscriberName), ""),
+			trigger.WithNewFilters(filters),
+			trigger.WithFilter(filter.Attributes),
+		}
 
-	asserter := f.Beta("New filters")
+		trigger.Install(triggerName, brokerName, triggerCfg...)(ctx, t)
 
+		broker.IsReady(brokerName)(ctx, t)
+		broker.IsAddressable(brokerName)(ctx, t)
+		trigger.IsReady(triggerName)(ctx, t)
+	})
+
+	assertDelivery(f, brokerName, subscriberName, eventContexts)
+}
+
+func assertDelivery(f *feature.Feature, brokerName, subscriberName string, eventContexts []CloudEventsContext) {
+	asserter := f.Beta("New filters")
 	for _, eventCtx := range eventContexts {
 		e := newEventFromEventContext(eventCtx)
 		eventSender := feature.MakeRandomK8sName("sender")

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

@@ -69,7 +69,8 @@ func TriggerDependencyAnnotation() *feature.Feature {
 	// Install the trigger
 	f.Setup("install trigger", trigger.Install(triggerName, brokerName, cfg...))
 
-	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
+	// trigger won't go ready until after the pingsource exists, because of the dependency annotation
+	f.Requirement("trigger goes ready", trigger.IsReady(triggerName))
 
 	f.Requirement("install pingsource", func(ctx context.Context, t feature.T) {
 		brokeruri, err := broker.Address(ctx, brokerName)