Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Create eventtypes on reply events to triggers and subscriptions #4077

Merged
merged 13 commits into from
Aug 26, 2024
Merged

Create eventtypes on reply events to triggers and subscriptions #4077

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
fec9ff7
feat: refactor contract to have shared feature flags message
Cali0707 Aug 20, 2024
ef536b0
feat: control plane provisions new feature flag message in contract
Cali0707 Aug 20, 2024
8f9ad45
feat: receiver uses new feature flags message in contract
Cali0707 Aug 20, 2024
d0c44c7
feat: handle autocreate on reply events
Cali0707 Aug 20, 2024
2c3dc6a
feat: set up dispatcher to handle et autocreate
Cali0707 Aug 20, 2024
3589fb7
test: add e2e test to verify eventtypes are created on reply
Cali0707 Aug 20, 2024
7560453
fix merge conflicts
Cali0707 Aug 20, 2024
8759c0c
fix: do not autocreate events on channel when owned by a broker
Cali0707 Aug 20, 2024
e26f729
fix(test): kafka sink oidc tests work with featureflags contract
Cali0707 Aug 20, 2024
2263a6b
fix: base response handler build issue is resolved
Cali0707 Aug 20, 2024
e07ee21
fix: consumer provisions feature flags
Cali0707 Aug 20, 2024
48f2554
fix merge conflicts
Cali0707 Aug 21, 2024
7192915
fix: reference is correctly provisioned for dispatchers in control pl…
Cali0707 Aug 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
294 changes: 179 additions & 115 deletions control-plane/pkg/contract/contract.pb.go
View file
Open in desktop

Large diffs are not rendered by default.

6 changes: 4 additions & 2 deletions control-plane/pkg/reconciler/broker/broker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -630,8 +630,10 @@ func (r *Reconciler) reconcilerBrokerResource(ctx context.Context, topic string,
Uid: string(broker.UID),
Topics: []string{topic},
Ingress: &contract.Ingress{
Path: receiver.PathFromObject(broker),
EnableAutoCreateEventTypes: features.IsEnabled(feature.EvenTypeAutoCreate),
Path: receiver.PathFromObject(broker),
},
FeatureFlags: &contract.FeatureFlags{
EnableEventTypeAutocreate: features.IsEnabled(feature.EvenTypeAutoCreate),
},
BootstrapServers: config.GetBootstrapServers(),
Reference: &contract.Reference{
Expand Down
31 changes: 30 additions & 1 deletion control-plane/pkg/reconciler/broker/broker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -253,6 +254,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -375,6 +377,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -439,6 +442,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -502,6 +506,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
BootstrapServers: bootstrapServers,
EgressConfig: &contract.EgressConfig{DeadLetter: ServiceURL},
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 2,
Expand Down Expand Up @@ -574,6 +579,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
BootstrapServers: bootstrapServers,
EgressConfig: &contract.EgressConfig{DeadLetter: ServiceURLFrom(BrokerNamespace, ServiceName)},
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 2,
Expand Down Expand Up @@ -694,6 +700,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
BootstrapServers: bootstrapServers,
EgressConfig: &contract.EgressConfig{DeadLetter: ServiceURL},
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -763,6 +770,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -857,6 +865,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -957,6 +966,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
BootstrapServers: bootstrapServers,
EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com/api"},
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -1062,6 +1072,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -1155,6 +1166,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 2,
Expand Down Expand Up @@ -1222,6 +1234,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -1334,6 +1347,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -1428,6 +1442,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Version: SecretResourceVersion,
},
},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -1661,7 +1676,9 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
EgressConfig: &contract.EgressConfig{DeadLetter: ServiceURL},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
FeatureFlags: FeatureFlagsETAutocreate(false),

Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
},
},
Generation: 2,
Expand Down Expand Up @@ -1772,6 +1789,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
BackoffPolicy: contract.BackoffPolicy_Exponential,
BackoffDelay: 2000,
},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 2,
Expand Down Expand Up @@ -1851,6 +1869,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
BackoffPolicy: contract.BackoffPolicy_Linear,
BackoffDelay: 2000,
},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 2,
Expand Down Expand Up @@ -1927,6 +1946,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
EgressConfig: &contract.EgressConfig{
DeadLetter: ServiceURL,
},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 2,
Expand Down Expand Up @@ -2006,6 +2026,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
BackoffPolicy: contract.BackoffPolicy_Linear,
BackoffDelay: env.DefaultBackoffDelayMs,
},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 2,
Expand Down Expand Up @@ -2068,6 +2089,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
EgressConfig: &contract.EgressConfig{DeadLetter: ServiceURL},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -2129,6 +2151,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
EgressConfig: &contract.EgressConfig{DeadLetter: ServiceURL},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -2206,6 +2229,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -2317,6 +2341,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Version: SecretResourceVersion,
},
},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -2429,6 +2454,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
}},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -2531,6 +2557,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -3084,6 +3111,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
EgressConfig: &contract.EgressConfig{DeadLetter: ServiceURL},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -3155,6 +3183,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
EgressConfig: &contract.EgressConfig{DeadLetter: ServiceURL},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down
1 change: 1 addition & 0 deletions control-plane/pkg/reconciler/broker/namespaced_broker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,6 +264,7 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
BootstrapServers: bootstrapServers,
Reference: BrokerReference(),
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down
18 changes: 15 additions & 3 deletions control-plane/pkg/reconciler/channel/channel.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -695,9 +695,11 @@
Uid: string(channel.UID),
Topics: []string{topic},
Ingress: &contract.Ingress{
Host: receiver.Host(channel.GetNamespace(), channel.GetName()),
EnableAutoCreateEventTypes: features.IsEnabled(feature.EvenTypeAutoCreate),
Path: receiver.Path(channel.GetNamespace(), channel.GetName()),
Host: receiver.Host(channel.GetNamespace(), channel.GetName()),
Path: receiver.Path(channel.GetNamespace(), channel.GetName()),
},
FeatureFlags: &contract.FeatureFlags{
EnableEventTypeAutocreate: features.IsEnabled(feature.EvenTypeAutoCreate) && !ownedByBroker(channel),
},
BootstrapServers: config.GetBootstrapServers(),
Reference: &contract.Reference{
Expand Down Expand Up @@ -862,3 +864,13 @@

return d
}

func ownedByBroker(channel *messagingv1beta1.KafkaChannel) bool {
for _, ref := range channel.OwnerReferences {
if strings.EqualFold(ref.Kind, "broker") {
return true

Check warning on line 871 in control-plane/pkg/reconciler/channel/channel.go

View check run for this annotation

Codecov / codecov/patch

control-plane/pkg/reconciler/channel/channel.go#L868-L871 

Added lines #L868 - L871 were not covered by tests
}
}

return false

Check warning on line 875 in control-plane/pkg/reconciler/channel/channel.go

View check run for this annotation

Codecov / codecov/patch

control-plane/pkg/reconciler/channel/channel.go#L875 

Added line #L875 was not covered by tests
}
Loading
Loading