[release-1.11] Cherry-pick E2E tests for broker: TLS key pair rotation #3287

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

@@ -2164,7 +2164,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-tls/broker-ingress-tls-certificate.yaml

@@ -26,12 +26,12 @@ spec:
       app.kubernetes.io/component: kafka-broker-receiver
       app.kubernetes.io/name: knative-eventing
 
-  duration: 2160h # 90d
-  renewBefore: 360h # 15d
+  # Use 0m0s so that we don't run into https://github.com/cert-manager/cert-manager/issues/6408 on the operator
+  duration: 2160h0m0s # 90d
+  renewBefore: 360h0m0s # 15d
   subject:
     organizations:
       - local
-  isCA: false
   privateKey:
     algorithm: RSA
     encoding: PKCS1

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

@@ -182,7 +182,6 @@ spec:
             secretName: kafka-broker-ingress-server-tls
             optional: true
 
-
       restartPolicy: Always
 ---
 
@@ -212,6 +211,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/config/channel-tls/channel-ingress-tls-certificate.yaml

@@ -26,12 +26,12 @@ spec:
       app.kubernetes.io/component: kafka-channel-receiver
       app.kubernetes.io/name: knative-eventing
 
-  duration: 2160h # 90d
-  renewBefore: 360h # 15d
+  # Use 0m0s so that we don't run into https://github.com/cert-manager/cert-manager/issues/6408 on the operator
+  duration: 2160h0m0s # 90d
+  renewBefore: 360h0m0s # 15d
   subject:
     organizations:
       - local
-  isCA: false
   privateKey:
     algorithm: RSA
     encoding: PKCS1

data-plane/config/sink-tls/sink-ingress-tls-certificate.yaml

@@ -26,12 +26,12 @@ spec:
       app.kubernetes.io/component: kafka-sink-receiver
       app.kubernetes.io/name: knative-eventing
 
-  duration: 2160h # 90d
-  renewBefore: 360h # 15d
+  # Use 0m0s so that we don't run into https://github.com/cert-manager/cert-manager/issues/6408 on the operator
+  duration: 2160h0m0s # 90d
+  renewBefore: 360h0m0s # 15d
   subject:
     organizations:
       - local
-  isCA: false
   privateKey:
     algorithm: RSA
     encoding: PKCS1

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

@@ -39,10 +39,10 @@ require (
 	github.com/google/gofuzz v1.2.0
 	github.com/hashicorp/go-cleanhttp v0.5.2
 	github.com/kedacore/keda/v2 v2.8.1
-	knative.dev/eventing v0.38.5
-	knative.dev/hack v0.0.0-20230712131415-ddae80293c43
-	knative.dev/pkg v0.0.0-20231023150739-56bfe0dd9626
-	knative.dev/reconciler-test v0.0.0-20231023114057-785e0bd2d9a2
+	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
 )
 

go.sum

@@ -1339,14 +1339,14 @@ 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/eventing v0.38.5 h1:NvSy3lek9IbLLWEot36NyAfNv7VkJNl38F1ItVL0D6s=
-knative.dev/eventing v0.38.5/go.mod h1:g+iAS+KBRSKULEPqoVnseMkObDeq3SJhqefbuIu8zY8=
-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/reconciler-test v0.0.0-20231023114057-785e0bd2d9a2 h1:Lenj/sGhPYZoCdl4bvoeZzA4Y1VS4LNEIWH1/HTU+6I=
-knative.dev/reconciler-test v0.0.0-20231023114057-785e0bd2d9a2/go.mod h1:HgugJUOhHZ3F6Tbhte92ecL0sBqJtCeJtd7K8jX+IJk=
+knative.dev/eventing v0.38.7-0.20231114105650-05b92c7f910a h1:gSGU4/w3NwVSmmcBG8B6H8AT647dMPLyqwn6LLpdp+I=
+knative.dev/eventing v0.38.7-0.20231114105650-05b92c7f910a/go.mod h1:a6XMiX0CEaQi9z698LZ9wVzseFK+4wgTExWhyKXzYqc=
+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=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=

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}'}}'