From 54ff0cb14470ab7f7e31eba5bee3fd23b256d9e8 Mon Sep 17 00:00:00 2001
From: Szilard Parrag <szilard.parrag@axoflow.com>
Date: Fri, 26 Apr 2024 09:08:25 +0200
Subject: [PATCH 1/3] feat(conf gen): create dedicated receiver for each tenant

Signed-off-by: Szilard Parrag <szilard.parrag@axoflow.com>
---
 .../otel_col_conf_test_fixtures/complex.yaml  | 179 ++++++++++++++----
 .../controller/telemetry/otel_conf_gen.go     |  49 +++--
 .../telemetry/otel_conf_gen_test.go           | 135 +++++++++----
 3 files changed, 269 insertions(+), 94 deletions(-)

diff --git a/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml b/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml
index 29d8d4c2..9942273d 100644
--- a/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml
+++ b/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml
@@ -1,5 +1,5 @@
 receivers:
-    filelog/kubernetes:
+    filelog/example-tenant-a:
         exclude:
             - /var/log/pods/*/otc-container/*.log
         include:
@@ -54,25 +54,86 @@ receivers:
             - from: attributes.uid
               to: resource["k8s.pod.uid"]
               type: move
+        retry_on_failure:
+            enabled: true
+            max_elapsed_time: 0
+        start_at: end
+    filelog/example-tenant-b:
+        exclude:
+            - /var/log/pods/*/otc-container/*.log
+        include:
+            - /var/log/pods/*/*/*.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
 exporters:
     logging/debug:
         verbosity: detailed
     loki/collector_loki-test-output:
-        endpoint: loki.example-tenant-ns.svc.cluster.local:4317
+        endpoint: loki.example-tenant-a-ns.svc.cluster.local:4317
         tls:
             insecure: true
     otlp/collector_otlp-test-output:
-        endpoint: receiver-collector.example-tenant-ns.svc.cluster.local:4317
+        endpoint: receiver-collector.example-tenant-a-ns.svc.cluster.local:4317
         tls:
             insecure: true
     otlp/collector_otlp-test-output-2:
-        endpoint: receiver-collector.example-tenant-ns.svc.cluster.local:4317
+        endpoint: receiver-collector.example-tenant-a-ns.svc.cluster.local:4317
         tls:
             insecure: true
     prometheus/message_metrics_exporter:
-      endpoint: 0.0.0.0:9999
-
+        endpoint: 0.0.0.0:9999
 processors:
     attributes/exporter_name_loki-test-output:
         actions:
@@ -107,11 +168,21 @@ processors:
             - action: insert
               key: subscription
               value: subscription-example-2
-    attributes/tenant_example-tenant:
+    attributes/subscription_subscription-example-3:
+        actions:
+            - action: insert
+              key: subscription
+              value: subscription-example-3
+    attributes/tenant_example-tenant-a:
         actions:
             - action: insert
               key: tenant
-              value: example-tenant
+              value: example-tenant-a
+    attributes/tenant_example-tenant-b:
+        actions:
+            - action: insert
+              key: tenant
+              value: example-tenant-b
     deltatocumulative: {}
     k8sattributes:
         auth_type: serviceAccount
@@ -134,15 +205,15 @@ processors:
                   from: resource_attribute
                 - name: k8s.pod.name
                   from: resource_attribute
+    memory_limiter:
+        check_interval: "1s"
+        limit_percentage: 75
+        spike_limit_percentage: 25
     resource/loki_exporter_loki-test-output:
         attributes:
             - action: insert
               key: loki.resource.labels
               value: k8s.pod.name, k8s.namespace.name
-    memory_limiter:
-        check_interval: "1s"
-        limit_percentage: 75
-        spike_limit_percentage: 25
 connectors:
     count/output_metrics:
         logs:
@@ -170,54 +241,82 @@ connectors:
                     - key: k8s.container.name
                     - key: k8s.pod.name
                     - key: k8s.pod.labels.app.kubernetes.io/name
-    routing/subscription_example-tenant-ns_subscription-example-1_outputs:
+    routing/subscription_example-tenant-a-ns_subscription-example-1_outputs:
         table:
             - statement: route()
-              pipelines: [logs/output_example-tenant-ns_subscription-example-1_collector_loki-test-output, logs/output_example-tenant-ns_subscription-example-1_collector_otlp-test-output]
-    routing/subscription_example-tenant-ns_subscription-example-2_outputs:
+              pipelines: [logs/output_example-tenant-a-ns_subscription-example-1_collector_loki-test-output, logs/output_example-tenant-a-ns_subscription-example-1_collector_otlp-test-output]
+    routing/subscription_example-tenant-a-ns_subscription-example-2_outputs:
         table:
             - statement: route()
-              pipelines: [logs/output_example-tenant-ns_subscription-example-2_collector_otlp-test-output-2]
-    routing/tenant_example-tenant_subscriptions:
+              pipelines: [logs/output_example-tenant-a-ns_subscription-example-2_collector_otlp-test-output-2]
+    routing/subscription_example-tenant-b-ns_subscription-example-3_outputs:
         table:
             - statement: route()
-              pipelines: [logs/tenant_example-tenant_subscription_example-tenant-ns_subscription-example-1]
+              pipelines: [logs/output_example-tenant-b-ns_subscription-example-3_collector_otlp-test-output-2]
+    routing/tenant_example-tenant-a_subscriptions:
+        table:
+            - statement: route()
+              pipelines: [logs/tenant_example-tenant-a_subscription_example-tenant-a-ns_subscription-example-1]
             - statement: 'route() '
-              pipelines: [logs/tenant_example-tenant_subscription_example-tenant-ns_subscription-example-2]
-    routing/tenants:
+              pipelines: [logs/tenant_example-tenant-a_subscription_example-tenant-a-ns_subscription-example-2]
+    routing/tenant_example-tenant-b_subscriptions:
+        table:
+            - statement: route()
+              pipelines: [logs/tenant_example-tenant-b_subscription_example-tenant-b-ns_subscription-example-3]
+    routing/tenants_example-tenant-a:
         table:
             - statement: 'route() where '
-              pipelines: [logs/tenant_example-tenant]
+              pipelines: [logs/tenant_example-tenant-a]
+    routing/tenants_example-tenant-b:
+        table:
+            - statement: 'route() where '
+              pipelines: [logs/tenant_example-tenant-b]
 service:
     pipelines:
-        logs/all:
-            receivers: [filelog/kubernetes]
+        logs/all_example-tenant-a:
+            receivers: [filelog/example-tenant-a]
+            processors: [memory_limiter, k8sattributes]
+            exporters: [routing/tenants_example-tenant-a]
+        logs/all_example-tenant-b:
+            receivers: [filelog/example-tenant-b]
             processors: [memory_limiter, k8sattributes]
-            exporters: [routing/tenants]
-        logs/output_example-tenant-ns_subscription-example-1_collector_loki-test-output:
-            receivers: [routing/subscription_example-tenant-ns_subscription-example-1_outputs]
+            exporters: [routing/tenants_example-tenant-b]
+        logs/output_example-tenant-a-ns_subscription-example-1_collector_loki-test-output:
+            receivers: [routing/subscription_example-tenant-a-ns_subscription-example-1_outputs]
             processors: [memory_limiter, attributes/exporter_name_loki-test-output, attributes/loki_exporter_loki-test-output, resource/loki_exporter_loki-test-output]
             exporters: [loki/collector_loki-test-output, count/output_metrics]
-        logs/output_example-tenant-ns_subscription-example-1_collector_otlp-test-output:
-            receivers: [routing/subscription_example-tenant-ns_subscription-example-1_outputs]
+        logs/output_example-tenant-a-ns_subscription-example-1_collector_otlp-test-output:
+            receivers: [routing/subscription_example-tenant-a-ns_subscription-example-1_outputs]
             processors: [memory_limiter, attributes/exporter_name_otlp-test-output]
             exporters: [otlp/collector_otlp-test-output, count/output_metrics]
-        logs/output_example-tenant-ns_subscription-example-2_collector_otlp-test-output-2:
-            receivers: [routing/subscription_example-tenant-ns_subscription-example-2_outputs]
+        logs/output_example-tenant-a-ns_subscription-example-2_collector_otlp-test-output-2:
+            receivers: [routing/subscription_example-tenant-a-ns_subscription-example-2_outputs]
+            processors: [memory_limiter, attributes/exporter_name_otlp-test-output-2]
+            exporters: [otlp/collector_otlp-test-output-2, count/output_metrics]
+        logs/output_example-tenant-b-ns_subscription-example-3_collector_otlp-test-output-2:
+            receivers: [routing/subscription_example-tenant-b-ns_subscription-example-3_outputs]
             processors: [memory_limiter, attributes/exporter_name_otlp-test-output-2]
             exporters: [otlp/collector_otlp-test-output-2, count/output_metrics]
-        logs/tenant_example-tenant:
-            receivers: [routing/tenants]
-            processors: [memory_limiter, attributes/tenant_example-tenant]
-            exporters: [routing/tenant_example-tenant_subscriptions, count/tenant_metrics]
-        logs/tenant_example-tenant_subscription_example-tenant-ns_subscription-example-1:
-            receivers: [routing/tenant_example-tenant_subscriptions]
+        logs/tenant_example-tenant-a:
+            receivers: [routing/tenants_example-tenant-a]
+            processors: [memory_limiter, attributes/tenant_example-tenant-a]
+            exporters: [routing/tenant_example-tenant-a_subscriptions, count/tenant_metrics]
+        logs/tenant_example-tenant-a_subscription_example-tenant-a-ns_subscription-example-1:
+            receivers: [routing/tenant_example-tenant-a_subscriptions]
             processors: [memory_limiter, attributes/subscription_subscription-example-1]
-            exporters: [routing/subscription_example-tenant-ns_subscription-example-1_outputs]
-        logs/tenant_example-tenant_subscription_example-tenant-ns_subscription-example-2:
-            receivers: [routing/tenant_example-tenant_subscriptions]
+            exporters: [routing/subscription_example-tenant-a-ns_subscription-example-1_outputs]
+        logs/tenant_example-tenant-a_subscription_example-tenant-a-ns_subscription-example-2:
+            receivers: [routing/tenant_example-tenant-a_subscriptions]
             processors: [memory_limiter, attributes/subscription_subscription-example-2]
-            exporters: [routing/subscription_example-tenant-ns_subscription-example-2_outputs]
+            exporters: [routing/subscription_example-tenant-a-ns_subscription-example-2_outputs]
+        logs/tenant_example-tenant-b:
+            receivers: [routing/tenants_example-tenant-b]
+            processors: [memory_limiter, attributes/tenant_example-tenant-b]
+            exporters: [routing/tenant_example-tenant-b_subscriptions, count/tenant_metrics]
+        logs/tenant_example-tenant-b_subscription_example-tenant-b-ns_subscription-example-3:
+            receivers: [routing/tenant_example-tenant-b_subscriptions]
+            processors: [memory_limiter, attributes/subscription_subscription-example-3]
+            exporters: [routing/subscription_example-tenant-b-ns_subscription-example-3_outputs]
         metrics/output:
             receivers: [count/output_metrics]
             processors: [memory_limiter, deltatocumulative]
diff --git a/internal/controller/telemetry/otel_conf_gen.go b/internal/controller/telemetry/otel_conf_gen.go
index 9b560948..c62742da 100644
--- a/internal/controller/telemetry/otel_conf_gen.go
+++ b/internal/controller/telemetry/otel_conf_gen.go
@@ -326,14 +326,15 @@ func buildRoutingTableItemForTenant(tenant v1alpha1.Tenant) RoutingConnectorTabl
 	return newItem
 }
 
-func generateRootRoutingConnector(tenants []v1alpha1.Tenant) RoutingConnector {
-	// Generate routing table's first hop that will sort it's input by tenant name
-	defaultRc := newRoutingConnector("routing/tenants")
+func generateRootRoutingConnectors(tenants []v1alpha1.Tenant) []RoutingConnector {
+	routingConnectors := make([]RoutingConnector, 0, len(tenants))
 	for _, tenant := range tenants {
+		defaultRcForTenant := newRoutingConnector(fmt.Sprintf("routing/tenants_%s", tenant.Name))
 		tableItem := buildRoutingTableItemForTenant(tenant)
-		defaultRc.AddRoutingConnectorTableElem(tableItem)
+		defaultRcForTenant.AddRoutingConnectorTableElem(tableItem)
+		routingConnectors = append(routingConnectors, defaultRcForTenant)
 	}
-	return defaultRc
+	return routingConnectors
 }
 
 func buildRoutingTableItemForSubscription(tenantName string, subscription v1alpha1.Subscription, index int) RoutingConnectorTableItem {
@@ -400,8 +401,10 @@ func (cfgInput *OtelColConfigInput) generateConnectors() map[string]any {
 
 	maps.Copy(connectors, countConnectors)
 
-	rootRoutingConnector := generateRootRoutingConnector(cfgInput.Tenants)
-	connectors[rootRoutingConnector.Name] = rootRoutingConnector
+	rootRoutingConnectors := generateRootRoutingConnectors(cfgInput.Tenants)
+	for _, rc := range rootRoutingConnectors {
+		connectors[rc.Name] = rc
+	}
 
 	for _, tenant := range cfgInput.Tenants {
 		rc := cfgInput.generateRoutingConnectorForTenantsSubscriptions(tenant.Name, cfgInput.TenantSubscriptionMap[tenant.Name])
@@ -616,31 +619,32 @@ func generateLokiExporterResourceProcessor() ResourceProcessor {
 
 }
 
-func generateRootPipeline() Pipeline {
-	return generatePipeline([]string{"filelog/kubernetes"}, []string{"k8sattributes"}, []string{"routing/tenants"})
+func generateRootPipeline(tenantName string) Pipeline {
+	receiverName := fmt.Sprintf("filelog/%s", tenantName)
+	exporterName := fmt.Sprintf("routing/tenants_%s", tenantName)
+	return generatePipeline([]string{receiverName}, []string{"k8sattributes"}, []string{exporterName})
 }
 
 func (cfgInput *OtelColConfigInput) generateNamedPipelines() map[string]Pipeline {
 	tenantCountConnectorName := "count/tenant_metrics"
 	outputCountConnectorName := "count/output_metrics"
-
 	var namedPipelines = make(map[string]Pipeline)
-	namedPipelines["logs/all"] = generateRootPipeline()
-
-	metricsPipelines := generateMetricsPipelines()
-
-	maps.Copy(namedPipelines, metricsPipelines)
 
 	tenants := []string{}
 	for tenant := range cfgInput.TenantSubscriptionMap {
+		tenantRootPipeline := fmt.Sprintf("logs/all_%s", tenant)
+		namedPipelines[tenantRootPipeline] = generateRootPipeline(tenant)
 		tenants = append(tenants, tenant)
 	}
 
+	metricsPipelines := generateMetricsPipelines()
+	maps.Copy(namedPipelines, metricsPipelines)
+
 	for _, tenant := range tenants {
 		// Generate a pipeline for the tenant
 		tenantPipelineName := fmt.Sprintf("logs/tenant_%s", tenant)
 		tenantRoutingName := fmt.Sprintf("routing/tenant_%s_subscriptions", tenant)
-		namedPipelines[tenantPipelineName] = generatePipeline([]string{"routing/tenants"}, []string{fmt.Sprintf("attributes/tenant_%s", tenant)}, []string{tenantRoutingName, tenantCountConnectorName})
+		namedPipelines[tenantPipelineName] = generatePipeline([]string{fmt.Sprintf("routing/tenants_%s", tenant)}, []string{fmt.Sprintf("attributes/tenant_%s", tenant)}, []string{tenantRoutingName, tenantCountConnectorName})
 
 		// Generate pipelines for the subscriptions for the tenant
 		for _, subscription := range cfgInput.TenantSubscriptionMap[tenant] {
@@ -838,6 +842,10 @@ func (cfgInput *OtelColConfigInput) generateDefaultKubernetesReceiver() map[stri
 		"include_file_path": true,
 		"include_file_name": false,
 		"operators":         operators,
+		"retry_on_failure": map[string]any{
+			"enabled":          true,
+			"max_elapsed_time": 0,
+		},
 	}
 
 	return k8sReceiver
@@ -854,13 +862,16 @@ func (cfgInput *OtelColConfigInput) ToIntermediateRepresentation(ctx context.Con
 	result.Processors = cfgInput.generateProcessors()
 
 	result.Receivers = make(map[string]any)
-	k8sReceiverName := "filelog/kubernetes" // only one instance for now
-	result.Receivers[k8sReceiverName] = cfgInput.generateDefaultKubernetesReceiver()
+	for tenant := range cfgInput.TenantSubscriptionMap {
+		k8sReceiverName := fmt.Sprintf("filelog/%s", tenant)
+		result.Receivers[k8sReceiverName] = cfgInput.generateDefaultKubernetesReceiver()
+	}
 
 	result.Connectors = cfgInput.generateConnectors()
 	result.Services.Pipelines.NamedPipelines = make(map[string]Pipeline)
 
 	result.Services.Pipelines.NamedPipelines = cfgInput.generateNamedPipelines()
+	result.Services.Telemetry = make(map[string]any)
 
 	if cfgInput.Debug {
 		result.Services.Telemetry = map[string]any{
@@ -880,8 +891,6 @@ func (cfgInput *OtelColConfigInput) ToIntermediateRepresentation(ctx context.Con
 		}
 	}
 
-	result.Services.Telemetry = make(map[string]any)
-
 	return &result
 }
 
diff --git a/internal/controller/telemetry/otel_conf_gen_test.go b/internal/controller/telemetry/otel_conf_gen_test.go
index 4a7f6c8a..244becd3 100644
--- a/internal/controller/telemetry/otel_conf_gen_test.go
+++ b/internal/controller/telemetry/otel_conf_gen_test.go
@@ -35,10 +35,10 @@ var otelColTargetYaml string
 func TestOtelColConfComplex(t *testing.T) {
 	// Required inputs
 	var subscriptions = map[v1alpha1.NamespacedName]v1alpha1.Subscription{
-		{Name: "subscription-example-1", Namespace: "example-tenant-ns"}: {
+		{Name: "subscription-example-1", Namespace: "example-tenant-a-ns"}: {
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      "subscription-example-1",
-				Namespace: "example-tenant-ns",
+				Namespace: "example-tenant-a-ns",
 			},
 			Spec: v1alpha1.SubscriptionSpec{
 				OTTL: "route()",
@@ -54,10 +54,25 @@ func TestOtelColConfComplex(t *testing.T) {
 				},
 			},
 		},
-		{Name: "subscription-example-2", Namespace: "example-tenant-ns"}: {
+		{Name: "subscription-example-2", Namespace: "example-tenant-a-ns"}: {
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      "subscription-example-2",
-				Namespace: "example-tenant-ns",
+				Namespace: "example-tenant-a-ns",
+			},
+			Spec: v1alpha1.SubscriptionSpec{
+				OTTL: "route()",
+				Outputs: []v1alpha1.NamespacedName{
+					{
+						Name:      "otlp-test-output-2",
+						Namespace: "collector",
+					},
+				},
+			},
+		},
+		{Name: "subscription-example-3", Namespace: "example-tenant-b-ns"}: {
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "subscription-example-3",
+				Namespace: "example-tenant-b-ns",
 			},
 			Spec: v1alpha1.SubscriptionSpec{
 				OTTL: "route()",
@@ -75,20 +90,62 @@ func TestOtelColConfComplex(t *testing.T) {
 		Tenants: []v1alpha1.Tenant{
 			{
 				ObjectMeta: metav1.ObjectMeta{
-					Name: "example-tenant",
+					Name: "example-tenant-a",
 				},
 				Spec: v1alpha1.TenantSpec{
 					SubscriptionNamespaceSelectors: []metav1.LabelSelector{
 						{
 							MatchLabels: map[string]string{
-								"nsSelector": "example-tenant",
+								"nsSelector": "example-tenant-a",
 							},
 						},
 					},
 					LogSourceNamespaceSelectors: []metav1.LabelSelector{
 						{
 							MatchLabels: map[string]string{
-								"nsSelector": "example-tenant",
+								"nsSelector": "example-tenant-a",
+							},
+						},
+					},
+				},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "example-tenant-b",
+				},
+				Spec: v1alpha1.TenantSpec{
+					SubscriptionNamespaceSelectors: []metav1.LabelSelector{
+						{
+							MatchLabels: map[string]string{
+								"nsSelector": "example-tenant-b",
+							},
+						},
+					},
+					LogSourceNamespaceSelectors: []metav1.LabelSelector{
+						{
+							MatchLabels: map[string]string{
+								"nsSelector": "example-tenant-b",
+							},
+						},
+					},
+				},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "example-tenant-a",
+				},
+				Spec: v1alpha1.TenantSpec{
+					SubscriptionNamespaceSelectors: []metav1.LabelSelector{
+						{
+							MatchLabels: map[string]string{
+								"nsSelector": "example-tenant-a",
+							},
+						},
+					},
+					LogSourceNamespaceSelectors: []metav1.LabelSelector{
+						{
+							MatchLabels: map[string]string{
+								"nsSelector": "example-tenant-a",
 							},
 						},
 					},
@@ -104,7 +161,7 @@ func TestOtelColConfComplex(t *testing.T) {
 				Spec: v1alpha1.OtelOutputSpec{
 					OTLP: &v1alpha1.OTLP{
 						GRPCClientConfig: v1alpha1.GRPCClientConfig{
-							Endpoint: "receiver-collector.example-tenant-ns.svc.cluster.local:4317",
+							Endpoint: "receiver-collector.example-tenant-a-ns.svc.cluster.local:4317",
 							TLSSetting: v1alpha1.TLSClientSetting{
 								Insecure: true,
 							},
@@ -120,7 +177,7 @@ func TestOtelColConfComplex(t *testing.T) {
 				Spec: v1alpha1.OtelOutputSpec{
 					OTLP: &v1alpha1.OTLP{
 						GRPCClientConfig: v1alpha1.GRPCClientConfig{
-							Endpoint: "receiver-collector.example-tenant-ns.svc.cluster.local:4317",
+							Endpoint: "receiver-collector.example-tenant-a-ns.svc.cluster.local:4317",
 							TLSSetting: v1alpha1.TLSClientSetting{
 								Insecure: true,
 							},
@@ -136,7 +193,7 @@ func TestOtelColConfComplex(t *testing.T) {
 				Spec: v1alpha1.OtelOutputSpec{
 					Loki: &v1alpha1.Loki{
 						HTTPClientConfig: v1alpha1.HTTPClientConfig{
-							Endpoint: "loki.example-tenant-ns.svc.cluster.local:4317",
+							Endpoint: "loki.example-tenant-a-ns.svc.cluster.local:4317",
 							TLSSetting: v1alpha1.TLSClientSetting{
 								Insecure: true,
 							},
@@ -161,9 +218,22 @@ func TestOtelColConfComplex(t *testing.T) {
 	inputCfg.SubscriptionOutputMap = subscriptionOutputMap
 
 	inputCfg.TenantSubscriptionMap = map[string][]v1alpha1.NamespacedName{}
-	tenant := inputCfg.Tenants[0]
+	tenantA := inputCfg.Tenants[0]
+	tenantASubscriptions := make(map[v1alpha1.NamespacedName]v1alpha1.Subscription)
+	for subName, sub := range inputCfg.Subscriptions {
+		if subName.Namespace != "example-tenant-b-ns" {
+			tenantASubscriptions[subName] = sub
+		}
+	}
+	inputCfg.TenantSubscriptionMap[tenantA.Name] = seqs.ToSlice(mapseqs.KeysOf(tenantASubscriptions))
 
-	inputCfg.TenantSubscriptionMap[tenant.Name] = seqs.ToSlice(mapseqs.KeysOf(inputCfg.Subscriptions))
+	tenantB := inputCfg.Tenants[1]
+	inputCfg.TenantSubscriptionMap[tenantB.Name] = []v1alpha1.NamespacedName{
+		{
+			Name:      "subscription-example-3",
+			Namespace: "example-tenant-b-ns",
+		},
+	}
 
 	// IR
 	generatedIR := inputCfg.ToIntermediateRepresentation(ctx)
@@ -199,7 +269,7 @@ func TestOtelColConfComplex(t *testing.T) {
 	}
 
 	if !reflect.DeepEqual(actualUniversalMap, expectedUniversalMap) {
-		t.Logf(`yaml mismatch:
+		t.Fatalf(`yaml mismatch:
 expected=
 ---
 %s
@@ -208,16 +278,6 @@ actual=
 ---
 %s
 ---`, otelColTargetYaml, actualYAML)
-		t.Fatalf(`yaml marshaling failed
-expected=
----
-%v
----,
-actual=
----
-%v
----`,
-			expectedUniversalMap, actualUniversalMap)
 	}
 }
 
@@ -228,7 +288,7 @@ func Test_generateRootRoutingConnector(t *testing.T) {
 	tests := []struct {
 		name string
 		args args
-		want RoutingConnector
+		want []RoutingConnector
 	}{
 		{
 			name: "two_tenants",
@@ -252,16 +312,23 @@ func Test_generateRootRoutingConnector(t *testing.T) {
 					},
 				},
 			},
-			want: RoutingConnector{
-				Name: "routing/tenants",
-				Table: []RoutingConnectorTableItem{
-					{
-						Statement: `route() where IsMatch(attributes["k8s.namespace.name"], "a") or IsMatch(attributes["k8s.namespace.name"], "b")`,
-						Pipelines: []string{"logs/tenant_tenantA"},
+			want: []RoutingConnector{
+				{
+					Name: "routing/tenants_tenantA",
+					Table: []RoutingConnectorTableItem{
+						{
+							Statement: `route() where IsMatch(attributes["k8s.namespace.name"], "a") or IsMatch(attributes["k8s.namespace.name"], "b")`,
+							Pipelines: []string{"logs/tenant_tenantA"},
+						},
 					},
-					{
-						Statement: `route() where IsMatch(attributes["k8s.namespace.name"], "c") or IsMatch(attributes["k8s.namespace.name"], "d")`,
-						Pipelines: []string{"logs/tenant_tenantB"},
+				},
+				{
+					Name: "routing/tenants_tenantB",
+					Table: []RoutingConnectorTableItem{
+						{
+							Statement: `route() where IsMatch(attributes["k8s.namespace.name"], "c") or IsMatch(attributes["k8s.namespace.name"], "d")`,
+							Pipelines: []string{"logs/tenant_tenantB"},
+						},
 					},
 				},
 			},
@@ -269,7 +336,7 @@ func Test_generateRootRoutingConnector(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := generateRootRoutingConnector(tt.args.tenants)
+			got := generateRootRoutingConnectors(tt.args.tenants)
 			assert.Equal(t, got, tt.want)
 		})
 	}

From f49c17560e42d8e3b0168597f3196ec3df94c1d5 Mon Sep 17 00:00:00 2001
From: Szilard Parrag <szilard.parrag@axoflow.com>
Date: Fri, 26 Apr 2024 16:01:03 +0200
Subject: [PATCH 2/3] fix(conf gen): remove extra hop in routing connector

Signed-off-by: Szilard Parrag <szilard.parrag@axoflow.com>
---
 .../otel_col_conf_test_fixtures/complex.yaml  | 28 ++-------
 .../controller/telemetry/otel_conf_gen.go     | 48 +++-----------
 .../telemetry/otel_conf_gen_test.go           | 62 -------------------
 3 files changed, 13 insertions(+), 125 deletions(-)

diff --git a/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml b/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml
index 9942273d..cbb8d63c 100644
--- a/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml
+++ b/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml
@@ -263,24 +263,16 @@ connectors:
         table:
             - statement: route()
               pipelines: [logs/tenant_example-tenant-b_subscription_example-tenant-b-ns_subscription-example-3]
-    routing/tenants_example-tenant-a:
-        table:
-            - statement: 'route() where '
-              pipelines: [logs/tenant_example-tenant-a]
-    routing/tenants_example-tenant-b:
-        table:
-            - statement: 'route() where '
-              pipelines: [logs/tenant_example-tenant-b]
 service:
     pipelines:
-        logs/all_example-tenant-a:
+        logs/tenant_example-tenant-a:
             receivers: [filelog/example-tenant-a]
-            processors: [memory_limiter, k8sattributes]
-            exporters: [routing/tenants_example-tenant-a]
-        logs/all_example-tenant-b:
+            processors: [memory_limiter, k8sattributes, attributes/tenant_example-tenant-a]
+            exporters: [routing/tenant_example-tenant-a_subscriptions, count/tenant_metrics]
+        logs/tenant_example-tenant-b:
             receivers: [filelog/example-tenant-b]
-            processors: [memory_limiter, k8sattributes]
-            exporters: [routing/tenants_example-tenant-b]
+            processors: [memory_limiter, k8sattributes, attributes/tenant_example-tenant-b]
+            exporters: [routing/tenant_example-tenant-b_subscriptions, count/tenant_metrics]
         logs/output_example-tenant-a-ns_subscription-example-1_collector_loki-test-output:
             receivers: [routing/subscription_example-tenant-a-ns_subscription-example-1_outputs]
             processors: [memory_limiter, attributes/exporter_name_loki-test-output, attributes/loki_exporter_loki-test-output, resource/loki_exporter_loki-test-output]
@@ -297,10 +289,6 @@ service:
             receivers: [routing/subscription_example-tenant-b-ns_subscription-example-3_outputs]
             processors: [memory_limiter, attributes/exporter_name_otlp-test-output-2]
             exporters: [otlp/collector_otlp-test-output-2, count/output_metrics]
-        logs/tenant_example-tenant-a:
-            receivers: [routing/tenants_example-tenant-a]
-            processors: [memory_limiter, attributes/tenant_example-tenant-a]
-            exporters: [routing/tenant_example-tenant-a_subscriptions, count/tenant_metrics]
         logs/tenant_example-tenant-a_subscription_example-tenant-a-ns_subscription-example-1:
             receivers: [routing/tenant_example-tenant-a_subscriptions]
             processors: [memory_limiter, attributes/subscription_subscription-example-1]
@@ -309,10 +297,6 @@ service:
             receivers: [routing/tenant_example-tenant-a_subscriptions]
             processors: [memory_limiter, attributes/subscription_subscription-example-2]
             exporters: [routing/subscription_example-tenant-a-ns_subscription-example-2_outputs]
-        logs/tenant_example-tenant-b:
-            receivers: [routing/tenants_example-tenant-b]
-            processors: [memory_limiter, attributes/tenant_example-tenant-b]
-            exporters: [routing/tenant_example-tenant-b_subscriptions, count/tenant_metrics]
         logs/tenant_example-tenant-b_subscription_example-tenant-b-ns_subscription-example-3:
             receivers: [routing/tenant_example-tenant-b_subscriptions]
             processors: [memory_limiter, attributes/subscription_subscription-example-3]
diff --git a/internal/controller/telemetry/otel_conf_gen.go b/internal/controller/telemetry/otel_conf_gen.go
index c62742da..2a6d30ca 100644
--- a/internal/controller/telemetry/otel_conf_gen.go
+++ b/internal/controller/telemetry/otel_conf_gen.go
@@ -309,34 +309,6 @@ func newRoutingConnector(name string) RoutingConnector {
 	return result
 }
 
-func buildRoutingTableItemForTenant(tenant v1alpha1.Tenant) RoutingConnectorTableItem {
-	conditions := make([]string, len(tenant.Status.LogSourceNamespaces))
-
-	for i, namespace := range tenant.Status.LogSourceNamespaces {
-		conditions[i] = fmt.Sprintf(`IsMatch(attributes["k8s.namespace.name"], %q)`, namespace)
-	}
-
-	conditionString := strings.Join(conditions, " or ")
-
-	newItem := RoutingConnectorTableItem{
-		Statement: fmt.Sprintf(`route() where %s`, conditionString),
-		Pipelines: []string{fmt.Sprintf("logs/tenant_%s", tenant.Name)},
-	}
-
-	return newItem
-}
-
-func generateRootRoutingConnectors(tenants []v1alpha1.Tenant) []RoutingConnector {
-	routingConnectors := make([]RoutingConnector, 0, len(tenants))
-	for _, tenant := range tenants {
-		defaultRcForTenant := newRoutingConnector(fmt.Sprintf("routing/tenants_%s", tenant.Name))
-		tableItem := buildRoutingTableItemForTenant(tenant)
-		defaultRcForTenant.AddRoutingConnectorTableElem(tableItem)
-		routingConnectors = append(routingConnectors, defaultRcForTenant)
-	}
-	return routingConnectors
-}
-
 func buildRoutingTableItemForSubscription(tenantName string, subscription v1alpha1.Subscription, index int) RoutingConnectorTableItem {
 
 	pipelineName := fmt.Sprintf("logs/tenant_%s_subscription_%s_%s", tenantName, subscription.Namespace, subscription.Name)
@@ -398,14 +370,8 @@ func (cfgInput *OtelColConfigInput) generateConnectors() map[string]any {
 	var connectors = make(map[string]any)
 
 	countConnectors := generateCountConnectors()
-
 	maps.Copy(connectors, countConnectors)
 
-	rootRoutingConnectors := generateRootRoutingConnectors(cfgInput.Tenants)
-	for _, rc := range rootRoutingConnectors {
-		connectors[rc.Name] = rc
-	}
-
 	for _, tenant := range cfgInput.Tenants {
 		rc := cfgInput.generateRoutingConnectorForTenantsSubscriptions(tenant.Name, cfgInput.TenantSubscriptionMap[tenant.Name])
 		connectors[rc.Name] = rc
@@ -620,19 +586,19 @@ func generateLokiExporterResourceProcessor() ResourceProcessor {
 }
 
 func generateRootPipeline(tenantName string) Pipeline {
+	tenantCountConnectorName := "count/tenant_metrics"
 	receiverName := fmt.Sprintf("filelog/%s", tenantName)
-	exporterName := fmt.Sprintf("routing/tenants_%s", tenantName)
-	return generatePipeline([]string{receiverName}, []string{"k8sattributes"}, []string{exporterName})
+	exporterName := fmt.Sprintf("routing/tenant_%s_subscriptions", tenantName)
+	return generatePipeline([]string{receiverName}, []string{"k8sattributes", fmt.Sprintf("attributes/tenant_%s", tenantName)}, []string{exporterName, tenantCountConnectorName})
 }
 
 func (cfgInput *OtelColConfigInput) generateNamedPipelines() map[string]Pipeline {
-	tenantCountConnectorName := "count/tenant_metrics"
 	outputCountConnectorName := "count/output_metrics"
 	var namedPipelines = make(map[string]Pipeline)
 
 	tenants := []string{}
 	for tenant := range cfgInput.TenantSubscriptionMap {
-		tenantRootPipeline := fmt.Sprintf("logs/all_%s", tenant)
+		tenantRootPipeline := fmt.Sprintf("logs/tenant_%s", tenant)
 		namedPipelines[tenantRootPipeline] = generateRootPipeline(tenant)
 		tenants = append(tenants, tenant)
 	}
@@ -642,13 +608,13 @@ func (cfgInput *OtelColConfigInput) generateNamedPipelines() map[string]Pipeline
 
 	for _, tenant := range tenants {
 		// Generate a pipeline for the tenant
-		tenantPipelineName := fmt.Sprintf("logs/tenant_%s", tenant)
+		tenantRootPipeline := fmt.Sprintf("logs/tenant_%s", tenant)
 		tenantRoutingName := fmt.Sprintf("routing/tenant_%s_subscriptions", tenant)
-		namedPipelines[tenantPipelineName] = generatePipeline([]string{fmt.Sprintf("routing/tenants_%s", tenant)}, []string{fmt.Sprintf("attributes/tenant_%s", tenant)}, []string{tenantRoutingName, tenantCountConnectorName})
+		namedPipelines[tenantRootPipeline] = generateRootPipeline(tenant)
 
 		// Generate pipelines for the subscriptions for the tenant
 		for _, subscription := range cfgInput.TenantSubscriptionMap[tenant] {
-			tenantSubscriptionPipelineName := fmt.Sprintf("%s_subscription_%s_%s", tenantPipelineName, subscription.Namespace, subscription.Name)
+			tenantSubscriptionPipelineName := fmt.Sprintf("%s_subscription_%s_%s", tenantRootPipeline, subscription.Namespace, subscription.Name)
 
 			namedPipelines[tenantSubscriptionPipelineName] = generatePipeline([]string{tenantRoutingName}, []string{fmt.Sprintf("attributes/subscription_%s", subscription.Name)}, []string{fmt.Sprintf("routing/subscription_%s_%s_outputs", subscription.Namespace, subscription.Name)})
 
diff --git a/internal/controller/telemetry/otel_conf_gen_test.go b/internal/controller/telemetry/otel_conf_gen_test.go
index 244becd3..232e0f61 100644
--- a/internal/controller/telemetry/otel_conf_gen_test.go
+++ b/internal/controller/telemetry/otel_conf_gen_test.go
@@ -280,68 +280,6 @@ actual=
 ---`, otelColTargetYaml, actualYAML)
 	}
 }
-
-func Test_generateRootRoutingConnector(t *testing.T) {
-	type args struct {
-		tenants []v1alpha1.Tenant
-	}
-	tests := []struct {
-		name string
-		args args
-		want []RoutingConnector
-	}{
-		{
-			name: "two_tenants",
-			args: args{
-				tenants: []v1alpha1.Tenant{
-					{
-						ObjectMeta: metav1.ObjectMeta{
-							Name: "tenantA",
-						},
-						Status: v1alpha1.TenantStatus{
-							LogSourceNamespaces: []string{"a", "b"},
-						},
-					},
-					{
-						ObjectMeta: metav1.ObjectMeta{
-							Name: "tenantB",
-						},
-						Status: v1alpha1.TenantStatus{
-							LogSourceNamespaces: []string{"c", "d"},
-						},
-					},
-				},
-			},
-			want: []RoutingConnector{
-				{
-					Name: "routing/tenants_tenantA",
-					Table: []RoutingConnectorTableItem{
-						{
-							Statement: `route() where IsMatch(attributes["k8s.namespace.name"], "a") or IsMatch(attributes["k8s.namespace.name"], "b")`,
-							Pipelines: []string{"logs/tenant_tenantA"},
-						},
-					},
-				},
-				{
-					Name: "routing/tenants_tenantB",
-					Table: []RoutingConnectorTableItem{
-						{
-							Statement: `route() where IsMatch(attributes["k8s.namespace.name"], "c") or IsMatch(attributes["k8s.namespace.name"], "d")`,
-							Pipelines: []string{"logs/tenant_tenantB"},
-						},
-					},
-				},
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got := generateRootRoutingConnectors(tt.args.tenants)
-			assert.Equal(t, got, tt.want)
-		})
-	}
-}
-
 func TestOtelColConfigInput_generateRoutingConnectorForTenantsSubscription(t *testing.T) {
 	type fields struct {
 		Tenants               []v1alpha1.Tenant

From 953a31a4fcfc6969ab2aeecccd52980d7b31d9da Mon Sep 17 00:00:00 2001
From: Szilard Parrag <szilard.parrag@axoflow.com>
Date: Tue, 30 Apr 2024 12:11:49 +0200
Subject: [PATCH 3/3] feat(conf gen): only watch namespaces selected by
 tenant's logSourcenamespaceselector field

Signed-off-by: Szilard Parrag <szilard.parrag@axoflow.com>
---
 .../otel_col_conf_test_fixtures/complex.yaml  |  5 +++--
 .../controller/telemetry/otel_conf_gen.go     | 21 ++++++++++++++-----
 .../telemetry/otel_conf_gen_test.go           | 10 +++++++++
 3 files changed, 29 insertions(+), 7 deletions(-)

diff --git a/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml b/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml
index cbb8d63c..c7e6c8ed 100644
--- a/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml
+++ b/internal/controller/telemetry/otel_col_conf_test_fixtures/complex.yaml
@@ -3,7 +3,7 @@ receivers:
         exclude:
             - /var/log/pods/*/otc-container/*.log
         include:
-            - /var/log/pods/*/*/*.log
+            - /var/log/pods/example-tenant-a_*/*/*.log
         include_file_name: false
         include_file_path: true
         operators:
@@ -62,7 +62,7 @@ receivers:
         exclude:
             - /var/log/pods/*/otc-container/*.log
         include:
-            - /var/log/pods/*/*/*.log
+            - /var/log/pods/example-tenant-b_*/*/*.log
         include_file_name: false
         include_file_path: true
         operators:
@@ -184,6 +184,7 @@ processors:
               key: tenant
               value: example-tenant-b
     deltatocumulative: {}
+                
     k8sattributes:
         auth_type: serviceAccount
         extract:
diff --git a/internal/controller/telemetry/otel_conf_gen.go b/internal/controller/telemetry/otel_conf_gen.go
index 2a6d30ca..48cc75a7 100644
--- a/internal/controller/telemetry/otel_conf_gen.go
+++ b/internal/controller/telemetry/otel_conf_gen.go
@@ -718,7 +718,7 @@ func (cfgInput *OtelColConfigInput) generateDefaultKubernetesProcessor() map[str
 	return k8sProcessor
 }
 
-func (cfgInput *OtelColConfigInput) generateDefaultKubernetesReceiver() map[string]any {
+func (cfgInput *OtelColConfigInput) generateDefaultKubernetesReceiver(namespaces []string) map[string]any {
 
 	// TODO: fix parser-crio
 	operators := []map[string]any{
@@ -801,8 +801,14 @@ func (cfgInput *OtelColConfigInput) generateDefaultKubernetesReceiver() map[stri
 		},
 	}
 
+	includeList := make([]string, 0, len(namespaces))
+	for _, ns := range namespaces {
+		include := fmt.Sprintf("/var/log/pods/%s_*/*/*.log", ns)
+		includeList = append(includeList, include)
+	}
+
 	k8sReceiver := map[string]any{
-		"include":           []string{"/var/log/pods/*/*/*.log"},
+		"include":           includeList,
 		"exclude":           []string{"/var/log/pods/*/otc-container/*.log"},
 		"start_at":          "end",
 		"include_file_path": true,
@@ -828,9 +834,14 @@ func (cfgInput *OtelColConfigInput) ToIntermediateRepresentation(ctx context.Con
 	result.Processors = cfgInput.generateProcessors()
 
 	result.Receivers = make(map[string]any)
-	for tenant := range cfgInput.TenantSubscriptionMap {
-		k8sReceiverName := fmt.Sprintf("filelog/%s", tenant)
-		result.Receivers[k8sReceiverName] = cfgInput.generateDefaultKubernetesReceiver()
+	for tenantName := range cfgInput.TenantSubscriptionMap {
+		if tenantIdx := slices.IndexFunc(cfgInput.Tenants, func(t v1alpha1.Tenant) bool {
+			return tenantName == t.Name
+		}); tenantIdx != -1 {
+			k8sReceiverName := fmt.Sprintf("filelog/%s", tenantName)
+			namespaces := cfgInput.Tenants[tenantIdx].Status.LogSourceNamespaces
+			result.Receivers[k8sReceiverName] = cfgInput.generateDefaultKubernetesReceiver(namespaces)
+		}
 	}
 
 	result.Connectors = cfgInput.generateConnectors()
diff --git a/internal/controller/telemetry/otel_conf_gen_test.go b/internal/controller/telemetry/otel_conf_gen_test.go
index 232e0f61..54651e58 100644
--- a/internal/controller/telemetry/otel_conf_gen_test.go
+++ b/internal/controller/telemetry/otel_conf_gen_test.go
@@ -108,6 +108,11 @@ func TestOtelColConfComplex(t *testing.T) {
 						},
 					},
 				},
+				Status: v1alpha1.TenantStatus{
+					LogSourceNamespaces: []string{
+						"example-tenant-a",
+					},
+				},
 			},
 			{
 				ObjectMeta: metav1.ObjectMeta{
@@ -129,6 +134,11 @@ func TestOtelColConfComplex(t *testing.T) {
 						},
 					},
 				},
+				Status: v1alpha1.TenantStatus{
+					LogSourceNamespaces: []string{
+						"example-tenant-b",
+					},
+				},
 			},
 			{
 				ObjectMeta: metav1.ObjectMeta{