[chore] configuration switching tests stabilisation (#1737)

* tests stabilisation

* moving internal.ResetLogsSink to proper place

Author: pszkamruk-splunk

functional_tests/configuration_switching/configuration_switching_test.go

@@ -206,6 +206,8 @@ func testIndexSwitch(t *testing.T) {
 			"Sourcetype":           nonDefaultSourcetype,
 		}
 		deployChartsAndApps(t, valuesFileName, replacements)
+		internal.ResetMetricsSink(t, hecMetricsConsumer)
+		internal.ResetLogsSink(t, agentLogsConsumer)
 
 		internal.WaitForLogs(t, 3, agentLogsConsumer)
 		logs := agentLogsConsumer.AllLogs()
@@ -221,6 +223,13 @@ func testIndexSwitch(t *testing.T) {
 }
 
 func testClusterReceiverEnabledOrDisabled(t *testing.T) {
+	testKubeConfig, setKubeConfig := os.LookupEnv("KUBECONFIG")
+	require.True(t, setKubeConfig, "the environment variable KUBECONFIG must be set")
+	config, err := clientcmd.BuildConfigFromFlags("", testKubeConfig)
+	require.NoError(t, err)
+	clientset, err := kubernetes.NewForConfig(config)
+	require.NoError(t, err)
+
 	valuesFileName := "values_cluster_receiver_switching.yaml.tmpl"
 	logsObjectsConsumer := globalSinks.logsObjectsConsumer
 	hostEp := internal.HostEndpoint(t)
@@ -230,25 +239,26 @@ func testClusterReceiverEnabledOrDisabled(t *testing.T) {
 	logsObjectsHecEndpoint := fmt.Sprintf("http://%s:%d/services/collector", hostEp, internal.HECObjectsReceiverPort)
 
 	t.Run("check cluster receiver disabled", func(t *testing.T) {
-		internal.ResetLogsSink(t, logsObjectsConsumer)
 		replacements := map[string]interface{}{
 			"ClusterReceiverEnabled": false,
 			"LogObjectsHecEndpoint":  logsObjectsHecEndpoint,
 		}
 		deployChartsAndApps(t, valuesFileName, replacements)
+		internal.WaitForTerminatingPods(t, clientset, internal.Namespace)
+		internal.ResetLogsSink(t, logsObjectsConsumer)
 		pods := listPodsInNamespace(t, internal.Namespace)
 		assert.Len(t, pods.Items, 1)
 		assert.True(t, strings.HasPrefix(pods.Items[0].Name, "sock-splunk-otel-collector-agent"))
 		internal.CheckNoEventsReceived(t, logsObjectsConsumer)
 	})
 
 	t.Run("check cluster receiver enabled", func(t *testing.T) {
-		internal.ResetLogsSink(t, logsObjectsConsumer)
 		replacements := map[string]interface{}{
 			"ClusterReceiverEnabled": true,
 			"LogObjectsHecEndpoint":  logsObjectsHecEndpoint,
 		}
 		deployChartsAndApps(t, valuesFileName, replacements)
+		internal.WaitForTerminatingPods(t, clientset, internal.Namespace)
 		internal.ResetLogsSink(t, logsObjectsConsumer)
 		pods := listPodsInNamespace(t, internal.Namespace)
 		assert.Len(t, pods.Items, 2)

functional_tests/internal/common.go

@@ -178,3 +178,23 @@ func AnnotateNamespace(t *testing.T, clientset *kubernetes.Clientset, name, key,
 	_, err = clientset.CoreV1().Namespaces().Update(context.TODO(), ns, metav1.UpdateOptions{})
 	require.NoError(t, err)
 }
+
+func WaitForTerminatingPods(t *testing.T, clientset *kubernetes.Clientset, namespace string) {
+	require.Eventually(t, func() bool {
+		pods, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{})
+		require.NoError(t, err)
+
+		terminatingPods := 0
+		for _, pod := range pods.Items {
+			if pod.Status.Phase == v1.PodRunning || pod.Status.Phase == v1.PodPending {
+				continue
+			}
+			// Check if the pod is terminating
+			if pod.DeletionTimestamp != nil {
+				terminatingPods++
+			}
+		}
+
+		return terminatingPods == 0
+	}, 2*time.Minute, 5*time.Second, "there are still terminating pods after 2 minutes")
+}