Fix receiver based routing overwriting data_stream.dataset

Author: carsonip

.chloggen/elasticsearchexporter_dynamic-routing-default.yaml

@@ -20,6 +20,7 @@ subtext:
   Deprecate and make `{logs,metrics,traces}_dynamic_index` config no-op.
   Config validation error on `{logs,metrics,traces}_dynamic_index::enabled` and `{logs,metrics,traces}_index` set at the same time, as users who rely on dynamic index should not set `{logs,metrics,traces}_index`.
   Remove `elasticsearch.index.{prefix,suffix}` handling. Replace it with `elasticsearch.index` handling that uses attribute value as index directly. Users rely on the previously supported `elasticsearch.index.prefix` and `elasticsearch.index.suffix` should migrate to a transform processor that sets `elasticsearch.index`.
+  Fix a bug where receiver-based routing overwrites data_stream.dataset.
 
 # If your change doesn't affect end users or the exported elements of any package,
 # you should instead start your pull request title with [chore] or use the "Skip Changelog" label.

exporter/elasticsearchexporter/data_stream_router.go

@@ -166,7 +166,7 @@ func routeRecord(
 		return elasticsearch.Index{Index: esIndex}, nil
 	}
 
-	dataset, _ := getFromAttributes(elasticsearch.DataStreamDataset, defaultDataStreamDataset, recordAttr, scopeAttr, resourceAttr)
+	dataset, datasetExists := getFromAttributes(elasticsearch.DataStreamDataset, defaultDataStreamDataset, recordAttr, scopeAttr, resourceAttr)
 	namespace, _ := getFromAttributes(elasticsearch.DataStreamNamespace, defaultDataStreamNamespace, recordAttr, scopeAttr, resourceAttr)
 
 	dsType := defaultDSType
@@ -178,13 +178,16 @@ func routeRecord(
 		}
 	}
 
-	// Receiver-based routing
-	// For example, hostmetricsreceiver (or hostmetricsreceiver.otel in the OTel output mode)
-	// for the scope name
-	// github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/internal/scraper/cpuscraper
-	if submatch := receiverRegex.FindStringSubmatch(scope.Name()); len(submatch) > 0 {
-		receiverName := submatch[1]
-		dataset = receiverName
+	// Only use receiver-based routing if dataset is not specified.
+	if !datasetExists {
+		// Receiver-based routing
+		// For example, hostmetricsreceiver (or hostmetricsreceiver.otel in the OTel output mode)
+		// for the scope name
+		// github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/internal/scraper/cpuscraper
+		if submatch := receiverRegex.FindStringSubmatch(scope.Name()); len(submatch) > 0 {
+			receiverName := submatch[1]
+			dataset = receiverName
+		}
 	}
 
 	// For dataset, the naming convention for datastream is expected to be "logs-[dataset].otel-[namespace]".

exporter/elasticsearchexporter/data_stream_router_test.go

@@ -65,8 +65,9 @@ func createRouteTests(dsType string) []routeTestCase {
 			want:      renderWantRoute(dsType, defaultDataStreamDataset, defaultDataStreamNamespace, MappingOTel),
 		},
 		{
-			name: "otel with elasticsearch.index",
-			mode: MappingOTel,
+			name:      "otel with elasticsearch.index",
+			mode:      MappingOTel,
+			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/should/be/ignored",
 			recordAttrs: map[string]any{
 				"elasticsearch.index": "my-index",
 			},
@@ -75,8 +76,9 @@ func createRouteTests(dsType string) []routeTestCase {
 			},
 		},
 		{
-			name: "otel with data_stream attrs",
-			mode: MappingOTel,
+			name:      "otel with data_stream attrs",
+			mode:      MappingOTel,
+			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/should/be/ignored",
 			recordAttrs: map[string]any{
 				"data_stream.dataset":   "foo",
 				"data_stream.namespace": "bar",