E2E tests for channel: TLS key pair rotation

control-plane/pkg/prober/prober.go

@@ -120,6 +120,15 @@ func probe(ctx context.Context, client httpClient, logger *zap.Logger, address s
 
 	if response.StatusCode != http.StatusOK {
 		logger.Info("Resource not ready", zap.Int("statusCode", response.StatusCode))
+		logger.Info("[haha] Response body", zap.Any("response", response.Body))
+		logger.Info("[haha] Response header", zap.Any("response", response.Header))
+		logger.Info("[haha] Response status", zap.Any("response", response.Status))
+		logger.Info("[haha] Response proto", zap.Any("response", response.Proto))
+		logger.Info("[haha] Response proto major", zap.Any("response", response.ProtoMajor))
+		logger.Info("[haha] Response proto minor", zap.Any("response", response.ProtoMinor))
+		logger.Info("[haha] Response transfer encoding", zap.Any("response", response.Request))
+		logger.Info("[haha] Response transfer encoding", zap.Any("response", response.ContentLength))
+
 		return StatusNotReady
 	}
 

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

@@ -697,12 +697,17 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 }
 
 func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messagingv1beta1.KafkaChannel) (*corev1.Service, error) {
-	expected, err := resources.MakeK8sService(channel, resources.ExternalService(r.DataPlaneNamespace, NewChannelIngressServiceName))
+	logger := kafkalogging.CreateReconcileMethodLogger(ctx, channel)
+	expected, err := resources.MakeK8sService(channel, logger, resources.ExternalService(r.DataPlaneNamespace, NewChannelIngressServiceName))
 	if err != nil {
 		return expected, fmt.Errorf("failed to create the channel service object: %w", err)
 	}
 
+	logger.Debug("[hahap] Channel service created", zap.Any("service", expected))
+
 	svc, err := r.ServiceLister.Services(channel.Namespace).Get(resources.MakeChannelServiceName(channel.Name))
+
+	logger.Debug("[hahap] Channel service got", zap.Any("service", svc))
 	if err != nil {
 		if apierrors.IsNotFound(err) {
 			_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Create(ctx, expected, metav1.CreateOptions{})

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

@@ -1,4 +1,4 @@
 /*
  * Copyright 2021 The Knative Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -18,6 +18,7 @@
 
 import (
 	"context"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 
 	"github.com/IBM/sarama"
 	"go.uber.org/zap"
@@ -81,6 +82,9 @@
 
 	logger := logging.FromContext(ctx)
 
+	featureStore := feature.NewStore(logging.FromContext(ctx).Named("feature-config-store"))
+	featureStore.WatchConfigs(watcher)
+
 	_, err := reconciler.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {
 		logger.Fatal("Failed to get or create data plane config map",
@@ -96,7 +100,13 @@
 		logger.Warn("Failed to get CA certs when at least one address uses TLS", zap.Error(err))
 	}
 
-	impl := kafkachannelreconciler.NewImpl(ctx, reconciler)
+	impl := kafkachannelreconciler.NewImpl(ctx, reconciler,
+		func(impl *controller.Impl) controller.Options {
+			return controller.Options{
+				ConfigStore:       featureStore,
+				PromoteFilterFunc: kafka.BrokerClassFilter(),
+			}
+		})
 	IPsLister := prober.IdentityIPsLister()
 	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 	reconciler.Prober, err = prober.NewComposite(ctx, "", "", IPsLister, impl.EnqueueKey, &caCerts)

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

@@ -86,8 +86,12 @@ func TestNewController(t *testing.T) {
 		configmap.NewStaticWatcher(&corev1.ConfigMap{
 			ObjectMeta: metav1.ObjectMeta{
 				Name: apisconfig.FlagsConfigName,
+			}}, &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "config-features",
 			},
 		}),
+
 		configs,
 	)
 	if controller == nil {

control-plane/pkg/reconciler/channel/resources/service.go

@@ -19,6 +19,8 @@ package resources
 import (
 	"fmt"
 
+	"go.uber.org/zap"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"knative.dev/pkg/kmeta"
@@ -28,8 +30,12 @@ import (
 )
 
 const (
-	portName           = "http"
-	portNumber         = 80
+	portName   = "http"
+	portNumber = 80
+
+	TLSportName   = "https"
+	TLSportNumber = 443
+
 	MessagingRoleLabel = "messaging.knative.dev/role"
 	MessagingRole      = "kafka-channel"
 
@@ -62,7 +68,7 @@ func ExternalService(namespace, service string) ServiceOption {
 // MakeK8sService creates a new K8s Service for a Channel resource. It also sets the appropriate
 // OwnerReferences on the resource so handleObject can discover the Channel resource that 'owns' it.
 // As well as being garbage collected when the Channel is deleted.
-func MakeK8sService(kc *v1beta1.KafkaChannel, opts ...ServiceOption) (*corev1.Service, error) {
+func MakeK8sService(kc *v1beta1.KafkaChannel, logger *zap.Logger, opts ...ServiceOption) (*corev1.Service, error) {
 	// Add annotations
 	svc := &corev1.Service{
 		TypeMeta: metav1.TypeMeta{
@@ -86,6 +92,11 @@ func MakeK8sService(kc *v1beta1.KafkaChannel, opts ...ServiceOption) (*corev1.Se
 					Protocol: corev1.ProtocolTCP,
 					Port:     portNumber,
 				},
+				{
+					Name:     TLSportName,
+					Protocol: corev1.ProtocolTCP,
+					Port:     TLSportNumber,
+				},
 			},
 		},
 	}

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

@@ -83,10 +83,12 @@ public IngressProducerReconcilableStore(
     }
 
     public IngressProducer resolve(String host, String path) {
+      Logger logger = LoggerFactory.getLogger(ProducerHolder.class);
         // Ignore the host when there's a path given in the request.
         // That means, we support these modes:
         // - Request coming to "/path" --> path is used for matching
         // - Request coming to "/" --> hostname is used for matching
+      logger.error("[haha java] print the pathMapper: {}", pathMapper);
         final var p = pathMapper.get(removeTrailingSlash(path));
         if (p != null) {
             return p;

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

@@ -206,6 +206,8 @@ public void stop(Promise<Void> stopPromise) throws Exception {
     @Override
     public void handle(HttpServerRequest request) {
 
+      logger.error("Received request: {}", request.path());
+
         final var requestContext = new RequestContext(request);
 
         // Look up for the ingress producer

test/e2e_new/channel_eventing_tls_test.go

@@ -0,0 +1,48 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2023 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
+ *
+ *     http://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.
+ */
+
+package e2e_new
+
+import (
+	"testing"
+	"time"
+
+	"knative.dev/eventing-kafka-broker/test/rekt/features"
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+)
+
+func TestChannelTLSCARotation(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+		eventshub.WithTLS(t),
+		environment.WithPollTimings(5*time.Second, 4*time.Minute),
+	)
+
+	env.Test(ctx, t, features.RotateChannelTLSCertificates())
+}

test/rekt/features/channel_tls.go

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2023 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
+ *
+ *     http://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.
+ */
+
+package features
+
+import (
+	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkachannel"
+	"knative.dev/eventing/test/rekt/features/featureflags"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+func RotateChannelTLSCertificates() *feature.Feature {
+	//
+	//ingressCertificateName := "kafka-channel-ingress-server-tls"
+	//ingressSecretName := "kafka-channel-ingress-server-tls"
+
+	channelName := feature.MakeRandomK8sName("channel")
+	//subscriptionName := feature.MakeRandomK8sName("sub")
+	sink := feature.MakeRandomK8sName("sink")
+	//source := feature.MakeRandomK8sName("source")
+
+	f := feature.NewFeatureNamed("Rotate Kafka Channel TLS certificate")
+
+	f.Prerequisite("transport encryption is strict", featureflags.TransportEncryptionStrict())
+	f.Prerequisite("should not run when Istio is enabled", featureflags.IstioDisabled())
+
+	//f.Setup("Rotate ingress certificate", certificate.Rotate(certificate.RotateCertificate{
+	//	Certificate: types.NamespacedName{
+	//		Namespace: system.Namespace(),
+	//		Name:      ingressCertificateName,
+	//	},
+	//}))
+
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiverTLS))
+	f.Setup("install channel", kafkachannel.Install(channelName,
+		kafkachannel.WithNumPartitions("3"),
+		kafkachannel.WithReplicationFactor("1"),
+		kafkachannel.WithRetentionDuration("P1D"),
+	))
+	f.Setup("channel is ready", kafkachannel.IsReady(channelName))
+
+	//f.Setup("install subscription", func(ctx context.Context, t feature.T) {
+	//	d := service.AsDestinationRef(sink)
+	//	d.CACerts = eventshub.GetCaCerts(ctx)
+	//	subscription.Install(subscriptionName,
+	//		subscription.WithChannel(kafkachannel.AsRef(channelName)),
+	//		subscription.WithSubscriberFromDestination(d))(ctx, t)
+	//})
+	//
+	//f.Setup("subscription is ready", subscription.IsReady(subscriptionName))
+	//
+	//f.Setup("Channel has HTTPS address", kafkachannel.ValidateAddress(channelName, addressable.AssertHTTPSAddress))
+	//
+	//event := cetest.FullEvent()
+	//event.SetID(uuid.New().String())
+	//
+	//f.Requirement("install source", eventshub.Install(source,
+	//	eventshub.StartSenderToResourceTLS(kafkachannel.GVR(), channelName, nil),
+	//	eventshub.InputEvent(event),
+	//	// Send multiple events so that we take into account that the certificate rotation might
+	//	// be detected by the server after some time.
+	//	eventshub.SendMultipleEvents(100, 3*time.Second),
+	//))
+	//
+	//f.Assert("Event sent", assert.OnStore(source).
+	//	MatchSentEvent(cetest.HasId(event.ID())).
+	//	AtLeast(1),
+	//)
+	//f.Assert("Event received", assert.OnStore(sink).
+	//	MatchReceivedEvent(cetest.HasId(event.ID())).
+	//	AtLeast(1),
+	//)
+	//f.Assert("Source match updated peer certificate", assert.OnStore(source).
+	//	MatchPeerCertificatesReceived(assert.MatchPeerCertificatesFromSecret(system.Namespace(), ingressSecretName, "tls.crt")).
+	//	AtLeast(1),
+	//)
+
+	return f
+}

test/rekt/resources/kafkachannel/kafkachannel.go

@@ -1,4 +1,4 @@
 /*
 Copyright 2021 The Knative Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
@@ -19,6 +19,8 @@
 import (
 	"context"
 	"embed"
+	"knative.dev/eventing/test/rekt/resources/addressable"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -34,6 +36,17 @@
 	return schema.GroupVersionResource{Group: "messaging.knative.dev", Version: "v1beta1", Resource: "kafkachannels"}
 }
 
+func GVK() schema.GroupVersionKind {
+	return schema.ParseGroupKind(EnvCfg.ChannelGK).WithVersion(EnvCfg.ChannelV)
+}
+
+var EnvCfg EnvConfig
+
+type EnvConfig struct {
+	ChannelGK string `envconfig:"CHANNEL_GROUP_KIND" default:"KafkaChannel.messaging.knative.dev" required:"true"`
+	ChannelV  string `envconfig:"CHANNEL_VERSION" default:"v1beta1" required:"true"`
+}
+
 // Install will create a KafkaChannel resource, using the latest version, augmented with the config fn options.
 func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 	cfg := map[string]interface{}{
@@ -84,3 +97,45 @@
 		cfg["retentionDuration"] = retentionDuration
 	}
 }
+
+// AsRef returns a KRef for a Channel without namespace.
+func AsRef(name string) *duckv1.KReference {
+	apiVersion, kind := GVK().ToAPIVersionAndKind()
+	return &duckv1.KReference{
+		Kind:       kind,
+		APIVersion: apiVersion,
+		Name:       name,
+	}
+}
+
+// AsRef returns a KRef for a Channel without namespace.
+func AsDestinationRef(name string) *duckv1.Destination {
+	apiVersion, kind := GVK().ToAPIVersionAndKind()
+	return &duckv1.Destination{
+		Ref: &duckv1.KReference{
+			Kind:       kind,
+			APIVersion: apiVersion,
+			Name:       name,
+		},
+	}
+}
+
+// Address returns a Channel's address.
+func Address(ctx context.Context, name string, timings ...time.Duration) (*duckv1.Addressable, error) {
+	return addressable.Address(ctx, GVR(), name, timings...)
+}
+
+// ValidateAddress validates the address retured by Address
+func ValidateAddress(name string, validate addressable.ValidateAddress, timings ...time.Duration) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		addr, err := Address(ctx, name, timings...)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		if err := validate(addr); err != nil {
+			t.Error(err)
+			return
+		}
+	}
+}