[chore, tests] configuration_switching tests improvements (#1593)

Author: pszkamruk-splunk

functional_tests/configuration_switching/configuration_switching_test.go

@@ -15,13 +15,12 @@ import (
 	"text/template"
 	"time"
 
-	"go.opentelemetry.io/collector/pdata/plog"
-	"go.opentelemetry.io/collector/pdata/pmetric"
-	"gopkg.in/yaml.v3"
-
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"gopkg.in/yaml.v3"
 	"helm.sh/helm/v3/pkg/action"
 	"helm.sh/helm/v3/pkg/kube"
 	corev1 "k8s.io/api/core/v1"
@@ -258,8 +257,15 @@ func testIndexSwitch(t *testing.T) {
 		var indices []string
 		logs := agentLogsConsumer.AllLogs()
 		sourcetypes, indices = getLogsIndexAndSourceType(logs)
-		assert.True(t, len(sourcetypes) > 1) // we are also receiving logs from other kind containers
-		assert.Contains(t, sourcetypes, "kube:container:kindnet-cni")
+		assert.Greater(t, len(sourcetypes), 1) // we are receiving logs from different containers
+		// check sourcetypes have same prefix
+		prefix := "kube:container:"
+		for _, element := range sourcetypes {
+			if !strings.HasPrefix(element, prefix) {
+				t.Errorf("Element does not start with the prefix %q: %s", prefix, element)
+			}
+		}
+		assert.NotContains(t, sourcetypes, nonDefaultSourcetype)
 		assert.True(t, len(indices) == 1)
 		assert.True(t, indices[0] == logsIndex)
 
@@ -278,7 +284,6 @@ func testIndexSwitch(t *testing.T) {
 		internal.ResetLogsSink(t, agentLogsConsumer)
 		internal.ResetMetricsSink(t, hecMetricsConsumer)
 
-		internal.WaitForMetrics(t, 3, hecMetricsConsumer)
 		internal.WaitForLogs(t, 3, agentLogsConsumer)
 		logs = agentLogsConsumer.AllLogs()
 		sourcetypes, indices = getLogsIndexAndSourceType(logs)
@@ -287,6 +292,8 @@ func testIndexSwitch(t *testing.T) {
 		assert.Contains(t, sourcetypes, nonDefaultSourcetype)
 		assert.True(t, len(indices) == 1)
 		assert.True(t, len(sourcetypes) == 1)
+
+		internal.WaitForMetrics(t, 3, hecMetricsConsumer)
 		mIndices = getMetricsIndex(hecMetricsConsumer.AllMetrics())
 		assert.True(t, len(mIndices) == 1)
 		assert.True(t, mIndices[0] == newMetricsIndex)
@@ -307,6 +314,7 @@ func testClusterReceiverEnabledOrDisabled(t *testing.T) {
 	logsObjectsHecEndpoint := fmt.Sprintf("http://%s:%d/services/collector", hostEp, internal.HECObjectsReceiverPort)
 
 	t.Run("check cluster receiver enabled", func(t *testing.T) {
+		internal.ResetLogsSink(t, logsObjectsConsumer)
 		replacements := map[string]interface{}{
 			"ClusterReceiverEnabled": false,
 			"LogObjectsHecEndpoint":  logsObjectsHecEndpoint,