[exporter/elasticsearch] OTel mode serialization

.chloggen/feature_elasticsearch_otel_model.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: enhancement
+
+# 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: Introduce an experimental OTel native mapping mode for logs
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [33290]
+
+# (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:
+
+# 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: []

exporter/elasticsearchexporter/README.md

@@ -127,6 +127,10 @@ behaviours, which may be configured through the following settings:
   - `mode` (default=none): The fields naming mode. valid modes are:
     - `none`: Use original fields and event structure from the OTLP event.
     - `ecs`: Try to map fields to [Elastic Common Schema (ECS)][ECS]
+    - `otel`: Elastic's preferred "OTel-native" mapping mode. Uses original fields and event structure from the OTLP event.
+          :warning: This mode's behavior is unstable, it is currently is experimental and undergoing changes.
+          There's a special treatment for the following attributes: `data_stream.type`, `data_stream.dataset`, `data_stream.namespace`. Instead of serializing these values under the `*attributes.*` namespace, they're put at the root of the document, to conform with the conventions of the data stream naming scheme that maps these as `constant_keyword` fields.
+
     - `raw`: Omit the `Attributes.` string prefixed to field names for log and 
              span attributes as well as omit the `Events.` string prefixed to
              field names for span events. 

exporter/elasticsearchexporter/config.go

@@ -179,6 +179,7 @@ type MappingMode int
 const (
 	MappingNone MappingMode = iota
 	MappingECS
+	MappingOTel
 	MappingRaw
 )
 
@@ -193,6 +194,8 @@ func (m MappingMode) String() string {
 		return ""
 	case MappingECS:
 		return "ecs"
+	case MappingOTel:
+		return "otel"
 	case MappingRaw:
 		return "raw"
 	default:
@@ -205,6 +208,7 @@ var mappingModes = func() map[string]MappingMode {
 	for _, m := range []MappingMode{
 		MappingNone,
 		MappingECS,
+		MappingOTel,
 		MappingRaw,
 	} {
 		table[strings.ToLower(m.String())] = m

exporter/elasticsearchexporter/data_stream_router.go

@@ -16,12 +16,14 @@ func routeWithDefaults(defaultDSType, defaultDSDataset, defaultDSNamespace strin
 	pcommon.Map,
 	pcommon.Map,
 	string,
+	bool,
 ) string {
 	return func(
 		recordAttr pcommon.Map,
 		scopeAttr pcommon.Map,
 		resourceAttr pcommon.Map,
 		fIndex string,
+		otel bool,
 	) string {
 		// Order:
 		// 1. read data_stream.* from attributes
@@ -37,6 +39,13 @@ func routeWithDefaults(defaultDSType, defaultDSDataset, defaultDSNamespace strin
 				return fmt.Sprintf("%s%s%s", prefix, fIndex, suffix)
 			}
 		}
+
+		// The naming convention for datastream is expected to be "logs-[dataset].otel-[namespace]".
+		// This is in order to match the soon to be built-in logs-*.otel-* index template.
+		if otel {
+			dataset += ".otel"
+		}
+
 		recordAttr.PutStr(dataStreamDataset, dataset)
 		recordAttr.PutStr(dataStreamNamespace, namespace)
 		recordAttr.PutStr(dataStreamType, defaultDSType)
@@ -51,9 +60,10 @@ func routeLogRecord(
 	scope pcommon.InstrumentationScope,
 	resource pcommon.Resource,
 	fIndex string,
+	otel bool,
 ) string {
 	route := routeWithDefaults(defaultDataStreamTypeLogs, defaultDataStreamDataset, defaultDataStreamNamespace)
-	return route(record.Attributes(), scope.Attributes(), resource.Attributes(), fIndex)
+	return route(record.Attributes(), scope.Attributes(), resource.Attributes(), fIndex, otel)
 }
 
 // routeDataPoint returns the name of the index to send the data point to according to data stream routing attributes.
@@ -63,9 +73,10 @@ func routeDataPoint(
 	scope pcommon.InstrumentationScope,
 	resource pcommon.Resource,
 	fIndex string,
+	otel bool,
 ) string {
 	route := routeWithDefaults(defaultDataStreamTypeMetrics, defaultDataStreamDataset, defaultDataStreamNamespace)
-	return route(dataPoint.Attributes(), scope.Attributes(), resource.Attributes(), fIndex)
+	return route(dataPoint.Attributes(), scope.Attributes(), resource.Attributes(), fIndex, otel)
 }
 
 // routeSpan returns the name of the index to send the span to according to data stream routing attributes.
@@ -75,7 +86,8 @@ func routeSpan(
 	scope pcommon.InstrumentationScope,
 	resource pcommon.Resource,
 	fIndex string,
+	otel bool,
 ) string {
 	route := routeWithDefaults(defaultDataStreamTypeTraces, defaultDataStreamDataset, defaultDataStreamNamespace)
-	return route(span.Attributes(), scope.Attributes(), resource.Attributes(), fIndex)
+	return route(span.Attributes(), scope.Attributes(), resource.Attributes(), fIndex, otel)
 }

exporter/elasticsearchexporter/data_stream_router_test.go

@@ -0,0 +1,78 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package elasticsearchexporter
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+type routeTestInfo struct {
+	name string
+	otel bool
+	want string
+}
+
+func createRouteTests(dsType string) []routeTestInfo {
+	renderWantRoute := func(dsType string, otel bool) string {
+		if otel {
+			return fmt.Sprintf("%s-%s.otel-%s", dsType, defaultDataStreamDataset, defaultDataStreamNamespace)
+		}
+		return fmt.Sprintf("%s-%s-%s", dsType, defaultDataStreamDataset, defaultDataStreamNamespace)
+	}
+
+	return []routeTestInfo{
+		{
+			name: "default",
+			otel: false,
+			want: renderWantRoute(dsType, false),
+		},
+		{
+			name: "otel",
+			otel: true,
+			want: renderWantRoute(dsType, true),
+		},
+	}
+}
+
+func TestRouteLogRecord(t *testing.T) {
+
+	tests := createRouteTests(defaultDataStreamTypeLogs)
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ds := routeLogRecord(plog.NewLogRecord(), plog.NewScopeLogs().Scope(), plog.NewResourceLogs().Resource(), "", tc.otel)
+			assert.Equal(t, tc.want, ds)
+		})
+	}
+}
+
+func TestRouteDataPoint(t *testing.T) {
+
+	tests := createRouteTests(defaultDataStreamTypeMetrics)
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ds := routeDataPoint(pmetric.NewNumberDataPoint(), plog.NewScopeLogs().Scope(), plog.NewResourceLogs().Resource(), "", tc.otel)
+			assert.Equal(t, tc.want, ds)
+		})
+	}
+}
+
+func TestRouteSpan(t *testing.T) {
+
+	tests := createRouteTests(defaultDataStreamTypeTraces)
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ds := routeSpan(ptrace.NewSpan(), plog.NewScopeLogs().Scope(), plog.NewResourceLogs().Resource(), "", tc.otel)
+			assert.Equal(t, tc.want, ds)
+		})
+	}
+}

exporter/elasticsearchexporter/exporter.go

@@ -30,6 +30,7 @@ type elasticsearchExporter struct {
 	logstashFormat LogstashFormatSettings
 	dynamicIndex   bool
 	model          mappingModel
+	otel           bool
 
 	bulkIndexer bulkIndexer
 }
@@ -49,6 +50,8 @@ func newExporter(
 		mode:  cfg.MappingMode(),
 	}
 
+	otel := model.mode == MappingOTel
+
 	userAgent := fmt.Sprintf(
 		"%s/%s (%s/%s)",
 		set.BuildInfo.Description,
@@ -66,6 +69,7 @@ func newExporter(
 		dynamicIndex:   dynamicIndex,
 		model:          model,
 		logstashFormat: cfg.LogstashFormat,
+		otel:           otel,
 	}, nil
 }
 
@@ -107,7 +111,7 @@ func (e *elasticsearchExporter) pushLogsData(ctx context.Context, ld plog.Logs)
 			scope := ill.Scope()
 			logs := ill.LogRecords()
 			for k := 0; k < logs.Len(); k++ {
-				if err := e.pushLogRecord(ctx, resource, logs.At(k), scope, session); err != nil {
+				if err := e.pushLogRecord(ctx, resource, rl.SchemaUrl(), logs.At(k), scope, ill.SchemaUrl(), session); err != nil {
 					if cerr := ctx.Err(); cerr != nil {
 						return cerr
 					}
@@ -130,13 +134,15 @@ func (e *elasticsearchExporter) pushLogsData(ctx context.Context, ld plog.Logs)
 func (e *elasticsearchExporter) pushLogRecord(
 	ctx context.Context,
 	resource pcommon.Resource,
+	resourceSchemaUrl string,
 	record plog.LogRecord,
 	scope pcommon.InstrumentationScope,
+	scopeSchemaUrl string,
 	bulkIndexerSession bulkIndexerSession,
 ) error {
 	fIndex := e.index
 	if e.dynamicIndex {
-		fIndex = routeLogRecord(record, scope, resource, fIndex)
+		fIndex = routeLogRecord(record, scope, resource, fIndex, e.otel)
 	}
 
 	if e.logstashFormat.Enabled {
@@ -147,7 +153,7 @@ func (e *elasticsearchExporter) pushLogRecord(
 		fIndex = formattedIndex
 	}
 
-	document, err := e.model.encodeLog(resource, record, scope)
+	document, err := e.model.encodeLog(resource, resourceSchemaUrl, record, scope, scopeSchemaUrl)
 	if err != nil {
 		return fmt.Errorf("failed to encode log event: %w", err)
 	}
@@ -279,7 +285,7 @@ func (e *elasticsearchExporter) getMetricDataPointIndex(
 ) (string, error) {
 	fIndex := e.index
 	if e.dynamicIndex {
-		fIndex = routeDataPoint(dataPoint, scope, resource, fIndex)
+		fIndex = routeDataPoint(dataPoint, scope, resource, fIndex, e.otel)
 	}
 
 	if e.logstashFormat.Enabled {
@@ -342,7 +348,7 @@ func (e *elasticsearchExporter) pushTraceRecord(
 ) error {
 	fIndex := e.index
 	if e.dynamicIndex {
-		fIndex = routeSpan(span, scope, resource, fIndex)
+		fIndex = routeSpan(span, scope, resource, fIndex, e.otel)
 	}
 
 	if e.logstashFormat.Enabled {

exporter/elasticsearchexporter/internal/objmodel/objmodel.go

@@ -244,19 +244,19 @@ func (doc *Document) Dedup() {
 // Serialize writes the document to the given writer. The serializer will create nested objects if dedot is true.
 //
 // NOTE: The documented MUST be sorted if dedot is true.
-func (doc *Document) Serialize(w io.Writer, dedot bool) error {
+func (doc *Document) Serialize(w io.Writer, dedot bool, otel bool) error {
 	v := json.NewVisitor(w)
-	return doc.iterJSON(v, dedot)
+	return doc.iterJSON(v, dedot, otel)
 }
 
-func (doc *Document) iterJSON(v *json.Visitor, dedot bool) error {
+func (doc *Document) iterJSON(v *json.Visitor, dedot bool, otel bool) error {
 	if dedot {
-		return doc.iterJSONDedot(v)
+		return doc.iterJSONDedot(v, otel)
 	}
-	return doc.iterJSONFlat(v)
+	return doc.iterJSONFlat(v, otel)
 }
 
-func (doc *Document) iterJSONFlat(w *json.Visitor) error {
+func (doc *Document) iterJSONFlat(w *json.Visitor, otel bool) error {
 	err := w.OnObjectStart(-1, structform.AnyType)
 	if err != nil {
 		return err
@@ -275,15 +275,22 @@ func (doc *Document) iterJSONFlat(w *json.Visitor) error {
 			return err
 		}
 
-		if err := fld.value.iterJSON(w, true); err != nil {
+		if err := fld.value.iterJSON(w, true, otel); err != nil {
 			return err
 		}
 	}
 
 	return nil
 }
 
-func (doc *Document) iterJSONDedot(w *json.Visitor) error {
+// Set of prefixes for the OTel attributes that needs to stay flattened
+var otelPrefixSet = map[string]struct{}{
+	"attributes.":          {},
+	"resource.attributes.": {},
+	"scope.attributes.":    {},
+}
+
+func (doc *Document) iterJSONDedot(w *json.Visitor, otel bool) error {
 	objPrefix := ""
 	level := 0
 
@@ -335,6 +342,16 @@ func (doc *Document) iterJSONDedot(w *json.Visitor) error {
 
 		// increase object level up to current field
 		for {
+
+			// Otel mode serialization
+			if otel {
+				// Check the prefix
+				_, isOtelPrefix := otelPrefixSet[objPrefix]
+				if isOtelPrefix {
+					break
+				}
+			}
+
 			start := len(objPrefix)
 			idx := strings.IndexByte(key[start:], '.')
 			if idx < 0 {
@@ -357,7 +374,7 @@ func (doc *Document) iterJSONDedot(w *json.Visitor) error {
 		if err := w.OnKey(fieldName); err != nil {
 			return err
 		}
-		if err := fld.value.iterJSON(w, true); err != nil {
+		if err := fld.value.iterJSON(w, true, otel); err != nil {
 			return err
 		}
 	}
@@ -460,7 +477,7 @@ func (v *Value) IsEmpty() bool {
 	}
 }
 
-func (v *Value) iterJSON(w *json.Visitor, dedot bool) error {
+func (v *Value) iterJSON(w *json.Visitor, dedot bool, otel bool) error {
 	switch v.kind {
 	case KindNil:
 		return w.OnNil()
@@ -483,13 +500,13 @@ func (v *Value) iterJSON(w *json.Visitor, dedot bool) error {
 		if len(v.doc.fields) == 0 {
 			return w.OnNil()
 		}
-		return v.doc.iterJSON(w, dedot)
+		return v.doc.iterJSON(w, dedot, otel)
 	case KindArr:
 		if err := w.OnArrayStart(-1, structform.AnyType); err != nil {
 			return err
 		}
 		for i := range v.arr {
-			if err := v.arr[i].iterJSON(w, dedot); err != nil {
+			if err := v.arr[i].iterJSON(w, dedot, otel); err != nil {
 				return err
 			}
 		}