test: refactor prometheus_metrics_test.go (#7063)

Refactors the prometheus_metrics sequential e2e test to rely on polling with a conditional instead of sleeping.
We've noticed flakes with this test where the magic sleep timings don't play well on certain machines and platforms.

Also adds KedaEventually which acts like ginkgo.Eventually, which allows to poll a condition until it is true.

Signed-off-by: Max Cao <macao@redhat.com>

Author: maxcao13

tests/helper/helper.go

@@ -70,6 +70,11 @@ const (
 	StrimziNamespace = "strimzi"
 )
 
+const (
+	WaitShort     = 20 * time.Second
+	IntervalShort = 1 * time.Second
+)
+
 const (
 	caCrtPath = "/tmp/keda-e2e-ca.crt"
 	caKeyPath = "/tmp/keda-e2e-ca.key"
@@ -1041,6 +1046,42 @@ func CheckKubectlGetResult(t *testing.T, kind string, name string, namespace str
 	assert.Equal(t, expected, unqoutedOutput)
 }
 
+// KedaEventually checks if the provided conditionFunc eventually returns true
+// (and no error) within the context's deadline. It polls the conditionFunc
+// at the given interval until the condition is met or the context times out.
+func KedaEventually(ctx context.Context, conditionFunc wait.ConditionWithContextFunc, interval time.Duration) error {
+	if interval <= 0 {
+		return fmt.Errorf("polling interval must be positive, got %v", interval)
+	}
+
+	ok, err := conditionFunc(ctx)
+	if err != nil {
+		return fmt.Errorf("eventually check failed on initial check: %w", err)
+	}
+	if ok {
+		return nil
+	}
+
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("eventually check failed: context deadline exceeded before condition was met")
+
+		case <-ticker.C:
+			ok, err := conditionFunc(ctx)
+			if err != nil {
+				return fmt.Errorf("eventually check failed during polling: %w", err)
+			}
+			if ok {
+				return nil
+			}
+		}
+	}
+}
+
 // KedaConsistently checks if the provided conditionFunc consistently returns true
 // (and no error) for the entire duration specified by the context's deadline.
 // It polls the conditionFunc at the given interval.

tests/sequential/prometheus_metrics/prometheus_metrics_test.go

@@ -11,7 +11,6 @@ import (
 	"os/exec"
 	"strings"
 	"testing"
-	"time"
 
 	prommodel "github.com/prometheus/client_model/go"
 	"github.com/prometheus/common/expfmt"
@@ -486,7 +485,7 @@ func getTemplateData() (templateData, []Template) {
 }
 
 func fetchAndParsePrometheusMetrics(t *testing.T, cmd string) map[string]*prommodel.MetricFamily {
-	out, _, err := ExecCommandOnSpecificPod(t, clientName, testNamespace, cmd)
+	out, _, err := ExecCommandOnSpecificPodWithoutTTY(t, clientName, testNamespace, cmd)
 	assert.NoErrorf(t, err, "cannot execute command - %s", err)
 
 	parser := expfmt.TextParser{}
@@ -498,6 +497,33 @@ func fetchAndParsePrometheusMetrics(t *testing.T, cmd string) map[string]*prommo
 	return families
 }
 
+// WaitForPrometheusMetric waits for a specific metric to appear in the Prometheus metrics endpoint
+// and validates that the MetricFamily it has certain conditions using the provided familyValidator function.
+// Returns the parsed MetricFamily.
+func WaitForPrometheusMetric(t *testing.T, metricToWaitFor string, familyValidator func(family *prommodel.MetricFamily) bool) map[string]*prommodel.MetricFamily {
+	contextWithTimeout, cancel := context.WithTimeout(context.Background(), WaitShort)
+	defer cancel()
+	var family map[string]*prommodel.MetricFamily
+	err := KedaEventually(contextWithTimeout, func(ctx context.Context) (bool, error) {
+		t.Logf("Waiting for metric %s", metricToWaitFor)
+		family = fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
+
+		if _, ok := family[metricToWaitFor]; ok {
+			if familyValidator(family[metricToWaitFor]) {
+				return true, nil
+			}
+			return false, nil
+		}
+		return false, nil
+	}, IntervalShort)
+
+	if err != nil {
+		t.Errorf("error waiting for metric %s: %v", metricToWaitFor, err)
+	}
+
+	return family
+}
+
 func testScalerMetricValue(t *testing.T) {
 	t.Log("--- testing scaler metric value ---")
 
@@ -528,110 +554,78 @@ func testScaledObjectErrors(t *testing.T, data templateData) {
 	t.Log("--- testing scaled object errors ---")
 
 	KubectlDeleteWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
-	time.Sleep(2 * time.Second)
 	KubectlApplyWithTemplate(t, data, "wrongScaledObjectTemplate", wrongScaledObjectTemplate)
 
-	// wait for 2 seconds as pollinginterval is 2
-	time.Sleep(20 * time.Second)
+	WaitForPrometheusMetric(t, "keda_scaled_object_errors_total", func(family *prommodel.MetricFamily) bool {
+		errCounterVal1 := getErrorMetricsValue(family)
 
-	family := fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
-	val, ok := family["keda_scaled_object_errors_total"]
-	assert.True(t, ok, "keda_scaled_object_errors_total not available")
-	if ok {
-		errCounterVal1 := getErrorMetricsValue(val)
+		// in the nested wait, we are just waiting for errCounterVal2 to eventually be greater that the first errCounterVal1
+		metrics2 := WaitForPrometheusMetric(t, "keda_scaled_object_errors_total", func(family *prommodel.MetricFamily) bool {
+			errCounterVal2 := getErrorMetricsValue(family)
+			return errCounterVal2 > errCounterVal1 && errCounterVal2 > 0
+		})
 
-		// wait for 2 seconds as pollinginterval is 2
-		time.Sleep(2 * time.Second)
+		errCounterVal2 := getErrorMetricsValue(metrics2["keda_scaled_object_errors_total"])
 
-		family = fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
-		val, ok := family["keda_scaled_object_errors_total"]
-		assert.True(t, ok, "keda_scaled_object_errors_total not available")
-		if ok {
-			errCounterVal2 := getErrorMetricsValue(val)
-			assert.NotEqual(t, errCounterVal2, float64(0))
-			assert.GreaterOrEqual(t, errCounterVal2, errCounterVal1)
-		}
-	}
+		// we don't have to check again, but extra validation is fine
+		return errCounterVal2 > errCounterVal1 && errCounterVal2 > 0
+	})
 
 	KubectlDeleteWithTemplate(t, data, "wrongScaledObjectTemplate", wrongScaledObjectTemplate)
-	time.Sleep(2 * time.Second)
 	KubectlApplyWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
 }
 
 func testScaledJobErrors(t *testing.T, data templateData) {
 	t.Log("--- testing scaled job errors ---")
 
 	KubectlDeleteWithTemplate(t, data, "scaledJobTemplate", scaledJobTemplate)
-	time.Sleep(2 * time.Second)
 	KubectlApplyWithTemplate(t, data, "wrongScaledJobTemplate", wrongScaledJobTemplate)
 
-	// wait for 2 seconds as pollinginterval is 2
-	time.Sleep(20 * time.Second)
+	WaitForPrometheusMetric(t, "keda_scaled_job_errors_total", func(family *prommodel.MetricFamily) bool {
+		errCounterVal1 := getErrorMetricsValue(family)
 
-	family := fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
-	val, ok := family["keda_scaled_job_errors_total"]
-	assert.True(t, ok, "keda_scaled_job_errors_total not available")
-	if ok {
-		errCounterVal1 := getErrorMetricsValue(val)
+		// in the nested wait, we are just waiting for errCounterVal2 to eventually be greater that the first errCounterVal1
+		metrics2 := WaitForPrometheusMetric(t, "keda_scaled_job_errors_total", func(family *prommodel.MetricFamily) bool {
+			errCounterVal2 := getErrorMetricsValue(family)
+			return errCounterVal2 > errCounterVal1 && errCounterVal2 > 0
+		})
 
-		// wait for 2 seconds as pollinginterval is 2
-		time.Sleep(2 * time.Second)
+		errCounterVal2 := getErrorMetricsValue(metrics2["keda_scaled_job_errors_total"])
 
-		family = fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
-		val, ok := family["keda_scaled_job_errors_total"]
-		assert.True(t, ok, "keda_scaled_job_errors_total not available")
-		if ok {
-			errCounterVal2 := getErrorMetricsValue(val)
-			assert.NotEqual(t, errCounterVal2, float64(0))
-			assert.GreaterOrEqual(t, errCounterVal2, errCounterVal1)
-		} else {
-			t.Errorf("metric keda_scaled_job_errors_total not available")
-		}
-	} else {
-		t.Errorf("metric keda_scaled_job_errors_total not available")
-	}
+		// we don't have to check again, but extra validation is fine
+		return errCounterVal2 > errCounterVal1 && errCounterVal2 > 0
+	})
 
 	KubectlDeleteWithTemplate(t, data, "wrongScaledJobTemplate", wrongScaledJobTemplate)
-	time.Sleep(2 * time.Second)
 	KubectlApplyWithTemplate(t, data, "scaledJobTemplate", scaledJobTemplate)
 }
 
 func testScalerErrors(t *testing.T, data templateData) {
 	t.Log("--- testing scaler errors ---")
 
 	KubectlDeleteWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
-	time.Sleep(2 * time.Second)
 	KubectlApplyWithTemplate(t, data, "wrongScaledObjectTemplate", wrongScaledObjectTemplate)
-
 	KubectlDeleteWithTemplate(t, data, "scaledJobTemplate", scaledJobTemplate)
-	time.Sleep(2 * time.Second)
 	KubectlApplyWithTemplate(t, data, "wrongScaledJobTemplate", wrongScaledJobTemplate)
 
-	family := fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
-	val, ok := family["keda_scaler_detail_errors_total"]
-	assert.True(t, ok, "keda_scaler_detail_errors_total not available")
-	if ok {
-		errCounterVal1 := getErrorMetricsValue(val)
+	WaitForPrometheusMetric(t, "keda_scaler_detail_errors_total", func(family *prommodel.MetricFamily) bool {
+		errCounterVal1 := getErrorMetricsValue(family)
 
-		// wait for 20 seconds to correctly fetch metrics.
-		time.Sleep(20 * time.Second)
+		// in the nested wait, we are just waiting for errCounterVal2 to eventually be greater that the first errCounterVal1
+		metrics2 := WaitForPrometheusMetric(t, "keda_scaler_detail_errors_total", func(family *prommodel.MetricFamily) bool {
+			errCounterVal2 := getErrorMetricsValue(family)
+			return errCounterVal2 > errCounterVal1 && errCounterVal2 > 0
+		})
 
-		family = fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
-		val, ok := family["keda_scaler_detail_errors_total"]
-		assert.True(t, ok, "keda_scaler_detail_errors_total not available")
-		if ok {
-			errCounterVal2 := getErrorMetricsValue(val)
-			assert.NotEqual(t, errCounterVal2, float64(0))
-			assert.GreaterOrEqual(t, errCounterVal2, errCounterVal1)
-		}
-	}
+		errCounterVal2 := getErrorMetricsValue(metrics2["keda_scaler_detail_errors_total"])
+
+		// we don't have to check again, but extra validation is fine
+		return errCounterVal2 > errCounterVal1 && errCounterVal2 > 0
+	})
 
 	KubectlDeleteWithTemplate(t, data, "wrongScaledJobTemplate", wrongScaledJobTemplate)
-	time.Sleep(2 * time.Second)
 	KubectlApplyWithTemplate(t, data, "scaledJobTemplate", scaledJobTemplate)
-
 	KubectlDeleteWithTemplate(t, data, "wrongScaledObjectTemplate", wrongScaledObjectTemplate)
-	time.Sleep(2 * time.Second)
 	KubectlApplyWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
 }
 
@@ -827,18 +821,41 @@ func testScaledObjectPausedMetric(t *testing.T, data templateData) {
 
 	// Pause the ScaledObject
 	KubectlApplyWithTemplate(t, data, "scaledObjectPausedTemplate", scaledObjectPausedTemplate)
-	time.Sleep(20 * time.Second)
 
 	// Check that the paused metric is now true
-	families := fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
+	families := WaitForPrometheusMetric(t, "keda_scaled_object_paused", func(family *prommodel.MetricFamily) bool {
+		metricValue := 0.0
+		metrics := family.GetMetric()
+		for _, metric := range metrics {
+			labels := metric.GetLabel()
+			for _, label := range labels {
+				if *label.Name == labelScaledObject && *label.Value == scaledObjectName {
+					metricValue = *metric.Gauge.Value
+				}
+			}
+		}
+
+		return metricValue == float64(1)
+	})
 	assertScaledObjectPausedMetric(t, families, scaledObjectName, true)
 
 	// Unpause the ScaledObject
 	KubectlApplyWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
-	time.Sleep(20 * time.Second)
 
 	// Check that the paused metric is back to false
-	families = fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
+	families = WaitForPrometheusMetric(t, "keda_scaled_object_paused", func(family *prommodel.MetricFamily) bool {
+		metricValue := 0.0
+		metrics := family.GetMetric()
+		for _, metric := range metrics {
+			labels := metric.GetLabel()
+			for _, label := range labels {
+				if *label.Name == labelScaledObject && *label.Value == scaledObjectName {
+					metricValue = *metric.Gauge.Value
+				}
+			}
+		}
+		return metricValue == float64(0)
+	})
 	assertScaledObjectPausedMetric(t, families, scaledObjectName, false)
 }
 
@@ -1350,62 +1367,64 @@ func testCloudEventEmitted(t *testing.T, data templateData) {
 	t.Log("--- testing cloudevent emitted ---")
 
 	KubectlDeleteWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
+
 	KubectlApplyWithTemplate(t, data, "cloudEventSourceTemplate", cloudEventSourceTemplate)
 	KubectlApplyWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
 
-	family := fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
+	defer KubectlDeleteWithTemplate(t, data, "cloudEventSourceTemplate", cloudEventSourceTemplate)
 
-	if val, ok := family["keda_cloudeventsource_events_emitted_total"]; ok {
-		var found bool
-		metrics := val.GetMetric()
-		for _, metric := range metrics {
+	familyValidator := func(family *prommodel.MetricFamily) bool {
+		labels := family.GetMetric()
+		found := false
+		for _, metric := range labels {
 			labels := metric.GetLabel()
 			if len(labels) >= 4 &&
 				ExtractPrometheusLabelValue("cloudeventsource", labels) == "prometheus-metrics-test-ce" &&
 				ExtractPrometheusLabelValue("eventsink", labels) == "http" &&
 				ExtractPrometheusLabelValue("namespace", labels) == "prometheus-metrics-test-ns" &&
-				ExtractPrometheusLabelValue("state", labels) == "emitted" {
-				assert.GreaterOrEqual(t, *metric.Counter.Value, float64(1))
+				ExtractPrometheusLabelValue("state", labels) == "emitted" &&
+				metric.GetCounter().GetValue() >= 1 {
 				found = true
 			}
 		}
-		assert.Equal(t, true, found)
-	} else {
-		t.Errorf("metric not available")
+		return found
 	}
+
+	families := WaitForPrometheusMetric(t, "keda_cloudeventsource_events_emitted_total", familyValidator)
+	metric := families["keda_cloudeventsource_events_emitted_total"]
+
+	assert.True(t, familyValidator(metric))
 }
 
 func testCloudEventEmittedError(t *testing.T, data templateData) {
 	t.Log("--- testing cloudevent emitted error ---")
 
 	KubectlDeleteWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
-	KubectlDeleteWithTemplate(t, data, "cloudEventSourceTemplate", cloudEventSourceTemplate)
+
 	KubectlApplyWithTemplate(t, data, "wrongCloudEventSourceTemplate", wrongCloudEventSourceTemplate)
-	time.Sleep(1 * time.Second)
 	KubectlApplyWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
-	time.Sleep(5 * time.Second)
 
-	family := fetchAndParsePrometheusMetrics(t, fmt.Sprintf("curl --insecure %s", kedaOperatorPrometheusURL))
+	defer KubectlDeleteWithTemplate(t, data, "wrongCloudEventSourceTemplate", wrongCloudEventSourceTemplate)
 
-	if val, ok := family["keda_cloudeventsource_events_emitted_total"]; ok {
-		var found bool
-		metrics := val.GetMetric()
-		for _, metric := range metrics {
+	familyValidator := func(family *prommodel.MetricFamily) bool {
+		labels := family.GetMetric()
+		found := false
+		for _, metric := range labels {
 			labels := metric.GetLabel()
 			if len(labels) >= 4 &&
 				ExtractPrometheusLabelValue("cloudeventsource", labels) == "prometheus-metrics-test-ce-w" &&
 				ExtractPrometheusLabelValue("eventsink", labels) == "http" &&
 				ExtractPrometheusLabelValue("namespace", labels) == "prometheus-metrics-test-ns" &&
-				ExtractPrometheusLabelValue("state", labels) == "failed" {
-				assert.GreaterOrEqual(t, *metric.Counter.Value, float64(5))
+				ExtractPrometheusLabelValue("state", labels) == "failed" &&
+				metric.GetCounter().GetValue() >= 5 {
 				found = true
 			}
 		}
-		assert.Equal(t, true, found)
-	} else {
-		t.Errorf("metric not available")
+		return found
 	}
 
-	KubectlDeleteWithTemplate(t, data, "wrongCloudEventSourceTemplate", wrongCloudEventSourceTemplate)
-	KubectlApplyWithTemplate(t, data, "cloudEventSourceTemplate", cloudEventSourceTemplate)
+	families := WaitForPrometheusMetric(t, "keda_cloudeventsource_events_emitted_total", familyValidator)
+	metric := families["keda_cloudeventsource_events_emitted_total"]
+
+	assert.True(t, familyValidator(metric))
 }