From 1129937b331992e5a3924efde04ae2c6926150e0 Mon Sep 17 00:00:00 2001
From: Szilard Parrag <szilard.parrag@axoflow.com>
Date: Wed, 8 Jan 2025 17:07:21 +0100
Subject: [PATCH 1/4] fix(conf gen test): check exporters/receivers

Signed-off-by: Szilard Parrag <szilard.parrag@axoflow.com>
---
 .../otel_col_conf_test_fixtures/complex.yaml  | 174 +++++++++++++++++-
 .../otel_conf_gen/otel_conf_gen_test.go       |  57 +++++-
 2 files changed, 224 insertions(+), 7 deletions(-)

diff --git a/internal/controller/telemetry/otel_conf_gen/otel_col_conf_test_fixtures/complex.yaml b/internal/controller/telemetry/otel_conf_gen/otel_col_conf_test_fixtures/complex.yaml
index c9c50a5..7a1087d 100644
--- a/internal/controller/telemetry/otel_conf_gen/otel_col_conf_test_fixtures/complex.yaml
+++ b/internal/controller/telemetry/otel_conf_gen/otel_col_conf_test_fixtures/complex.yaml
@@ -55,7 +55,57 @@ connectors:
     - condition: "true"
       pipelines:
       - logs/tenant_example-tenant-b_subscription_example-tenant-b-ns_subscription-example-3
-exporters: {}
+exporters:
+  fluentforwardexporter/collector_fluentforward-test-output:
+    endpoint:
+      tcp_addr: fluentd.example-tenant-b-ns.svc.cluster.local:24224
+      validate_tcp_resolution: false
+    retry_on_failure:
+      enabled: true
+      max_elapsed_time: 0
+    sending_queue:
+      enabled: true
+      queue_size: 100
+      storage: file_storage/example-tenant-b
+    tls:
+      insecure: true
+  otlp/collector_otlp-test-output:
+    auth:
+      authenticator: bearertokenauth/collector_otlp-test-output
+    endpoint: receiver-collector.example-tenant-a-ns.svc.cluster.local:4317
+    retry_on_failure:
+      enabled: true
+      max_elapsed_time: 0
+    sending_queue:
+      enabled: true
+      queue_size: 100
+      storage: file_storage/example-tenant-a
+    tls:
+      insecure: true
+  otlp/collector_otlp-test-output-2:
+    endpoint: receiver-collector.example-tenant-a-ns.svc.cluster.local:4317
+    retry_on_failure:
+      enabled: true
+      max_elapsed_time: 0
+    sending_queue:
+      enabled: true
+      queue_size: 100
+      storage: file_storage/example-tenant-a
+    tls:
+      insecure: true
+  otlphttp/collector_loki-test-output:
+    endpoint: loki.example-tenant-a-ns.svc.cluster.local:4317
+    retry_on_failure:
+      enabled: true
+      max_elapsed_time: 0
+    sending_queue:
+      enabled: true
+      queue_size: 100
+      storage: file_storage/example-tenant-a
+    tls:
+      insecure: true
+  prometheus/message_metrics_exporter:
+    endpoint: :9999
 extensions:
   bearertokenauth/collector_otlp-test-output:
     token: testtoken
@@ -175,7 +225,127 @@ processors:
     check_interval: 1s
     limit_percentage: 75
     spike_limit_percentage: 25
-receivers: {}
+receivers:
+  filelog/example-tenant-a:
+    exclude:
+    - /var/log/pods/*/otc-container/*.log
+    include:
+    - /var/log/pods/example-tenant-a_*/*/*.log
+    include_file_name: false
+    include_file_path: true
+    operators:
+    - id: get-format
+      routes:
+      - expr: body matches "^\\{"
+        output: parser-docker
+      - expr: body matches "^[^ Z]+Z"
+        output: parser-containerd
+      type: router
+    - id: parser-containerd
+      output: extract_metadata_from_filepath
+      regex: ^(?P<time>[^ ^Z]+Z) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) ?(?P<log>.*)$
+      timestamp:
+        layout: '%Y-%m-%dT%H:%M:%S.%LZ'
+        parse_from: attributes.time
+      type: regex_parser
+    - id: parser-docker
+      output: extract_metadata_from_filepath
+      timestamp:
+        layout: '%Y-%m-%dT%H:%M:%S.%LZ'
+        parse_from: attributes.time
+      type: json_parser
+    - cache:
+        size: 128
+      id: extract_metadata_from_filepath
+      parse_from: attributes["log.file.path"]
+      regex: ^.*\/(?P<namespace>[^_]+)_(?P<pod_name>[^_]+)_(?P<uid>[a-f0-9-]+)\/(?P<container_name>[^\/]+)\/(?P<restart_count>\d+)\.log$
+      type: regex_parser
+    - from: attributes.log
+      to: body
+      type: move
+    - from: attributes.stream
+      to: attributes["log.iostream"]
+      type: move
+    - from: attributes.container_name
+      to: resource["k8s.container.name"]
+      type: move
+    - from: attributes.namespace
+      to: resource["k8s.namespace.name"]
+      type: move
+    - from: attributes.pod_name
+      to: resource["k8s.pod.name"]
+      type: move
+    - from: attributes.restart_count
+      to: resource["k8s.container.restart_count"]
+      type: move
+    - from: attributes.uid
+      to: resource["k8s.pod.uid"]
+      type: move
+    retry_on_failure:
+      enabled: true
+      max_elapsed_time: 0
+    start_at: end
+    storage: file_storage/example-tenant-a
+  filelog/example-tenant-b:
+    exclude:
+    - /var/log/pods/*/otc-container/*.log
+    include:
+    - /var/log/pods/example-tenant-b_*/*/*.log
+    include_file_name: false
+    include_file_path: true
+    operators:
+    - id: get-format
+      routes:
+      - expr: body matches "^\\{"
+        output: parser-docker
+      - expr: body matches "^[^ Z]+Z"
+        output: parser-containerd
+      type: router
+    - id: parser-containerd
+      output: extract_metadata_from_filepath
+      regex: ^(?P<time>[^ ^Z]+Z) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) ?(?P<log>.*)$
+      timestamp:
+        layout: '%Y-%m-%dT%H:%M:%S.%LZ'
+        parse_from: attributes.time
+      type: regex_parser
+    - id: parser-docker
+      output: extract_metadata_from_filepath
+      timestamp:
+        layout: '%Y-%m-%dT%H:%M:%S.%LZ'
+        parse_from: attributes.time
+      type: json_parser
+    - cache:
+        size: 128
+      id: extract_metadata_from_filepath
+      parse_from: attributes["log.file.path"]
+      regex: ^.*\/(?P<namespace>[^_]+)_(?P<pod_name>[^_]+)_(?P<uid>[a-f0-9-]+)\/(?P<container_name>[^\/]+)\/(?P<restart_count>\d+)\.log$
+      type: regex_parser
+    - from: attributes.log
+      to: body
+      type: move
+    - from: attributes.stream
+      to: attributes["log.iostream"]
+      type: move
+    - from: attributes.container_name
+      to: resource["k8s.container.name"]
+      type: move
+    - from: attributes.namespace
+      to: resource["k8s.namespace.name"]
+      type: move
+    - from: attributes.pod_name
+      to: resource["k8s.pod.name"]
+      type: move
+    - from: attributes.restart_count
+      to: resource["k8s.container.restart_count"]
+      type: move
+    - from: attributes.uid
+      to: resource["k8s.pod.uid"]
+      type: move
+    retry_on_failure:
+      enabled: true
+      max_elapsed_time: 0
+    start_at: end
+    storage: file_storage/example-tenant-b
 service:
   extensions:
   - bearertokenauth/collector_otlp-test-output
diff --git a/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go b/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go
index 6cacd51..3695a4c 100644
--- a/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go
+++ b/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go
@@ -15,6 +15,8 @@
 package otel_conf_gen
 
 import (
+	k8syaml "sigs.k8s.io/yaml"
+
 	"context"
 	_ "embed"
 	"reflect"
@@ -26,7 +28,6 @@ import (
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"sigs.k8s.io/yaml"
 
 	"github.com/kube-logging/telemetry-controller/api/telemetry/v1alpha1"
 	"github.com/kube-logging/telemetry-controller/internal/controller/telemetry/pipeline"
@@ -38,6 +39,36 @@ import (
 //go:embed otel_col_conf_test_fixtures/complex.yaml
 var otelColTargetYAML string
 
+type UniversalMap = map[string]any
+
+func workAroundExportersMarshaling(cfg otelv1beta1.Config) (UniversalMap, error) {
+	actualExporters, err := cfg.Exporters.MarshalJSON()
+	if err != nil {
+		return UniversalMap{}, err
+	}
+
+	var actualUniversalExporters UniversalMap
+	if err := k8syaml.Unmarshal(actualExporters, &actualUniversalExporters); err != nil {
+		return UniversalMap{}, err
+	}
+
+	return actualUniversalExporters, nil
+}
+
+func workAroundReceiversMarshaling(cfg otelv1beta1.Config) (UniversalMap, error) {
+	actualReceivers, err := cfg.Receivers.MarshalJSON()
+	if err != nil {
+		return UniversalMap{}, err
+	}
+
+	var actualUniversalReceivers map[string]any
+	if err := k8syaml.Unmarshal(actualReceivers, &actualUniversalReceivers); err != nil {
+		return UniversalMap{}, err
+	}
+
+	return actualUniversalReceivers, nil
+}
+
 func TestOtelColConfComplex(t *testing.T) {
 	// Required inputs
 	var subscriptions = map[v1alpha1.NamespacedName]v1alpha1.Subscription{
@@ -308,20 +339,36 @@ func TestOtelColConfComplex(t *testing.T) {
 
 	// Config
 
-	// The receiver and exporter entries are not serialized because of tags on the underlying data structure. The tests won't contain them, this is a known issue.
+	// The receiver and exporter entries are not (properly) serialized because of tags on the underlying data structure, this is a known issue.
+	// The following hacks are workarounds around the different tags and their handlings.
+	// Receiver and exporter entries are marshalled, then unmarshalled and added to the actual rendered config.
 	generatedConfig, _ := inputCfg.AssembleConfig(context.TODO())
-	actualYAMLBytes, err := yaml.Marshal(generatedConfig)
+
+	actualUniversalExporters, err := workAroundExportersMarshaling(generatedConfig)
+	if err != nil {
+		t.Fatalf("error: %v", err)
+	}
+
+	actualUniversalReceivers, err := workAroundReceiversMarshaling(generatedConfig)
+	if err != nil {
+		t.Fatalf("error: %v", err)
+	}
+
+	actualYAMLBytes, err := k8syaml.Marshal(generatedConfig)
 	if err != nil {
 		t.Fatalf("error %v", err)
 	}
 
 	var actualUniversalMap map[string]any
-	if err := yaml.Unmarshal(actualYAMLBytes, &actualUniversalMap); err != nil {
+	if err := k8syaml.Unmarshal(actualYAMLBytes, &actualUniversalMap); err != nil {
 		t.Fatalf("error: %v", err)
 	}
 
+	actualUniversalMap["exporters"] = actualUniversalExporters
+	actualUniversalMap["receivers"] = actualUniversalReceivers
+
 	var expectedUniversalMap map[string]any
-	if err := yaml.Unmarshal([]byte(otelColTargetYAML), &expectedUniversalMap); err != nil {
+	if err := k8syaml.Unmarshal([]byte(otelColTargetYAML), &expectedUniversalMap); err != nil {
 		t.Fatalf("error: %v", err)
 	}
 

From b9c13570e67e7f8c9b1b986e64fd8f34cf25c540 Mon Sep 17 00:00:00 2001
From: Szilard Parrag <szilard.parrag@axoflow.com>
Date: Wed, 8 Jan 2025 17:08:01 +0100
Subject: [PATCH 2/4] fix(conf gen): make sure omitempty works properly for
 countconnector

Signed-off-by: Szilard Parrag <szilard.parrag@axoflow.com>
---
 .../components/connector/count_connector.go   | 44 ++++++++++---------
 1 file changed, 23 insertions(+), 21 deletions(-)

diff --git a/internal/controller/telemetry/pipeline/components/connector/count_connector.go b/internal/controller/telemetry/pipeline/components/connector/count_connector.go
index 7657e35..16f458d 100644
--- a/internal/controller/telemetry/pipeline/components/connector/count_connector.go
+++ b/internal/controller/telemetry/pipeline/components/connector/count_connector.go
@@ -14,13 +14,15 @@
 
 package connector
 
+import "github.com/kube-logging/telemetry-controller/internal/controller/telemetry/utils"
+
 type CountConnectorAttributeConfig struct {
-	Key          string `json:"key,omitempty"`
-	DefaultValue string `json:"default_value,omitempty"`
+	Key          *string `json:"key,omitempty"`
+	DefaultValue *string `json:"default_value,omitempty"`
 }
 
 type CountConnectorMetricInfo struct {
-	Description        string                          `json:"description,omitempty"`
+	Description        *string                         `json:"description,omitempty"`
 	Conditions         []string                        `json:"conditions,omitempty"`
 	Attributes         []CountConnectorAttributeConfig `json:"attributes,omitempty"`
 	ResourceAttributes []CountConnectorAttributeConfig `json:"resource_attributes,omitempty"`
@@ -31,28 +33,28 @@ func GenerateCountConnectors() map[string]any {
 	countConnectors["count/tenant_metrics"] = map[string]any{
 		"logs": map[string]CountConnectorMetricInfo{
 			"telemetry_controller_tenant_log_count": {
-				Description: "The number of logs from each tenant pipeline.",
+				Description: utils.ToPtr("The number of logs from each tenant pipeline."),
 				Attributes: []CountConnectorAttributeConfig{{
-					Key: "tenant",
+					Key: utils.ToPtr("tenant"),
 				}},
 				ResourceAttributes: []CountConnectorAttributeConfig{
 					{
-						Key: "k8s.namespace.name",
+						Key: utils.ToPtr("k8s.namespace.name"),
 					},
 					{
-						Key: "k8s.node.name",
+						Key: utils.ToPtr("k8s.node.name"),
 					},
 					{
-						Key: "k8s.container.name",
+						Key: utils.ToPtr("k8s.container.name"),
 					},
 					{
-						Key: "k8s.pod.name",
+						Key: utils.ToPtr("k8s.pod.name"),
 					},
 					{
-						Key: "k8s.pod.labels.app.kubernetes.io/name",
+						Key: utils.ToPtr("k8s.pod.labels.app.kubernetes.io/name"),
 					},
 					{
-						Key: "k8s.pod.labels.app",
+						Key: utils.ToPtr("k8s.pod.labels.app"),
 					},
 				},
 			},
@@ -62,33 +64,33 @@ func GenerateCountConnectors() map[string]any {
 	countConnectors["count/output_metrics"] = map[string]any{
 		"logs": map[string]CountConnectorMetricInfo{
 			"telemetry_controller_output_log_count": {
-				Description: "The number of logs sent out from each exporter.",
+				Description: utils.ToPtr("The number of logs sent out from each exporter."),
 				Attributes: []CountConnectorAttributeConfig{
 					{
-						Key: "tenant",
+						Key: utils.ToPtr("tenant"),
 					}, {
-						Key: "subscription",
+						Key: utils.ToPtr("subscription"),
 					}, {
-						Key: "exporter",
+						Key: utils.ToPtr("exporter"),
 					}},
 				ResourceAttributes: []CountConnectorAttributeConfig{
 					{
-						Key: "k8s.namespace.name",
+						Key: utils.ToPtr("k8s.namespace.name"),
 					},
 					{
-						Key: "k8s.node.name",
+						Key: utils.ToPtr("k8s.node.name"),
 					},
 					{
-						Key: "k8s.container.name",
+						Key: utils.ToPtr("k8s.container.name"),
 					},
 					{
-						Key: "k8s.pod.name",
+						Key: utils.ToPtr("k8s.pod.name"),
 					},
 					{
-						Key: "k8s.pod.labels.app.kubernetes.io/name",
+						Key: utils.ToPtr("k8s.pod.labels.app.kubernetes.io/name"),
 					},
 					{
-						Key: "k8s.pod.labels.app",
+						Key: utils.ToPtr("k8s.pod.labels.app"),
 					},
 				},
 			},

From 290ce1406b0ccdc8f849e365f18948e89f0bfa12 Mon Sep 17 00:00:00 2001
From: Szilard Parrag <szilard.parrag@axoflow.com>
Date: Thu, 9 Jan 2025 14:58:38 +0100
Subject: [PATCH 3/4] chore(confgen): refactor test execution steps

Co-authored-by: Szilard Parrag <szilard.parrag@axoflow.com>
Co-authored-by: Bence Csati <bence.csati@axoflow.com>
Signed-off-by: Szilard Parrag <szilard.parrag@axoflow.com>
---
 .../otel_conf_gen/otel_conf_gen_test.go       | 158 ++++++++++--------
 1 file changed, 89 insertions(+), 69 deletions(-)

diff --git a/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go b/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go
index 3695a4c..14a3927 100644
--- a/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go
+++ b/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go
@@ -15,10 +15,9 @@
 package otel_conf_gen
 
 import (
-	k8syaml "sigs.k8s.io/yaml"
-
 	"context"
 	_ "embed"
+	"fmt"
 	"reflect"
 	"testing"
 	"time"
@@ -28,6 +27,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8syaml "sigs.k8s.io/yaml"
 
 	"github.com/kube-logging/telemetry-controller/api/telemetry/v1alpha1"
 	"github.com/kube-logging/telemetry-controller/internal/controller/telemetry/pipeline"
@@ -39,36 +39,6 @@ import (
 //go:embed otel_col_conf_test_fixtures/complex.yaml
 var otelColTargetYAML string
 
-type UniversalMap = map[string]any
-
-func workAroundExportersMarshaling(cfg otelv1beta1.Config) (UniversalMap, error) {
-	actualExporters, err := cfg.Exporters.MarshalJSON()
-	if err != nil {
-		return UniversalMap{}, err
-	}
-
-	var actualUniversalExporters UniversalMap
-	if err := k8syaml.Unmarshal(actualExporters, &actualUniversalExporters); err != nil {
-		return UniversalMap{}, err
-	}
-
-	return actualUniversalExporters, nil
-}
-
-func workAroundReceiversMarshaling(cfg otelv1beta1.Config) (UniversalMap, error) {
-	actualReceivers, err := cfg.Receivers.MarshalJSON()
-	if err != nil {
-		return UniversalMap{}, err
-	}
-
-	var actualUniversalReceivers map[string]any
-	if err := k8syaml.Unmarshal(actualReceivers, &actualUniversalReceivers); err != nil {
-		return UniversalMap{}, err
-	}
-
-	return actualUniversalReceivers, nil
-}
-
 func TestOtelColConfComplex(t *testing.T) {
 	// Required inputs
 	var subscriptions = map[v1alpha1.NamespacedName]v1alpha1.Subscription{
@@ -342,51 +312,32 @@ func TestOtelColConfComplex(t *testing.T) {
 	// The receiver and exporter entries are not (properly) serialized because of tags on the underlying data structure, this is a known issue.
 	// The following hacks are workarounds around the different tags and their handlings.
 	// Receiver and exporter entries are marshalled, then unmarshalled and added to the actual rendered config.
-	generatedConfig, _ := inputCfg.AssembleConfig(context.TODO())
-
-	actualUniversalExporters, err := workAroundExportersMarshaling(generatedConfig)
-	if err != nil {
-		t.Fatalf("error: %v", err)
-	}
-
-	actualUniversalReceivers, err := workAroundReceiversMarshaling(generatedConfig)
-	if err != nil {
-		t.Fatalf("error: %v", err)
-	}
-
-	actualYAMLBytes, err := k8syaml.Marshal(generatedConfig)
+	actualUniversalMap, expectedUniversalMap, err := createUniversalMaps(t, inputCfg, otelColTargetYAML)
 	if err != nil {
-		t.Fatalf("error %v", err)
+		t.Fatalf("Error creating universal maps: %v", err)
 	}
 
-	var actualUniversalMap map[string]any
-	if err := k8syaml.Unmarshal(actualYAMLBytes, &actualUniversalMap); err != nil {
-		t.Fatalf("error: %v", err)
-	}
-
-	actualUniversalMap["exporters"] = actualUniversalExporters
-	actualUniversalMap["receivers"] = actualUniversalReceivers
-
-	var expectedUniversalMap map[string]any
-	if err := k8syaml.Unmarshal([]byte(otelColTargetYAML), &expectedUniversalMap); err != nil {
-		t.Fatalf("error: %v", err)
-	}
-
-	// use dyff for YAML comparison
+	// Compare using cmp.Diff for detailed differences
 	if diff := cmp.Diff(expectedUniversalMap, actualUniversalMap); diff != "" {
-		t.Logf("mismatch:\n---%s\n---\n", diff)
+		t.Errorf("config mismatch (-expected +actual):\n%s", diff)
 	}
 
+	// Additional DeepEqual check with full YAML output on failure
 	if !reflect.DeepEqual(actualUniversalMap, expectedUniversalMap) {
+		actualYAML, err := k8syaml.Marshal(actualUniversalMap)
+		if err != nil {
+			t.Fatalf("failed to marshal actual config: %v", err)
+		}
+
 		t.Fatalf(`yaml mismatch:
-expected=
----
-%s
----
-actual=
----
-%s
----`, otelColTargetYAML, string(actualYAMLBytes))
+	expected=
+	---
+	%s
+	---
+	actual=
+	---
+	%s
+	---`, otelColTargetYAML, string(actualYAML))
 	}
 }
 
@@ -755,3 +706,72 @@ func TestOtelColConfigInput_generateNamedPipelines(t *testing.T) {
 		})
 	}
 }
+
+type UniversalMap = map[string]any
+
+// createUniversalMaps creates the actual and expected universal maps for comparison
+func createUniversalMaps(t *testing.T, inputCfg OtelColConfigInput, expectedYAML string) (UniversalMap, UniversalMap, error) {
+	t.Helper()
+
+	generatedConfig, _ := inputCfg.AssembleConfig(context.TODO())
+	actualUniversalMap, err := buildActualConfig(t, generatedConfig)
+	if err != nil {
+		return nil, nil, fmt.Errorf("building actual config: %w", err)
+	}
+
+	var expectedUniversalMap UniversalMap
+	if err := k8syaml.Unmarshal([]byte(expectedYAML), &expectedUniversalMap); err != nil {
+		return nil, nil, fmt.Errorf("unmarshaling expected YAML: %w", err)
+	}
+
+	return actualUniversalMap, expectedUniversalMap, nil
+}
+
+// buildActualConfig handles the special marshaling requirements for the config
+func buildActualConfig(t *testing.T, generatedConfig otelv1beta1.Config) (UniversalMap, error) {
+	t.Helper()
+
+	actualYAMLBytes, err := k8syaml.Marshal(generatedConfig)
+	if err != nil {
+		return nil, fmt.Errorf("marshaling generated config: %w", err)
+	}
+
+	var actualUniversalMap UniversalMap
+	if err := k8syaml.Unmarshal(actualYAMLBytes, &actualUniversalMap); err != nil {
+		return nil, fmt.Errorf("unmarshaling base config: %w", err)
+	}
+
+	// Handle special cases for exporters and receivers due to serialization issues
+	exporters, err := marshalToUniversalMap(t, generatedConfig.Exporters)
+	if err != nil {
+		return nil, fmt.Errorf("handling exporters: %w", err)
+	}
+	actualUniversalMap["exporters"] = exporters
+
+	// Handle receivers
+	receivers, err := marshalToUniversalMap(t, generatedConfig.Receivers)
+	if err != nil {
+		return nil, fmt.Errorf("handling receivers: %w", err)
+	}
+	actualUniversalMap["receivers"] = receivers
+
+	return actualUniversalMap, nil
+}
+
+// marshalToUniversalMap is a generic helper function that handles marshaling
+// of any JSON-marshalable type to UniversalMap
+func marshalToUniversalMap(t *testing.T, anyConfig otelv1beta1.AnyConfig) (UniversalMap, error) {
+	t.Helper()
+
+	data, err := anyConfig.MarshalJSON()
+	if err != nil {
+		return nil, fmt.Errorf("marshaling to JSON: %w", err)
+	}
+
+	var result UniversalMap
+	if err := k8syaml.Unmarshal(data, &result); err != nil {
+		return nil, fmt.Errorf("unmarshaling to UniversalMap: %w", err)
+	}
+
+	return result, nil
+}

From f2ed535bac5cad807ae15ee4b50200c652bec026 Mon Sep 17 00:00:00 2001
From: Szilard Parrag <szilard.parrag@axoflow.com>
Date: Mon, 13 Jan 2025 09:12:36 +0100
Subject: [PATCH 4/4] fix(conf gen test): Outputs should belong to a single
 tenant

The current test case breaks this invariant, that's why it had to be changed.
Once the Output visibility issue is solved we can refactor/rethink this test case as well
since this is getting convoluted.

Signed-off-by: Szilard Parrag <szilard.parrag@axoflow.com>
---
 .../otel_col_conf_test_fixtures/complex.yaml  | 24 +++++++++++++++----
 .../otel_conf_gen/otel_conf_gen_test.go       | 20 +++++++++++++++-
 2 files changed, 39 insertions(+), 5 deletions(-)

diff --git a/internal/controller/telemetry/otel_conf_gen/otel_col_conf_test_fixtures/complex.yaml b/internal/controller/telemetry/otel_conf_gen/otel_col_conf_test_fixtures/complex.yaml
index 7a1087d..20f396a 100644
--- a/internal/controller/telemetry/otel_conf_gen/otel_col_conf_test_fixtures/complex.yaml
+++ b/internal/controller/telemetry/otel_conf_gen/otel_col_conf_test_fixtures/complex.yaml
@@ -43,7 +43,7 @@ connectors:
     - condition: "true"
       pipelines:
       - logs/output_example-tenant-b-ns_subscription-example-3_collector_fluentforward-test-output
-      - logs/output_example-tenant-b-ns_subscription-example-3_collector_otlp-test-output-2
+      - logs/output_example-tenant-b-ns_subscription-example-3_collector_otlp-test-output-3
   routing/tenant_example-tenant-a_subscriptions:
     table:
     - condition: "true"
@@ -93,6 +93,17 @@ exporters:
       storage: file_storage/example-tenant-a
     tls:
       insecure: true
+  otlp/collector_otlp-test-output-3:
+    endpoint: "receiver-collector.example-tenant-b-ns.svc.cluster.local:4317"
+    retry_on_failure:
+      enabled: true
+      max_elapsed_time: 0.0
+    sending_queue:
+      enabled: true
+      queue_size: 100.0
+      storage: "file_storage/example-tenant-b"
+    tls:
+      insecure: true
   otlphttp/collector_loki-test-output:
     endpoint: loki.example-tenant-a-ns.svc.cluster.local:4317
     retry_on_failure:
@@ -136,6 +147,11 @@ processors:
     - action: insert
       key: exporter
       value: otlp/collector_otlp-test-output-2
+  attributes/exporter_name_otlp-test-output-3:
+    actions:
+    - action: "insert"
+      key: "exporter"
+      value: "otlp/collector_otlp-test-output-3"
   attributes/metricattributes:
     actions:
     - action: insert
@@ -389,13 +405,13 @@ service:
       - attributes/exporter_name_fluentforward-test-output
       receivers:
       - routing/subscription_example-tenant-b-ns_subscription-example-3_outputs
-    logs/output_example-tenant-b-ns_subscription-example-3_collector_otlp-test-output-2:
+    logs/output_example-tenant-b-ns_subscription-example-3_collector_otlp-test-output-3:
       exporters:
-      - otlp/collector_otlp-test-output-2
+      - otlp/collector_otlp-test-output-3
       - count/output_metrics
       processors:
       - memory_limiter
-      - attributes/exporter_name_otlp-test-output-2
+      - attributes/exporter_name_otlp-test-output-3
       receivers:
       - routing/subscription_example-tenant-b-ns_subscription-example-3_outputs
     logs/tenant_example-tenant-a:
diff --git a/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go b/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go
index 14a3927..2e602a1 100644
--- a/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go
+++ b/internal/controller/telemetry/otel_conf_gen/otel_conf_gen_test.go
@@ -91,7 +91,7 @@ func TestOtelColConfComplex(t *testing.T) {
 				Condition: "true",
 				Outputs: []v1alpha1.NamespacedName{
 					{
-						Name:      "otlp-test-output-2",
+						Name:      "otlp-test-output-3",
 						Namespace: "collector",
 					},
 					{
@@ -246,6 +246,24 @@ func TestOtelColConfComplex(t *testing.T) {
 						},
 					},
 				},
+				{
+					Output: v1alpha1.Output{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:      "otlp-test-output-3",
+							Namespace: "collector",
+						},
+						Spec: v1alpha1.OutputSpec{
+							OTLPGRPC: &v1alpha1.OTLPGRPC{
+								GRPCClientConfig: v1alpha1.GRPCClientConfig{
+									Endpoint: utils.ToPtr("receiver-collector.example-tenant-b-ns.svc.cluster.local:4317"),
+									TLSSetting: &v1alpha1.TLSClientSetting{
+										Insecure: true,
+									},
+								},
+							},
+						},
+					},
+				},
 				{
 					Output: v1alpha1.Output{
 						ObjectMeta: metav1.ObjectMeta{