Merge pull request #2302 from kube-logging/fix/e2e-flakes-cap-failnow

fix(e2e): remove the elasticsearch kill loop and five flake sources

Author: csatib02

CLAUDE.md

@@ -154,7 +154,8 @@ v1alpha1 is the legacy API; conversion functions exist to v1beta1 (which is the
 ## Testing
 
 - Unit/integration tests use `envtest` (embedded Kubernetes API server + etcd); no cluster needed
-- E2E tests in `e2e/` use KIND and cover scenarios: fluentd-aggregator, fluentbit-multitenant, syslog-ng-aggregator
+- E2E tests in `e2e/` use KIND: one directory per suite, each its own Go package and test binary, provisioning its own KIND cluster. Run a single suite with `make test-e2e E2E_TEST=<suite-dir>`. Shared helpers live in `e2e/common/` and `e2e/internal/` and are excluded from suite selection
+- E2E knobs (all `make` overrides): `E2E_TEST_TIMEOUT` (per suite binary, default 20m), `E2E_SUITE_PARALLEL` (clusters one suite binary builds at once, default 2 — raising it starves the aggregators on a 4-vCPU runner), `KIND_COMMAND_TIMEOUT` (per kind invocation; derived from `E2E_TEST_TIMEOUT` when unset)
 - Coverage profile config in `.testcoverage.yml`; tool: `go-test-coverage`
 - Test framework: Ginkgo + Gomega for BDD-style tests; testify for unit tests
 

Makefile

@@ -53,6 +53,11 @@ OPERATOR_IMG_DEBUG ?= controller:debug
 CRD_OPTIONS ?= crd:maxDescLen=0
 
 E2E_TEST_TIMEOUT ?= 20m
+
+# Clusters one suite binary builds at once. Caps peak concurrency, which
+# otherwise follows the core count and starves the aggregators.
+E2E_SUITE_PARALLEL ?= 2
+
 TEST_COV_DIR := $(shell mkdir -p build/_test_coverage && realpath build/_test_coverage)
 
 CONTROLLER_GEN := ${BIN}/controller-gen
@@ -242,7 +247,7 @@ test-e2e-nodeps:
 		KIND_IMAGE="$(KIND_IMAGE)" \
 		PROJECT_DIR="$(PWD)" \
 		E2E_TEST_COV_DIR=${TEST_COV_DIR} \
-		go test -count=1 -v -timeout ${E2E_TEST_TIMEOUT} $$(go list ./${E2E_TEST}/... | grep -vE '/e2e/(common|internal)(/|$$)')
+		go test -count=1 -v -parallel ${E2E_SUITE_PARALLEL} -timeout ${E2E_TEST_TIMEOUT} $$(go list ./${E2E_TEST}/... | grep -vE '/e2e/(common|internal)(/|$$)')
 		go tool covdata textfmt -i=${TEST_COV_DIR}/covdatafiles -o ${TEST_COV_DIR}/coverage_e2e.out
 	@echo "--- E2E test coverage report"
 	go tool covdata percent -i=${TEST_COV_DIR}/covdatafiles

e2e/common/cluster.go

@@ -15,6 +15,7 @@
 package common
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"os"
@@ -59,6 +60,11 @@ func WithCluster(name string, t *testing.T, fn func(*testing.T, Cluster), before
 	ctrl.SetLogger(zapLogger)
 
 	cluster, err := GetTestCluster(name, opts...)
+	if err != nil {
+		// The cluster is created before the client can fail, and the deferred
+		// teardown below is not registered yet.
+		assert.NoError(t, DeleteTestCluster(name))
+	}
 	RequireNoError(t, err)
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -134,12 +140,18 @@ func (c kindCluster) CollectTestCoverageFiles(ns string, loggingOperatorName str
 		return errors.WrapIfWithDetails(err, "Error in sending signal to logging-operator", cmdOut)
 	}
 	testCovDir := os.Getenv("E2E_TEST_COV_DIR")
-	cmd = exec.Command("sh", "-c", fmt.Sprintf(
-		"kubectl --kubeconfig %s -n %s exec deployment/%s -- tar -cf - /covdatafiles | tar -xf - -C %s",
-		c.KubeConfigFilePath(), ns, loggingOperatorName, testCovDir))
-	cmdOut, err = cmd.Output()
+	archive := CmdEnv(exec.Command("kubectl", "-n", ns,
+		"exec", fmt.Sprintf("deployment/%s", loggingOperatorName), "--",
+		"tar", "-cf", "-", "/covdatafiles"), c)
+	tarball, err := archive.Output()
 	if err != nil {
-		return errors.WrapIfWithDetails(err, "Error in collecting test coverage files", cmdOut)
+		return errors.WrapIfWithDetails(err, "Error in reading test coverage files", tarball)
+	}
+
+	extract := exec.Command("tar", "-xf", "-", "-C", testCovDir)
+	extract.Stdin = bytes.NewReader(tarball)
+	if cmdOut, err := extract.CombinedOutput(); err != nil {
+		return errors.WrapIfWithDetails(err, "Error in extracting test coverage files", cmdOut)
 	}
 	return nil
 }

e2e/common/kind.go

@@ -15,8 +15,11 @@
 package common
 
 import (
+	"fmt"
 	"strings"
 
+	"emperror.dev/errors"
+
 	"github.com/kube-logging/logging-operator/e2e/internal/kind"
 )
 
@@ -28,13 +31,24 @@ const KindClusterCreationTimeout = "3m"
 var kindCLI = kind.New()
 
 func KindClusterKubeconfig(name string) ([]byte, error) {
-	err := kindCLI.CreateCluster(kind.CreateClusterOptions{
+	create := kind.CreateClusterOptions{
 		Name: name,
 		Wait: KindClusterCreationTimeout,
-	})
-	if err != nil && !isClusterAlreadyExistsError(err) {
+	}
+
+	err := kindCLI.CreateCluster(create)
+	if err != nil && isClusterAlreadyExistsError(err) {
+		// Adopting a leftover would hand the suite an unknown operator and data.
+		fmt.Printf("kind cluster %q already exists, recreating it\n", name)
+		if err := kindCLI.DeleteCluster(kind.DeleteClusterOptions{Name: name}); err != nil {
+			return nil, errors.WrapIfWithDetails(err, "deleting a leftover kind cluster", "clusterName", name)
+		}
+		err = kindCLI.CreateCluster(create)
+	}
+	if err != nil {
 		return nil, err
 	}
+
 	return kindCLI.GetKubeconfig(kind.GetKubeconfigOptions{
 		Name: name,
 	})

e2e/elasticsearch-multiversion/elasticsearch_multiversion_test.go

@@ -70,6 +70,29 @@ func init() {
 	}
 }
 
+// logContainerRestarts names a restart loop while a readiness wait is still
+// running, so it is a reported number rather than an unexplained stall.
+func logContainerRestarts(t *testing.T, c common.Cluster, ctx context.Context, ns, app string) {
+	var pods corev1.PodList
+	if err := c.GetClient().List(ctx, &pods, client.InNamespace(ns), client.MatchingLabels{"app": app}); err != nil {
+		t.Logf("listing %s pods failed: %v", app, err)
+		return
+	}
+
+	for _, pod := range pods.Items {
+		for _, status := range pod.Status.ContainerStatuses {
+			if status.RestartCount == 0 {
+				continue
+			}
+			reason := "unknown"
+			if terminated := status.LastTerminationState.Terminated; terminated != nil {
+				reason = fmt.Sprintf("%s, exit %d", terminated.Reason, terminated.ExitCode)
+			}
+			t.Logf("%s/%s restarted %d times, last termination: %s", pod.Name, status.Name, status.RestartCount, reason)
+		}
+	}
+}
+
 // esReadyBudget caps a wait strictly below the enclosing -timeout, so an
 // Elasticsearch deployment that never becomes ready fails by name instead of
 // letting the package binary panic and report only a goroutine dump.
@@ -210,20 +233,12 @@ func TestElasticsearch_MultiVersion(t *testing.T) {
 										corev1.ResourceCPU:    resource.MustParse("500m"),
 									},
 									Limits: corev1.ResourceList{
-										corev1.ResourceMemory: resource.MustParse("1Gi"),
+										corev1.ResourceMemory: resource.MustParse("1536Mi"),
 										corev1.ResourceCPU:    resource.MustParse("1000m"),
 									},
 								},
-								LivenessProbe: &corev1.Probe{
-									ProbeHandler: corev1.ProbeHandler{
-										HTTPGet: &corev1.HTTPGetAction{
-											Path: "/_cluster/health",
-											Port: intstr.FromInt(9200),
-										},
-									},
-									InitialDelaySeconds: 60,
-									PeriodSeconds:       10,
-								},
+								// No liveness probe: readiness already gates the wait, and a
+								// 60s + 3x10s deadline killed the JVM mid-boot on a loaded runner.
 								ReadinessProbe: &corev1.Probe{
 									ProbeHandler: corev1.ProbeHandler{
 										HTTPGet: &corev1.HTTPGetAction{
@@ -332,20 +347,12 @@ func TestElasticsearch_MultiVersion(t *testing.T) {
 										corev1.ResourceCPU:    resource.MustParse("500m"),
 									},
 									Limits: corev1.ResourceList{
-										corev1.ResourceMemory: resource.MustParse("1Gi"),
+										corev1.ResourceMemory: resource.MustParse("1536Mi"),
 										corev1.ResourceCPU:    resource.MustParse("1000m"),
 									},
 								},
-								LivenessProbe: &corev1.Probe{
-									ProbeHandler: corev1.ProbeHandler{
-										HTTPGet: &corev1.HTTPGetAction{
-											Path: "/_cluster/health",
-											Port: intstr.FromInt(9200),
-										},
-									},
-									InitialDelaySeconds: 60,
-									PeriodSeconds:       10,
-								},
+								// No liveness probe: readiness already gates the wait, and a
+								// 60s + 3x10s deadline killed the JVM mid-boot on a loaded runner.
 								ReadinessProbe: &corev1.Probe{
 									ProbeHandler: corev1.ProbeHandler{
 										HTTPGet: &corev1.HTTPGetAction{
@@ -454,20 +461,12 @@ func TestElasticsearch_MultiVersion(t *testing.T) {
 										corev1.ResourceCPU:    resource.MustParse("500m"),
 									},
 									Limits: corev1.ResourceList{
-										corev1.ResourceMemory: resource.MustParse("1Gi"),
+										corev1.ResourceMemory: resource.MustParse("1536Mi"),
 										corev1.ResourceCPU:    resource.MustParse("1000m"),
 									},
 								},
-								LivenessProbe: &corev1.Probe{
-									ProbeHandler: corev1.ProbeHandler{
-										HTTPGet: &corev1.HTTPGetAction{
-											Path: "/_cluster/health",
-											Port: intstr.FromInt(9200),
-										},
-									},
-									InitialDelaySeconds: 60,
-									PeriodSeconds:       10,
-								},
+								// No liveness probe: readiness already gates the wait, and a
+								// 60s + 3x10s deadline killed the JVM mid-boot on a loaded runner.
 								ReadinessProbe: &corev1.Probe{
 									ProbeHandler: corev1.ProbeHandler{
 										HTTPGet: &corev1.HTTPGetAction{
@@ -486,20 +485,16 @@ func TestElasticsearch_MultiVersion(t *testing.T) {
 		}
 		common.RequireNoError(t, c.GetClient().Create(ctx, es9Deployment))
 
-		t.Log("Waiting for Elasticsearch 7 deployment to be ready...")
-		require.Eventually(t, func() bool {
-			return wait.DeploymentAvailable(t, c.GetClient(), ctx, ns, "elasticsearch7")()
-		}, esReadyBudget(t), 10*time.Second)
-
-		t.Log("Waiting for Elasticsearch 8 deployment to be ready...")
-		require.Eventually(t, func() bool {
-			return wait.DeploymentAvailable(t, c.GetClient(), ctx, ns, "elasticsearch8")()
-		}, esReadyBudget(t), 10*time.Second)
-
-		t.Log("Waiting for Elasticsearch 9 deployment to be ready...")
-		require.Eventually(t, func() bool {
-			return wait.DeploymentAvailable(t, c.GetClient(), ctx, ns, "elasticsearch9")()
-		}, esReadyBudget(t), 10*time.Second)
+		for _, name := range []string{"elasticsearch7", "elasticsearch8", "elasticsearch9"} {
+			t.Logf("Waiting for %s deployment to be ready...", name)
+			require.Eventually(t, func() bool {
+				if wait.DeploymentAvailable(t, c.GetClient(), ctx, ns, name)() {
+					return true
+				}
+				logContainerRestarts(t, c, ctx, ns, name)
+				return false
+			}, esReadyBudget(t), 10*time.Second)
+		}
 
 		logging := v1beta1.Logging{
 			ObjectMeta: metav1.ObjectMeta{

e2e/fluentd-aggregator-detached/fluentd_aggregator_detached_test.go

@@ -259,7 +259,10 @@ func TestFluentdAggregator_detached_MultiWorker(t *testing.T) {
 				return false
 			}
 			if len(logging.Status.FluentdConfigName) == 0 || logging.Status.FluentdConfigName != fluentd.Name {
-				common.RequireNoError(t, c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&logging), &logging))
+				if err := c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&logging), &logging); err != nil {
+					t.Logf("reading the Logging failed, retrying: %v", err)
+					return false
+				}
 				t.Logf("logging should use the detached fluentd configuration (name=%s), found: %v", fluentd.Name, logging.Status.FluentdConfigName)
 				return false
 			}
@@ -268,15 +271,25 @@ func TestFluentdAggregator_detached_MultiWorker(t *testing.T) {
 				return false
 			}
 			var detachedFluentds v1beta1.FluentdConfigList
-			common.RequireNoError(t, c.GetClient().List(ctx, &detachedFluentds))
+			if err := c.GetClient().List(ctx, &detachedFluentds); err != nil {
+				t.Logf("listing the detached fluentd configurations failed, retrying: %v", err)
+				return false
+			}
 			if len(detachedFluentds.Items) != 2 {
-				// Add a new detached fluentd that is not going to be used
-				common.RequireNoError(t, c.GetClient().Create(ctx, &excessFluentd))
+				// Add a new detached fluentd that is not going to be used.
+				// AlreadyExists is tolerated: the list above can be stale on a retry.
+				if err := client.IgnoreAlreadyExists(c.GetClient().Create(ctx, &excessFluentd)); err != nil {
+					t.Logf("creating the excess detached fluentd failed, retrying: %v", err)
+					return false
+				}
 				t.Log("creating excess detached fluentd")
 				return false
 			} else if isValid := wait.CheckExcessFluentdStatus(t, c.GetClient(), ctx, &excessFluentd); !isValid && len(detachedFluentds.Items) == 2 {
 				t.Log("checking excess detached fluentd status")
-				common.RequireNoError(t, c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&excessFluentd), &excessFluentd))
+				if err := c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&excessFluentd), &excessFluentd); err != nil {
+					t.Logf("reading the excess detached fluentd failed, retrying: %v", err)
+					return false
+				}
 				return false
 			}
 

e2e/fluentd-aggregator/fluentd_aggregator_test.go

@@ -407,7 +407,10 @@ func TestFluentdAggregator_ConfigChecks(t *testing.T) {
 		output.Spec.FileOutput.Path = "/tmp/zzz"
 		common.RequireNoError(t, c.GetClient().Patch(ctx, &output, patch))
 		require.Eventually(t, func() bool {
-			common.RequireNoError(t, c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&logging), &logging))
+			if err := c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&logging), &logging); err != nil {
+				t.Logf("reading the Logging failed, retrying: %v", err)
+				return false
+			}
 			if logging.Status.ProblemsCount > 0 {
 				for _, problem := range logging.Status.Problems {
 					if configCheckFailure.MatchString(problem) {
@@ -425,7 +428,10 @@ func TestFluentdAggregator_ConfigChecks(t *testing.T) {
 		output.Spec.FileOutput.Path = "/tmp/logs/${tag}/%Y/%m/%d.%H.%M"
 		common.RequireNoError(t, c.GetClient().Patch(ctx, &output, patch))
 		require.Eventually(t, func() bool {
-			common.RequireNoError(t, c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&logging), &logging))
+			if err := c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&logging), &logging); err != nil {
+				t.Logf("reading the Logging failed, retrying: %v", err)
+				return false
+			}
 			if logging.Status.ProblemsCount > 0 {
 				for _, problem := range logging.Status.Problems {
 					if configCheckFailure.MatchString(problem) {

e2e/syslog-ng-aggregator-detached/syslog_ng_aggregator_detached_test.go

@@ -294,7 +294,9 @@ func TestSyslogNGDetachedIsRunningAndForwardingLogs(t *testing.T) {
 			}
 			if logging.Status.SyslogNGConfigName != syslogNG.Name {
 				t.Logf("logging should use the detached SyslogNG configuration (name=%s), found: %v", syslogNG.Name, logging.Status.SyslogNGConfigName)
-				common.RequireNoError(t, c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&logging), &logging))
+				if err := c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&logging), &logging); err != nil {
+					t.Logf("reading the Logging failed, retrying: %v", err)
+				}
 				return false
 			}
 
@@ -303,15 +305,24 @@ func TestSyslogNGDetachedIsRunningAndForwardingLogs(t *testing.T) {
 				return false
 			}
 			var detachedSyslogNGs v1beta1.SyslogNGConfigList
-			common.RequireNoError(t, c.GetClient().List(ctx, &detachedSyslogNGs))
+			if err := c.GetClient().List(ctx, &detachedSyslogNGs); err != nil {
+				t.Logf("listing the detached syslog-ng configurations failed, retrying: %v", err)
+				return false
+			}
 			if len(detachedSyslogNGs.Items) != 2 {
-				// Add a new detached syslogng that is not going to be used
-				common.RequireNoError(t, c.GetClient().Create(ctx, &excessSyslogNG))
+				// Add a new detached syslogng that is not going to be used.
+				// AlreadyExists is tolerated: the list above can be stale on a retry.
+				if err := client.IgnoreAlreadyExists(c.GetClient().Create(ctx, &excessSyslogNG)); err != nil {
+					t.Logf("creating the excess detached syslog-ng failed, retrying: %v", err)
+					return false
+				}
 				t.Log("creating excess detached syslog-ng")
 				return false
 			} else if isValid := wait.CheckExcessSyslogNGStatus(t, c.GetClient(), ctx, &excessSyslogNG); !isValid && len(detachedSyslogNGs.Items) == 2 {
 				t.Log("checking excess detached SyslogNG status")
-				common.RequireNoError(t, c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&excessSyslogNG), &excessSyslogNG))
+				if err := c.GetClient().Get(ctx, utils.ObjectKeyFromObjectMeta(&excessSyslogNG), &excessSyslogNG); err != nil {
+					t.Logf("reading the excess detached syslog-ng failed, retrying: %v", err)
+				}
 				return false
 			}