[elasticsearchexporter] Consistently store the structured body of log…

…s and events in `otel` mode (#37387)

The OTel community has reached a consensus that all structured data for
events defined by semantic conventions should be stored in attributes
(open-telemetry/semantic-conventions#1651 (comment)).
The body for events may contain opaque/external data, which sounds like
a job for the flattened field type.

Therefore, it no longer makes sense to map the body for logs and events
differently.

The corresponding Elasticsearch mapping change has been merged already:
elastic/elasticsearch#120547

---------

Co-authored-by: Carson Ip <[email protected]>

.chloggen/elasticsearchexporter_log_body.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: breaking
+
+# 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: Consistently store the structured body of logs and events in `body.structured` in `otel` mode
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37387]
+
+# (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: [user]

exporter/elasticsearchexporter/exporter_test.go

@@ -438,7 +438,7 @@ func TestExporterLogs(t *testing.T) {
 					m.PutEmptyMap("inner").PutStr("foo", "bar")
 					return vm
 				}(),
-				wantDocument: []byte(`{"@timestamp":"0.0","attributes":{"attr.foo":"attr.foo.value"},"data_stream":{"dataset":"attr.dataset.otel","namespace":"resource.attribute.namespace","type":"logs"},"observed_timestamp":"0.0","resource":{"attributes":{"resource.attr.foo":"resource.attr.foo.value"}},"scope":{},"body":{"flattened":{"true":true,"false":false,"inner":{"foo":"bar"}}}}`),
+				wantDocument: []byte(`{"@timestamp":"0.0","attributes":{"attr.foo":"attr.foo.value"},"data_stream":{"dataset":"attr.dataset.otel","namespace":"resource.attribute.namespace","type":"logs"},"observed_timestamp":"0.0","resource":{"attributes":{"resource.attr.foo":"resource.attr.foo.value"}},"scope":{},"body":{"structured":{"true":true,"false":false,"inner":{"foo":"bar"}}}}`),
 			},
 			{
 				body: func() pcommon.Value {
@@ -461,7 +461,7 @@ func TestExporterLogs(t *testing.T) {
 					s.AppendEmpty().SetEmptyMap().PutStr("foo", "bar")
 					return vs
 				}(),
-				wantDocument: []byte(`{"@timestamp":"0.0","attributes":{"attr.foo":"attr.foo.value"},"data_stream":{"dataset":"attr.dataset.otel","namespace":"resource.attribute.namespace","type":"logs"},"observed_timestamp":"0.0","resource":{"attributes":{"resource.attr.foo":"resource.attr.foo.value"}},"scope":{},"body":{"flattened":{"value":["foo",false,{"foo":"bar"}]}}}`),
+				wantDocument: []byte(`{"@timestamp":"0.0","attributes":{"attr.foo":"attr.foo.value"},"data_stream":{"dataset":"attr.dataset.otel","namespace":"resource.attribute.namespace","type":"logs"},"observed_timestamp":"0.0","resource":{"attributes":{"resource.attr.foo":"resource.attr.foo.value"}},"scope":{},"body":{"structured":{"value":["foo",false,{"foo":"bar"}]}}}`),
 			},
 			{
 				body: func() pcommon.Value {

exporter/elasticsearchexporter/pdata_serializer.go

@@ -200,17 +200,14 @@ func serializeLog(resource pcommon.Resource, resourceSchemaURL string, scope pco
 	writeSpanIDField(v, "span_id", record.SpanID())
 	writeAttributes(v, record.Attributes(), false)
 	writeIntFieldSkipDefault(v, "dropped_attributes_count", int64(record.DroppedAttributesCount()))
-	isEvent := false
 	if record.EventName() != "" {
-		isEvent = true
 		writeStringFieldSkipDefault(v, "event_name", record.EventName())
 	} else if eventNameAttr, ok := record.Attributes().Get("event.name"); ok && eventNameAttr.Str() != "" {
-		isEvent = true
 		writeStringFieldSkipDefault(v, "event_name", eventNameAttr.Str())
 	}
 	writeResource(v, resource, resourceSchemaURL, false)
 	writeScope(v, scope, scopeSchemaURL, false)
-	writeLogBody(v, record, isEvent)
+	writeLogBody(v, record)
 	_ = v.OnObjectFinished()
 	return nil
 }
@@ -227,21 +224,14 @@ func writeDataStream(v *json.Visitor, idx esIndex) {
 	_ = v.OnObjectFinished()
 }
 
-func writeLogBody(v *json.Visitor, record plog.LogRecord, isEvent bool) {
+func writeLogBody(v *json.Visitor, record plog.LogRecord) {
 	if record.Body().Type() == pcommon.ValueTypeEmpty {
 		return
 	}
 	_ = v.OnKey("body")
 	_ = v.OnObjectStart(-1, structform.AnyType)
 
-	// Determine if this log record is an event, as they are mapped differently
-	// https://github.com/open-telemetry/semantic-conventions/blob/main/docs/general/events.md
-	var bodyType string
-	if isEvent {
-		bodyType = "structured"
-	} else {
-		bodyType = "flattened"
-	}
+	bodyType := "structured"
 	body := record.Body()
 	switch body.Type() {
 	case pcommon.ValueTypeMap:

exporter/elasticsearchexporter/pdata_serializer_test.go

@@ -103,7 +103,7 @@ func TestSerializeLog(t *testing.T) {
 				"resource":           map[string]any{},
 				"scope":              map[string]any{},
 				"body": map[string]any{
-					"flattened": map[string]any{
+					"structured": map[string]any{
 						"foo.bar": "baz",
 					},
 				},