test: add test for splitted routing

Signed-off-by: Dominik Rosiek <[email protected]>

tests/integration/features.go

@@ -541,7 +541,7 @@ func GetAdditionalPartiallyLogsFeature() features.Feature {
 			tickDuration,
 			false,
 		)).
-		Assess("logs from log generator daemonset present", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
+		Assess("logs from log generator daemonset not present", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
 			0,
 			map[string]string{
 				"namespace":      internal.LogsGeneratorName,
@@ -553,7 +553,7 @@ func GetAdditionalPartiallyLogsFeature() features.Feature {
 			true,
 		)).
 		Assess("expected container log metadata is present for log generator deployment", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
-			0,
+			logsGeneratorCount,
 			map[string]string{
 				"cluster": internal.ClusterName,
 				// TODO: uncomment this after v4 release
@@ -585,8 +585,72 @@ func GetAdditionalPartiallyLogsFeature() features.Feature {
 			tickDuration,
 			true,
 		)).
-		Assess("expected container log metadata is present for log generator daemonset", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
+		Assess("logs from node systemd not present", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
+			0,
+			map[string]string{
+				"cluster":         "kubernetes",
+				"_sourceName":     internal.NotUndefinedRegex,
+				"_sourceCategory": "kubernetes/system",
+				"_sourceHost":     internal.NodeNameRegex,
+			},
+			waitDuration,
+			tickDuration,
+			true,
+		)).
+		Assess("logs from kubelet not present", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
+			0,
+			map[string]string{
+				"cluster":         "kubernetes",
+				"_sourceName":     "k8s_kubelet",
+				"_sourceCategory": "kubernetes/kubelet",
+				"_sourceHost":     internal.NodeNameRegex,
+			},
+			waitDuration,
+			tickDuration,
+			true,
+		)).
+		Feature()
+}
+
+func GetPartialLogsFeature() features.Feature {
+	return features.New("partial logs").
+		Setup(stepfuncs.GenerateLogs(
+			stepfuncs.LogsGeneratorDeployment,
+			logsGeneratorCount,
+			internal.LogsGeneratorName,
+			internal.LogsGeneratorNamespace,
+			internal.LogsGeneratorImage,
+		)).
+		Setup(stepfuncs.GenerateLogs(
+			stepfuncs.LogsGeneratorDaemonSet,
+			logsGeneratorCount,
+			internal.LogsGeneratorName,
+			internal.LogsGeneratorNamespace,
+			internal.LogsGeneratorImage,
+		)).
+		Assess("logs from log generator deployment are not present", stepfuncs.WaitUntilExpectedExactLogsPresent(
 			0,
+			map[string]string{
+				"namespace":      internal.LogsGeneratorName,
+				"pod_labels_app": internal.LogsGeneratorName,
+				"deployment":     internal.LogsGeneratorName,
+			},
+			waitDuration,
+			tickDuration,
+			true,
+		)).
+		Assess("logs from log generator daemonset present", stepfuncs.WaitUntilExpectedLogsPresent(
+			logsGeneratorCount,
+			map[string]string{
+				"namespace":      internal.LogsGeneratorName,
+				"pod_labels_app": internal.LogsGeneratorName,
+				"daemonset":      internal.LogsGeneratorName,
+			},
+			waitDuration,
+			tickDuration,
+		)).
+		Assess("expected container log metadata is present for log generator daemonset", stepfuncs.WaitUntilExpectedLogsPresent(
+			logsGeneratorCount,
 			map[string]string{
 				// TODO: uncomment this after v4 release
 				// or make it depend on the metadata provider
@@ -615,9 +679,85 @@ func GetAdditionalPartiallyLogsFeature() features.Feature {
 			},
 			waitDuration,
 			tickDuration,
+		)).
+		Assess("logs from node systemd present", stepfuncs.WaitUntilExpectedLogsPresent(
+			10, // we don't really control this, just want to check if the logs show up
+			map[string]string{
+				"cluster":         "kubernetes",
+				"_sourceName":     internal.NotUndefinedRegex,
+				"_sourceCategory": "kubernetes/system",
+				"_sourceHost":     internal.NodeNameRegex,
+			},
+			waitDuration,
+			tickDuration,
+		)).
+		Assess("logs from kubelet present", stepfuncs.WaitUntilExpectedLogsPresent(
+			1, // we don't really control this, just want to check if the logs show up
+			map[string]string{
+				"cluster":         "kubernetes",
+				"_sourceName":     "k8s_kubelet",
+				"_sourceCategory": "kubernetes/kubelet",
+				"_sourceHost":     internal.NodeNameRegex,
+			},
+			waitDuration,
+			tickDuration,
+		)).
+		Assess("logs from log generator deployment present", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
+			logsGeneratorCount,
+			map[string]string{
+				"namespace":      internal.LogsGeneratorName,
+				"pod_labels_app": internal.LogsGeneratorName,
+				"deployment":     internal.LogsGeneratorName,
+			},
+			waitDuration,
+			tickDuration,
+			false,
+		)).
+		Assess("logs from log generator daemonset not present", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
+			0,
+			map[string]string{
+				"namespace":      internal.LogsGeneratorName,
+				"pod_labels_app": internal.LogsGeneratorName,
+				"daemonset":      internal.LogsGeneratorName,
+			},
+			waitDuration,
+			tickDuration,
+			true,
+		)).
+		Assess("expected container log metadata is present for log generator deployment", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
+			logsGeneratorCount,
+			map[string]string{
+				"cluster": internal.ClusterName,
+				// TODO: uncomment this after v4 release
+				// or make it depend on the metadata provider
+				// "_collector":     internal.ClusterName,
+				"namespace":      internal.LogsGeneratorName,
+				"pod_labels_app": internal.LogsGeneratorName,
+				"container":      internal.LogsGeneratorName,
+				"deployment":     internal.LogsGeneratorName,
+				"pod":            fmt.Sprintf("%s%s", internal.LogsGeneratorName, internal.PodDeploymentSuffixRegex),
+				"host":           internal.NodeNameRegex,
+				"node":           internal.NodeNameRegex,
+				"_sourceName": fmt.Sprintf(
+					"%s\\.%s%s\\.%s",
+					internal.LogsGeneratorNamespace,
+					internal.LogsGeneratorName,
+					internal.PodDeploymentSuffixRegex,
+					internal.LogsGeneratorName,
+				),
+				"_sourceCategory": fmt.Sprintf(
+					"%s/%s/%s", // dashes instead of hyphens due to sourceCategoryReplaceDash
+					internal.ClusterName,
+					strings.ReplaceAll(internal.LogsGeneratorNamespace, "-", "/"),
+					strings.ReplaceAll(internal.LogsGeneratorName, "-", "/"), // this is the pod name prefix, in this case the deployment name
+				),
+				"_sourceHost": internal.EmptyRegex,
+			},
+			waitDuration,
+			tickDuration,
 			true,
 		)).
-		Assess("logs from node systemd present", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
+		Assess("logs from node systemd not present", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
 			0,
 			map[string]string{
 				"cluster":         "kubernetes",
@@ -629,7 +769,7 @@ func GetAdditionalPartiallyLogsFeature() features.Feature {
 			tickDuration,
 			true,
 		)).
-		Assess("logs from kubelet present", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
+		Assess("logs from kubelet not present", stepfuncs.WaitUntilExpectedAdditionalLogsPresent(
 			0,
 			map[string]string{
 				"cluster":         "kubernetes",
@@ -641,6 +781,21 @@ func GetAdditionalPartiallyLogsFeature() features.Feature {
 			tickDuration,
 			true,
 		)).
+		Teardown(
+			func(ctx context.Context, t *testing.T, envConf *envconf.Config) context.Context {
+				opts := *ctxopts.KubectlOptions(ctx)
+				opts.Namespace = internal.LogsGeneratorNamespace
+				terrak8s.RunKubectl(t, &opts, "delete", "deployment", internal.LogsGeneratorName)
+				return ctx
+			}).
+		Teardown(
+			func(ctx context.Context, t *testing.T, envConf *envconf.Config) context.Context {
+				opts := *ctxopts.KubectlOptions(ctx)
+				opts.Namespace = internal.LogsGeneratorNamespace
+				terrak8s.RunKubectl(t, &opts, "delete", "daemonset", internal.LogsGeneratorName)
+				return ctx
+			}).
+		Teardown(stepfuncs.KubectlDeleteNamespaceOpt(internal.LogsGeneratorNamespace)).
 		Feature()
 }
 

tests/integration/helm_ot_routing_additional_partially_test.go

@@ -7,7 +7,7 @@ import (
 	"testing"
 )
 
-func Test_Helm_Routing_Additional_Partially_OT(t *testing.T) {
+func Test_Helm_Routing_Additional_Partially(t *testing.T) {
 
 	installChecks := []featureCheck{
 		CheckOtelcolMetadataLogsInstall,

tests/integration/helm_ot_routing_partial_test.go

@@ -0,0 +1,25 @@
+//go:build allversions
+// +build allversions
+
+package integration
+
+import (
+	"testing"
+)
+
+func Test_Helm_Routing_Partial(t *testing.T) {
+
+	installChecks := []featureCheck{
+		CheckOtelcolMetadataLogsInstall,
+		CheckOtelcolLogsCollectorInstall,
+	}
+
+	featInstall := GetInstallFeature(installChecks)
+
+	featLogs := GetPartialLogsFeature()
+
+	featDeployMock := DeployAdditionalSumologicMock()
+	featDeleteMock := DeleteAdditionalSumologicMock()
+
+	testenv.Test(t, featInstall, featDeployMock, featLogs, featDeleteMock)
+}

tests/integration/internal/stepfuncs/assess_funcs.go

@@ -359,6 +359,22 @@ func WaitUntilExpectedLogsPresent(
 	expectedLogsMetadata map[string]string,
 	waitDuration time.Duration,
 	tickDuration time.Duration,
+) features.Func {
+	return WaitUntilExpectedExactLogsPresent(
+		expectedLogsCount,
+		expectedLogsMetadata,
+		waitDuration,
+		tickDuration,
+		false,
+	)
+}
+
+func WaitUntilExpectedExactLogsPresent(
+	expectedLogsCount uint,
+	expectedLogsMetadata map[string]string,
+	waitDuration time.Duration,
+	tickDuration time.Duration,
+	exactValue bool,
 ) features.Func {
 	return func(ctx context.Context, t *testing.T, envConf *envconf.Config) context.Context {
 		k8s_internal.WaitUntilSumologicMockAvailable(ctx, t, waitDuration, tickDuration)
@@ -372,14 +388,24 @@ func WaitUntilExpectedLogsPresent(
 				log.ErrorS(err, "failed getting log counts from sumologic-mock")
 				return false
 			}
-			if logsCount < expectedLogsCount {
+			if !exactValue && logsCount < expectedLogsCount {
 				log.InfoS(
 					"received logs, less than expected",
 					"received", logsCount,
 					"expected", expectedLogsCount,
 				)
 				return false
 			}
+
+			if exactValue && logsCount != expectedLogsCount {
+				log.InfoS(
+					"received logs, not the same like expected",
+					"received", logsCount,
+					"expected", expectedLogsCount,
+				)
+				return false
+			}
+
 			log.InfoS(
 				"received enough logs",
 				"received", logsCount,

tests/integration/values/values_helm_routing_partial.yaml

@@ -0,0 +1,12 @@
+sumologic:
+  logs:
+    otelcol:
+      extraExporters:
+        sumologic/additional-sumologic-mock:
+          endpoint: http://sumologic-mock.additional-sumologic-mock:3000/receiver
+      routing:
+        fallbackExporters:
+          - sumologic
+        table:
+          - exporter: sumologic/additional-sumologic-mock
+            statement: route() where resource.attributes["deployment"] == "logs-generator"