Optimize and fix geo attribute serialization

Author: felixbarny

exporter/elasticsearchexporter/pdata_serializer.go

@@ -289,33 +289,37 @@ func writeAttributes(v *json.Visitor, attributes pcommon.Map, stringifyMapValues
 	if attributes.Len() == 0 {
 		return
 	}
-	geoAttributes := mergeGeolocation(attributes)
+
 	_ = v.OnKey("attributes")
 	_ = v.OnObjectStart(-1, structform.AnyType)
 	attributes.Range(func(k string, val pcommon.Value) bool {
 		switch k {
 		case dataStreamType, dataStreamDataset, dataStreamNamespace, mappingHintsAttrKey:
 			return true
 		}
-		if strings.HasSuffix(k, ".geo.location.lat") || strings.HasSuffix(k, ".geo.location.lon") {
+		if isGeoAttribute(k, val) {
 			return true
 		}
 		_ = v.OnKey(k)
 		writeValue(v, val, stringifyMapValues)
 		return true
 	})
-	geoAttributes.Range(func(k string, val pcommon.Value) bool {
-		_ = v.OnKey(k)
-		writeValue(v, val, stringifyMapValues)
-		return true
-	})
+	writeGeolocationAttributes(v, attributes)
 	_ = v.OnObjectFinished()
 }
 
-// mergeGeolocation returns a new map that to merges all `geo.location.{lon,lat}`,
-// and namespaced `*.geo.location.{lon,lat}` attributes from the provided map to unnamespaced and namespaced `geo.location`.
-// This is to match the geo_point type in Elasticsearch.
-func mergeGeolocation(attributes pcommon.Map) pcommon.Map {
+func isGeoAttribute(k string, val pcommon.Value) bool {
+	if val.Type() != pcommon.ValueTypeDouble {
+		return false
+	}
+	switch k {
+	case "geo.location.lat", "geo.location.lon":
+		return true
+	}
+	return strings.HasSuffix(k, ".geo.location.lat") || strings.HasSuffix(k, ".geo.location.lon")
+}
+
+func writeGeolocationAttributes(v *json.Visitor, attributes pcommon.Map) {
 	const (
 		lonKey    = "geo.location.lon"
 		latKey    = "geo.location.lat"
@@ -362,28 +366,27 @@ func mergeGeolocation(attributes pcommon.Map) pcommon.Map {
 		return true
 	})
 
-	geoAttributes := pcommon.NewMap()
 	for prefix, geo := range prefixToGeo {
 		if geo.lonSet && geo.latSet {
 			key := prefix + mergedKey
 			// Geopoint expressed as an array with the format: [lon, lat]
-			s := geoAttributes.PutEmptySlice(key)
-			s.EnsureCapacity(2)
-			s.AppendEmpty().SetDouble(geo.lon)
-			s.AppendEmpty().SetDouble(geo.lat)
+			_ = v.OnKey(key)
+			_ = v.OnArrayStart(-1, structform.AnyType)
+			_ = v.OnFloat64(geo.lon)
+			_ = v.OnFloat64(geo.lat)
+			_ = v.OnArrayFinished()
 			continue
 		}
 		// Place the attributes back if lon and lat are not present together
 		if geo.lonSet {
-			key := prefix + lonKey
-			geoAttributes.PutDouble(key, geo.lon)
+			_ = v.OnKey(prefix + lonKey)
+			_ = v.OnFloat64(geo.lon)
 		}
 		if geo.latSet {
-			key := prefix + latKey
-			geoAttributes.PutDouble(key, geo.lat)
+			_ = v.OnKey(prefix + latKey)
+			_ = v.OnFloat64(geo.lat)
 		}
 	}
-	return geoAttributes
 }
 
 func writeMap(v *json.Visitor, m pcommon.Map, stringifyMapValues bool) {

exporter/elasticsearchexporter/pdata_serializer_test.go

@@ -10,7 +10,6 @@ import (
 	"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"
@@ -117,9 +116,17 @@ func TestSerializeLog(t *testing.T) {
 		{
 			name: "geo attributes",
 			logCustomizer: func(_ pcommon.Resource, _ pcommon.InstrumentationScope, record plog.LogRecord) {
-				record.Attributes().PutDouble("foo.geo.location.lon", 1)
-				record.Attributes().PutDouble("foo.geo.location.lat", 2)
-				record.Attributes().PutDouble("bar.geo.location.lat", 3)
+				record.Attributes().PutDouble("geo.location.lon", 1.1)
+				record.Attributes().PutDouble("geo.location.lat", 2.2)
+				record.Attributes().PutDouble("foo.bar.geo.location.lon", 3.3)
+				record.Attributes().PutDouble("foo.bar.geo.location.lat", 4.4)
+				record.Attributes().PutDouble("a.geo.location.lon", 5.5)
+				record.Attributes().PutDouble("b.geo.location.lat", 6.6)
+				record.Attributes().PutDouble("unrelatedgeo.location.lon", 7.7)
+				record.Attributes().PutDouble("unrelatedgeo.location.lat", 8.8)
+				record.Attributes().PutDouble("d", 9.9)
+				record.Attributes().PutStr("e.geo.location.lon", "foo")
+				record.Attributes().PutStr("e.geo.location.lat", "bar")
 			},
 			wantErr: false,
 			expected: map[string]any{
@@ -129,8 +136,15 @@ func TestSerializeLog(t *testing.T) {
 				"resource":           map[string]any{},
 				"scope":              map[string]any{},
 				"attributes": map[string]any{
-					"foo.geo.location":     []any{json.Number("1.0"), json.Number("2.0")},
-					"bar.geo.location.lat": json.Number("3.0"),
+					"geo.location":              []any{json.Number("1.1"), json.Number("2.2")},
+					"foo.bar.geo.location":      []any{json.Number("3.3"), json.Number("4.4")},
+					"a.geo.location.lon":        json.Number("5.5"),
+					"b.geo.location.lat":        json.Number("6.6"),
+					"unrelatedgeo.location.lon": json.Number("7.7"),
+					"unrelatedgeo.location.lat": json.Number("8.8"),
+					"d":                         json.Number("9.9"),
+					"e.geo.location.lon":        "foo",
+					"e.geo.location.lat":        "bar",
 				},
 			},
 		},
@@ -202,31 +216,3 @@ func TestSerializeMetricsConflict(t *testing.T) {
 		},
 	}, result, eventAsJSON)
 }
-
-func TestMergeGeolocation(t *testing.T) {
-	attributes := map[string]any{
-		"geo.location.lon":          1.1,
-		"geo.location.lat":          2.2,
-		"foo.bar.geo.location.lon":  3.3,
-		"foo.bar.geo.location.lat":  4.4,
-		"a.geo.location.lon":        5.5,
-		"b.geo.location.lat":        6.6,
-		"unrelatedgeo.location.lon": 7.7,
-		"unrelatedgeo.location.lat": 8.8,
-		"d":                         9.9,
-		"e.geo.location.lon":        "foo",
-		"e.geo.location.lat":        "bar",
-	}
-	wantAttributes := map[string]any{
-		"a.geo.location.lon":   5.5,
-		"b.geo.location.lat":   6.6,
-		"geo.location":         []any{1.1, 2.2},
-		"foo.bar.geo.location": []any{3.3, 4.4},
-	}
-	input := pcommon.NewMap()
-	err := input.FromRaw(attributes)
-	require.NoError(t, err)
-	output := mergeGeolocation(input)
-	after := output.AsRaw()
-	assert.Equal(t, wantAttributes, after)
-}