[BUILD] Make e2e tests output less verbose and fix flaky NodePort unit-tests

Author: amuraru

Makefile

@@ -53,6 +53,16 @@ all: test manager ## Run 'test' and 'manager' targets.
 .PHONY: check
 check: test lint ## Run tests and linters
 
+.PHONY: clean
+clean: ## Clean build artifacts and test binaries
+	@echo "Cleaning build artifacts..."
+	@if [ -d "bin" ]; then \
+		chmod -R u+w bin/ 2>/dev/null || true; \
+		rm -rf bin/; \
+	fi
+	@rm -f cover.out
+	@rm -f manager_image_patch.yaml
+
 bin/golangci-lint: bin/golangci-lint-${GOLANGCI_VERSION} ## Symlink golangi-lint-<version> into versionless golangci-lint.
 	@ln -sf golangci-lint-${GOLANGCI_VERSION} bin/golangci-lint
 bin/golangci-lint-${GOLANGCI_VERSION}: ## Download versioned golangci-lint.
@@ -95,7 +105,6 @@ install-kustomize: ## Install kustomize.
 test: generate fmt vet bin/setup-envtest
 	cd api && go test ./...
 	KUBEBUILDER_ASSETS=$$($(BIN_DIR)/setup-envtest --print path --bin-dir $(BIN_DIR) use $(ENVTEST_K8S_VERSION)) \
-	GINKGO_FLAKE_ATTEMPTS=3 \
 	go test ./... \
 		-coverprofile cover.out \
 		-v \

controllers/tests/common_test.go

@@ -132,6 +132,10 @@ func createMinimalKafkaClusterCR(name, namespace string) *banzaicloudv1beta1.Kaf
 }
 
 func waitForClusterRunningState(ctx context.Context, kafkaCluster *banzaicloudv1beta1.KafkaCluster, namespace string) {
+	waitForClusterRunningStateWithTimeout(ctx, kafkaCluster, namespace, 240*time.Second)
+}
+
+func waitForClusterRunningStateWithTimeout(ctx context.Context, kafkaCluster *banzaicloudv1beta1.KafkaCluster, namespace string, timeout time.Duration) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -161,7 +165,7 @@ func waitForClusterRunningState(ctx context.Context, kafkaCluster *banzaicloudv1
 			}
 		}
 	}()
-	Eventually(ch, 240*time.Second, 50*time.Millisecond).Should(Receive())
+	Eventually(ch, timeout, 50*time.Millisecond).Should(Receive())
 }
 
 func getMockedKafkaClientForCluster(kafkaCluster *banzaicloudv1beta1.KafkaCluster) (kafkaclient.KafkaClient, func()) {

controllers/tests/kafkacluster_controller_externalnodeport_test.go

@@ -27,7 +27,6 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/intstr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/banzaicloud/koperator/api/v1beta1"
@@ -52,9 +51,6 @@ var (
 	}
 )
 
-var allocatedNodePorts []int32
-var safePort int32
-
 var _ = Describe("KafkaClusterNodeportExternalAccess", Ordered, Serial, func() {
 	var (
 		count              uint64 = 0
@@ -89,7 +85,7 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", Ordered, Serial, func() {
 		err = k8sClient.Create(ctx, kafkaCluster)
 		Expect(err).NotTo(HaveOccurred())
 
-		waitForClusterRunningState(ctx, kafkaCluster, namespace)
+		waitForClusterRunningStateWithTimeout(ctx, kafkaCluster, namespace, 120*time.Second)
 
 	})
 
@@ -121,13 +117,6 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", Ordered, Serial, func() {
 
 	When("hostnameOverride is configured with externalStartingPort 0", func() {
 		BeforeEach(func() {
-			allocatedNodePorts = nil
-			// Pre-allocate ports even when using auto-assignment to prevent conflicts
-			// Allocate 3 consecutive ports for the 3 brokers to avoid race conditions
-			safePort = GetNodePort(3)
-			for i := int32(0); i < 3; i++ {
-				allocatedNodePorts = append(allocatedNodePorts, safePort+i)
-			}
 			kafkaCluster.Spec.ListenersConfig.ExternalListeners = []v1beta1.ExternalListenerConfig{
 				{
 					CommonListenerSpec: v1beta1.CommonListenerSpec{
@@ -204,31 +193,18 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", Ordered, Serial, func() {
 				},
 			}))
 		})
-		AfterEach(func() {
-			for _, port := range allocatedNodePorts {
-				ReleaseNodePort(port)
-			}
-			allocatedNodePorts = nil
-		})
 	})
 
 	When("NodePortExternalIP is configured", func() {
 		BeforeEach(func() {
-			allocatedNodePorts = nil
-			safePort = GetNodePort(3)
-			// Allocate all 3 ports for the 3 brokers to avoid conflicts
-			for i := int32(0); i < 3; i++ {
-				allocatedNodePorts = append(allocatedNodePorts, safePort+i)
-			}
-			// update the external listener config with a nodeport listener
 			kafkaCluster.Spec.ListenersConfig.ExternalListeners = []v1beta1.ExternalListenerConfig{
 				{
 					CommonListenerSpec: v1beta1.CommonListenerSpec{
 						Name:          "test",
 						ContainerPort: 9733,
 						Type:          "plaintext",
 					},
-					ExternalStartingPort: safePort,
+					ExternalStartingPort: 0,
 					AccessMethod:         corev1.ServiceTypeNodePort,
 				},
 			}
@@ -268,13 +244,6 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", Ordered, Serial, func() {
 			}
 		})
 
-		AfterEach(func() {
-			for _, port := range allocatedNodePorts {
-				ReleaseNodePort(port)
-			}
-			allocatedNodePorts = nil
-		})
-
 		It("reconciles the service successfully", func(ctx SpecContext) {
 			var svc corev1.Service
 			svcName := fmt.Sprintf("%s-0-test", kafkaClusterCRName)
@@ -302,13 +271,8 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", Ordered, Serial, func() {
 			Expect(svc.Spec.Ports[0].Port).To(BeEquivalentTo(9733))
 			Expect(svc.Spec.Ports[0].TargetPort.IntVal).To(BeEquivalentTo(9733))
 
-			Expect(svc.Spec.Ports).To(ConsistOf(corev1.ServicePort{
-				Name:       "broker-0",
-				Protocol:   corev1.ProtocolTCP,
-				Port:       9733,
-				TargetPort: intstr.FromInt(9733),
-				NodePort:   safePort,
-			}))
+			Expect(svc.Spec.Ports).NotTo(BeEmpty())
+			Expect(svc.Spec.Ports[0].NodePort).To(BeNumerically(">", 0))
 
 			// check status
 			err := k8sClient.Get(ctx, types.NamespacedName{
@@ -360,20 +324,14 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", Ordered, Serial, func() {
 
 	When("hostnameOverride is configured", func() {
 		BeforeEach(func() {
-			allocatedNodePorts = nil
-			safePort = GetNodePort(3)
-			// Allocate all 3 ports for the 3 brokers to avoid conflicts
-			for i := int32(0); i < 3; i++ {
-				allocatedNodePorts = append(allocatedNodePorts, safePort+i)
-			}
 			kafkaCluster.Spec.ListenersConfig.ExternalListeners = []v1beta1.ExternalListenerConfig{
 				{
 					CommonListenerSpec: v1beta1.CommonListenerSpec{
 						Name:          "test",
 						ContainerPort: 9733,
 						Type:          "plaintext",
 					},
-					ExternalStartingPort: safePort,
+					ExternalStartingPort: 0,
 					IngressServiceSettings: v1beta1.IngressServiceSettings{
 						HostnameOverride: ".external.nodeport.com",
 					},
@@ -389,7 +347,17 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", Ordered, Serial, func() {
 			}, kafkaCluster)
 			Expect(err).NotTo(HaveOccurred())
 
-			expectedPort := safePort
+			// Get the actual allocated ports from Services
+			assignedNodePortPerBroker := make(map[int32]int32, len(kafkaCluster.Spec.Brokers))
+			for _, broker := range kafkaCluster.Spec.Brokers {
+				var svc corev1.Service
+				err := k8sClient.Get(ctx, types.NamespacedName{
+					Name:      fmt.Sprintf(kafka.NodePortServiceTemplate, kafkaCluster.GetName(), broker.Id, "test"),
+					Namespace: kafkaCluster.GetNamespace(),
+				}, &svc)
+				Expect(err).NotTo(HaveOccurred())
+				assignedNodePortPerBroker[broker.Id] = svc.Spec.Ports[0].NodePort
+			}
 
 			Expect(kafkaCluster.Status.ListenerStatuses).To(Equal(v1beta1.ListenerStatuses{
 				InternalListeners: map[string]v1beta1.ListenerStatusList{
@@ -416,26 +384,21 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", Ordered, Serial, func() {
 					"test": {
 						{
 							Name:    "broker-0",
-							Address: fmt.Sprintf("%s-0-test.kafka-nodeport-%d.external.nodeport.com:%d", kafkaCluster.Name, count, expectedPort),
+							Address: fmt.Sprintf("%s-0-test.kafka-nodeport-%d.external.nodeport.com:%d", kafkaCluster.Name, count, assignedNodePortPerBroker[0]),
 						},
 						{
 							Name:    "broker-1",
-							Address: fmt.Sprintf("%s-1-test.kafka-nodeport-%d.external.nodeport.com:%d", kafkaCluster.Name, count, expectedPort+1),
+							Address: fmt.Sprintf("%s-1-test.kafka-nodeport-%d.external.nodeport.com:%d", kafkaCluster.Name, count, assignedNodePortPerBroker[1]),
 						},
 						{
 							Name:    "broker-2",
-							Address: fmt.Sprintf("%s-2-test.kafka-nodeport-%d.external.nodeport.com:%d", kafkaCluster.Name, count, expectedPort+2),
+							Address: fmt.Sprintf("%s-2-test.kafka-nodeport-%d.external.nodeport.com:%d", kafkaCluster.Name, count, assignedNodePortPerBroker[2]),
 						},
 					},
 				},
 			}))
 		})
-		AfterEach(func() {
-			for _, port := range allocatedNodePorts {
-				ReleaseNodePort(port)
-			}
-			allocatedNodePorts = nil
-		})
+
 	})
 })
 

controllers/tests/suite_test.go

@@ -31,18 +31,13 @@ limitations under the License.
 package tests
 
 import (
-	"context"
-	"fmt"
-	"math/rand/v2"
 	"path/filepath"
-	"sync"
 	"testing"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
-	corev1 "k8s.io/api/core/v1"
 	apiv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -107,6 +102,8 @@ var _ = BeforeSuite(func(ctx SpecContext) {
 	}
 	apiServer := testEnv.ControlPlane.GetAPIServer()
 	apiServer.Configure().Set("service-cluster-ip-range", "10.0.0.0/16")
+	// Give kube-apiserver more room for automatic NodePort allocation and avoid the small static band.
+	apiServer.Configure().Set("service-node-port-range", "30128-65000")
 
 	var cfg *rest.Config
 	var err error
@@ -252,90 +249,8 @@ var _ = BeforeSuite(func(ctx SpecContext) {
 })
 
 var _ = AfterSuite(func() {
+
 	By("tearing down the test environment")
 	err := testEnv.Stop()
 	Expect(err).ToNot(HaveOccurred())
 })
-
-var (
-	nodePortMutex sync.Mutex
-	nodePorts     = make(map[int32]bool)
-)
-
-func GetNodePort(portAmount int32) int32 {
-	portAmount--
-	nodePortMutex.Lock()
-	defer nodePortMutex.Unlock()
-
-	const minPort, maxPort = 30000, 32767
-	portRange := maxPort - minPort + 1
-
-	fmt.Println("GetNodePort: Looking for an available nodeport")
-
-	// Always refresh the nodePorts map from actual Kubernetes services
-	if k8sClient != nil {
-		var serviceList corev1.ServiceList
-		if err := k8sClient.List(context.Background(), &serviceList); err == nil {
-			fmt.Printf("GetNodePort: Found %d services to check for nodeports\n", len(serviceList.Items))
-			// Clear the map first to get fresh state
-			nodePorts = make(map[int32]bool)
-			for _, service := range serviceList.Items {
-				if service.Spec.Type == corev1.ServiceTypeNodePort {
-					for _, port := range service.Spec.Ports {
-						if port.NodePort > 0 {
-							nodePorts[port.NodePort] = true
-							fmt.Printf("GetNodePort: Found existing nodeport %d in service %s/%s\n",
-								port.NodePort, service.Namespace, service.Name)
-						}
-					}
-				}
-			}
-		} else {
-			fmt.Printf("ERROR: Failed to list services: %v\n", err)
-		}
-	} else {
-		fmt.Println("WARNING: k8sClient not initialized yet skipping Kubernetes service check")
-	}
-
-	attempts := 0
-	for attempts = 0; attempts < 200; attempts++ {
-		port := minPort + rand.Int32N(int32(portRange))
-
-		// Avoid the problematic range around 32030 that often causes conflicts
-		if port >= 32025 && port <= 32035 {
-			continue
-		}
-
-		// Ensure the port range doesn't cross into the high conflict zone
-		if port+portAmount >= 32025 && port <= 32035 {
-			continue
-		}
-
-		allAvailable := true
-		for i := int32(0); i <= portAmount; i++ {
-			if nodePorts[port+i] {
-				allAvailable = false
-				break
-			}
-		}
-
-		if allAvailable {
-			for i := int32(0); i <= portAmount; i++ {
-				nodePorts[port+i] = true
-			}
-			fmt.Printf("GetNodePort: Successfully allocated NodePort range %d-%d after %d attempts\n",
-				port, port+portAmount, attempts+1)
-			return port
-		}
-	}
-
-	fmt.Printf("WARNING: No free NodePorts found after %d attempts, returning 0 for auto-assignment\n", attempts)
-	return 0
-}
-
-func ReleaseNodePort(port int32) {
-	nodePortMutex.Lock()
-	defer nodePortMutex.Unlock()
-	delete(nodePorts, port)
-	fmt.Printf("Released NodePort %d\n", port)
-}

tests/e2e/const.go

@@ -52,19 +52,19 @@ const (
 	zookeeperClusterName       = "zookeeper-server"
 	managedByHelmLabelTemplate = "app.kubernetes.io/managed-by=Helm,app.kubernetes.io/instance=%s"
 
-	cruiseControlPodReadinessTimeout       = 50 * time.Second
-	kafkaClusterResourceReadinessTimeout   = 60 * time.Second
-	defaultDeletionTimeout                 = 20 * time.Second
-	defaultPodReadinessWaitTime            = 10 * time.Second
-	defaultTopicCreationWaitTime           = 10 * time.Second
-	defaultUserCreationWaitTime            = 10 * time.Second
+	cruiseControlPodReadinessTimeout       = 300 * time.Second // Increased for kind environments
+	kafkaClusterResourceReadinessTimeout   = 300 * time.Second // Increased for kind environments
+	defaultDeletionTimeout                 = 60 * time.Second  // Increased for kind environments
+	defaultPodReadinessWaitTime            = 180 * time.Second // Increased for kind environments
+	defaultTopicCreationWaitTime           = 60 * time.Second  // Increased for kind environments
+	defaultUserCreationWaitTime            = 60 * time.Second  // Increased for kind environments
 	kafkaClusterCreateTimeout              = 1800 * time.Second
 	kafkaClusterResourceCleanupTimeout     = 600 * time.Second
-	kcatDeleetionTimeout                   = 40 * time.Second
-	zookeeperClusterCreateTimeout          = 4 * time.Minute
-	zookeeperClusterResourceCleanupTimeout = 60 * time.Second
-	externalConsumerTimeout                = 5 * time.Second
-	externalProducerTimeout                = 5 * time.Second
+	kcatDeleetionTimeout                   = 120 * time.Second // Increased for kind environments
+	zookeeperClusterCreateTimeout          = 10 * time.Minute  // Increased for kind environments
+	zookeeperClusterResourceCleanupTimeout = 180 * time.Second // Increased for kind environments
+	externalConsumerTimeout                = 120 * time.Second // Increased for kind environments
+	externalProducerTimeout                = 120 * time.Second // Increased for kind environments
 
 	zookeeperClusterReplicaCount = 1