[exporter/elasticsearch] Fix data loss due to metric grouping regression

.chloggen/elasticsearchexporter_metrics-grouping-data-loss.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: bug_fix
+
+# 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: Fix data loss caused by incorrect metric grouping in ECS and OTel mode
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37898]
+
+# (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: Fix data loss when the same metric exists across different resources or scopes. Data points / metrics were incorrectly grouped together, leading to data loss with warning logs e.g. "metric with name '***' has already been serialized".
+
+# 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/exporter.go

@@ -285,7 +285,7 @@ func (e *elasticsearchExporter) pushMetricsData(
 						groupedDataPoints = make(map[uint32]*dataPointsGroup)
 						groupedDataPointsByIndex[index] = groupedDataPoints
 					}
-					dpHash := hasher.hashDataPoint(dp)
+					dpHash := hasher.hashDataPoint(resource, scope, dp)
 					dpGroup, ok := groupedDataPoints[dpHash]
 					if !ok {
 						groupedDataPoints[dpHash] = &dataPointsGroup{

exporter/elasticsearchexporter/exporter_test.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
 	"math"
 	"net/http"
 	"runtime"
@@ -931,17 +932,6 @@ func TestExporterMetrics(t *testing.T) {
 	})
 
 	t.Run("publish with metrics grouping", func(t *testing.T) {
-		rec := newBulkRecorder()
-		server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
-			rec.Record(docs)
-			return itemsAllOK(docs)
-		})
-
-		exporter := newTestMetricsExporter(t, server.URL, func(cfg *Config) {
-			cfg.MetricsIndex = "metrics.index"
-			cfg.Mapping.Mode = "ecs"
-		})
-
 		addToMetricSlice := func(metricSlice pmetric.MetricSlice) {
 			fooMetric := metricSlice.AppendEmpty()
 			fooMetric.SetName("metric.foo")
@@ -957,7 +947,7 @@ func TestExporterMetrics(t *testing.T) {
 			barMetric := metricSlice.AppendEmpty()
 			barMetric.SetName("metric.bar")
 			barDps := barMetric.SetEmptyGauge().DataPoints()
-			barDp := barDps.AppendEmpty()
+			barDp := barDps.AppendEmpty() // dp without attribute
 			barDp.SetDoubleValue(1.0)
 			barOtherDp := barDps.AppendEmpty()
 			fillAttributeMap(barOtherDp.Attributes(), map[string]any{
@@ -975,62 +965,109 @@ func TestExporterMetrics(t *testing.T) {
 			bazMetric.SetName("metric.baz")
 			bazDps := bazMetric.SetEmptyGauge().DataPoints()
 			bazDp := bazDps.AppendEmpty()
-			bazDp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(3600, 0)))
+			bazDp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(3600, 0))) // dp with different timestamp
 			bazDp.SetDoubleValue(1.0)
 		}
 
 		metrics := pmetric.NewMetrics()
-		resourceMetrics := metrics.ResourceMetrics().AppendEmpty()
-		fillAttributeMap(resourceMetrics.Resource().Attributes(), map[string]any{
+		resourceA := metrics.ResourceMetrics().AppendEmpty()
+		fillAttributeMap(resourceA.Resource().Attributes(), map[string]any{
 			elasticsearch.DataStreamNamespace: "resource.namespace",
 		})
-		scopeA := resourceMetrics.ScopeMetrics().AppendEmpty()
-		addToMetricSlice(scopeA.Metrics())
+		scopeAA := resourceA.ScopeMetrics().AppendEmpty()
+		addToMetricSlice(scopeAA.Metrics())
 
-		scopeB := resourceMetrics.ScopeMetrics().AppendEmpty()
-		fillAttributeMap(scopeB.Scope().Attributes(), map[string]any{
-			elasticsearch.DataStreamDataset: "scope.b",
+		scopeAB := resourceA.ScopeMetrics().AppendEmpty()
+		fillAttributeMap(scopeAB.Scope().Attributes(), map[string]any{
+			elasticsearch.DataStreamDataset: "scope.ab", // routes to a different index and should not be grouped together
 		})
-		addToMetricSlice(scopeB.Metrics())
+		addToMetricSlice(scopeAB.Metrics())
 
-		mustSendMetrics(t, exporter, metrics)
+		scopeAC := resourceA.ScopeMetrics().AppendEmpty()
+		fillAttributeMap(scopeAC.Scope().Attributes(), map[string]any{
+			// ecs: scope attributes are ignored, and duplicates are dropped silently.
+			// otel: scope attributes are dimensions and should result in a separate group.
+			"some.scope.attribute": "scope.ac",
+		})
+		addToMetricSlice(scopeAC.Metrics())
 
-		expected := []itemRequest{
-			{
-				Action:   []byte(`{"create":{"_index":"metrics-generic-bar"}}`),
-				Document: []byte(`{"@timestamp":"1970-01-01T00:00:00.000000000Z","data_stream":{"dataset":"generic","namespace":"bar","type":"metrics"},"dp":{"attribute":"dp.attribute.value"},"metric":{"bar":1.0}}`),
-			},
-			{
-				Action:   []byte(`{"create":{"_index":"metrics-generic-resource.namespace"}}`),
-				Document: []byte(`{"@timestamp":"1970-01-01T00:00:00.000000000Z","data_stream":{"dataset":"generic","namespace":"resource.namespace","type":"metrics"},"dp":{"attribute":"dp.attribute.value"},"metric":{"bar":1.0,"foo":1.0}}`),
-			},
-			{
-				Action:   []byte(`{"create":{"_index":"metrics-generic-resource.namespace"}}`),
-				Document: []byte(`{"@timestamp":"1970-01-01T00:00:00.000000000Z","data_stream":{"dataset":"generic","namespace":"resource.namespace","type":"metrics"},"metric":{"bar":1.0,"foo":1}}`),
-			},
-			{
-				Action:   []byte(`{"create":{"_index":"metrics-generic-resource.namespace"}}`),
-				Document: []byte(`{"@timestamp":"1970-01-01T01:00:00.000000000Z","data_stream":{"dataset":"generic","namespace":"resource.namespace","type":"metrics"},"metric":{"baz":1.0}}`),
-			},
-			{
-				Action:   []byte(`{"create":{"_index":"metrics-scope.b-bar"}}`),
-				Document: []byte(`{"@timestamp":"1970-01-01T00:00:00.000000000Z","data_stream":{"dataset":"scope.b","namespace":"bar","type":"metrics"},"dp":{"attribute":"dp.attribute.value"},"metric":{"bar":1.0}}`),
-			},
-			{
-				Action:   []byte(`{"create":{"_index":"metrics-scope.b-resource.namespace"}}`),
-				Document: []byte(`{"@timestamp":"1970-01-01T00:00:00.000000000Z","data_stream":{"dataset":"scope.b","namespace":"resource.namespace","type":"metrics"},"dp":{"attribute":"dp.attribute.value"},"metric":{"bar":1.0,"foo":1.0}}`),
-			},
-			{
-				Action:   []byte(`{"create":{"_index":"metrics-scope.b-resource.namespace"}}`),
-				Document: []byte(`{"@timestamp":"1970-01-01T00:00:00.000000000Z","data_stream":{"dataset":"scope.b","namespace":"resource.namespace","type":"metrics"},"metric":{"bar":1.0,"foo":1}}`),
-			},
-			{
-				Action:   []byte(`{"create":{"_index":"metrics-scope.b-resource.namespace"}}`),
-				Document: []byte(`{"@timestamp":"1970-01-01T01:00:00.000000000Z","data_stream":{"dataset":"scope.b","namespace":"resource.namespace","type":"metrics"},"metric":{"baz":1.0}}`),
-			},
+		resourceB := metrics.ResourceMetrics().AppendEmpty()
+		fillAttributeMap(resourceB.Resource().Attributes(), map[string]any{
+			"my.resource": "resource.b",
+		})
+		scopeBA := resourceB.ScopeMetrics().AppendEmpty()
+		addToMetricSlice(scopeBA.Metrics())
+
+		scopeBB := resourceB.ScopeMetrics().AppendEmpty()
+		scopeBB.Scope().SetName("scope.bb")
+		addToMetricSlice(scopeBB.Metrics())
+
+		// identical resource
+		resourceAnotherB := metrics.ResourceMetrics().AppendEmpty()
+		fillAttributeMap(resourceAnotherB.Resource().Attributes(), map[string]any{
+			"my.resource": "resource.b",
+		})
+		addToMetricSlice(resourceAnotherB.ScopeMetrics().AppendEmpty().Metrics())
+
+		assertDocsInIndices := func(t *testing.T, wantDocsPerIndex map[string]int, rec *bulkRecorder) {
+			var sum int
+			for v := range maps.Values(wantDocsPerIndex) {
+				sum += v
+			}
+			rec.WaitItems(sum)
+
+			actualDocsPerIndex := make(map[string]int)
+			for _, item := range rec.Items() {
+				idx := gjson.GetBytes(item.Action, "create._index")
+				actualDocsPerIndex[idx.String()] += 1
+			}
+			assert.Equal(t, wantDocsPerIndex, actualDocsPerIndex)
 		}
 
-		assertRecordedItems(t, expected, rec, false)
+		t.Run("ecs", func(t *testing.T) {
+			rec := newBulkRecorder()
+			server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
+				rec.Record(docs)
+				return itemsAllOK(docs)
+			})
+
+			exporter := newTestMetricsExporter(t, server.URL, func(cfg *Config) {
+				cfg.MetricsIndex = "metrics.index"
+				cfg.Mapping.Mode = "ecs"
+			})
+
+			mustSendMetrics(t, exporter, metrics)
+
+			assertDocsInIndices(t, map[string]int{
+				"metrics-generic-bar":                 2, // AA, BA
+				"metrics-generic-resource.namespace":  3,
+				"metrics-scope.ab-bar":                1,
+				"metrics-scope.ab-resource.namespace": 3,
+				"metrics-generic-default":             3,
+			}, rec)
+		})
+
+		t.Run("otel", func(t *testing.T) {
+			rec := newBulkRecorder()
+			server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
+				rec.Record(docs)
+				return itemsAllOK(docs)
+			})
+
+			exporter := newTestMetricsExporter(t, server.URL, func(cfg *Config) {
+				cfg.Mapping.Mode = "otel"
+			})
+
+			mustSendMetrics(t, exporter, metrics)
+
+			assertDocsInIndices(t, map[string]int{
+				"metrics-generic.otel-bar":                 4, // AA->bar, AC->bar, BA->bar, BB->bar
+				"metrics-generic.otel-resource.namespace":  6, // AA, AC
+				"metrics-scope.ab.otel-bar":                1, // AB->bar
+				"metrics-scope.ab.otel-resource.namespace": 3, // AB
+				"metrics-generic.otel-default":             6, // BA, BB
+			}, rec)
+		})
 	})
 
 	t.Run("publish histogram", func(t *testing.T) {
@@ -1365,40 +1402,6 @@ func TestExporterMetrics(t *testing.T) {
 		assertRecordedItems(t, expected, rec, false)
 	})
 
-	t.Run("otel mode grouping of equal resources", func(t *testing.T) {
-		rec := newBulkRecorder()
-		server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
-			rec.Record(docs)
-			return itemsAllOK(docs)
-		})
-
-		exporter := newTestMetricsExporter(t, server.URL, func(cfg *Config) {
-			cfg.Mapping.Mode = "otel"
-		})
-
-		metrics := pmetric.NewMetrics()
-		for _, n := range []string{"m1", "m2"} {
-			resourceMetric := metrics.ResourceMetrics().AppendEmpty()
-			scopeMetric := resourceMetric.ScopeMetrics().AppendEmpty()
-
-			sumMetric := scopeMetric.Metrics().AppendEmpty()
-			sumMetric.SetName(n)
-			sumDP := sumMetric.SetEmptySum().DataPoints().AppendEmpty()
-			sumDP.SetIntValue(0)
-		}
-
-		mustSendMetrics(t, exporter, metrics)
-
-		expected := []itemRequest{
-			{
-				Action:   []byte(`{"create":{"_index":"metrics-generic.otel-default","dynamic_templates":{"metrics.m1":"gauge_long","metrics.m2":"gauge_long"}}}`),
-				Document: []byte(`{"@timestamp":"0.0","data_stream":{"dataset":"generic.otel","namespace":"default","type":"metrics"},"metrics":{"m1":0,"m2":0},"resource":{},"scope":{}}`),
-			},
-		}
-
-		assertRecordedItems(t, expected, rec, false)
-	})
-
 	t.Run("otel mode aggregate_metric_double hint", func(t *testing.T) {
 		rec := newBulkRecorder()
 		server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {

exporter/elasticsearchexporter/metric_grouping.go

@@ -19,7 +19,7 @@ import (
 // dataPointHasher is an interface for hashing data points by their identity,
 // for grouping into a single document.
 type dataPointHasher interface {
-	hashDataPoint(datapoints.DataPoint) uint32
+	hashDataPoint(pcommon.Resource, pcommon.InstrumentationScope, datapoints.DataPoint) uint32
 }
 
 func newDataPointHasher(mode MappingMode) dataPointHasher {
@@ -39,9 +39,11 @@ type (
 	otelDataPointHasher struct{}
 )
 
-func (h ecsDataPointHasher) hashDataPoint(dp datapoints.DataPoint) uint32 {
+func (h ecsDataPointHasher) hashDataPoint(resource pcommon.Resource, _ pcommon.InstrumentationScope, dp datapoints.DataPoint) uint32 {
 	hasher := fnv.New32a()
 
+	mapHashExcludeReservedAttrs(hasher, resource.Attributes())
+
 	timestampBuf := make([]byte, 8)
 	binary.LittleEndian.PutUint64(timestampBuf, uint64(dp.Timestamp()))
 	hasher.Write(timestampBuf)
@@ -51,9 +53,13 @@ func (h ecsDataPointHasher) hashDataPoint(dp datapoints.DataPoint) uint32 {
 	return hasher.Sum32()
 }
 
-func (h otelDataPointHasher) hashDataPoint(dp datapoints.DataPoint) uint32 {
+func (h otelDataPointHasher) hashDataPoint(resource pcommon.Resource, scope pcommon.InstrumentationScope, dp datapoints.DataPoint) uint32 {
 	hasher := fnv.New32a()
 
+	mapHashExcludeReservedAttrs(hasher, resource.Attributes(), elasticsearch.MappingHintsAttrKey)
+	hasher.Write([]byte(scope.Name()))
+	mapHashExcludeReservedAttrs(hasher, scope.Attributes(), elasticsearch.MappingHintsAttrKey)
+
 	timestampBuf := make([]byte, 8)
 	binary.LittleEndian.PutUint64(timestampBuf, uint64(dp.Timestamp()))
 	hasher.Write(timestampBuf)

exporter/elasticsearchexporter/model_test.go

@@ -101,7 +101,7 @@ func TestEncodeMetric(t *testing.T) {
 	dps := m.Sum().DataPoints()
 	for i := 0; i < dps.Len(); i++ {
 		dp := datapoints.NewNumber(m, dps.At(i))
-		dpHash := hasher.hashDataPoint(dp)
+		dpHash := hasher.hashDataPoint(rm.Resource(), sm.Scope(), dp)
 		dataPoints, ok := groupedDataPoints[dpHash]
 		if !ok {
 			groupedDataPoints[dpHash] = []datapoints.DataPoint{dp}