[exporter/elasticsearch] OTel mode serialization

exporter/elasticsearchexporter/README.md

@@ -126,6 +126,9 @@ 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`: Try to map fields defined in the
+          [OpenTelemetry Semantic Conventions](https://github.com/open-telemetry/semantic-conventions) (version 1.22.0)
+          to Elastic's preffered "OTel-native" convention. :warning: This mode's behavior is unstable, it is currently is experimental and undergoing changes
     - `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

@@ -81,7 +81,8 @@ type LogstashFormatSettings struct {
 }
 
 type DynamicIndexSetting struct {
-	Enabled bool `mapstructure:"enabled"`
+	Enabled bool   `mapstructure:"enabled"`
+	Mode    string `mapstructure:"mode"` // prefix_suffix or data_stream
 }
 
 // AuthenticationSettings defines user authentication related settings.
@@ -176,6 +177,7 @@ type MappingMode int
 const (
 	MappingNone MappingMode = iota
 	MappingECS
+	MappingOTel
 	MappingRaw
 )
 
@@ -190,6 +192,8 @@ func (m MappingMode) String() string {
 		return ""
 	case MappingECS:
 		return "ecs"
+	case MappingOTel:
+		return "otel"
 	case MappingRaw:
 		return "raw"
 	default:
@@ -202,6 +206,7 @@ var mappingModes = func() map[string]MappingMode {
 	for _, m := range []MappingMode{
 		MappingNone,
 		MappingECS,
+		MappingOTel,
 		MappingRaw,
 	} {
 		table[strings.ToLower(m.String())] = m

exporter/elasticsearchexporter/exporter.go

@@ -24,13 +24,39 @@ type elasticsearchExporter struct {
 	component.TelemetrySettings
 	userAgent string
 
-	config         *Config
-	index          string
-	logstashFormat LogstashFormatSettings
-	dynamicIndex   bool
-	model          mappingModel
+	config           *Config
+	index            string
+	logstashFormat   LogstashFormatSettings
+	dynamicIndex     bool
+	dynamicIndexMode dynIdxMode
 
 	bulkIndexer *esBulkIndexerCurrent
+	model       mappingModel
+	mode        MappingMode
+}
+
+type dynIdxMode int // dynamic index mode
+
+const (
+	dynIdxModePrefixSuffix dynIdxMode = iota
+	dynIdxModeDataStream
+)
+
+const (
+	sDynIdxModePrefixSuffix  = "prefix_suffix"
+	sDynIdxModedimDataStream = "data_stream"
+)
+
+var errUnsupportedDynamicIndexMappingMode = errors.New("unsupported dynamic indexing mode")
+
+func parseDIMode(s string) (dynIdxMode, error) {
+	switch s {
+	case "", sDynIdxModePrefixSuffix:
+		return dynIdxModePrefixSuffix, nil
+	case sDynIdxModedimDataStream:
+		return dynIdxModeDataStream, nil
+	}
+	return dynIdxModePrefixSuffix, errUnsupportedDynamicIndexMappingMode
 }
 
 func newExporter(
@@ -43,6 +69,11 @@ func newExporter(
 		return nil, err
 	}
 
+	dimMode, err := parseDIMode(cfg.LogsDynamicIndex.Mode)
+	if err != nil {
+		return nil, fmt.Errorf("%w: %s", err, cfg.LogsDynamicIndex.Mode)
+	}
+
 	model := &encodeModel{
 		dedup: cfg.Mapping.Dedup,
 		dedot: cfg.Mapping.Dedot,
@@ -61,11 +92,13 @@ func newExporter(
 		TelemetrySettings: set.TelemetrySettings,
 		userAgent:         userAgent,
 
-		config:         cfg,
-		index:          index,
-		dynamicIndex:   dynamicIndex,
-		model:          model,
-		logstashFormat: cfg.LogstashFormat,
+		config:           cfg,
+		index:            index,
+		dynamicIndex:     dynamicIndex,
+		dynamicIndexMode: dimMode,
+		model:            model,
+		mode:             cfg.MappingMode(),
+		logstashFormat:   cfg.LogstashFormat,
 	}, nil
 }
 
@@ -102,7 +135,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); err != nil {
+				if err := e.pushLogRecord(ctx, resource, rl.SchemaUrl(), logs.At(k), scope, ill.SchemaUrl()); err != nil {
 					if cerr := ctx.Err(); cerr != nil {
 						return cerr
 					}
@@ -116,7 +149,12 @@ func (e *elasticsearchExporter) pushLogsData(ctx context.Context, ld plog.Logs)
 	return errors.Join(errs...)
 }
 
-func (e *elasticsearchExporter) pushLogRecord(ctx context.Context, resource pcommon.Resource, record plog.LogRecord, scope pcommon.InstrumentationScope) error {
+func (e *elasticsearchExporter) pushLogRecord(ctx context.Context,
+	resource pcommon.Resource,
+	resourceSchemaUrl string,
+	record plog.LogRecord,
+	scope pcommon.InstrumentationScope,
+	scopeSchemaUrl string) error {
 	fIndex := e.index
 	if e.dynamicIndex {
 		fIndex = routeLogRecord(record, scope, resource, fIndex)
@@ -130,7 +168,7 @@ func (e *elasticsearchExporter) pushLogRecord(ctx context.Context, resource pcom
 		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)
 	}

exporter/elasticsearchexporter/internal/objmodel/objmodel.go

@@ -245,19 +245,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
@@ -276,15 +276,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.":          struct{}{},
+	"resource.attributes.": struct{}{},
+	"scope.attributes.":    struct{}{},
+}
+
+func (doc *Document) iterJSONDedot(w *json.Visitor, otel bool) error {
 	objPrefix := ""
 	level := 0
 
@@ -336,6 +343,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 {
@@ -358,7 +375,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
 		}
 	}
@@ -462,7 +479,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()
@@ -485,13 +502,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
 			}
 		}

exporter/elasticsearchexporter/internal/objmodel/objmodel_test.go

@@ -320,7 +320,7 @@ func TestDocument_Serialize_Flat(t *testing.T) {
 			assert.NoError(t, m.FromRaw(test.attrs))
 			doc := DocumentFromAttributes(m)
 			doc.Dedup()
-			err := doc.Serialize(&buf, false)
+			err := doc.Serialize(&buf, false, false)
 			require.NoError(t, err)
 
 			assert.Equal(t, test.want, buf.String())
@@ -381,14 +381,50 @@ func TestDocument_Serialize_Dedot(t *testing.T) {
 			assert.NoError(t, m.FromRaw(test.attrs))
 			doc := DocumentFromAttributes(m)
 			doc.Dedup()
-			err := doc.Serialize(&buf, true)
+			err := doc.Serialize(&buf, true, false)
 			require.NoError(t, err)
 
 			assert.Equal(t, test.want, buf.String())
 		})
 	}
 }
 
+func TestDocument_Serialize_Otel(t *testing.T) {
+	tests := map[string]struct {
+		attrs map[string]any
+		want  string
+	}{
+		"otel": {
+			attrs: map[string]any{
+				"@timestamp":                        "2024-03-18T21:09:53.645578000Z",
+				"attributes.auditd.log.op":          "PAM:session_open",
+				"attributes.auditd.log.record_type": "USER_START",
+				"attributes.auditd.log.sequence":    6082,
+				"attributes.auditd.log.subj":        "unconfined",
+				"attributes.auditd.log.uid":         "1000",
+				"scope.attributes.bar.one":          "boo",
+				"scope.attributes.foo.two":          "bar",
+				"resource.attributes.blah.num":      234,
+				"resource.attributes.blah.str":      "something",
+			},
+			want: `{"@timestamp":"2024-03-18T21:09:53.645578000Z","attributes":{"auditd.log.op":"PAM:session_open","auditd.log.record_type":"USER_START","auditd.log.sequence":6082,"auditd.log.subj":"unconfined","auditd.log.uid":"1000"},"resource":{"attributes":{"blah.num":234,"blah.str":"something"}},"scope":{"attributes":{"bar.one":"boo","foo.two":"bar"}}}`,
+		},
+	}
+
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			var buf strings.Builder
+			m := pcommon.NewMap()
+			assert.NoError(t, m.FromRaw(test.attrs))
+			doc := DocumentFromAttributes(m)
+			doc.Dedup() // Call Dedup for predictable order
+			err := doc.Serialize(&buf, true, true)
+			require.NoError(t, err)
+			assert.Equal(t, test.want, buf.String())
+		})
+	}
+}
+
 func TestValue_Serialize(t *testing.T) {
 	tests := map[string]struct {
 		value Value
@@ -427,7 +463,7 @@ func TestValue_Serialize(t *testing.T) {
 	for name, test := range tests {
 		t.Run(name, func(t *testing.T) {
 			var buf strings.Builder
-			err := test.value.iterJSON(json.NewVisitor(&buf), false)
+			err := test.value.iterJSON(json.NewVisitor(&buf), false, false)
 			require.NoError(t, err)
 			assert.Equal(t, test.want, buf.String())
 		})