[exporter/elasticsearch] OTel mode serialization

exporter/elasticsearchexporter/attribute.go

@@ -7,57 +7,19 @@ import "go.opentelemetry.io/collector/pdata/pcommon"
 
 // dynamic index attribute key constants
 const (
-	indexPrefix = "elasticsearch.index.prefix"
-	indexSuffix = "elasticsearch.index.suffix"
-
-	dataStreamType      = "data_stream.type"
-	dataStreamDataset   = "data_stream.dataset"
-	dataStreamNamespace = "data_stream.namespace"
-
-	defaultDataStreamType      = "logs"
-	defaultDataStreamNamespace = "default"
-	defaultDataStreamDataset   = "generic"
+	indexPrefix                  = "elasticsearch.index.prefix"
+	indexSuffix                  = "elasticsearch.index.suffix"
+	dataStreamDataset            = "data_stream.dataset"
+	dataStreamNamespace          = "data_stream.namespace"
+	dataStreamType               = "data_stream.type"
+	defaultDataStreamDataset     = "generic"
+	defaultDataStreamNamespace   = "default"
+	defaultDataStreamTypeLogs    = "logs"
+	defaultDataStreamTypeMetrics = "metrics"
+	defaultDataStreamTypeTraces  = "traces"
 )
 
-// resource is higher priotized than record attribute
-type attrGetter interface {
-	Attributes() pcommon.Map
-}
-
-// retrieve attribute out of resource, scope, and record (span or log, if not found in resource)
-// Deprecated: Use getFromAttributesNew instead.
-func getFromAttributes(name string, resource, scope, record attrGetter) string {
-	var str string
-	val, exist := resource.Attributes().Get(name)
-	if !exist {
-		val, exist = scope.Attributes().Get(name)
-		if !exist {
-			val, exist = record.Attributes().Get(name)
-			if exist {
-				str = val.AsString()
-			}
-		}
-		if exist {
-			str = val.AsString()
-		}
-	}
-	if exist {
-		str = val.AsString()
-	}
-	return str
-}
-
-// retrieve attribute out of resource, scope, and record (span or log, if not found in resource)
-// returns default value if result is empty string
-func getFromAttributesWithDefault(name string, resource, scope, record attrGetter, defaultValue string) string {
-	s := getFromAttributes(name, resource, scope, record)
-	if s == "" {
-		s = defaultValue
-	}
-	return s
-}
-
-func getFromAttributesNew(name string, defaultValue string, attributeMaps ...pcommon.Map) string {
+func getFromAttributes(name string, defaultValue string, attributeMaps ...pcommon.Map) (string, bool) {
 	for _, attributeMap := range attributeMaps {
 		if value, exists := attributeMap.Get(name); exists {
 			return value.AsString(), true

exporter/elasticsearchexporter/exporter.go

@@ -8,7 +8,6 @@ import (
 	"errors"
 	"fmt"
 	"runtime"
-	"strings"
 	"time"
 
 	"go.opentelemetry.io/collector/component"
@@ -158,24 +157,7 @@ func (e *elasticsearchExporter) pushLogRecord(ctx context.Context,
 	scopeSchemaUrl string) error {
 	fIndex := e.index
 	if e.dynamicIndex {
-		switch e.dynamicIndexMode {
-		case dynIdxModePrefixSuffix:
-			prefix := getFromAttributes(indexPrefix, resource, scope, record)
-			suffix := getFromAttributes(indexSuffix, resource, scope, record)
-			fIndex = fmt.Sprintf("%s%s%s", prefix, fIndex, suffix)
-		case dynIdxModeDataStream:
-			const otelSuffix = ".otel"
-			typ := getFromAttributesWithDefault(dataStreamType, resource, scope, record, defaultDataStreamType)
-			dataset := getFromAttributesWithDefault(dataStreamDataset, resource, scope, record, defaultDataStreamDataset)
-			namespace := getFromAttributesWithDefault(dataStreamNamespace, resource, scope, record, defaultDataStreamNamespace)
-
-			// The naming convension for datastream is expected to be the following
-			// "logs-[dataset].otel-[namespace]"
-			if e.mode == MappingOTel && !strings.HasSuffix(dataset, otelSuffix) {
-				dataset += otelSuffix
-			}
-			fIndex = fmt.Sprintf("%s-%s-%s", typ, dataset, namespace)
-		}
+		fIndex = routeLogRecord(record, scope, resource, fIndex)
 	}
 
 	if e.logstashFormat.Enabled {

exporter/elasticsearchexporter/model.go

@@ -64,7 +64,6 @@ var resourceAttrsToPreserve = map[string]bool{
 
 type mappingModel interface {
 	encodeLog(pcommon.Resource, string, plog.LogRecord, pcommon.InstrumentationScope, string) ([]byte, error)
-	encodeMetrics(pcommon.Resource, pmetric.MetricSlice, pcommon.InstrumentationScope) ([][]byte, error)
 	encodeSpan(pcommon.Resource, ptrace.Span, pcommon.InstrumentationScope) ([]byte, error)
 	upsertMetricDataPoint(map[uint32]objmodel.Document, pcommon.Resource, pcommon.InstrumentationScope, pmetric.Metric, pmetric.NumberDataPoint) error
 	encodeDocument(objmodel.Document) ([]byte, error)
@@ -274,27 +273,23 @@ func (m *encodeModel) encodeDocument(document objmodel.Document) ([]byte, error)
 	}
 
 	var buf bytes.Buffer
-	err := document.Serialize(&buf, m.dedot)
+	err := document.Serialize(&buf, m.dedot, m.mode == MappingOTel)
 	if err != nil {
 		return nil, err
 	}
 	return buf.Bytes(), nil
 }
 
-	for _, doc := range docs {
-		if m.dedup {
-			doc.Dedup()
-		} else if m.dedot {
-			doc.Sort()
-		}
-
-		var buf bytes.Buffer
-		err := doc.Serialize(&buf, m.dedot, m.mode == MappingOTel)
-		if err != nil {
-			return nil, err
-		}
-
-		res = append(res, buf.Bytes())
+func (m *encodeModel) upsertMetricDataPoint(documents map[uint32]objmodel.Document, resource pcommon.Resource, _ pcommon.InstrumentationScope, metric pmetric.Metric, dp pmetric.NumberDataPoint) error {
+	hash := metricHash(dp.Timestamp(), dp.Attributes())
+	var (
+		document objmodel.Document
+		ok       bool
+	)
+	if document, ok = documents[hash]; !ok {
+		encodeAttributesECSMode(&document, resource.Attributes(), resourceAttrsConversionMap, resourceAttrsToPreserve)
+		document.AddTimestamp("@timestamp", dp.Timestamp())
+		document.AddAttributes("", dp.Attributes())
 	}
 
 	switch dp.ValueType() {