[elasticsearchexporter] Declare MutatesData: false

.chloggen/elasticsearchexporter_mutates-data-false.yaml

@@ -0,0 +1,27 @@
+# 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: elasticsearchexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Declare MutatesData: false"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37234]
+
+# (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: When multiple exporters are used, the collector doesn't need to clone the incoming data anymore
+
+# 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/data_stream_router.go

@@ -46,15 +46,15 @@ func routeWithDefaults(defaultDSType string) func(
 	string,
 	bool,
 	string,
-) string {
+) esIndex {
 	return func(
 		recordAttr pcommon.Map,
 		scopeAttr pcommon.Map,
 		resourceAttr pcommon.Map,
 		fIndex string,
 		otel bool,
 		scopeName string,
-	) string {
+	) esIndex {
 		// Order:
 		// 1. read data_stream.* from attributes
 		// 2. read elasticsearch.index.* from attributes
@@ -67,7 +67,7 @@ func routeWithDefaults(defaultDSType string) func(
 			prefix, prefixExists := getFromAttributes(indexPrefix, "", resourceAttr, scopeAttr, recordAttr)
 			suffix, suffixExists := getFromAttributes(indexSuffix, "", resourceAttr, scopeAttr, recordAttr)
 			if prefixExists || suffixExists {
-				return fmt.Sprintf("%s%s%s", prefix, fIndex, suffix)
+				return esIndex{Index: fmt.Sprintf("%s%s%s", prefix, fIndex, suffix)}
 			}
 		}
 
@@ -89,15 +89,30 @@ func routeWithDefaults(defaultDSType string) func(
 
 		dataset = sanitizeDataStreamField(dataset, disallowedDatasetRunes, datasetSuffix)
 		namespace = sanitizeDataStreamField(namespace, disallowedNamespaceRunes, "")
+		return newDataStream(defaultDSType, dataset, namespace)
+	}
+}
 
-		recordAttr.PutStr(dataStreamDataset, dataset)
-		recordAttr.PutStr(dataStreamNamespace, namespace)
-		recordAttr.PutStr(dataStreamType, defaultDSType)
+type esIndex struct {
+	Index     string
+	Type      string
+	Dataset   string
+	Namespace string
+}
 
-		return fmt.Sprintf("%s-%s-%s", defaultDSType, dataset, namespace)
+func newDataStream(typ, dataset, namespace string) esIndex {
+	return esIndex{
+		Index:     fmt.Sprintf("%s-%s-%s", typ, dataset, namespace),
+		Type:      typ,
+		Dataset:   dataset,
+		Namespace: namespace,
 	}
 }
 
+func (i esIndex) isDataStream() bool {
+	return i.Type != "" && i.Dataset != "" && i.Namespace != ""
+}
+
 var (
 	// routeLogRecord returns the name of the index to send the log record to according to data stream routing related attributes.
 	// This function may mutate record attributes.

exporter/elasticsearchexporter/data_stream_router_test.go

@@ -4,7 +4,6 @@
 package elasticsearchexporter
 
 import (
-	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -15,15 +14,15 @@ type routeTestCase struct {
 	name      string
 	otel      bool
 	scopeName string
-	want      string
+	want      esIndex
 }
 
 func createRouteTests(dsType string) []routeTestCase {
-	renderWantRoute := func(dsType, dsDataset string, otel bool) string {
+	renderWantRoute := func(dsType, dsDataset string, otel bool) esIndex {
 		if otel {
-			return fmt.Sprintf("%s-%s.otel-%s", dsType, dsDataset, defaultDataStreamNamespace)
+			dsDataset += ".otel"
 		}
-		return fmt.Sprintf("%s-%s-%s", dsType, dsDataset, defaultDataStreamNamespace)
+		return newDataStream(dsType, dsDataset, defaultDataStreamNamespace)
 	}
 
 	return []routeTestCase{

exporter/elasticsearchexporter/exporter.go

@@ -162,27 +162,27 @@ func (e *elasticsearchExporter) pushLogRecord(
 	scopeSchemaURL string,
 	bulkIndexerSession bulkIndexerSession,
 ) error {
-	fIndex := e.index
+	fIndex := esIndex{Index: e.index}
 	if e.dynamicIndex {
-		fIndex = routeLogRecord(record.Attributes(), scope.Attributes(), resource.Attributes(), fIndex, e.otel, scope.Name())
+		fIndex = routeLogRecord(record.Attributes(), scope.Attributes(), resource.Attributes(), e.index, e.otel, scope.Name())
 	}
 
 	if e.logstashFormat.Enabled {
-		formattedIndex, err := generateIndexWithLogstashFormat(fIndex, &e.logstashFormat, time.Now())
+		formattedIndex, err := generateIndexWithLogstashFormat(fIndex.Index, &e.logstashFormat, time.Now())
 		if err != nil {
 			return err
 		}
-		fIndex = formattedIndex
+		fIndex = esIndex{Index: formattedIndex}
 	}
 
 	buf := e.bufferPool.NewPooledBuffer()
-	err := e.model.encodeLog(resource, resourceSchemaURL, record, scope, scopeSchemaURL, buf.Buffer)
+	err := e.model.encodeLog(resource, resourceSchemaURL, record, scope, scopeSchemaURL, fIndex, buf.Buffer)
 	if err != nil {
 		buf.Recycle()
 		return fmt.Errorf("failed to encode log event: %w", err)
 	}
 	// not recycling after Add returns an error as we don't know if it's already recycled
-	return bulkIndexerSession.Add(ctx, fIndex, buf, nil)
+	return bulkIndexerSession.Add(ctx, fIndex.Index, buf, nil)
 }
 
 func (e *elasticsearchExporter) pushMetricsData(
@@ -209,7 +209,7 @@ func (e *elasticsearchExporter) pushMetricsData(
 			var validationErrs []error // log instead of returning these so that upstream does not retry
 			scopeMetrics := scopeMetrics.At(j)
 			scope := scopeMetrics.Scope()
-			groupedDataPointsByIndex := make(map[string]map[uint32][]dataPoint)
+			groupedDataPointsByIndex := make(map[esIndex]map[uint32][]dataPoint)
 			for k := 0; k < scopeMetrics.Metrics().Len(); k++ {
 				metric := scopeMetrics.Metrics().At(k)
 
@@ -293,13 +293,13 @@ func (e *elasticsearchExporter) pushMetricsData(
 			for fIndex, groupedDataPoints := range groupedDataPointsByIndex {
 				for _, dataPoints := range groupedDataPoints {
 					buf := e.bufferPool.NewPooledBuffer()
-					dynamicTemplates, err := e.model.encodeMetrics(resource, resourceMetric.SchemaUrl(), scope, scopeMetrics.SchemaUrl(), dataPoints, &validationErrs, buf.Buffer)
+					dynamicTemplates, err := e.model.encodeMetrics(resource, resourceMetric.SchemaUrl(), scope, scopeMetrics.SchemaUrl(), dataPoints, &validationErrs, fIndex, buf.Buffer)
 					if err != nil {
 						buf.Recycle()
 						errs = append(errs, err)
 						continue
 					}
-					if err := session.Add(ctx, fIndex, buf, dynamicTemplates); err != nil {
+					if err := session.Add(ctx, fIndex.Index, buf, dynamicTemplates); err != nil {
 						// not recycling after Add returns an error as we don't know if it's already recycled
 						if cerr := ctx.Err(); cerr != nil {
 							return cerr
@@ -327,18 +327,18 @@ func (e *elasticsearchExporter) getMetricDataPointIndex(
 	resource pcommon.Resource,
 	scope pcommon.InstrumentationScope,
 	dataPoint dataPoint,
-) (string, error) {
-	fIndex := e.index
+) (esIndex, error) {
+	fIndex := esIndex{Index: e.index}
 	if e.dynamicIndex {
-		fIndex = routeDataPoint(dataPoint.Attributes(), scope.Attributes(), resource.Attributes(), fIndex, e.otel, scope.Name())
+		fIndex = routeDataPoint(dataPoint.Attributes(), scope.Attributes(), resource.Attributes(), e.index, e.otel, scope.Name())
 	}
 
 	if e.logstashFormat.Enabled {
-		formattedIndex, err := generateIndexWithLogstashFormat(fIndex, &e.logstashFormat, time.Now())
+		formattedIndex, err := generateIndexWithLogstashFormat(fIndex.Index, &e.logstashFormat, time.Now())
 		if err != nil {
-			return "", err
+			return esIndex{}, err
 		}
-		fIndex = formattedIndex
+		fIndex = esIndex{Index: formattedIndex}
 	}
 	return fIndex, nil
 }
@@ -402,27 +402,27 @@ func (e *elasticsearchExporter) pushTraceRecord(
 	scopeSchemaURL string,
 	bulkIndexerSession bulkIndexerSession,
 ) error {
-	fIndex := e.index
+	fIndex := esIndex{Index: e.index}
 	if e.dynamicIndex {
-		fIndex = routeSpan(span.Attributes(), scope.Attributes(), resource.Attributes(), fIndex, e.otel, span.Name())
+		fIndex = routeSpan(span.Attributes(), scope.Attributes(), resource.Attributes(), e.index, e.otel, span.Name())
 	}
 
 	if e.logstashFormat.Enabled {
-		formattedIndex, err := generateIndexWithLogstashFormat(fIndex, &e.logstashFormat, time.Now())
+		formattedIndex, err := generateIndexWithLogstashFormat(fIndex.Index, &e.logstashFormat, time.Now())
 		if err != nil {
 			return err
 		}
-		fIndex = formattedIndex
+		fIndex = esIndex{Index: formattedIndex}
 	}
 
 	buf := e.bufferPool.NewPooledBuffer()
-	err := e.model.encodeSpan(resource, resourceSchemaURL, span, scope, scopeSchemaURL, buf.Buffer)
+	err := e.model.encodeSpan(resource, resourceSchemaURL, span, scope, scopeSchemaURL, fIndex, buf.Buffer)
 	if err != nil {
 		buf.Recycle()
 		return fmt.Errorf("failed to encode trace record: %w", err)
 	}
 	// not recycling after Add returns an error as we don't know if it's already recycled
-	return bulkIndexerSession.Add(ctx, fIndex, buf, nil)
+	return bulkIndexerSession.Add(ctx, fIndex.Index, buf, nil)
 }
 
 func (e *elasticsearchExporter) pushSpanEvent(
@@ -435,24 +435,24 @@ func (e *elasticsearchExporter) pushSpanEvent(
 	scopeSchemaURL string,
 	bulkIndexerSession bulkIndexerSession,
 ) error {
-	fIndex := e.index
+	fIndex := esIndex{Index: e.index}
 	if e.dynamicIndex {
-		fIndex = routeSpanEvent(spanEvent.Attributes(), scope.Attributes(), resource.Attributes(), fIndex, e.otel, scope.Name())
+		fIndex = routeSpanEvent(spanEvent.Attributes(), scope.Attributes(), resource.Attributes(), e.index, e.otel, scope.Name())
 	}
 
 	if e.logstashFormat.Enabled {
-		formattedIndex, err := generateIndexWithLogstashFormat(fIndex, &e.logstashFormat, time.Now())
+		formattedIndex, err := generateIndexWithLogstashFormat(fIndex.Index, &e.logstashFormat, time.Now())
 		if err != nil {
 			return err
 		}
-		fIndex = formattedIndex
+		fIndex = esIndex{Index: formattedIndex}
 	}
 	buf := e.bufferPool.NewPooledBuffer()
-	e.model.encodeSpanEvent(resource, resourceSchemaURL, span, spanEvent, scope, scopeSchemaURL, buf.Buffer)
+	e.model.encodeSpanEvent(resource, resourceSchemaURL, span, spanEvent, scope, scopeSchemaURL, fIndex, buf.Buffer)
 	if buf.Buffer.Len() == 0 {
 		buf.Recycle()
 		return nil
 	}
 	// not recycling after Add returns an error as we don't know if it's already recycled
-	return bulkIndexerSession.Add(ctx, fIndex, buf, nil)
+	return bulkIndexerSession.Add(ctx, fIndex.Index, buf, nil)
 }

exporter/elasticsearchexporter/exporter_test.go

@@ -1196,19 +1196,19 @@ func TestExporterMetrics(t *testing.T) {
 		expected := []itemRequest{
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.metric.foo":"histogram"}}}`),
-				Document: []byte(`{"@timestamp":"0.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"attributes":{},"metrics":{"metric.foo":{"counts":[1,2,3,4],"values":[0.5,1.5,2.5,3.0]}},"resource":{},"scope":{}}`),
+				Document: []byte(`{"@timestamp":"0.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.foo":{"counts":[1,2,3,4],"values":[0.5,1.5,2.5,3.0]}},"resource":{},"scope":{}}`),
 			},
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.metric.foo":"histogram"}}}`),
-				Document: []byte(`{"@timestamp":"3600000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"attributes":{},"metrics":{"metric.foo":{"counts":[4,5,6,7],"values":[2.0,4.5,5.5,6.0]}},"resource":{},"scope":{}}`),
+				Document: []byte(`{"@timestamp":"3600000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.foo":{"counts":[4,5,6,7],"values":[2.0,4.5,5.5,6.0]}},"resource":{},"scope":{}}`),
 			},
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.metric.sum":"gauge_double"}}}`),
-				Document: []byte(`{"@timestamp":"3600000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"attributes":{},"metrics":{"metric.sum":1.5},"resource":{},"scope":{},"start_timestamp":"7200000.0"}`),
+				Document: []byte(`{"@timestamp":"3600000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.sum":1.5},"resource":{},"scope":{},"start_timestamp":"7200000.0"}`),
 			},
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.metric.summary":"summary"}}}`),
-				Document: []byte(`{"@timestamp":"10800000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"attributes":{},"metrics":{"metric.summary":{"sum":1.5,"value_count":1}},"resource":{},"scope":{},"start_timestamp":"10800000.0"}`),
+				Document: []byte(`{"@timestamp":"10800000.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"metric.summary":{"sum":1.5,"value_count":1}},"resource":{},"scope":{},"start_timestamp":"10800000.0"}`),
 			},
 		}
 
@@ -1277,7 +1277,7 @@ func TestExporterMetrics(t *testing.T) {
 		expected := []itemRequest{
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.sum":"gauge_long","metrics.summary":"summary"}}}`),
-				Document: []byte(`{"@timestamp":"0.0","_doc_count":10,"data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"attributes":{},"metrics":{"sum":0,"summary":{"sum":1.0,"value_count":10}},"resource":{},"scope":{}}`),
+				Document: []byte(`{"@timestamp":"0.0","_doc_count":10,"data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"sum":0,"summary":{"sum":1.0,"value_count":10}},"resource":{},"scope":{}}`),
 			},
 		}
 
@@ -1370,7 +1370,7 @@ func TestExporterMetrics(t *testing.T) {
 		expected := []itemRequest{
 			{
 				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.foo.bar":"gauge_long","metrics.foo":"gauge_long","metrics.foo.bar.baz":"gauge_long"}}}`),
-				Document: []byte(`{"@timestamp":"0.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"attributes":{},"metrics":{"foo":0,"foo.bar":0,"foo.bar.baz":0},"resource":{},"scope":{}}`),
+				Document: []byte(`{"@timestamp":"0.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"foo":0,"foo.bar":0,"foo.bar.baz":0},"resource":{},"scope":{}}`),
 			},
 		}
 
@@ -1867,6 +1867,7 @@ func mustSendLogRecords(t *testing.T, exporter exporter.Logs, records ...plog.Lo
 }
 
 func mustSendLogs(t *testing.T, exporter exporter.Logs, logs plog.Logs) {
+	logs.MarkReadOnly()
 	err := exporter.ConsumeLogs(context.Background(), logs)
 	require.NoError(t, err)
 }
@@ -1896,6 +1897,7 @@ func mustSendMetricGaugeDataPoints(t *testing.T, exporter exporter.Metrics, data
 }
 
 func mustSendMetrics(t *testing.T, exporter exporter.Metrics, metrics pmetric.Metrics) {
+	metrics.MarkReadOnly()
 	err := exporter.ConsumeMetrics(context.Background(), metrics)
 	require.NoError(t, err)
 }
@@ -1911,6 +1913,7 @@ func mustSendSpans(t *testing.T, exporter exporter.Traces, spans ...ptrace.Span)
 }
 
 func mustSendTraces(t *testing.T, exporter exporter.Traces, traces ptrace.Traces) {
+	traces.MarkReadOnly()
 	err := exporter.ConsumeTraces(context.Background(), traces)
 	require.NoError(t, err)
 }

exporter/elasticsearchexporter/factory.go

@@ -166,7 +166,7 @@ func exporterhelperOptions(
 	shutdown component.ShutdownFunc,
 ) []exporterhelper.Option {
 	opts := []exporterhelper.Option{
-		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: true}),
+		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
 		exporterhelper.WithStart(start),
 		exporterhelper.WithShutdown(shutdown),
 		exporterhelper.WithQueue(cfg.QueueSettings),

exporter/elasticsearchexporter/integrationtest/exporter_bench_test.go

@@ -65,11 +65,12 @@ func benchmarkLogs(b *testing.B, batchSize int, mappingMode string) {
 	require.NoError(b, err)
 	require.NoError(b, exporter.Start(ctx, componenttest.NewNopHost()))
 
+	logs, _ := runnerCfg.provider.GenerateLogs()
+	logs.MarkReadOnly()
 	b.ReportAllocs()
 	b.ResetTimer()
 	b.StopTimer()
 	for i := 0; i < b.N; i++ {
-		logs, _ := runnerCfg.provider.GenerateLogs()
 		b.StartTimer()
 		require.NoError(b, exporter.ConsumeLogs(ctx, logs))
 		b.StopTimer()
@@ -94,11 +95,12 @@ func benchmarkMetrics(b *testing.B, batchSize int, mappingMode string) {
 	require.NoError(b, err)
 	require.NoError(b, exporter.Start(ctx, componenttest.NewNopHost()))
 
+	metrics, _ := runnerCfg.provider.GenerateMetrics()
+	metrics.MarkReadOnly()
 	b.ReportAllocs()
 	b.ResetTimer()
 	b.StopTimer()
 	for i := 0; i < b.N; i++ {
-		metrics, _ := runnerCfg.provider.GenerateMetrics()
 		b.StartTimer()
 		require.NoError(b, exporter.ConsumeMetrics(ctx, metrics))
 		b.StopTimer()
@@ -123,11 +125,12 @@ func benchmarkTraces(b *testing.B, batchSize int, mappingMode string) {
 	require.NoError(b, err)
 	require.NoError(b, exporter.Start(ctx, componenttest.NewNopHost()))
 
+	traces, _ := runnerCfg.provider.GenerateTraces()
+	traces.MarkReadOnly()
 	b.ReportAllocs()
 	b.ResetTimer()
 	b.StopTimer()
 	for i := 0; i < b.N; i++ {
-		traces, _ := runnerCfg.provider.GenerateTraces()
 		b.StartTimer()
 		require.NoError(b, exporter.ConsumeTraces(ctx, traces))
 		b.StopTimer()