[release-v1.11] Cherry-pick E2E tests for broker: TLS key pair rotation

knative-extensions#3287 (#869)

* E2E tests for broker: TLS key pair rotation (knative-extensions#3287)

* progress save

* add the new cert rotation test

* Update the rekt test

* Format the files

* workspace save

* maven

* Clean up

* Clean up

* Format

* Fix the rekt test CA cert issue on Source

Co-authored-by: Pierangelo Di Pilato <[email protected]>

* Format fix

* Re-order the prerequisite

* Enable the strict transportation mode in TLS rekt tests

* Fix the reviewDog comment

* Update test/e2e_new/broker_eventing_tls_test.go

Co-authored-by: Pierangelo Di Pilato <[email protected]>

* Fix the format of the shell script

* Fix the format of the shell script

* Update the Strict feature flag

* Add the code to inject the feature flag to the context in the broker controller

* Run go import

* Fix the controller tests

* Set the default value to all other feature flags when we are updating the transport-encryption feature flag

---------

Co-authored-by: Pierangelo Di Pilato <[email protected]>
Co-authored-by: Pierangelo Di Pilato <[email protected]>

* Upgrade eventing reconciler-test to release-1.11

* Cherry-pick the TLS commit

* Update eventing dependency

---------

Co-authored-by: Pierangelo Di Pilato <[email protected]>
Co-authored-by: Pierangelo Di Pilato <[email protected]>

control-plane/pkg/reconciler/broker/controller.go

@@ -88,15 +88,21 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 		)
 	}
 
+	featureStore := feature.NewStore(logging.FromContext(ctx).Named("feature-config-store"))
+	featureStore.WatchConfigs(watcher)
+
 	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.BrokerClass, func(impl *controller.Impl) controller.Options {
-		return controller.Options{PromoteFilterFunc: kafka.BrokerClassFilter()}
+		return controller.Options{
+			ConfigStore:       featureStore,
+			PromoteFilterFunc: kafka.BrokerClassFilter()}
 	})
 
 	reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: reconciler.DataPlaneNamespace, Name: env.IngressName})
 
 	features := feature.FromContext(ctx)
 	caCerts, err := reconciler.getCaCerts()
+
 	if err != nil && (features.IsStrictTransportEncryption() || features.IsPermissiveTransportEncryption()) {
 		// We only need to warn here as the broker won't reconcile properly without the proper certs because the prober won't succeed
 		logger.Warn("Failed to get CA certs when at least one address uses TLS", zap.Error(err))

control-plane/pkg/reconciler/broker/controller_test.go

@@ -89,6 +89,10 @@ func TestNewController(t *testing.T) {
 			ObjectMeta: metav1.ObjectMeta{
 				Name: apisconfig.FlagsConfigName,
 			},
+		}, &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "config-features",
+			},
 		}),
 		env,
 	)

control-plane/pkg/reconciler/consumergroup/consumergroup_test.go

@@ -2170,7 +2170,7 @@ func TestFinalizeKind(t *testing.T) {
 				Eventf(
 					corev1.EventTypeWarning,
 					"InternalError",
-					"failed to delete consumer group offset: unable to delete the consumer group my.group.id: "+sarama.ErrClusterAuthorizationFailed.Error() + " (retry num 1)",
+					"failed to delete consumer group offset: unable to delete the consumer group my.group.id: "+sarama.ErrClusterAuthorizationFailed.Error()+" (retry num 1)",
 				),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{

data-plane/config/broker/500-receiver.yaml

@@ -180,7 +180,6 @@ spec:
             secretName: kafka-broker-ingress-server-tls
             optional: true
 
-
       restartPolicy: Always
 ---
 
@@ -210,6 +209,10 @@ spec:
       port: 8080
       protocol: TCP
       targetPort: 8080
+    - name: https-container
+      port: 8443
+      protocol: TCP
+      targetPort: 8443
     - name: http-metrics
       port: 9090
       protocol: TCP

data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java

@@ -62,35 +62,42 @@ public class Metrics {
     public static final boolean DISABLE_KAFKA_CLIENTS_METRICS =
             Boolean.parseBoolean(System.getenv("DISABLE_KAFKA_CLIENTS_METRICS"));
 
-    // There are different thread polls usable, mainly, each with its own drawbacks for our use case:
-    //   - cached thread pools
-    //   - fixed thread pools
+    // There are different thread polls usable, mainly, each with its own drawbacks
+    // for our use case:
+    // - cached thread pools
+    // - fixed thread pools
     //
-    // A cached thread might grow unbounded and since creating, updating and deleting resources
-    // trigger the usage of this executor, a bad actor might start continuously creating, updating
+    // A cached thread might grow unbounded and since creating, updating and
+    // deleting resources
+    // trigger the usage of this executor, a bad actor might start continuously
+    // creating, updating
     // and deleting resources which will cause resource exhaustion.
     //
-    // A fixed thread poll doesn't give the best possible latency for every resource, but it's
+    // A fixed thread poll doesn't give the best possible latency for every
+    // resource, but it's
     // bounded, so we keep the resource usage under control.
-    // We might want to provide configs to make it bigger than a single thread but a single thread
+    // We might want to provide configs to make it bigger than a single thread but a
+    // single thread
     // to start with is good enough for now.
     public static final ExecutorService meterBinderExecutor = Executors.newSingleThreadExecutor();
 
     static {
         Runtime.getRuntime().addShutdownHook(new Thread(meterBinderExecutor::shutdown));
     }
 
-    // Micrometer employs a naming convention that separates lowercase words with a '.' (dot) character.
-    // Different monitoring systems have different recommendations regarding naming convention, and some naming
+    // Micrometer employs a naming convention that separates lowercase words with a
+    // '.' (dot) character.
+    // Different monitoring systems have different recommendations regarding naming
+    // convention, and some naming
     // conventions may be incompatible for one system and not another.
-    // Each Micrometer implementation for a monitoring system comes with a naming convention that transforms lowercase
+    // Each Micrometer implementation for a monitoring system comes with a naming
+    // convention that transforms lowercase
     // dot notation names to the monitoring system’s recommended naming convention.
-    // Additionally, this naming convention implementation sanitizes metric names and tags of special characters that
+    // Additionally, this naming convention implementation sanitizes metric names
+    // and tags of special characters that
     // are disallowed by the monitoring system.
 
-    /**
-     * In prometheus format --> http_events_sent_total
-     */
+    /** In prometheus format --> http_events_sent_total */
     public static final String HTTP_EVENTS_SENT_COUNT = "http.events.sent";
 
     /**
@@ -221,8 +228,8 @@ public static MeterRegistry getRegistry() {
      * Register the given consumer to the global meter registry.
      *
      * @param consumer consumer to bind to the global registry.
-     * @param <K>      Record key type.
-     * @param <V>      Record value type.
+     * @param <K> Record key type.
+     * @param <V> Record value type.
      * @return A meter binder to close once the consumer is closed.
      */
     public static <K, V> AsyncCloseable register(final Consumer<K, V> consumer) {
@@ -233,8 +240,8 @@ public static <K, V> AsyncCloseable register(final Consumer<K, V> consumer) {
      * Register the given producer to the global meter registry.
      *
      * @param producer Consumer to bind to the global registry.
-     * @param <K>      Record key type.
-     * @param <V>      Record value type.
+     * @param <K> Record key type.
+     * @param <V> Record value type.
      * @return A meter binder to close once the producer is closed.
      */
     public static <K, V> AsyncCloseable register(final Producer<K, V> producer) {
@@ -270,7 +277,8 @@ private static AsyncCloseable register(final Supplier<KafkaClientMetrics> metric
                 };
 
             } catch (final RejectedExecutionException ex) {
-                // if this task cannot be accepted for execution when the executor has been shutdown.
+                // if this task cannot be accepted for execution when the executor has been
+                // shutdown.
                 logger.warn("Failed to bind metrics for Kafka client", ex);
             }
         }

data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java

@@ -30,9 +30,8 @@
 import dev.knative.eventing.kafka.broker.receiver.impl.handler.MethodNotAllowedHandler;
 import dev.knative.eventing.kafka.broker.receiver.impl.handler.ProbeHandler;
 import dev.knative.eventing.kafka.broker.receiver.main.ReceiverEnv;
-import io.fabric8.kubernetes.client.*;
 import io.vertx.core.*;
-import io.vertx.core.buffer.*;
+import io.vertx.core.buffer.Buffer;
 import io.vertx.core.eventbus.MessageConsumer;
 import io.vertx.core.http.HttpServer;
 import io.vertx.core.http.HttpServerOptions;

data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverVerticleFactory.java

@@ -36,7 +36,7 @@ class ReceiverVerticleFactory implements Supplier<Verticle> {
     private final HttpServerOptions httpServerOptions;
     private final HttpServerOptions httpsServerOptions;
 
-    private final String secretVolumePath = "/etc/receiver-secret-volume";
+    private final String secretVolumePath = "/etc/receiver-tls-secret";
 
     private final IngressRequestHandler ingressRequestHandler;
 

data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTest.java

@@ -15,10 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.receiver.impl;
 
-import static io.netty.handler.codec.http.HttpResponseStatus.ACCEPTED;
-import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
-import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
-import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpResponseStatus.*;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.mock;

go.mod

@@ -40,9 +40,9 @@ require (
 	github.com/hashicorp/go-cleanhttp v0.5.2
 	github.com/kedacore/keda/v2 v2.8.1
 	go.opencensus.io v0.24.0
-	knative.dev/eventing v0.38.6-0.20231024083920-1bdedcc2016a
-	knative.dev/hack v0.0.0-20230712131415-ddae80293c43
-	knative.dev/pkg v0.0.0-20231023150739-56bfe0dd9626
+	knative.dev/eventing v0.38.7-0.20231114105650-05b92c7f910a
+	knative.dev/hack v0.0.0-20231123073118-c0f04e812cfe
+	knative.dev/pkg v0.0.0-20231103161548-f5b42e8dea44
 	knative.dev/reconciler-test v0.0.0-20231121134246-52a86c40f40c
 	sigs.k8s.io/controller-runtime v0.12.3
 )
@@ -165,4 +165,4 @@ require (
 
 replace github.com/cloudevents/sdk-go/sql/v2 => github.com/cloudevents/sdk-go/sql/v2 v2.0.0-20220930150014-52b12276cc4a
 
-replace knative.dev/eventing => github.com/openshift-knative/eventing v0.99.1-0.20230901062933-33b0c175e2ab
+replace knative.dev/eventing => github.com/openshift-knative/eventing v0.99.1-0.20231204165534-958f85c43970

go.sum

@@ -615,8 +615,8 @@ github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1y
 github.com/onsi/gomega v1.10.2/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/onsi/gomega v1.16.0/go.mod h1:HnhC7FXeEQY45zxNK3PPoIUhzk/80Xly9PcubAlGdZY=
 github.com/onsi/gomega v1.23.0 h1:/oxKu9c2HVap+F3PfKort2Hw5DEU+HGlW8n+tguWsys=
-github.com/openshift-knative/eventing v0.99.1-0.20230901062933-33b0c175e2ab h1:RFwz3GUPdibID2DhFqz/0ihcRTYsTaf2zJIHoFuC8Z4=
-github.com/openshift-knative/eventing v0.99.1-0.20230901062933-33b0c175e2ab/go.mod h1:u/Ss1HNDfCmp+D+y862iSYMzEX07hahXEGscB5mrFds=
+github.com/openshift-knative/eventing v0.99.1-0.20231204165534-958f85c43970 h1:fcZVDFDjMgxG30PaSSjSE2pT9+TXqWD98KN6V2WYzY8=
+github.com/openshift-knative/eventing v0.99.1-0.20231204165534-958f85c43970/go.mod h1:Y+9JJFZt1ilIcjhPEbaMP/hWX2Fg9g3oixT7oQrOnkE=
 github.com/openzipkin/zipkin-go v0.2.2/go.mod h1:NaW6tEwdmWMaCDZzg8sh+IBNOxHMPnhQw8ySjnjRyN4=
 github.com/openzipkin/zipkin-go v0.3.0 h1:XtuXmOLIXLjiU2XduuWREDT0LOKtSgos/g7i7RYyoZQ=
 github.com/openzipkin/zipkin-go v0.3.0/go.mod h1:4c3sLeE8xjNqehmF5RpAFLPLJxXscc0R4l6Zg0P1tTQ=
@@ -1341,10 +1341,10 @@ 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-20221108210102-8e77b1f39fe2 h1:GfD9OzL11kvZN5iArC6oTS7RTj7oJOIfnislxYlqTj8=
 k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
-knative.dev/hack v0.0.0-20230712131415-ddae80293c43 h1:3SE06uNfSFGm/5XS+0trbyCUpgsOaBeyhPQU8FPNFz8=
-knative.dev/hack v0.0.0-20230712131415-ddae80293c43/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
-knative.dev/pkg v0.0.0-20231023150739-56bfe0dd9626 h1:qFE+UDBRg6cpF5LbA0sv1XK4XZ36Z7aTRCa+HcuxnNQ=
-knative.dev/pkg v0.0.0-20231023150739-56bfe0dd9626/go.mod h1:g+UCgSKQ2f15kHYu/V3CPtoKo5F1x/2Y1ot0NSK7gA0=
+knative.dev/hack v0.0.0-20231123073118-c0f04e812cfe h1:8MMQg9UvxCLiOqWnWm6+kiYyV81Are8ocj7fX6qpgrk=
+knative.dev/hack v0.0.0-20231123073118-c0f04e812cfe/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
+knative.dev/pkg v0.0.0-20231103161548-f5b42e8dea44 h1:2gjHbqg8K9k1KJtLgxsTvzxovXOhozcrk3AzzJmjsA0=
+knative.dev/pkg v0.0.0-20231103161548-f5b42e8dea44/go.mod h1:g+UCgSKQ2f15kHYu/V3CPtoKo5F1x/2Y1ot0NSK7gA0=
 knative.dev/reconciler-test v0.0.0-20231121134246-52a86c40f40c h1:MDmiP3HYihbl5gkNJJOC/adSup3y8nHZlmSMwHZ9heU=
 knative.dev/reconciler-test v0.0.0-20231121134246-52a86c40f40c/go.mod h1:Yw7Jkv+7PjDitG6CUkakWc/5SZa8Tm/sgXfaFy305Ng=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=

test/config-transport-encryption/features.yaml

@@ -0,0 +1,32 @@
+# Copyright 2021 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-features
+  namespace: knative-eventing
+  labels:
+    knative.dev/config-propagation: original
+    knative.dev/config-category: eventing
+data:
+  authentication.oidc: "disabled"
+  delivery-retryafter: "disabled"
+  delivery-timeout: "enabled"
+  eventtype-auto-create: "disabled"
+  kreference-group: "disabled"
+  kreference-mapping: "disabled"
+  new-trigger-filters: "enabled"
+  strict-subscriber: "disabled"
+  transport-encryption: "Strict"

test/e2e-common.sh

@@ -96,9 +96,13 @@ function knative_eventing() {
     kubectl apply -f "${EVENTING_CONFIG}/eventing-crds.yaml"
     kubectl apply -f "${EVENTING_CONFIG}/eventing-core.yaml"
     kubectl apply -f "${EVENTING_CONFIG}/eventing-tls-networking.yaml"
+
   else
     echo ">> Install Knative Eventing from ${KNATIVE_EVENTING_RELEASE}"
     kubectl apply -f "${KNATIVE_EVENTING_RELEASE}"
+
+    echo ">> Intall Knative Eventing TLS from ${KNATIVE_EVENTING_RELEASE_TLS}"
+    kubectl apply -f "${KNATIVE_EVENTING_RELEASE_TLS}"
   fi
 
   ! kubectl patch horizontalpodautoscalers.autoscaling -n knative-eventing eventing-webhook -p '{"spec": {"minReplicas": '${REPLICAS}'}}'