Add metrics grouping for otel mode

exporter/elasticsearchexporter/internal/objmodel/objmodel.go

@@ -283,6 +283,7 @@ func (doc *Document) iterJSONFlat(w *json.Visitor, otel bool) error {
 	return nil
 }
 
+// FIXME: metrics.* as well
 // Set of prefixes for the OTel attributes that needs to stay flattened
 var otelPrefixSet = map[string]struct{}{
 	"attributes.":          {},

exporter/elasticsearchexporter/model.go

@@ -83,6 +83,7 @@ type encodeModel struct {
 
 type dataPoint interface {
 	Timestamp() pcommon.Timestamp
+	StartTimestamp() pcommon.Timestamp
 	Attributes() pcommon.Map
 }
 
@@ -321,6 +322,94 @@ func (m *encodeModel) upsertMetricDataPointValue(documents map[uint32]objmodel.D
 	return nil
 }
 
+func (m *encodeModel) upsertMetricDataPointValueOTelMode(documents map[uint32]objmodel.Document, resource pcommon.Resource, resourceSchemaUrl string, scope pcommon.InstrumentationScope, scopeSchemaUrl string, metric pmetric.Metric, dp dataPoint, value pcommon.Value) error {
+	// documents is per-resource. Therefore, there is no need to hash resource attributes
+	hash := metricOTelHash(dp, scope.Attributes(), metric.Unit())
+	var (
+		document objmodel.Document
+		ok       bool
+	)
+	if document, ok = documents[hash]; !ok {
+		document.AddTimestamp("@timestamp", dp.Timestamp())
+		document.AddTimestamp("start_timestamp", dp.StartTimestamp())
+		document.AddString("unit", metric.Unit())
+
+		// At this point the data_stream attributes are expected to be in the record attributes,
+		// updated by the router.
+		// Move them to the top of the document and remove them from the record
+		attributeMap := dp.Attributes()
+
+		forEachDataStreamKey := func(fn func(key string)) {
+			for _, key := range datastreamKeys {
+				fn(key)
+			}
+		}
+
+		forEachDataStreamKey(func(key string) {
+			if value, exists := attributeMap.Get(key); exists {
+				document.AddAttribute(key, value)
+				attributeMap.Remove(key)
+			}
+		})
+
+		document.AddAttributes("attributes", attributeMap)
+
+		// Resource
+		resourceMapVal := pcommon.NewValueMap()
+		resourceMap := resourceMapVal.Map()
+		resourceMap.PutStr("schema_url", resourceSchemaUrl)
+		resourceMap.PutInt("dropped_attributes_count", int64(resource.DroppedAttributesCount()))
+		resourceAttrMap := resourceMap.PutEmptyMap("attributes")
+
+		resource.Attributes().CopyTo(resourceAttrMap)
+
+		// Remove data_stream attributes from the resources attributes if present
+		forEachDataStreamKey(func(key string) {
+			resourceAttrMap.Remove(key)
+		})
+
+		document.Add("resource", objmodel.ValueFromAttribute(resourceMapVal))
+
+		// Scope
+		scopeMapVal := pcommon.NewValueMap()
+		scopeMap := scopeMapVal.Map()
+		if scope.Name() != "" {
+			scopeMap.PutStr("name", scope.Name())
+		}
+		if scope.Version() != "" {
+			scopeMap.PutStr("version", scope.Version())
+		}
+		if scopeSchemaUrl != "" {
+			scopeMap.PutStr("schema_url", scopeSchemaUrl)
+		}
+		if scope.DroppedAttributesCount() > 0 {
+			scopeMap.PutInt("dropped_attributes_count", int64(scope.DroppedAttributesCount()))
+		}
+		scopeAttributes := scope.Attributes()
+		if scopeAttributes.Len() > 0 {
+			scopeAttrMap := scopeMap.PutEmptyMap("attributes")
+			scopeAttributes.CopyTo(scopeAttrMap)
+
+			// Remove data_stream attributes from the scope attributes if present
+			forEachDataStreamKey(func(key string) {
+				scopeAttrMap.Remove(key)
+			})
+		}
+
+		if scopeMap.Len() > 0 {
+			document.Add("scope", objmodel.ValueFromAttribute(scopeMapVal))
+		}
+	}
+
+	metricsMap := pcommon.NewValueMap()
+	v := metricsMap.SetEmptyMap().PutEmpty(metric.Name())
+	value.CopyTo(v)
+	document.Add("metrics", objmodel.ValueFromAttribute(metricsMap))
+
+	documents[hash] = document
+	return nil
+}
+
 func histogramToValue(dp pmetric.HistogramDataPoint) (pcommon.Value, error) {
 	// Histogram conversion function is from
 	// https://github.com/elastic/apm-data/blob/3b28495c3cbdc0902983134276eb114231730249/input/otlp/metrics.go#L277
@@ -586,6 +675,24 @@ func metricHash(timestamp pcommon.Timestamp, attributes pcommon.Map) uint32 {
 	return hasher.Sum32()
 }
 
+func metricOTelHash(dp dataPoint, scopeAttrs pcommon.Map, unit string) uint32 {
+	hasher := fnv.New32a()
+
+	timestampBuf := make([]byte, 8)
+	binary.LittleEndian.PutUint64(timestampBuf, uint64(dp.Timestamp()))
+	hasher.Write(timestampBuf)
+
+	binary.LittleEndian.PutUint64(timestampBuf, uint64(dp.StartTimestamp()))
+	hasher.Write(timestampBuf)
+
+	hasher.Write([]byte(unit))
+
+	mapHash(hasher, scopeAttrs)
+	mapHash(hasher, dp.Attributes())
+
+	return hasher.Sum32()
+}
+
 func mapHash(hasher hash.Hash, m pcommon.Map) {
 	m.Range(func(k string, v pcommon.Value) bool {
 		hasher.Write([]byte(k))