Remove batcher and related config in favor of sending queue

exporter/elasticsearchexporter/README.md

@@ -94,12 +94,12 @@ sending_queue:
   queue_size: 10
   batch:
     flush_timeout: 10s
-    min_size: 5e+6 // 5MB
-    max_size: 10e+6 // 10MB
+    min_size: 1e+6 // 1MB
+    max_size: 5e+6 // 5MB
     sizer: bytes
 ```
 
-The default configurations are chosen to be closer to the previous defaults with the exporter's inbuilt batching. The [`exporterhelper` documentation][exporterhelper] provides more details on the `sending_queue` settings.
+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
 
@@ -292,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
@@ -303,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/config.go

@@ -476,6 +476,12 @@ func handleDeprecatedConfig(cfg *Config, logger *zap.Logger) {
 	if cfg.TracesDynamicIndex.Enabled {
 		logger.Warn("traces_dynamic_index::enabled has been deprecated, and will be removed in a future version. It is now a no-op. Dynamic document routing is now the default. See Elasticsearch Exporter README.")
 	}
+	if cfg.Flush.Bytes > 0 || cfg.Flush.Interval > 0 {
+		logger.Warn("flush settings are now deprecated and ignored. Please use `sending_queue` instead.")
+	}
+	if cfg.NumWorkers > 0 {
+		logger.Warn("`num_workers` are now deprecated and ignored. Please use `sending_queue` instead.")
+	}
 }
 
 func handleTelemetryConfig(cfg *Config, logger *zap.Logger) {

exporter/elasticsearchexporter/config_test.go

@@ -65,8 +65,8 @@ func TestConfig(t *testing.T) {
 					Batch: configoptional.Some(exporterhelper.BatchConfig{
 						FlushTimeout: 10 * time.Second,
 						Sizer:        exporterhelper.RequestSizerTypeBytes,
-						MinSize:      5000000,
-						MaxSize:      10000000,
+						MinSize:      1000000,
+						MaxSize:      5000000,
 					}),
 				},
 				Endpoints: []string{
@@ -148,8 +148,8 @@ func TestConfig(t *testing.T) {
 					Batch: configoptional.Some(exporterhelper.BatchConfig{
 						FlushTimeout: 10 * time.Second,
 						Sizer:        exporterhelper.RequestSizerTypeBytes,
-						MinSize:      5000000,
-						MaxSize:      10000000,
+						MinSize:      1000000,
+						MaxSize:      5000000,
 					}),
 				},
 				Endpoints: []string{"http://localhost:9200"},
@@ -221,8 +221,8 @@ func TestConfig(t *testing.T) {
 					Batch: configoptional.Some(exporterhelper.BatchConfig{
 						FlushTimeout: 10 * time.Second,
 						Sizer:        exporterhelper.RequestSizerTypeBytes,
-						MinSize:      5000000,
-						MaxSize:      10000000,
+						MinSize:      1000000,
+						MaxSize:      5000000,
 					}),
 				},
 				Endpoints: []string{"http://localhost:9200"},

exporter/elasticsearchexporter/factory.go

@@ -46,8 +46,8 @@ func createDefaultConfig() component.Config {
 	qs.QueueSize = 10
 	qs.Batch = configoptional.Some(exporterhelper.BatchConfig{
 		FlushTimeout: 10 * time.Second,
-		MinSize:      5e+6,
-		MaxSize:      10e+6,
+		MinSize:      1e+6,
+		MaxSize:      5e+6,
 		Sizer:        exporterhelper.RequestSizerTypeBytes,
 	})
 
@@ -220,19 +220,11 @@ func exporterhelperOptions(
 ) []exporterhelper.Option {
 	// not setting capabilities as they will default to non-mutating but will be updated
 	// by the base-exporter to mutating if batching is enabled.
-	opts := []exporterhelper.Option{
+	return []exporterhelper.Option{
 		exporterhelper.WithStart(start),
 		exporterhelper.WithShutdown(shutdown),
-		exporterhelper.WithQueueBatch(cfg.QueueBatchConfig, qbs),
+		xexporterhelper.WithQueueBatch(cfg.QueueBatchConfig, qbs),
+		// Effectively disable timeout_sender because timeout is enforced in bulk indexer.
+		exporterhelper.WithTimeout(exporterhelper.TimeoutConfig{Timeout: 0}),
 	}
-
-	// Effectively disable timeout_sender because timeout is enforced in bulk indexer.
-	//
-	// We keep timeout_sender enabled in the async mode (sending_queue not enabled OR sending
-	// queue enabled but batching not enabled OR based on the deprecated batcher setting), to
-	// ensure sending data to the background workers will not block indefinitely.
-	if cfg.QueueBatchConfig.Enabled {
-		opts = append(opts, exporterhelper.WithTimeout(exporterhelper.TimeoutConfig{Timeout: 0}))
-	}
-	return opts
 }

exporter/elasticsearchexporter/go.mod

@@ -26,7 +26,6 @@ require (
 	go.opentelemetry.io/collector/config/configoptional v0.136.0
 	go.opentelemetry.io/collector/confmap v1.42.0
 	go.opentelemetry.io/collector/confmap/xconfmap v0.136.0
-	go.opentelemetry.io/collector/consumer v1.42.0
 	go.opentelemetry.io/collector/consumer/consumererror v0.136.0
 	go.opentelemetry.io/collector/exporter v1.42.0
 	go.opentelemetry.io/collector/exporter/exporterhelper v0.136.0
@@ -88,6 +87,7 @@ require (
 	go.opentelemetry.io/collector/config/configmiddleware v1.42.0 // indirect
 	go.opentelemetry.io/collector/config/configretry v1.42.0 // indirect
 	go.opentelemetry.io/collector/config/configtls v1.42.0 // indirect
+	go.opentelemetry.io/collector/consumer v1.42.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror/xconsumererror v0.136.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumertest v0.136.0 // indirect
 	go.opentelemetry.io/collector/consumer/xconsumer v0.136.0 // indirect

exporter/elasticsearchexporter/integrationtest/datareceiver.go

@@ -67,7 +67,7 @@ type esDataReceiver struct {
 	receiver          receiver.Logs
 	endpoint          string
 	decodeBulkRequest bool
-	batcherEnabled    *bool
+	enableBatching    bool
 	t                 testing.TB
 }
 
@@ -92,9 +92,9 @@ func withDecodeBulkRequest(decode bool) dataReceiverOption {
 	}
 }
 
-func withBatcherEnabled(enabled bool) dataReceiverOption {
+func withBatching(enabled bool) dataReceiverOption {
 	return func(r *esDataReceiver) {
-		r.batcherEnabled = &enabled
+		r.enableBatching = enabled
 	}
 }
 
@@ -153,9 +153,6 @@ func (es *esDataReceiver) GenConfigYAMLStr() string {
     logs_index: %s
     metrics_index: %s
     traces_index: %s
-    sending_queue:
-      enabled: true
-      block_on_overflow: true
     mapping:
       mode: otel
     retry:
@@ -169,17 +166,24 @@ func (es *esDataReceiver) GenConfigYAMLStr() string {
 		es.endpoint, TestLogsIndex, TestMetricsIndex, TestTracesIndex,
 	)
 
-	if es.batcherEnabled == nil {
+	if es.enableBatching {
 		cfgFormat += `
-    flush:
-      interval: 1s`
+    sending_queue:
+      enabled: true
+      block_on_overflow: true
+      batch:
+        flush_timeout: 1s
+        sizer: bytes`
 	} else {
-		cfgFormat += fmt.Sprintf(`
-    batcher:
-      flush_timeout: 1s
-      enabled: %v`,
-			*es.batcherEnabled,
-		)
+		// Batching is disabled using `min_size` as we are setting batching
+		// as a default behavior.
+		cfgFormat += `
+    sending_queue:
+      enabled: true
+      block_on_overflow: true
+      batch:
+        min_size: 0
+        sizer: bytes`
 	}
 	return cfgFormat + "\n"
 }

exporter/elasticsearchexporter/integrationtest/exporter_test.go

@@ -21,11 +21,7 @@ func TestExporter(t *testing.T) {
 		for _, tc := range []struct {
 			name string
 
-			// batcherEnabled enables/disables the batch sender. If this is
-			// nil, then the exporter buffers data itself (legacy behavior),
-			// whereas if it is non-nil then the exporter will not perform
-			// any buffering itself.
-			batcherEnabled *bool
+			enableBatching bool
 
 			// restartCollector restarts the OTEL collector. Restarting
 			// the collector allows durability testing of the ES exporter
@@ -37,24 +33,20 @@ func TestExporter(t *testing.T) {
 			{name: "es_intermittent_http_error", mockESErr: errElasticsearch{httpStatus: http.StatusServiceUnavailable}},
 			{name: "es_intermittent_doc_error", mockESErr: errElasticsearch{httpStatus: http.StatusOK, httpDocStatus: http.StatusTooManyRequests}},
 
-			{name: "batcher_enabled", batcherEnabled: ptrTo(true)},
-			{name: "batcher_enabled_es_intermittent_http_error", batcherEnabled: ptrTo(true), mockESErr: errElasticsearch{httpStatus: http.StatusServiceUnavailable}},
-			{name: "batcher_enabled_es_intermittent_doc_error", batcherEnabled: ptrTo(true), mockESErr: errElasticsearch{httpStatus: http.StatusOK, httpDocStatus: http.StatusTooManyRequests}},
-			{name: "batcher_disabled", batcherEnabled: ptrTo(false)},
-			{name: "batcher_disabled_es_intermittent_http_error", batcherEnabled: ptrTo(false), mockESErr: errElasticsearch{httpStatus: http.StatusServiceUnavailable}},
-			{name: "batcher_disabled_es_intermittent_doc_error", batcherEnabled: ptrTo(false), mockESErr: errElasticsearch{httpStatus: http.StatusOK, httpDocStatus: http.StatusTooManyRequests}},
+			{name: "enable sending_queue batching", enableBatching: true},
+			{name: "batcher_enabled_es_intermittent_http_error", enableBatching: true, mockESErr: errElasticsearch{httpStatus: http.StatusServiceUnavailable}},
+			{name: "batcher_enabled_es_intermittent_doc_error", enableBatching: true, mockESErr: errElasticsearch{httpStatus: http.StatusOK, httpDocStatus: http.StatusTooManyRequests}},
+			{name: "batcher_disabled", enableBatching: false},
+			{name: "batcher_disabled_es_intermittent_http_error", enableBatching: false, mockESErr: errElasticsearch{httpStatus: http.StatusServiceUnavailable}},
+			{name: "batcher_disabled_es_intermittent_doc_error", enableBatching: false, mockESErr: errElasticsearch{httpStatus: http.StatusOK, httpDocStatus: http.StatusTooManyRequests}},
 
 			/* TODO: Below tests should be enabled after https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/30792 is fixed
 			{name: "collector_restarts", restartCollector: true},
 			{name: "collector_restart_with_es_intermittent_failure", mockESErr: true, restartCollector: true},
 			*/
 		} {
 			t.Run(fmt.Sprintf("%s/%s", eventType, tc.name), func(t *testing.T) {
-				var opts []dataReceiverOption
-				if tc.batcherEnabled != nil {
-					opts = append(opts, withBatcherEnabled(*tc.batcherEnabled))
-				}
-				runner(t, eventType, tc.restartCollector, tc.mockESErr, opts...)
+				runner(t, eventType, tc.restartCollector, tc.mockESErr, withBatching(tc.enableBatching))
 			})
 		}
 	}
@@ -145,7 +137,3 @@ func runner(t *testing.T, eventType string, restartCollector bool, mockESErr err
 	)
 	tc.ValidateData()
 }
-
-func ptrTo[T any](t T) *T {
-	return &t
-}