[release-1.11] Cherry-pick E2E tests for channel: TLS key pair rotation

#3406 (#3497) * use the composite prober with the channel (#3252) * using the composite prober for the channel * edit channel_test.go file * change channel_test.go * done changes in V2 channel * edit in controllerv2.go * edit addresses * 2nd edit addresses * Update prober pod port Co-authored-by: Calum Murray <[email protected]> * Update prober pod port Co-authored-by: Calum Murray <[email protected]> * Update prober pod port for TLS Co-authored-by: Calum Murray <[email protected]> * Update prober pod port for TLS Co-authored-by: Calum Murray <[email protected]> --------- Co-authored-by: Pierangelo Di Pilato <[email protected]> Co-authored-by: Calum Murray <[email protected]> * E2E tests for channel: TLS key pair rotation (#3406) * Save work progress * Expose the TLS port * Adding the logger to see what is happening * Java - Adding the debugging information * Adding the path to the contract * Comment out the certificate rotation test portion * Resolve the source certificate not found issue * Fix the issue in the test * Update control-plane/pkg/prober/prober.go Co-authored-by: Calum Murray <[email protected]> * Update control-plane/pkg/reconciler/channel/channel.go Co-authored-by: Calum Murray <[email protected]> * Update control-plane/pkg/reconciler/channel/channel.go Co-authored-by: Calum Murray <[email protected]> * Update control-plane/pkg/reconciler/channel/resources/service.go Co-authored-by: Calum Murray <[email protected]> * Update data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java Co-authored-by: Calum Murray <[email protected]> * Fix the inconsistent varable name * Fix the failed build issue * Remove the logger * Run formatting * Update data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java Co-authored-by: Calum Murray <[email protected]> * Remove the logger * Code gen * Update control-plane/pkg/reconciler/channel/channel.go Co-authored-by: Pierangelo Di Pilato <[email protected]> * Remove the uncessary code * Fix the failing reconciler tests due to the missing newly added filed in the test * Format fix * Instead of using channel service name, we directly use channel name for Path * Instead of using channel service name, we directly use channel name for Path --------- Co-authored-by: Calum Murray <[email protected]> Co-authored-by: Pierangelo Di Pilato <[email protected]> * Update the function name --------- Co-authored-by: Rahul kumar <[email protected]> Co-authored-by: Pierangelo Di Pilato <[email protected]> Co-authored-by: Calum Murray <[email protected]>
knative-extensions · Dec 5, 2023 · 82651c9 · 82651c9
1 parent 3076046
commit 82651c9
Show file tree

Hide file tree

Showing 13 changed files with 358 additions and 43 deletions.
diff --git a/control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel.yaml b/control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel.yaml
@@ -155,8 +155,28 @@ spec:
                   required:
                     - url
                   properties:
+                    name:
+                      type: string
                     url:
                       type: string
+                    CACerts:
+                      type: string
+                    audience:
+                      type: string
+                addresses:
+                  description: Kafka Sink is Addressable. It exposes the endpoints as URIs to get events delivered into the Kafka topic.
+                  type: array
+                  items:
+                    type: object
+                    properties:
+                      name:
+                        type: string
+                      url:
+                        type: string
+                      CACerts:
+                        type: string
+                      audience:
+                        type: string
                 annotations:
                   description: Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.
                   type: object

diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
@@ -93,7 +93,7 @@ type Reconciler struct {
 	ServiceLister      corelisters.ServiceLister
 	SubscriptionLister messaginglisters.SubscriptionLister
 
-	Prober prober.Prober
+	Prober prober.NewProber
 
 	IngressHost string
 
@@ -316,7 +316,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		}
 
 		httpAddress := receiver.ChannelHTTPAddress(channelHttpHost)
-		httpsAddress := receiver.HTTPSAddress(channelHttpsHost, channelService, caCerts)
+		httpsAddress := receiver.HTTPSAddress(channelHttpsHost, channel, caCerts)
 		// Permissive mode:
 		// - status.address http address with path-based routing
 		// - status.addresses:
@@ -334,7 +334,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 			return err
 		}
 
-		httpsAddress := receiver.HTTPSAddress(channelHttpsHost, channelService, caCerts)
+		httpsAddress := receiver.HTTPSAddress(channelHttpsHost, channel, caCerts)
 		addressableStatus.Addresses = []duckv1.Addressable{httpsAddress}
 		addressableStatus.Address = &httpsAddress
 	} else {
@@ -343,9 +343,8 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		addressableStatus.Addresses = []duckv1.Addressable{httpAddress}
 	}
 
-	address := addressableStatus.Address.URL.URL()
-	proberAddressable := prober.Addressable{
-		Address: address,
+	proberAddressable := prober.NewAddressable{
+		AddressStatus: &addressableStatus,
 		ResourceKey: types.NamespacedName{
 			Namespace: channel.GetNamespace(),
 			Name:      channel.GetName(),
@@ -426,9 +425,12 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 	// 	See (under discussions KIPs, unlikely to be accepted as they are):
 	// 	- https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446
 	// 	- https://cwiki.apache.org/confluence/display/KAFKA/KIP-286%3A+producer.send%28%29+should+not+block+on+metadata+update
-	address := receiver.Address(r.IngressHost, channel)
-	proberAddressable := prober.Addressable{
-		Address: address,
+	address := receiver.HTTPAddress(r.IngressHost, channel)
+	proberAddressable := prober.NewAddressable{
+		AddressStatus: &duckv1.AddressStatus{
+			Address:   &address,
+			Addresses: []duckv1.Addressable{address},
+		},
 		ResourceKey: types.NamespacedName{
 			Namespace: channel.GetNamespace(),
 			Name:      channel.GetName(),
@@ -667,6 +669,7 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 		Ingress: &contract.Ingress{
 			Host:                       receiver.Host(channel.GetNamespace(), channel.GetName()),
 			EnableAutoCreateEventTypes: feature.FromContext(ctx).IsEnabled(feature.EvenTypeAutoCreate),
+			Path:                       receiver.Path(channel.GetNamespace(), channel.GetName()),
 		},
 		BootstrapServers: config.GetBootstrapServers(),
 		Reference: &contract.Reference{

diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -140,7 +140,7 @@ func TestReconcileKind(t *testing.T) {
 				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 			},
 			OtherTestData: map[string]interface{}{
-				testProber: probertesting.MockProber(prober.StatusNotReady),
+				testProber: probertesting.MockNewProber(prober.StatusNotReady),
 			},
 		},
 		{
@@ -157,7 +157,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			WantErr: true,
 			OtherTestData: map[string]interface{}{
-				testProber: probertesting.MockProber(prober.StatusReady),
+				testProber: probertesting.MockNewProber(prober.StatusReady),
 			},
 		},
 		{
@@ -188,6 +188,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
@@ -262,6 +263,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							EgressConfig: &contract.EgressConfig{
@@ -338,6 +340,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
@@ -377,7 +380,7 @@ func TestReconcileKind(t *testing.T) {
 				finalizerUpdatedEvent,
 			},
 			OtherTestData: map[string]interface{}{
-				testProber: probertesting.MockProber(prober.StatusNotReady),
+				testProber: probertesting.MockNewProber(prober.StatusNotReady),
 			},
 		},
 		{
@@ -408,6 +411,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
@@ -447,7 +451,7 @@ func TestReconcileKind(t *testing.T) {
 				finalizerUpdatedEvent,
 			},
 			OtherTestData: map[string]interface{}{
-				testProber: probertesting.MockProber(prober.StatusUnknown),
+				testProber: probertesting.MockNewProber(prober.StatusUnknown),
 			},
 		},
 		{
@@ -482,6 +486,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{},
@@ -554,6 +559,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
@@ -631,6 +637,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
@@ -713,6 +720,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
@@ -795,6 +803,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
@@ -883,6 +892,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
@@ -1173,6 +1183,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
@@ -1270,6 +1281,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
@@ -1376,6 +1388,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
@@ -1479,6 +1492,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
@@ -1557,6 +1571,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
@@ -1626,6 +1641,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
@@ -1738,6 +1754,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
@@ -1823,6 +1840,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							EgressConfig: &contract.EgressConfig{
@@ -1864,7 +1882,7 @@ func TestReconcileKind(t *testing.T) {
 						WithChannelAddresses([]duckv1.Addressable{
 							{
 								Name:    pointer.String("https"),
-								URL:     httpsURL(ChannelServiceName, ChannelNamespace),
+								URL:     httpsURL(ChannelName, ChannelNamespace),
 								CACerts: pointer.String(testCaCerts),
 							},
 							{
@@ -1925,6 +1943,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 							EgressConfig: &contract.EgressConfig{
@@ -1966,13 +1985,13 @@ func TestReconcileKind(t *testing.T) {
 						WithChannelAddresses([]duckv1.Addressable{
 							{
 								Name:    pointer.String("https"),
-								URL:     httpsURL(ChannelServiceName, ChannelNamespace),
+								URL:     httpsURL(ChannelName, ChannelNamespace),
 								CACerts: pointer.String(testCaCerts),
 							},
 						}),
 						WithChannelAddress(duckv1.Addressable{
 							Name:    pointer.String("https"),
-							URL:     httpsURL(ChannelServiceName, ChannelNamespace),
+							URL:     httpsURL(ChannelName, ChannelNamespace),
 							CACerts: pointer.String(testCaCerts),
 						}),
 						WithChannelAddessable(),
@@ -2014,6 +2033,7 @@ func TestFinalizeKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 								Host: receiver.Host(ChannelNamespace, ChannelName),
 							},
 						},
@@ -2048,7 +2068,7 @@ func TestFinalizeKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
 			OtherTestData: map[string]interface{}{
-				testProber: probertesting.MockProber(prober.StatusNotReady),
+				testProber: probertesting.MockNewProber(prober.StatusNotReady),
 			},
 		},
 	}
@@ -2058,9 +2078,9 @@ func TestFinalizeKind(t *testing.T) {
 
 func useTable(t *testing.T, table TableTest, env config.Env) {
 	table.Test(t, NewFactory(&env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
-		proberMock := probertesting.MockProber(prober.StatusReady)
+		proberMock := probertesting.MockNewProber(prober.StatusReady)
 		if p, ok := row.OtherTestData[testProber]; ok {
-			proberMock = p.(prober.Prober)
+			proberMock = p.(prober.NewProber)
 		}
 
 		var featureFlags *apisconfig.KafkaFeatureFlags