Merge pull request #2126 from kube-logging/feat/es-version-compat

feat: use forked elasticsearch gem

Author: csatib02

e2e/common/cond/conditions.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cisco-open/operator-tools/pkg/utils"
 	"github.com/kube-logging/logging-operator/e2e/common"
 	"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -191,3 +192,50 @@ func CheckExcessSyslogNGStatus(t *testing.T, c *common.Cluster, ctx *context.Con
 
 	return true
 }
+
+// DeploymentAvailable returns a condition function that checks if a deployment
+// is available with all replicas ready.
+func DeploymentAvailable(t *testing.T, c client.Client, ctx *context.Context, namespace, name string) func() bool {
+	return func() bool {
+		deployment := &appsv1.Deployment{}
+		if err := c.Get(*ctx, client.ObjectKey{
+			Name:      name,
+			Namespace: namespace,
+		}, deployment); err != nil {
+			t.Logf("Failed to get deployment %s/%s: %v", namespace, name, err)
+			return false
+		}
+
+		if deployment.Spec.Replicas == nil {
+			return false
+		}
+		desiredReplicas := *deployment.Spec.Replicas
+
+		if deployment.Status.ReadyReplicas != desiredReplicas {
+			t.Logf("Deployment %s/%s: %d/%d replicas ready",
+				namespace, name, deployment.Status.ReadyReplicas, desiredReplicas)
+			return false
+		}
+
+		if deployment.Status.AvailableReplicas != desiredReplicas {
+			t.Logf("Deployment %s/%s: %d/%d replicas available",
+				namespace, name, deployment.Status.AvailableReplicas, desiredReplicas)
+			return false
+		}
+
+		for _, condition := range deployment.Status.Conditions {
+			if condition.Type == appsv1.DeploymentAvailable {
+				if condition.Status == corev1.ConditionTrue {
+					t.Logf("Deployment %s/%s is available", namespace, name)
+					return true
+				}
+				t.Logf("Deployment %s/%s Available condition is %s: %s",
+					namespace, name, condition.Status, condition.Message)
+				return false
+			}
+		}
+
+		t.Logf("Deployment %s/%s has no Available condition", namespace, name)
+		return false
+	}
+}

e2e/common/helpers.go

@@ -20,13 +20,15 @@ import (
 	"os"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"emperror.dev/errors"
 	"github.com/spf13/cast"
 	"github.com/stretchr/testify/assert"
-	v12 "k8s.io/api/core/v1"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
@@ -78,7 +80,7 @@ func LoggingInfra(
 	producerLabels map[string]string) {
 
 	output := v1beta1.ClusterOutput{
-		ObjectMeta: v1.ObjectMeta{
+		ObjectMeta: metav1.ObjectMeta{
 			Name:      "http",
 			Namespace: nsInfra,
 		},
@@ -96,7 +98,7 @@ func LoggingInfra(
 
 	RequireNoError(t, c.Create(ctx, &output))
 	flow := v1beta1.ClusterFlow{
-		ObjectMeta: v1.ObjectMeta{
+		ObjectMeta: metav1.ObjectMeta{
 			Name:      "flow",
 			Namespace: nsInfra,
 		},
@@ -115,7 +117,7 @@ func LoggingInfra(
 	RequireNoError(t, c.Create(ctx, &flow))
 
 	agent := v1beta1.FluentbitAgent{
-		ObjectMeta: v1.ObjectMeta{
+		ObjectMeta: metav1.ObjectMeta{
 			Name: "infra",
 		},
 		Spec: v1beta1.FluentbitSpec{
@@ -135,7 +137,7 @@ func LoggingInfra(
 	RequireNoError(t, c.Create(ctx, &agent))
 
 	logging := v1beta1.Logging{
-		ObjectMeta: v1.ObjectMeta{
+		ObjectMeta: metav1.ObjectMeta{
 			Name: "infra",
 			Labels: map[string]string{
 				"tenant": "infra",
@@ -158,10 +160,10 @@ func LoggingInfra(
 					Tag:        NodeExporterTag,
 				},
 				DisablePvc: true,
-				Resources: v12.ResourceRequirements{
-					Requests: v12.ResourceList{
-						v12.ResourceCPU:    resource.MustParse("50m"),
-						v12.ResourceMemory: resource.MustParse("50M"),
+				Resources: corev1.ResourceRequirements{
+					Requests: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("50m"),
+						corev1.ResourceMemory: resource.MustParse("50M"),
 					},
 				},
 			},
@@ -181,7 +183,7 @@ func LoggingTenant(
 	buffer *output.Buffer,
 	producerLabels map[string]string) {
 	output := v1beta1.Output{
-		ObjectMeta: v1.ObjectMeta{
+		ObjectMeta: metav1.ObjectMeta{
 			Name:      "http",
 			Namespace: nsTenant,
 		},
@@ -197,7 +199,7 @@ func LoggingTenant(
 
 	RequireNoError(t, c.Create(ctx, &output))
 	flow := v1beta1.Flow{
-		ObjectMeta: v1.ObjectMeta{
+		ObjectMeta: metav1.ObjectMeta{
 			Name:      "flow",
 			Namespace: nsTenant,
 		},
@@ -216,7 +218,7 @@ func LoggingTenant(
 	RequireNoError(t, c.Create(ctx, &flow))
 
 	logging := v1beta1.Logging{
-		ObjectMeta: v1.ObjectMeta{
+		ObjectMeta: metav1.ObjectMeta{
 			Name: "tenant",
 			Labels: map[string]string{
 				"tenant": "tenant",
@@ -240,10 +242,10 @@ func LoggingTenant(
 					Tag:        NodeExporterTag,
 				},
 				DisablePvc: true,
-				Resources: v12.ResourceRequirements{
-					Requests: v12.ResourceList{
-						v12.ResourceCPU:    resource.MustParse("50m"),
-						v12.ResourceMemory: resource.MustParse("50M"),
+				Resources: corev1.ResourceRequirements{
+					Requests: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("50m"),
+						corev1.ResourceMemory: resource.MustParse("50M"),
 					},
 				},
 			},
@@ -254,20 +256,68 @@ func LoggingTenant(
 
 func LoggingRoute(ctx context.Context, t *testing.T, c client.Client) {
 	ap := v1beta1.LoggingRoute{
-		ObjectMeta: v1.ObjectMeta{
+		ObjectMeta: metav1.ObjectMeta{
 			Name: "tenants",
 		},
 		Spec: v1beta1.LoggingRouteSpec{
 			Source: "infra",
-			Targets: v1.LabelSelector{
-				MatchExpressions: []v1.LabelSelectorRequirement{
+			Targets: metav1.LabelSelector{
+				MatchExpressions: []metav1.LabelSelectorRequirement{
 					{
 						Key:      "tenant",
-						Operator: v1.LabelSelectorOpExists,
+						Operator: metav1.LabelSelectorOpExists,
 					},
 				},
 			},
 		},
 	}
 	RequireNoError(t, c.Create(ctx, &ap))
 }
+
+// WaitForPodReady waits for a pod to be in Running phase and Ready condition
+func WaitForPodReady(ctx context.Context, c client.Client, pod *corev1.Pod, pollInterval, pollTimeout time.Duration) error {
+	return wait.PollUntilContextTimeout(ctx, pollInterval, pollTimeout, true, wait.ConditionWithContextFunc(func(ctx context.Context) (bool, error) {
+		var updatedPod corev1.Pod
+		err := c.Get(ctx, client.ObjectKeyFromObject(pod), &updatedPod)
+		if client.IgnoreNotFound(err) != nil {
+			return false, fmt.Errorf("failed to get pod status: %w", err)
+		}
+
+		isReady := updatedPod.Status.Phase == corev1.PodRunning
+		for _, cond := range updatedPod.Status.Conditions {
+			if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
+				return true, nil
+			}
+		}
+		return isReady, nil
+	}))
+}
+
+// SetupCurlPod creates a curl pod for testing HTTP endpoints and waits for it to be ready
+func SetupCurlPod(ctx context.Context, c client.Client, namespace, name string, pollInterval, pollTimeout time.Duration) (*corev1.Pod, error) {
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name:    "curl",
+					Image:   "curlimages/curl:latest",
+					Command: []string{"sleep", "3600"},
+				},
+			},
+		},
+	}
+
+	if err := c.Create(ctx, pod); err != nil {
+		return nil, fmt.Errorf("failed to create curl pod: %w", err)
+	}
+
+	if err := WaitForPodReady(ctx, c, pod, pollInterval, pollTimeout); err != nil {
+		return nil, fmt.Errorf("failed to wait for curl pod to be ready: %w", err)
+	}
+
+	return pod, nil
+}