[exporter/opensearch] Add bodymap mode support for logs to opensearch exporter (#42206)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Add support for bodymap for opensearch exporter

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
#41654
<!--Describe what testing was performed and which tests were added.-->
#### Testing

Add unit tests for the bodymap mode 
<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

Author: flomedja

.chloggen/support-bodymap-mapping-mode-for-opensearch.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: opensearchexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for bodymap mapping mode
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [41654]
+
+# (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: The bodymap mapping mode supports only logs and uses the body of a log record as the exact content of the OpenSearch document, without any transformation.
+
+# 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/opensearchexporter/README.md

@@ -93,6 +93,70 @@ This configuration will create:
 
 If `service.name` is missing, the fallback value is used (e.g., `otel-logs-default-service-2024.06.07`).
 
+### OpenSearch document mapping
+
+
+The mapping mode can be controlled via the scope attribute `opensearch.mapping.mode`. 
+
+The OpenSearch exporter supports several document schemas and preprocessing behaviors, which may be configured through the following settings:
+
+- `mapping`:
+  - `mode` (default=`ss4o`): Configures the field mappings. Supported modes are:
+    - `ss4o`: Exports logs in the [Simple Schema for Observability](https://opensearch.org/docs/latest/observing-your-data/ss4o/) standard.
+    - `ecs`: Maps fields defined in the OpenTelemetry Semantic Conventions to the [Elastic Common Schema](https://www.elastic.co/guide/en/ecs/current/index.html)
+    - `flatten_attributes`: Uses the ECS mapping but flattens all resource and log attributes in the record to the top-level.
+    - `bodymap`: uses the "body" of a log record as the exact content of the OpenSearch document, without any transformation. This mapping mode is intended for use cases where the client wishes to have complete control over the OpenSearch document structure.
+  - `timestamp_field`: (optional) Field to store the timestamp in. If not set, uses the default `@timestamp`.
+  - `unix_timestamp`: (optional) Whether to store the timestamp in epoch milliseconds.
+  - `dedup`: (optional) removes fields from the document, that have duplicate keys. The filtering only keeps the last value for a key.
+  - `dedot`: (optional) convert dotted keys into nested JSON objects.
+
+#### SS4O mapping mode
+
+The default [`Simple Schema for Observability`](https://docs.opensearch.org/latest/observing-your-data/ss4o/) mapping mode.
+
+In `ss4o` mapping mode, the OpenSearch exporter stores documents using the SS4O schema, which is designed for observability data in OpenSearch. Documents use standardized field names and structure to facilitate integration with OpenSearch dashboards and tools.
+
+| Signal    | Supported          |
+| --------- | ------------------ |
+| Logs      | :white_check_mark: |
+| Traces    | :white_check_mark: |
+
+#### ECS mapping mode
+
+> [!WARNING]
+> The ECS mapping mode is currently undergoing changes, and its behaviour is unstable.
+
+In `ecs` mapping mode, the OpenSearch exporter maps fields from [OpenTelemetry Semantic Conventions](https://opentelemetry.io/docs/reference/specification/trace/semantic_conventions/) to [Elastic Common Schema (ECS)](https://www.elastic.co/guide/en/ecs/current/index.html) where possible. This mode may be used for compatibility with dashboards and tools that expect ECS.
+
+| Signal    | `ecs`              |
+| --------- | ------------------ |
+| Logs      | :white_check_mark: |
+| Traces    | :no_entry_sign:    |
+
+#### Flatten attributes mapping mode
+
+> [!WARNING]
+> The Flatten attributes mapping mode is currently undergoing changes, and its behaviour is unstable.
+
+In `flatten_attributes` mapping mode, the OpenSearch exporter uses the ECS mapping but flattens all resource and log attributes in the record to the top-level of the document.
+
+| Signal    | `flatten_attributes` |
+| --------- | -------------------- |
+| Logs      | :white_check_mark:   |
+| Traces    | :no_entry_sign:      |
+
+#### Bodymap mapping mode
+
+In `bodymap` mapping mode, the OpenSearch exporter supports only logs and uses the "body" of a log record as the exact content of the OpenSearch document, without any transformation. This mapping mode is intended for use cases where the client wishes to have complete control over the OpenSearch document structure.
+
+The bodymap mapping mode only supports log records where the body is of type Map. If the log body is not a Map, encoding will fail with an error. This ensures that only structured map data can be used as the document content in bodymap mode.
+
+| Signal    | `bodymap`           |
+| --------- | ------------------- |
+| Logs      | :white_check_mark:  |
+| Traces    | :no_entry_sign:     |
+
 ### HTTP Connection Options
 
 OpenSearch export supports standard [HTTP client settings](https://github.com/open-telemetry/opentelemetry-collector/tree/main/config/confighttp#client-configuration).

exporter/opensearchexporter/config.go

@@ -85,6 +85,11 @@ type MappingsSettings struct {
 	//
 	//   flatten_attributes: uses the ECS mapping but flattens all resource and
 	//   log attributes in the record to the top-level.
+	//
+	//   bodymap: supports only logs and uses the "body" of a log record as the exact content
+	//   of the OpenSearch document, without any transformation.
+	//   This mapping mode is intended for use cases where the client wishes to have complete control over the
+	//   OpenSearch document structure.
 	Mode string `mapstructure:"mode"`
 
 	// Additional field mappings.
@@ -112,6 +117,7 @@ const (
 	MappingSS4O MappingMode = iota
 	MappingECS
 	MappingFlattenAttributes
+	MappingBodyMap
 )
 
 func (m MappingMode) String() string {
@@ -122,6 +128,8 @@ func (m MappingMode) String() string {
 		return "ecs"
 	case MappingFlattenAttributes:
 		return "flatten_attributes"
+	case MappingBodyMap:
+		return "bodymap"
 	default:
 		return "ss4o"
 	}
@@ -133,6 +141,7 @@ var mappingModes = func() map[string]MappingMode {
 		MappingECS,
 		MappingSS4O,
 		MappingFlattenAttributes,
+		MappingBodyMap,
 	} {
 		table[strings.ToLower(m.String())] = m
 	}

exporter/opensearchexporter/encoder.go

@@ -6,15 +6,21 @@ package opensearchexporter // import "github.com/open-telemetry/opentelemetry-co
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
+	"fmt"
 	"time"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter/internal/objmodel"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter/internal/pool"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter/internal/serializer"
 )
 
+var errInvalidTypeForBodyMapMode = errors.New("invalid log record body type for 'bodymap' mapping mode")
+
 type mappingModel interface {
 	encodeLog(resource pcommon.Resource,
 		scope pcommon.InstrumentationScope,
@@ -26,6 +32,39 @@ type mappingModel interface {
 		record ptrace.Span) ([]byte, error)
 }
 
+type bodyMapMappingModel struct {
+	bufferPool *pool.BufferPool
+}
+
+func (*bodyMapMappingModel) encodeTrace(
+	_ pcommon.Resource,
+	_ pcommon.InstrumentationScope,
+	_ string,
+	_ ptrace.Span,
+) ([]byte, error) {
+	return nil, fmt.Errorf("mapping mode '%s' does not support encoding traces", MappingBodyMap.String())
+}
+
+func (m *bodyMapMappingModel) encodeLog(
+	_ pcommon.Resource,
+	_ pcommon.InstrumentationScope,
+	_ string,
+	record plog.LogRecord,
+) ([]byte, error) {
+	body := record.Body()
+	if body.Type() != pcommon.ValueTypeMap {
+		return nil, fmt.Errorf("%w: %q", errInvalidTypeForBodyMapMode, body.Type().String())
+	}
+	pooledBuf := m.bufferPool.NewPooledBuffer()
+	defer pooledBuf.Recycle()
+
+	serializer.Map(body.Map(), pooledBuf.Buffer)
+	// Copy bytes to avoid holding reference to pooled buffer
+	result := make([]byte, pooledBuf.Buffer.Len())
+	copy(result, pooledBuf.Buffer.Bytes())
+	return result, nil
+}
+
 // encodeModel supports multiple encoding OpenTelemetry signals to multiple schemas.
 type encodeModel struct {
 	dedup             bool