forked from knative-extensions/eventing-kafka-broker
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[release-1.14] Refactor NewFiltersFeatureSet to prevent starvation (k…
…native-extensions#1086) * Define InstallKafkaBrokerStepFn for newfilters.NewFiltersFeatureSet The NewFiltersFeatureSet now requires a different type after upgrading dependencies. * Refactor NewFiltersFeatureSet to prevent starvation * Add patch
- Loading branch information
Showing
6 changed files
with
249 additions
and
46 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,168 @@ | ||
| commit 33d3d89e28a64d97d615c017b984974cac28cc9f | ||
| Author: Martin Gencur <mgencur@redhat.com> | ||
| Date: Fri May 10 10:21:41 2024 +0200 | ||
|
|
||
| Refactor NewFiltersFeatureSet to prevent starvation | ||
|
|
||
| diff --git a/vendor/knative.dev/eventing/test/rekt/features/new_trigger_filters/feature.go b/vendor/knative.dev/eventing/test/rekt/features/new_trigger_filters/feature.go | ||
| index 6da3b0a69..ee95bab4e 100644 | ||
| --- a/vendor/knative.dev/eventing/test/rekt/features/new_trigger_filters/feature.go | ||
| +++ b/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 | ||
| } | ||
| diff --git a/vendor/knative.dev/eventing/test/rekt/features/new_trigger_filters/filters.go b/vendor/knative.dev/eventing/test/rekt/features/new_trigger_filters/filters.go | ||
| index f9a83d1f4..b07d66f81 100644 | ||
| --- a/vendor/knative.dev/eventing/test/rekt/features/new_trigger_filters/filters.go | ||
| +++ b/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") | ||
| diff --git a/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go b/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go | ||
| index b3abbda98..d5c618514 100644 | ||
| --- a/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go | ||
| +++ b/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go | ||
| @@ -278,10 +278,6 @@ func InstallMTBroker(name string) *feature.Feature { | ||
| return f | ||
| } | ||
|
|
||
| -func InstallMTBrokerIntoFeature(f *feature.Feature) string { | ||
| - brokerName := feature.MakeRandomK8sName("broker") | ||
| - f.Setup(fmt.Sprintf("Install broker %q", brokerName), Install(brokerName, WithEnvConfig()...)) | ||
| - f.Setup("Broker is ready", IsReady(brokerName)) | ||
| - f.Setup("Broker is addressable", k8s.IsAddressable(GVR(), brokerName)) | ||
| - return brokerName | ||
| +func InstallMTBrokerStepFn(brokerName string) feature.StepFn { | ||
| + return Install(brokerName, WithEnvConfig()...) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters