elasticsearchexporter: mapping mode from scope

.chloggen/elasticsearchexporter-scopeattr-mappingmode.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: Add support for extracting mapping mode from a scope attribute.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39110]
+
+# (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/README.md

@@ -189,8 +189,12 @@ The mapping mode can also be controlled via the client metadata key `X-Elastic-M
 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.
+processors or exporter configurations will maintain client metadata.
+
+Finally, the mapping mode can be controlled via the scope attribute `elastic.mapping.mode`.
+If specified, this takes precedence over the `X-Elastic-Mapping-Mode` client metadata.
+If any scope has an invalid mapping mode, the exporter will reject the entire batch.
+The attribute will be excluded from the final document.
 
 See below for a description of each mapping mode.
 

exporter/elasticsearchexporter/bulkindexer.go

@@ -26,7 +26,7 @@ import (
 
 type bulkIndexer interface {
 	// StartSession starts a new bulk indexing session.
-	StartSession(context.Context) (bulkIndexerSession, error)
+	StartSession(context.Context) bulkIndexerSession
 
 	// Close closes the bulk indexer, ending any in-progress
 	// sessions and stopping any background processing.
@@ -111,15 +111,16 @@ type syncBulkIndexer struct {
 
 // StartSession creates a new docappender.BulkIndexer, and wraps
 // it with a syncBulkIndexerSession.
-func (s *syncBulkIndexer) StartSession(context.Context) (bulkIndexerSession, error) {
+func (s *syncBulkIndexer) StartSession(context.Context) bulkIndexerSession {
 	bi, err := docappender.NewBulkIndexer(s.config)
 	if err != nil {
-		return nil, err
+		// This should never happen in practice:
+		// NewBulkIndexer should only fail if the
+		// config is invalid, and we expect it to
+		// always be valid at this point.
+		return errBulkIndexerSession{err: err}
 	}
-	return &syncBulkIndexerSession{
-		s:  s,
-		bi: bi,
-	}, nil
+	return &syncBulkIndexerSession{s: s, bi: bi}
 }
 
 // Close is a no-op.
@@ -241,8 +242,8 @@ type asyncBulkIndexerSession struct {
 }
 
 // StartSession returns a new asyncBulkIndexerSession.
-func (a *asyncBulkIndexer) StartSession(context.Context) (bulkIndexerSession, error) {
-	return asyncBulkIndexerSession{a}, nil
+func (a *asyncBulkIndexer) StartSession(context.Context) bulkIndexerSession {
+	return asyncBulkIndexerSession{a}
 }
 
 // Close closes the asyncBulkIndexer and any active sessions.
@@ -508,14 +509,10 @@ type wgTrackingBulkIndexer struct {
 	wg *sync.WaitGroup
 }
 
-func (w *wgTrackingBulkIndexer) StartSession(ctx context.Context) (bulkIndexerSession, error) {
+func (w *wgTrackingBulkIndexer) StartSession(ctx context.Context) bulkIndexerSession {
 	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
+	session := w.bulkIndexer.StartSession(ctx)
+	return &wgTrackingBulkIndexerSession{bulkIndexerSession: session, wg: w.wg}
 }
 
 type wgTrackingBulkIndexerSession struct {
@@ -527,3 +524,17 @@ func (w *wgTrackingBulkIndexerSession) End() {
 	defer w.wg.Done()
 	w.bulkIndexerSession.End()
 }
+
+type errBulkIndexerSession struct {
+	err error
+}
+
+func (s errBulkIndexerSession) Add(context.Context, string, string, string, io.WriterTo, map[string]string, string) error {
+	return fmt.Errorf("creating bulk indexer session failed, cannot add item: %w", s.err)
+}
+
+func (s errBulkIndexerSession) End() {}
+
+func (s errBulkIndexerSession) Flush(context.Context) error {
+	return fmt.Errorf("creating bulk indexer session failed, cannot flush: %w", s.err)
+}

exporter/elasticsearchexporter/bulkindexer_test.go

@@ -101,13 +101,14 @@ func TestAsyncBulkIndexer_flush(t *testing.T) {
 
 			bulkIndexer, err := newAsyncBulkIndexer(zap.NewNop(), client, &tt.config, false)
 			require.NoError(t, err)
-			session, err := bulkIndexer.StartSession(context.Background())
-			require.NoError(t, err)
 
+			session := bulkIndexer.StartSession(context.Background())
 			assert.NoError(t, session.Add(context.Background(), "foo", "", "", strings.NewReader(`{"foo": "bar"}`), nil, docappender.ActionCreate))
 			// should flush
 			time.Sleep(100 * time.Millisecond)
 			assert.Equal(t, int64(1), bulkIndexer.stats.docsIndexed.Load())
+			assert.NoError(t, session.Flush(context.Background()))
+			session.End()
 			assert.NoError(t, bulkIndexer.Close(context.Background()))
 		})
 	}
@@ -201,9 +202,7 @@ func TestAsyncBulkIndexer_flush_error(t *testing.T) {
 			require.NoError(t, err)
 			defer bulkIndexer.Close(context.Background())
 
-			session, err := bulkIndexer.StartSession(context.Background())
-			require.NoError(t, err)
-
+			session := bulkIndexer.StartSession(context.Background())
 			assert.NoError(t, session.Add(context.Background(), "foo", "", "", strings.NewReader(`{"foo": "bar"}`), nil, docappender.ActionCreate))
 			// should flush
 			time.Sleep(100 * time.Millisecond)
@@ -213,6 +212,8 @@ func TestAsyncBulkIndexer_flush_error(t *testing.T) {
 			for _, wantField := range tt.wantFields {
 				assert.Equal(t, 1, messages.FilterField(wantField).Len(), "message with field not found; observed.All()=%v", observed.All())
 			}
+			assert.NoError(t, session.Flush(context.Background()))
+			session.End()
 		})
 	}
 }
@@ -288,10 +289,11 @@ func TestAsyncBulkIndexer_logRoundTrip(t *testing.T) {
 func runBulkIndexerOnce(t *testing.T, config *Config, client *elasticsearch.Client) *asyncBulkIndexer {
 	bulkIndexer, err := newAsyncBulkIndexer(zap.NewNop(), client, config, false)
 	require.NoError(t, err)
-	session, err := bulkIndexer.StartSession(context.Background())
-	require.NoError(t, err)
 
+	session := bulkIndexer.StartSession(context.Background())
 	assert.NoError(t, session.Add(context.Background(), "foo", "", "", strings.NewReader(`{"foo": "bar"}`), nil, docappender.ActionCreate))
+	assert.NoError(t, session.Flush(context.Background()))
+	session.End()
 	assert.NoError(t, bulkIndexer.Close(context.Background()))
 
 	return bulkIndexer
@@ -315,10 +317,11 @@ func TestSyncBulkIndexer_flushBytes(t *testing.T) {
 	require.NoError(t, err)
 
 	bi := newSyncBulkIndexer(zap.NewNop(), client, &cfg, false)
-	session, err := bi.StartSession(context.Background())
-	require.NoError(t, err)
 
+	session := bi.StartSession(context.Background())
 	assert.NoError(t, session.Add(context.Background(), "foo", "", "", strings.NewReader(`{"foo": "bar"}`), nil, docappender.ActionCreate))
 	assert.Equal(t, int64(1), reqCnt.Load()) // flush due to flush::bytes
+	assert.NoError(t, session.Flush(context.Background()))
+	session.End()
 	assert.NoError(t, bi.Close(context.Background()))
 }

exporter/elasticsearchexporter/config.go

@@ -207,8 +207,12 @@ type RetrySettings struct {
 type MappingsSettings struct {
 	// Mode configures the default document mapping mode.
 	//
-	// The mode may be overridden by the client metadata key
-	// X-Elastic-Mapping-Mode, if specified.
+	// The mode may be overridden in two ways:
+	//  - by the client metadata key X-Elastic-Mapping-Mode, if specified
+	//  - by the scope attribute elastic.mapping.mode, if specified
+	//
+	// The order of precedence is:
+	//   scope attribute > client metadata > default mode.
 	Mode string `mapstructure:"mode"`
 
 	// AllowedModes controls the allowed document mapping modes