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

Jump to bottom

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 6 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
288 changes: 177 additions & 111 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 @@ -634,8 +634,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: feature.FromContext(ctx).IsEnabled(feature.EvenTypeAutoCreate),
Path: receiver.PathFromObject(broker),
},
FeatureFlags: &contract.FeatureFlags{
EnableEventTypeAutocreate: feature.FromContext(ctx).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 @@ -438,6 +441,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 @@ -500,6 +504,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 @@ -572,6 +577,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 @@ -692,6 +698,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 @@ -761,6 +768,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 @@ -855,6 +863,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 @@ -955,6 +964,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 @@ -1060,6 +1070,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 @@ -1153,6 +1164,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 @@ -1220,6 +1232,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 @@ -1331,6 +1344,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 @@ -1425,6 +1439,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Version: SecretResourceVersion,
},
},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -1658,7 +1673,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 @@ -1769,6 +1786,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 @@ -1848,6 +1866,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 @@ -1924,6 +1943,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 @@ -2003,6 +2023,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 @@ -2065,6 +2086,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 @@ -2126,6 +2148,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 @@ -2203,6 +2226,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 @@ -2296,6 +2320,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
Version: SecretResourceVersion,
},
},
FeatureFlags: FeatureFlagsETAutocreate(false),
},
},
Generation: 1,
Expand Down Expand Up @@ -2388,6 +2413,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 @@ -2471,6 +2497,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 @@ -3023,6 +3050,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 @@ -3094,6 +3122,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
8 changes: 5 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 @@ -690,9 +690,11 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
Uid: string(channel.UID),
Topics: []string{topic},
Ingress: &contract.Ingress{
Host: receiver.Host(channel.GetNamespace(), channel.GetName()),
EnableAutoCreateEventTypes: feature.FromContext(ctx).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: feature.FromContext(ctx).IsEnabled(feature.EvenTypeAutoCreate),
},
BootstrapServers: config.GetBootstrapServers(),
Reference: &contract.Reference{
Expand Down
Loading