Remove batcher and related config in favor of sending queue

.chloggen/esexporter-remove-batcher.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: Remove batcher and related config in favor of sending queue
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [42718]
+
+# (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: Previously deprecated `batcher` configuration is removed. `num_consumers` and `flush` are now deprecated and ignored if defined as they conflict with `sending_queue` configurations.
+
+# 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

@@ -82,52 +82,24 @@ All other defaults are as defined by [confighttp].
 
 ### Queuing and batching
 
-The exporter is transitioning from its own internal batching to OpenTelemetry's standard
-queueing and batching. The below sections describe the current default and the latest
-configuration option for queueing and batching available via the `sending_queue` configuration.
-
-#### Internal batching by Elasticsearch exporter
-
-By default, the exporter will perform its own buffering and batching, as configured through the
-`flush` config. In this case both `sending_queue` and `batcher` will be unused. The exporter
-will perform its own buffering and batching and will issue async requests to Elasticsearch in
-all cases other than if any of the following conditions are met:
-
-- `sending_queue::batch` is defined (irrespective of `sending_queue` being enabled or not)
-- `batcher::enabled` is defined (set to `true` or `false`)
-
-In a future release when the `sending_queue` config is stable, and has feature parity
-with the exporter's existing `flush` config, it will be enabled by default.
-
-Using the `sending_queue` functionality provides several benefits over the default behavior:
- - With a persistent queue, or no queue at all, `sending_queue` enables at least once delivery.
-   On the other hand, with the default behavior, the exporter will accept data and process it
-   asynchronously, which interacts poorly with queueing.
- - By ensuring the exporter makes requests to Elasticsearch synchronously (batching disabled),
-   client metadata can be passed through to Elasticsearch requests,
-   e.g. by using the [`headers_setter` extension](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/extension/headerssetterextension/README.md).
-
-#### Queueing and batching using sending queue
-
 The Elasticsearch exporter supports the common [`sending_queue` settings][exporterhelper] which
-supports both queueing and batching. However, the sending queue is currently disabled by
-default. Sending queue can be enabled by setting `sending_queue::enabled` to `true`. The batching support in sending queue is also disabled by default. Batching can be enabled by defining `sending_queue::batch`.
+supports both queueing and batching. The default sending queue is configured to do async batching
+with the following configuration:
 
-The [`exporterhelper` documentation][exporterhelper] provides more details on the `sending_queue` settings.
-
-#### Deprecated batcher config
-
-> [!WARNING]
-> The `batcher` config is now deprecated and will be removed in an upcoming version. Check the [queueing and batching](#queueing-and-batching) section for using the `sending_queue` setting that supersedes `batcher`. In the interim, `batcher` configurations are still valid, however, they will be ignored if `sending_queue::batch` is defined even if `sending_queue` is not enabled.
-
-The Elasticsearch exporter supports the [common `batcher` settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/internal/queue_sender.go).
+```yaml
+sending_queue:
+  enabled: true
+  sizer: requests
+  num_consumers: 10
+  queue_size: 10
+  batch:
+    flush_timeout: 10s
+    min_size: 1e+6 // 1MB
+    max_size: 5e+6 // 5MB
+    sizer: bytes
+```
 
-- `batcher`:
-  - `enabled` (default=unset): Enable batching of requests into 1 or more bulk requests. On a batcher flush, it is possible for a batched request to be translated to more than 1 bulk request due to `flush::bytes`.
-  - `sizer` (default=items): Unit of `min_size` and `max_size`. Currently supports only "items", in the future will also support "bytes".
-  - `min_size` (default=5000): Minimum batch size to be exported to Elasticsearch, measured in units according to `batcher::sizer`.
-  - `max_size` (default=0): Maximum batch size to be exported to Elasticsearch, measured in units according to `batcher::sizer`. To limit bulk request size, configure `flush::bytes` instead. :warning: It is recommended to keep `max_size` as 0 as a non-zero value may lead to broken metrics grouping and indexing rejections.
-  - `flush_timeout` (default=10s): Maximum time of the oldest item spent inside the batcher buffer, aka "max age of batcher buffer". A batcher flush will happen regardless of the size of content in batcher buffer.
+The default configurations are chosen to be closer to the defaults with the exporter's previous inbuilt batching feature. The [`exporterhelper` documentation][exporterhelper] provides more details on the `sending_queue` settings.
 
 ### Elasticsearch document routing
 
@@ -320,8 +292,8 @@ This can be configured through the following settings:
 The Elasticsearch exporter uses the [Elasticsearch Bulk API] for indexing documents.
 The behaviour of this bulk indexing can be configured with the following settings:
 
-- `num_workers` (default=runtime.NumCPU()): Number of workers publishing bulk requests concurrently. Note this is not applicable if `batcher::enabled` is `true` or `false`.
-- `flush`: Event bulk indexer buffer flush settings
+- `num_workers` (default=runtime.NumCPU()): This config is deprecated now and will be ignored, use the `sending_queue` config. Number of workers publishing bulk requests concurrently. Note this is not applicable if `batcher::enabled` is `true` or `false`.
+- `flush`: This config is deprecated now and will be ignored, use `sending_queue` config. Event bulk indexer buffer flush settings
   - `bytes` (default=5000000): Write buffer flush size limit before compression. A bulk request will be sent immediately when its buffer exceeds this limit. This value should be much lower than [Elasticsearch's `http.max_content_length`](https://www.elastic.co/guide/en/elasticsearch/reference/current/modules-network.html#http-settings) config to avoid HTTP 413 Entity Too Large error. It is recommended to keep this value under 5MB.
   - `interval` (default=10s): Write buffer flush time limit.
 - `retry`: Elasticsearch bulk request retry settings
@@ -331,9 +303,18 @@ The behaviour of this bulk indexing can be configured with the following setting
   - `initial_interval` (default=100ms): Initial waiting time if a HTTP request failed.
   - `max_interval` (default=1m): Max waiting time if a HTTP request failed.
   - `retry_on_status` (default=[429]): Status codes that trigger request or document level retries. Request level retry and document level retry status codes are shared and cannot be configured separately. To avoid duplicates, it defaults to `[429]`.
-
-> [!NOTE]
-> The `flush::interval` config will be ignored when `batcher::enabled` config is explicitly set to `true` or `false`.
+- `sending_queue`: Configures the queueing and batching behaviour. Below are the defaults (which may vary from standard defaults), for full configuration check the [exporterheler docs][exporterhelper].
+  - `enabled` (default=true): Enable queueing and batching behaviour.
+  - `num_consumers` (default=10): Number of consumers that dequeue batches.
+  - `wait_for_result` (default=false): If `true`, blocks incoming requests until processed.
+  - `block_on_overflow` (default=false): If `true`, blocks the request until the queue has space.
+  - `sizer` (default=requests): Measure queueing by requests.
+  - `queue_size` (default=10): Maximum size the queue can accept.
+  - `batch`:
+    - `flush_timeout` (default=10s): Time after which batch is exported irrespective of other settings.
+    - `sizer` (default=bytes): Size batches by bytes.
+    - `min_size` (default=1MB): Min size of the batch.
+    - `max_size` (default=5MB): Max size of the batch.
 
 #### Bulk indexing error response
 

exporter/elasticsearchexporter/bulkindexer.go

@@ -70,11 +70,8 @@ func newBulkIndexer(
 	requireDataStream bool,
 	tb *metadata.TelemetryBuilder,
 	logger *zap.Logger,
-) (bulkIndexer, error) {
-	if config.Batcher.enabledSet || (config.QueueBatchConfig.Enabled && config.QueueBatchConfig.Batch.HasValue()) {
-		return newSyncBulkIndexer(client, config, requireDataStream, tb, logger), nil
-	}
-	return newAsyncBulkIndexer(client, config, requireDataStream, tb, logger)
+) bulkIndexer {
+	return newSyncBulkIndexer(client, config, requireDataStream, tb, logger)
 }
 
 func bulkIndexerConfig(client esapi.Transport, config *Config, requireDataStream bool) docappender.BulkIndexerConfig {
@@ -121,7 +118,6 @@ func newSyncBulkIndexer(
 	return &syncBulkIndexer{
 		config:                bulkIndexerConfig(client, config, requireDataStream),
 		flushTimeout:          config.Timeout,
-		flushBytes:            config.Flush.Bytes,
 		retryConfig:           config.Retry,
 		metadataKeys:          config.MetadataKeys,
 		telemetryBuilder:      tb,
@@ -133,7 +129,6 @@ func newSyncBulkIndexer(
 type syncBulkIndexer struct {
 	config                docappender.BulkIndexerConfig
 	flushTimeout          time.Duration
-	flushBytes            int
 	retryConfig           RetrySettings
 	metadataKeys          []string
 	telemetryBuilder      *metadata.TelemetryBuilder
@@ -185,11 +180,6 @@ func (s *syncBulkIndexerSession) Add(ctx context.Context, index, docID, pipeline
 			getAttributesFromMetadataKeys(ctx, s.s.metadataKeys)...),
 		),
 	)
-	// flush bytes should operate on uncompressed length
-	// as Elasticsearch http.max_content_length measures uncompressed length.
-	if s.bi.UncompressedLen() >= s.s.flushBytes {
-		return s.Flush(ctx)
-	}
 	return nil
 }
 
@@ -237,166 +227,6 @@ func (s *syncBulkIndexerSession) Flush(ctx context.Context) error {
 	}
 }
 
-func newAsyncBulkIndexer(
-	client esapi.Transport,
-	config *Config,
-	requireDataStream bool,
-	tb *metadata.TelemetryBuilder,
-	logger *zap.Logger,
-) (*asyncBulkIndexer, error) {
-	numWorkers := config.NumWorkers
-	if numWorkers == 0 {
-		numWorkers = runtime.NumCPU()
-	}
-
-	pool := &asyncBulkIndexer{
-		wg:               sync.WaitGroup{},
-		items:            make(chan docappender.BulkIndexerItem, config.NumWorkers),
-		telemetryBuilder: tb,
-	}
-	pool.wg.Add(numWorkers)
-
-	for i := 0; i < numWorkers; i++ {
-		bi, err := docappender.NewBulkIndexer(bulkIndexerConfig(client, config, requireDataStream))
-		if err != nil {
-			return nil, err
-		}
-		w := asyncBulkIndexerWorker{
-			indexer:               bi,
-			items:                 pool.items,
-			flushInterval:         config.Flush.Interval,
-			flushTimeout:          config.Timeout,
-			flushBytes:            config.Flush.Bytes,
-			telemetryBuilder:      tb,
-			logger:                logger,
-			failedDocsInputLogger: newFailedDocsInputLogger(logger, config),
-		}
-		go func() {
-			defer pool.wg.Done()
-			w.run()
-		}()
-	}
-	return pool, nil
-}
-
-type asyncBulkIndexer struct {
-	items            chan docappender.BulkIndexerItem
-	wg               sync.WaitGroup
-	telemetryBuilder *metadata.TelemetryBuilder
-}
-
-type asyncBulkIndexerSession struct {
-	*asyncBulkIndexer
-}
-
-// StartSession returns a new asyncBulkIndexerSession.
-func (a *asyncBulkIndexer) StartSession(context.Context) bulkIndexerSession {
-	return asyncBulkIndexerSession{a}
-}
-
-// Close closes the asyncBulkIndexer and any active sessions.
-func (a *asyncBulkIndexer) Close(ctx context.Context) error {
-	close(a.items)
-	doneCh := make(chan struct{})
-	go func() {
-		a.wg.Wait()
-		close(doneCh)
-	}()
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case <-doneCh:
-		return nil
-	}
-}
-
-// Add adds an item to the async bulk indexer session.
-//
-// Adding an item after a call to Close() will panic.
-func (s asyncBulkIndexerSession) Add(ctx context.Context, index, docID, pipeline string, document io.WriterTo, dynamicTemplates map[string]string, action string) error {
-	item := docappender.BulkIndexerItem{
-		Index:            index,
-		Body:             document,
-		DocumentID:       docID,
-		DynamicTemplates: dynamicTemplates,
-		Action:           action,
-		Pipeline:         pipeline,
-	}
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case s.items <- item:
-	}
-	s.telemetryBuilder.ElasticsearchDocsReceived.Add(ctx, 1)
-	return nil
-}
-
-// End is a no-op.
-func (asyncBulkIndexerSession) End() {
-}
-
-// Flush is a no-op.
-func (asyncBulkIndexerSession) Flush(context.Context) error {
-	return nil
-}
-
-type asyncBulkIndexerWorker struct {
-	indexer       *docappender.BulkIndexer
-	items         <-chan docappender.BulkIndexerItem
-	flushInterval time.Duration
-	flushTimeout  time.Duration
-	flushBytes    int
-
-	logger                *zap.Logger
-	failedDocsInputLogger *zap.Logger
-	telemetryBuilder      *metadata.TelemetryBuilder
-}
-
-func (w *asyncBulkIndexerWorker) run() {
-	flushTick := time.NewTicker(w.flushInterval)
-	defer flushTick.Stop()
-	for {
-		select {
-		case item, ok := <-w.items:
-			// if channel is closed, flush and return
-			if !ok {
-				w.flush()
-				return
-			}
-
-			if err := w.indexer.Add(item); err != nil {
-				w.logger.Error("error adding item to bulk indexer", zap.Error(err))
-			}
-
-			// flush bytes should operate on uncompressed length
-			// as Elasticsearch http.max_content_length measures uncompressed length.
-			if w.indexer.UncompressedLen() >= w.flushBytes {
-				w.flush()
-				flushTick.Reset(w.flushInterval)
-			}
-		case <-flushTick.C:
-			// bulk indexer needs to be flushed every flush interval because
-			// there may be pending bytes in bulk indexer buffer due to e.g. document level 429
-			w.flush()
-		}
-	}
-}
-
-func (w *asyncBulkIndexerWorker) flush() {
-	// TODO (lahsivjar): Should use proper context else client metadata will not be accessible
-	ctx := context.Background()
-	// ignore error as we they should be already logged and for async we don't propagate errors
-	_ = flushBulkIndexer(
-		ctx,
-		w.indexer,
-		w.flushTimeout,
-		nil, // async bulk indexer cannot propagate client context/metadata
-		w.telemetryBuilder,
-		w.logger,
-		w.failedDocsInputLogger,
-	)
-}
-
 func flushBulkIndexer(
 	ctx context.Context,
 	bi *docappender.BulkIndexer,
@@ -662,36 +492,20 @@ func (b *bulkIndexers) start(
 	}
 
 	for _, mode := range allowedMappingModes {
-		var bi bulkIndexer
-		bi, err = newBulkIndexer(esClient, cfg, mode == MappingOTel, b.telemetryBuilder, set.Logger)
-		if err != nil {
-			return err
-		}
+		bi := newBulkIndexer(esClient, cfg, mode == MappingOTel, b.telemetryBuilder, set.Logger)
 		b.modes[mode] = &wgTrackingBulkIndexer{bulkIndexer: bi, wg: &b.wg}
 	}
 
-	profilingEvents, err := newBulkIndexer(esClient, cfg, true, b.telemetryBuilder, set.Logger)
-	if err != nil {
-		return err
-	}
+	profilingEvents := newBulkIndexer(esClient, cfg, true, b.telemetryBuilder, set.Logger)
 	b.profilingEvents = &wgTrackingBulkIndexer{bulkIndexer: profilingEvents, wg: &b.wg}
 
-	profilingStackTraces, err := newBulkIndexer(esClient, cfg, false, b.telemetryBuilder, set.Logger)
-	if err != nil {
-		return err
-	}
+	profilingStackTraces := newBulkIndexer(esClient, cfg, false, b.telemetryBuilder, set.Logger)
 	b.profilingStackTraces = &wgTrackingBulkIndexer{bulkIndexer: profilingStackTraces, wg: &b.wg}
 
-	profilingStackFrames, err := newBulkIndexer(esClient, cfg, false, b.telemetryBuilder, set.Logger)
-	if err != nil {
-		return err
-	}
+	profilingStackFrames := newBulkIndexer(esClient, cfg, false, b.telemetryBuilder, set.Logger)
 	b.profilingStackFrames = &wgTrackingBulkIndexer{bulkIndexer: profilingStackFrames, wg: &b.wg}
 
-	profilingExecutables, err := newBulkIndexer(esClient, cfg, false, b.telemetryBuilder, set.Logger)
-	if err != nil {
-		return err
-	}
+	profilingExecutables := newBulkIndexer(esClient, cfg, false, b.telemetryBuilder, set.Logger)
 	b.profilingExecutables = &wgTrackingBulkIndexer{bulkIndexer: profilingExecutables, wg: &b.wg}
 	return nil
 }