[elasticsearchexporter] support dynamic mapping modes (#38114)

#### Description

Add support for controlling the mapping mode via a client metadata key
(typically an HTTP or gRPC header), X-Elastic-Mapping-Mode. This will
override the mode specified in `mapping::mode`, which now serves as the
default.

We also introduce `mapping::allowed_modes` config which restricts
mapping modes to those listed. This will allow an administrator to lock
down the allowed mapping modes that can be specified through client
metadata.

Because we want to set "require_data_stream=true" for otel mode, and we
no longer know ahead of time which mapping mode will be used, we now
create a BulkIndexer per mapping mode.

#### Link to tracking issue

Closes
#36092

#### Testing

- Added unit tests
- Tested against a live collector with telemetrygen, passing
`--otlp-header X-Elastic-Mapping-Mode=\"<mode>\"`

#### Documentation

Updated README.

---------

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

Author: axw

.chloggen/mappingmode-context.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: Support specifying mapping mode via client metadata
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [36092]
+
+# (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: Add config `mapping::allowed_modes` to restrict mapping modes configurable from client metadata.
+
+# 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/README.md

@@ -157,28 +157,41 @@ This can be customised through the following settings:
 The Elasticsearch exporter supports several document schemas and preprocessing
 behaviours, which may be configured through the following settings:
 
-- `mapping`: Events are encoded to JSON. The `mapping` allows users to
-  configure additional mapping rules.
-  - `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.
-      - 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.
-      - `data_stream.dataset` will always be appended with `.otel`. It is recommended to use with `*_dynamic_index.enabled: true` to route documents to data stream `${data_stream.type}-${data_stream.dataset}-${data_stream.namespace}`.
-      - Span events are stored in separate documents. They will be routed with `data_stream.type` set to `logs` if `traces_dynamic_index::enabled` is `true`.
-
-    - `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.
-    - `bodymap`: Provides fine-grained control over the final documents to be ingested.
-            :warning: This mode's behavior is unstable, it is currently experimental and undergoing changes.
-            It works only for logs where the log record body is a map. Each LogRecord
-            body is serialized to JSON as-is and becomes a separate document for ingestion.
-            If the log record body is not a map, the exporter will log a warning and drop the log record.
+- `mapping`:
+  - `mode` (default=none): The default mapping mode. Valid modes are:
+    - `none`
+    - `ecs`
+    - `otel`
+    - `raw`
+    - `bodymap`
+  - `allowed_modes` (defaults to all mapping modes): A list of allowed mapping modes.
+
+The mapping mode can also be controlled via the client metadata key `X-Elastic-Mapping-Mode`,
+e.g. via HTTP headers, gRPC metadata. This will override the configured `mapping::mode`.
+It is possible to restrict which mapping modes may be requested by configuring
+`mapping::allowed_modes`, which defaults to all mapping modes. Keep in mind that not all
+processors or exporter configurations will maintain client
+metadata.
+
+See below for a description of each mapping mode.
 
 #### OTel mapping mode
 
-In `otel` mapping mode, the Elasticsearch Exporter stores documents in an OTel-native schema.
+In `otel` mapping mode, the Elasticsearch Exporter stores documents in Elastic's preferred
+"OTel-native" schema. In this mapping mode, documents use the original attribute names and
+closely follows the event structure from the OTLP events.
+
+There is special treatment for the following attributes: `data_stream.type`, `data_stream.dataset`,
+and `data_stream.namespace`. Instead of serializing these values under the `*attributes.*` namespace,
+they are 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.
+
+`data_stream.dataset` will always be appended with `.otel`. It is recommended to use with
+`*_dynamic_index::enabled: true` (e.g. `logs_dynamic_index::enabled`) to route documents to data stream
+`${data_stream.type}-${data_stream.dataset}-${data_stream.namespace}`.
+
+Span events are stored in separate documents. They will be routed with `data_stream.type` set to
+`logs` if `traces_dynamic_index::enabled` is `true`.
 
 | Signal    | Supported          |
 | --------- | ------------------ |

exporter/elasticsearchexporter/bulkindexer.go

@@ -6,6 +6,7 @@ package elasticsearchexporter // import "github.com/open-telemetry/opentelemetry
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"runtime"
 	"strings"
@@ -15,7 +16,9 @@ import (
 
 	"github.com/elastic/go-docappender/v2"
 	"github.com/elastic/go-elasticsearch/v8/esapi"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configcompression"
+	"go.opentelemetry.io/collector/exporter"
 	"go.uber.org/zap"
 )
 
@@ -362,3 +365,143 @@ func getErrorHint(index, errorType string) string {
 	}
 	return ""
 }
+
+type bulkIndexers struct {
+	// wg tracks active sessions
+	wg sync.WaitGroup
+
+	// NOTE(axw) when we get rid of the async bulk indexer there would be
+	// no reason for having one per mode or for different document types.
+	// Instead, the caller can create separate sessions as needed, and we
+	// can either have one for required_data_stream=true and one for false,
+	// or callers can set this per document.
+
+	modes                [NumMappingModes]bulkIndexer
+	profilingEvents      bulkIndexer // For profiling-events-*
+	profilingStackTraces bulkIndexer // For profiling-stacktraces
+	profilingStackFrames bulkIndexer // For profiling-stackframes
+	profilingExecutables bulkIndexer // For profiling-executables
+}
+
+func (b *bulkIndexers) start(
+	ctx context.Context,
+	cfg *Config,
+	set exporter.Settings,
+	host component.Host,
+	allowedMappingModes map[string]MappingMode,
+) error {
+	userAgent := fmt.Sprintf(
+		"%s/%s (%s/%s)",
+		set.BuildInfo.Description,
+		set.BuildInfo.Version,
+		runtime.GOOS,
+		runtime.GOARCH,
+	)
+
+	esClient, err := newElasticsearchClient(ctx, cfg, host, set.TelemetrySettings, userAgent)
+	if err != nil {
+		return err
+	}
+
+	for _, mode := range allowedMappingModes {
+		var bi bulkIndexer
+		bi, err = newBulkIndexer(set.TelemetrySettings.Logger, esClient, cfg, mode == MappingOTel)
+		if err != nil {
+			return err
+		}
+		b.modes[mode] = &wgTrackingBulkIndexer{bulkIndexer: bi, wg: &b.wg}
+	}
+
+	profilingEvents, err := newBulkIndexer(set.Logger, esClient, cfg, true)
+	if err != nil {
+		return err
+	}
+	b.profilingEvents = &wgTrackingBulkIndexer{bulkIndexer: profilingEvents, wg: &b.wg}
+
+	profilingStackTraces, err := newBulkIndexer(set.Logger, esClient, cfg, false)
+	if err != nil {
+		return err
+	}
+	b.profilingStackTraces = &wgTrackingBulkIndexer{bulkIndexer: profilingStackTraces, wg: &b.wg}
+
+	profilingStackFrames, err := newBulkIndexer(set.Logger, esClient, cfg, false)
+	if err != nil {
+		return err
+	}
+	b.profilingStackFrames = &wgTrackingBulkIndexer{bulkIndexer: profilingStackFrames, wg: &b.wg}
+
+	profilingExecutables, err := newBulkIndexer(set.Logger, esClient, cfg, false)
+	if err != nil {
+		return err
+	}
+	b.profilingExecutables = &wgTrackingBulkIndexer{bulkIndexer: profilingExecutables, wg: &b.wg}
+	return nil
+}
+
+func (b *bulkIndexers) shutdown(ctx context.Context) error {
+	for _, bi := range b.modes {
+		if bi == nil {
+			continue
+		}
+		if err := bi.Close(ctx); err != nil {
+			return err
+		}
+	}
+	if b.profilingEvents != nil {
+		if err := b.profilingEvents.Close(ctx); err != nil {
+			return err
+		}
+	}
+	if b.profilingStackTraces != nil {
+		if err := b.profilingStackTraces.Close(ctx); err != nil {
+			return err
+		}
+	}
+	if b.profilingStackFrames != nil {
+		if err := b.profilingStackFrames.Close(ctx); err != nil {
+			return err
+		}
+	}
+	if b.profilingExecutables != nil {
+		if err := b.profilingExecutables.Close(ctx); err != nil {
+			return err
+		}
+	}
+
+	doneCh := make(chan struct{})
+	go func() {
+		b.wg.Wait()
+		close(doneCh)
+	}()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-doneCh:
+	}
+	return nil
+}
+
+type wgTrackingBulkIndexer struct {
+	bulkIndexer
+	wg *sync.WaitGroup
+}
+
+func (w *wgTrackingBulkIndexer) StartSession(ctx context.Context) (bulkIndexerSession, error) {
+	w.wg.Add(1)
+	session, err := w.bulkIndexer.StartSession(ctx)
+	if err != nil {
+		w.wg.Done()
+		return nil, err
+	}
+	return &wgTrackingBulkIndexerSession{bulkIndexerSession: session, wg: w.wg}, nil
+}
+
+type wgTrackingBulkIndexerSession struct {
+	bulkIndexerSession
+	wg *sync.WaitGroup
+}
+
+func (w *wgTrackingBulkIndexerSession) End() {
+	defer w.wg.Done()
+	w.bulkIndexerSession.End()
+}