[elasticsearchexporter] Support 'encoding.format' scope attribute for dataset routing (#42844)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Adds an additional way to set dataset based on scope attributes set by
econding extensions. The main use case we are supporting with this right
now is awslogsencodingextension which sets `encoding.format` like
`aws.cloudtrail`, `aws.vpcflow`, `aws.elbaccess`. See
#42899
for more detail.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Unit test added. 

<!--Describe the documentation added.-->
#### Documentation

README updated.

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Co-authored-by: Carson Ip <[email protected]>

Author: tommyers-elastic

.chloggen/esexporter-encoding-format-routing.yaml

@@ -0,0 +1,28 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: exporter/elasticsearch
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Support experimental 'encoding.format' scope attribute for dataset routing.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [42844]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# 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.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]
+

exporter/elasticsearchexporter/README.md

@@ -111,8 +111,11 @@ where `data_stream.type` is `logs` for log records, `metrics` for data points, a
 In a special case with `mapping::mode: bodymap`, `data_stream.type` field (valid values: `logs`, `metrics`) can be dynamically set from attributes.
 The resulting documents will contain the corresponding `data_stream.*` fields, see restrictions applied to [Data Stream Fields](https://www.elastic.co/guide/en/ecs/current/ecs-data_stream.html).
    1. `data_stream.dataset` or `data_stream.namespace` in attributes (precedence: log record / data point / span attribute > scope attribute > resource attribute)
-   2. Otherwise, if scope name matches regex `/receiver/(\w*receiver)`, `data_stream.dataset` will be capture group #1
-   3. Otherwise, `data_stream.dataset` falls back to `generic` and `data_stream.namespace` falls back to `default`. 
+   2. Otherwise, if a scope attribute with the name `encoding.format` exists and contains a string value, `data_stream.dataset` will be set to this value. 
+
+      Note that whilst enabled by default, this behaviour is considered experimental. Some encoding extensions set this field (e.g. [awslogsencodingextension](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/extension/encoding/awslogsencodingextension)), but it is not yet part of Semantic Conventions. There is the potential that the name of this routing field evolves as the [discussion progresses in SemConv](https://github.com/open-telemetry/semantic-conventions/issues/2854). 
+   3. Otherwise, if scope name matches regex `/receiver/(\w*receiver)`, `data_stream.dataset` will be capture group #1
+   4. Otherwise, `data_stream.dataset` falls back to `generic` and `data_stream.namespace` falls back to `default`. 
 
 [^3]: See additional handling in [Document routing exceptions for OTel data mode](#document-routing-exceptions-for-otel-data-mode)
 

exporter/elasticsearchexporter/data_stream_router.go

@@ -26,9 +26,10 @@ var selfTelemetryScopeNames = map[string]bool{
 }
 
 const (
-	maxDataStreamBytes       = 100
-	disallowedNamespaceRunes = "\\/*?\"<>| ,#:"
-	disallowedDatasetRunes   = "-\\/*?\"<>| ,#:"
+	maxDataStreamBytes          = 100
+	disallowedNamespaceRunes    = "\\/*?\"<>| ,#:"
+	disallowedDatasetRunes      = "-\\/*?\"<>| ,#:"
+	encodingFormatAttributeName = "encoding.format"
 )
 
 // Sanitize the datastream fields (dataset, namespace) to apply restrictions
@@ -166,7 +167,7 @@ func routeRecord(
 	// Order:
 	// 1. elasticsearch.index from attributes
 	// 2. read data_stream.* from attributes
-	// 3. receiver-based routing
+	// 3. scope-based routing
 	// 4. use default hardcoded data_stream.*
 	if esIndex, esIndexExists := getFromAttributes(elasticsearch.IndexAttributeName, "", recordAttr, scopeAttr, resourceAttr); esIndexExists {
 		// Advanced users can route documents by setting IndexAttributeName in a processor earlier in the pipeline.
@@ -186,20 +187,10 @@ func routeRecord(
 		}
 	}
 
-	// Only use receiver-based routing if dataset is not specified.
+	// Only use scope-based routing if dataset is not specified.
 	if !datasetExists {
-		if selfTelemetryScopeNames[scope.Name()] {
-			// For collector self-telemetry, use a fixed dataset name
-			dataset = collectorSelfTelemetryDataStreamDataset
-		} else {
-			// 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
-			loc := receiverRegex.FindStringSubmatchIndex(scope.Name())
-			if len(loc) == 4 {
-				dataset = scope.Name()[loc[2]:loc[3]]
-			}
+		if ds, ok := applyScopeRouting(scope); ok {
+			dataset = ds
 		}
 	}
 
@@ -214,3 +205,37 @@ func routeRecord(
 	namespace = sanitizeDataStreamField(namespace, disallowedNamespaceRunes, "")
 	return elasticsearch.NewDataStreamIndex(dsType, dataset, namespace), nil
 }
+
+func applyScopeRouting(scope pcommon.InstrumentationScope) (string, bool) {
+	// Priority:
+	// 1. self-telemetry
+	// 2. encoding-based routing
+	// 3. receiver-based routing
+
+	// For collector self-telemetry, use a fixed dataset name
+	if selfTelemetryScopeNames[scope.Name()] {
+		return collectorSelfTelemetryDataStreamDataset, true
+	}
+
+	// Encoding-based routing
+	// Encoding extensions may set the `encoding.format` scope attribute according to log types.
+	// For example, awslogsencodingextension sets `aws.elbaccess`, `aws.vpcflow`, etc.
+	if format, ok := scope.Attributes().Get(encodingFormatAttributeName); ok {
+		if format.Type() == pcommon.ValueTypeStr {
+			if stringVal := format.Str(); stringVal != "" {
+				return stringVal, true
+			}
+		}
+	}
+
+	// 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
+	loc := receiverRegex.FindStringSubmatchIndex(scope.Name())
+	if len(loc) == 4 {
+		return scope.Name()[loc[2]:loc[3]], true
+	}
+
+	return "", false
+}

exporter/elasticsearchexporter/data_stream_router_test.go

@@ -17,6 +17,7 @@ type routeTestCase struct {
 	name        string
 	mode        MappingMode
 	scopeName   string
+	scopeAttrs  map[string]any
 	recordAttrs map[string]any
 	want        elasticsearch.Index
 }
@@ -40,66 +41,6 @@ func createRouteTests(dsType string) []routeTestCase {
 			mode: MappingOTel,
 			want: renderWantRoute(dsType, defaultDataStreamDataset, defaultDataStreamNamespace, MappingOTel),
 		},
-		{
-			name:      "default with receiver scope name",
-			mode:      MappingNone,
-			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/internal/scraper/cpuscraper",
-			want:      renderWantRoute(dsType, "hostmetricsreceiver", defaultDataStreamNamespace, MappingNone),
-		},
-		{
-			name:      "otel with receiver scope name",
-			mode:      MappingOTel,
-			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/internal/scraper/cpuscraper",
-			want:      renderWantRoute(dsType, "hostmetricsreceiver", defaultDataStreamNamespace, MappingOTel),
-		},
-		{
-			name:      "default with non-receiver scope name",
-			mode:      MappingNone,
-			scopeName: "some_other_scope_name",
-			want:      renderWantRoute(dsType, defaultDataStreamDataset, defaultDataStreamNamespace, MappingNone),
-		},
-		{
-			name:      "otel with non-receiver scope name",
-			mode:      MappingOTel,
-			scopeName: "some_other_scope_name",
-			want:      renderWantRoute(dsType, defaultDataStreamDataset, defaultDataStreamNamespace, MappingOTel),
-		},
-		{
-			name:      "receiver without a receiver name",
-			mode:      MappingOTel,
-			scopeName: "some.scope.name/receiver/receiver/should/be/ignored",
-			want:      renderWantRoute(dsType, defaultDataStreamDataset, defaultDataStreamNamespace, MappingOTel),
-		},
-		{
-			name:      "otel collector self-telemetry for receivers",
-			mode:      MappingOTel,
-			scopeName: "go.opentelemetry.io/collector/receiver/receiverhelper",
-			want:      renderWantRoute(dsType, collectorSelfTelemetryDataStreamDataset, defaultDataStreamNamespace, MappingOTel),
-		},
-		{
-			name:      "otel collector self-telemetry for scrapers",
-			mode:      MappingOTel,
-			scopeName: "go.opentelemetry.io/collector/scraper/scraperhelper",
-			want:      renderWantRoute(dsType, collectorSelfTelemetryDataStreamDataset, defaultDataStreamNamespace, MappingOTel),
-		},
-		{
-			name:      "otel collector self-telemetry for processors",
-			mode:      MappingOTel,
-			scopeName: "go.opentelemetry.io/collector/processor/processorhelper",
-			want:      renderWantRoute(dsType, collectorSelfTelemetryDataStreamDataset, defaultDataStreamNamespace, MappingOTel),
-		},
-		{
-			name:      "otel collector self-telemetry for exporters",
-			mode:      MappingOTel,
-			scopeName: "go.opentelemetry.io/collector/exporter/exporterhelper",
-			want:      renderWantRoute(dsType, collectorSelfTelemetryDataStreamDataset, defaultDataStreamNamespace, MappingOTel),
-		},
-		{
-			name:      "otel collector self-telemetry for service",
-			mode:      MappingOTel,
-			scopeName: "go.opentelemetry.io/collector/service",
-			want:      renderWantRoute(dsType, collectorSelfTelemetryDataStreamDataset, defaultDataStreamNamespace, MappingOTel),
-		},
 		{
 			name:      "otel with elasticsearch.index",
 			mode:      MappingOTel,
@@ -121,6 +62,27 @@ func createRouteTests(dsType string) []routeTestCase {
 			},
 			want: renderWantRoute(dsType, "foo", "bar", MappingOTel),
 		},
+		{
+			name:      "default with scope-based routing for self-telemetry (sanity)",
+			mode:      MappingNone,
+			scopeName: "go.opentelemetry.io/collector/receiver/receiverhelper",
+			want:      renderWantRoute(dsType, collectorSelfTelemetryDataStreamDataset, defaultDataStreamNamespace, MappingNone),
+		},
+		{
+			name:      "otel with scope-based routing for encoding (sanity)",
+			mode:      MappingOTel,
+			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension",
+			scopeAttrs: map[string]any{
+				"encoding.format": "aws.cloudtrail",
+			},
+			want: renderWantRoute(dsType, "aws.cloudtrail", defaultDataStreamNamespace, MappingOTel),
+		},
+		{
+			name:      "otel with scope-based routing for receivers (sanity)",
+			mode:      MappingOTel,
+			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/internal/scraper/cpuscraper",
+			want:      renderWantRoute(dsType, "hostmetricsreceiver", defaultDataStreamNamespace, MappingOTel),
+		},
 	}
 }
 
@@ -132,6 +94,7 @@ func TestRouteLogRecord(t *testing.T) {
 			router := dynamicDocumentRouter{mode: tc.mode}
 			scope := pcommon.NewInstrumentationScope()
 			scope.SetName(tc.scopeName)
+			fillAttributeMap(scope.Attributes(), tc.scopeAttrs)
 
 			recordAttrMap := pcommon.NewMap()
 			fillAttributeMap(recordAttrMap, tc.recordAttrs)
@@ -179,6 +142,7 @@ func TestRouteDataPoint(t *testing.T) {
 			router := dynamicDocumentRouter{mode: tc.mode}
 			scope := pcommon.NewInstrumentationScope()
 			scope.SetName(tc.scopeName)
+			fillAttributeMap(scope.Attributes(), tc.scopeAttrs)
 
 			recordAttrMap := pcommon.NewMap()
 			fillAttributeMap(recordAttrMap, tc.recordAttrs)
@@ -198,6 +162,7 @@ func TestRouteSpan(t *testing.T) {
 			router := dynamicDocumentRouter{mode: tc.mode}
 			scope := pcommon.NewInstrumentationScope()
 			scope.SetName(tc.scopeName)
+			fillAttributeMap(scope.Attributes(), tc.scopeAttrs)
 
 			recordAttrMap := pcommon.NewMap()
 			fillAttributeMap(recordAttrMap, tc.recordAttrs)
@@ -208,3 +173,143 @@ func TestRouteSpan(t *testing.T) {
 		})
 	}
 }
+
+func TestApplyRouting(t *testing.T) {
+	tests := []struct {
+		name        string
+		scopeName   string
+		scopeAttrs  map[string]any
+		wantDataset string
+		wantFound   bool
+	}{
+		{
+			name:        "no routing applied for default scope",
+			scopeName:   "",
+			wantDataset: "",
+			wantFound:   false,
+		},
+		{
+			name:        "no routing applied for non-receiver scope name",
+			scopeName:   "some_other_scope_name",
+			wantDataset: "",
+			wantFound:   false,
+		},
+		{
+			name:        "receiver-based routing with hostmetricsreceiver",
+			scopeName:   "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/internal/scraper/cpuscraper",
+			wantDataset: "hostmetricsreceiver",
+			wantFound:   true,
+		},
+		{
+			name:        "receiver without a receiver name",
+			scopeName:   "some.scope.name/receiver/receiver/should/be/ignored",
+			wantDataset: "",
+			wantFound:   false,
+		},
+		{
+			name:        "otel collector self-telemetry for receivers",
+			scopeName:   "go.opentelemetry.io/collector/receiver/receiverhelper",
+			wantDataset: collectorSelfTelemetryDataStreamDataset,
+			wantFound:   true,
+		},
+		{
+			name:        "otel collector self-telemetry for scrapers",
+			scopeName:   "go.opentelemetry.io/collector/scraper/scraperhelper",
+			wantDataset: collectorSelfTelemetryDataStreamDataset,
+			wantFound:   true,
+		},
+		{
+			name:        "otel collector self-telemetry for processors",
+			scopeName:   "go.opentelemetry.io/collector/processor/processorhelper",
+			wantDataset: collectorSelfTelemetryDataStreamDataset,
+			wantFound:   true,
+		},
+		{
+			name:        "otel collector self-telemetry for exporters",
+			scopeName:   "go.opentelemetry.io/collector/exporter/exporterhelper",
+			wantDataset: collectorSelfTelemetryDataStreamDataset,
+			wantFound:   true,
+		},
+		{
+			name:        "otel collector self-telemetry for service",
+			scopeName:   "go.opentelemetry.io/collector/service",
+			wantDataset: collectorSelfTelemetryDataStreamDataset,
+			wantFound:   true,
+		},
+		{
+			name:      "encoding-based routing with aws.cloudtrail",
+			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension",
+			scopeAttrs: map[string]any{
+				"encoding.format": "aws.cloudtrail",
+			},
+			wantDataset: "aws.cloudtrail",
+			wantFound:   true,
+		},
+		{
+			name:      "encoding-based routing with aws.vpcflow",
+			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension",
+			scopeAttrs: map[string]any{
+				"encoding.format": "aws.vpcflow",
+			},
+			wantDataset: "aws.vpcflow",
+			wantFound:   true,
+		},
+		{
+			name:      "encoding-based routing takes precedence over receiver-based routing",
+			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/internal/scraper/cpuscraper",
+			scopeAttrs: map[string]any{
+				"encoding.format": "aws.vpcflow",
+			},
+			wantDataset: "aws.vpcflow",
+			wantFound:   true,
+		},
+		{
+			name:      "self-telemetry takes precedence over encoding-based routing",
+			scopeName: "go.opentelemetry.io/collector/receiver/receiverhelper",
+			scopeAttrs: map[string]any{
+				"encoding.format": "aws.cloudtrail",
+			},
+			wantDataset: collectorSelfTelemetryDataStreamDataset,
+			wantFound:   true,
+		},
+		{
+			name:      "encoding format that is wrong type is ignored",
+			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension",
+			scopeAttrs: map[string]any{
+				"encoding.format": true,
+			},
+			wantDataset: "",
+			wantFound:   false,
+		},
+		{
+			name:      "non-encoding scope attributes are ignored",
+			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension",
+			scopeAttrs: map[string]any{
+				"some_other_attr": "should_be_ignored",
+			},
+			wantDataset: "",
+			wantFound:   false,
+		},
+		{
+			name:      "empty encoding.format scope attribute is ignored",
+			scopeName: "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension",
+			scopeAttrs: map[string]any{
+				"encoding.format": "",
+			},
+			wantDataset: "",
+			wantFound:   false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			scope := pcommon.NewInstrumentationScope()
+			scope.SetName(tc.scopeName)
+			fillAttributeMap(scope.Attributes(), tc.scopeAttrs)
+
+			dataset, found := applyScopeRouting(scope)
+			assert.Equal(t, tc.wantDataset, dataset)
+			assert.Equal(t, tc.wantFound, found)
+		})
+	}
+}