Log validation error when metric with same name has already been serialized

Author: felixbarny

exporter/elasticsearchexporter/pdata_serializer.go

@@ -6,6 +6,7 @@ package elasticsearchexporter // import "github.com/open-telemetry/opentelemetry
 import (
 	"bytes"
 	"encoding/hex"
+	"fmt"
 	"strings"
 
 	"github.com/elastic/go-structform"
@@ -48,8 +49,21 @@ func serializeDataPoints(v *json.Visitor, dataPoints []dataPoint, validationErro
 
 	dynamicTemplates := make(map[string]string, len(dataPoints))
 	var docCount uint64
+	metricNames := make(map[string]bool, len(dataPoints))
 	for _, dp := range dataPoints {
 		metric := dp.Metric()
+		if _, present := metricNames[metric.Name()]; present {
+			*validationErrors = append(
+				*validationErrors,
+				fmt.Errorf(
+					"metric with name '%s' has already been serialized in document with timestamp %s",
+					metric.Name(),
+					dp.Timestamp().AsTime().UTC().Format(tsLayout),
+				),
+			)
+			continue
+		}
+		metricNames[metric.Name()] = true
 		value, err := dp.Value()
 		if dp.HasMappingHint(hintDocCount) {
 			docCount = dp.DocCount()

exporter/elasticsearchexporter/pdata_serializer_test.go

@@ -6,12 +6,14 @@ package elasticsearchexporter
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
 )
 
 func TestSerializeLog(t *testing.T) {
@@ -159,6 +161,48 @@ func TestSerializeLog(t *testing.T) {
 	}
 }
 
+func TestSerializeMetricsConflict(t *testing.T) {
+	resourceMetrics := pmetric.NewResourceMetrics()
+	scopeMetrics := resourceMetrics.ScopeMetrics().AppendEmpty()
+	var dataPoints []dataPoint
+	metric1 := scopeMetrics.Metrics().AppendEmpty()
+	metric2 := scopeMetrics.Metrics().AppendEmpty()
+	for _, m := range []pmetric.Metric{metric1, metric2} {
+		m.SetName("foo")
+		dp := m.SetEmptyGauge().DataPoints().AppendEmpty()
+		dp.SetIntValue(42)
+		dataPoints = append(dataPoints, newNumberDataPoint(m, dp))
+	}
+
+	var validationErrors []error
+	var buf bytes.Buffer
+	_, err := serializeMetrics(resourceMetrics.Resource(), "", scopeMetrics.Scope(), "", dataPoints, &validationErrors, &buf)
+	if err != nil {
+		t.Errorf("serializeMetrics() error = %v", err)
+	}
+	b := buf.Bytes()
+	eventAsJSON := string(b)
+	var result any
+	decoder := json.NewDecoder(bytes.NewBuffer(b))
+	decoder.UseNumber()
+	if err := decoder.Decode(&result); err != nil {
+		t.Error(err)
+	}
+
+	assert.Len(t, validationErrors, 1)
+	assert.Equal(t, fmt.Errorf("metric with name 'foo' has already been serialized in document with timestamp 1970-01-01T00:00:00.000000000Z"), validationErrors[0])
+
+	assert.Equal(t, map[string]any{
+		"@timestamp":  "1970-01-01T00:00:00.000000000Z",
+		"data_stream": map[string]any{},
+		"resource":    map[string]any{},
+		"scope":       map[string]any{},
+		"metrics": map[string]any{
+			"foo": json.Number("42"),
+		},
+	}, result, eventAsJSON)
+}
+
 func TestMergeGeolocation(t *testing.T) {
 	attributes := map[string]any{
 		"geo.location.lon":          1.1,