[exporter/elasticsearch] Use exporterhelper/batchsender for reliability

.chloggen/elasticsearchexporter_batchsender.yaml

@@ -0,0 +1,33 @@
+# 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: Improve reliability when used with persistent queue. Remove support for `flush.*`, use `batcher.*` instead.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32377]
+
+# (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: |
+  Move buffering from bulk indexer to batch sender to improve reliability.
+  With this change, there should be no event loss when used with persistent queue in the event of a collector crash.
+  Introduce `batcher.*` to configure the batch sender which is now enabled by default.
+  Option `flush.bytes` is now ignored. Exporter will not start if `flush.bytes` is configured. Use the new `batcher.min_size_items` option to control the minimum number of items (log records, spans) to trigger a flush.
+  Option `flush.interval` is now ignored. Use the new `batcher.flush_timeout` option to control max age of buffer.
+  Queue sender `sending_queue.enabled` defaults to `true`.
+
+# 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

@@ -22,7 +22,7 @@ This exporter supports sending OpenTelemetry logs and traces to [Elasticsearch](
   [ID](https://www.elastic.co/guide/en/cloud/current/ec-cloud-id.html) of the
   Elastic Cloud Cluster to publish events to. The `cloudid` can be used instead
   of `endpoints`.
-- `num_workers` (default=runtime.NumCPU()): Number of workers publishing bulk requests concurrently.
+- `num_workers` (default=runtime.NumCPU()): Maximum number of concurrent bulk requests.
 - `index` (DEPRECATED, please use `logs_index` for logs, `traces_index` for traces): The
   [index](https://www.elastic.co/guide/en/elasticsearch/reference/current/indices.html)
   or [data stream](https://www.elastic.co/guide/en/elasticsearch/reference/current/data-streams.html)
@@ -50,9 +50,14 @@ This exporter supports sending OpenTelemetry logs and traces to [Elasticsearch](
   - `prefix_separator`(default=`-`): Set a separator between logstash_prefix and date.
   - `date_format`(default=`%Y.%m.%d`): Time format (based on strftime) to generate the second part of the Index name.
 - `pipeline` (optional): Optional [Ingest pipeline](https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest.html) ID used for processing documents published by the exporter.
-- `flush`: Event bulk indexer buffer flush settings
-  - `bytes` (default=5000000): Write buffer flush size limit.
-  - `interval` (default=30s): Write buffer flush time limit.
+- `batcher`: Exporter batching settings
+  - `enabled` (default=true): Enable batching of requests into a single bulk request.
+  - `min_size_items` (default=125): Minimum number of log records / spans in the buffer to trigger a flush immediately. 
+  - `max_size_items` (default=0): Maximum number of log records / spans in a request. 0 means there is no limit on the maximum number.
+  - `flush_timeout` (default=30s): Maximum time of the oldest item spent inside the buffer, aka "max age of buffer". A flush will happen regardless of the size of content in buffer.
+- `flush`: (DEPRECATED) Event bulk indexer buffer flush settings
+  - `bytes` (DEPRECATED, use `batcher.min_size_items` instead): Write buffer flush size limit. WARNING: This configuration is ignored.
+  - `interval` (DEPRECATED, use `batcher.flush_timeout` instead): Max age of a document in the buffer. A flush will happen regardless of the size of content in buffer.
 - `retry`: Elasticsearch bulk request retry settings
   - `enabled` (default=true): Enable/Disable request retry on error. Failed requests are retried with exponential backoff.
   - `max_requests` (default=3): Number of HTTP request retries.
@@ -77,10 +82,11 @@ This exporter supports sending OpenTelemetry logs and traces to [Elasticsearch](
     will reject documents that have duplicate fields.
   - `dedot` (default=true): When enabled attributes with `.` will be split into
     proper json objects.
-- `sending_queue`
-  - `enabled` (default = false)
-  - `num_consumers` (default = 10): Number of consumers that dequeue batches; ignored if `enabled` is `false`
-  - `queue_size` (default = 1000): Maximum number of batches kept in queue; ignored if `enabled` is `false`;
+- `sending_queue`: Queue sender settings
+  - `enabled` (default=true)
+  - `num_consumers` (default=100): Number of consumers that dequeue batches. A combined batch cannot contain more batches than the number of consumers.
+  - `queue_size` (default=1000): Maximum number of batches kept in queue.
+  - `storage` (optional): If not empty, it enables the persistent storage and uses the component specified as a storage extension for the persistent queue. When persistent queue is used, there should be no event loss even on collector crashes.
 ### HTTP settings
 
 - `read_buffer_size` (default=0): Read buffer size of HTTP client.

exporter/elasticsearchexporter/config.go

@@ -14,12 +14,18 @@ import (
 
 	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/config/configtls"
+	"go.opentelemetry.io/collector/exporter/exporterbatcher"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 )
 
 // Config defines configuration for Elastic exporter.
 type Config struct {
 	exporterhelper.QueueSettings `mapstructure:"sending_queue"`
+
+	// Experimental: This configuration is at the early stage of development and may change without backward compatibility
+	// until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.
+	BatcherConfig exporterbatcher.Config `mapstructure:"batcher"`
+
 	// Endpoints holds the Elasticsearch URLs the exporter should send events to.
 	//
 	// This setting is required if CloudID is not set and if the
@@ -58,10 +64,11 @@ type Config struct {
 	// https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest.html
 	Pipeline string `mapstructure:"pipeline"`
 
-	ClientConfig   `mapstructure:",squash"`
-	Discovery      DiscoverySettings      `mapstructure:"discover"`
-	Retry          RetrySettings          `mapstructure:"retry"`
-	Flush          FlushSettings          `mapstructure:"flush"`
+	ClientConfig `mapstructure:",squash"`
+	Discovery    DiscoverySettings `mapstructure:"discover"`
+	Retry        RetrySettings     `mapstructure:"retry"`
+	Flush        FlushSettings     `mapstructure:"flush"` // Deprecated: use `batcher` instead.
+
 	Mapping        MappingsSettings       `mapstructure:"mapping"`
 	LogstashFormat LogstashFormatSettings `mapstructure:"logstash_format"`
 }
@@ -133,9 +140,13 @@ type DiscoverySettings struct {
 // all events already serialized into the send-buffer.
 type FlushSettings struct {
 	// Bytes sets the send buffer flushing limit.
+	//
+	// Deprecated: This configuration is ignored. Use `batcher.min_size_items` instead.
 	Bytes int `mapstructure:"bytes"`
 
 	// Interval configures the max age of a document in the send buffer.
+	//
+	// Deprecated: This configuration is ignored. Use `batcher.flush_timeout` instead.
 	Interval time.Duration `mapstructure:"interval"`
 }
 

exporter/elasticsearchexporter/config_test.go

@@ -13,7 +13,9 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
+	"go.opentelemetry.io/collector/exporter/exporterbatcher"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/exporter/exporterqueue"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/elasticsearchexporter/internal/metadata"
 )
@@ -47,9 +49,9 @@ func TestConfig(t *testing.T) {
 			configFile: "config.yaml",
 			expected: &Config{
 				QueueSettings: exporterhelper.QueueSettings{
-					Enabled:      false,
-					NumConsumers: exporterhelper.NewDefaultQueueSettings().NumConsumers,
-					QueueSize:    exporterhelper.NewDefaultQueueSettings().QueueSize,
+					Enabled:      true,
+					NumConsumers: 100,
+					QueueSize:    exporterqueue.NewDefaultConfig().QueueSize,
 				},
 				Endpoints:   []string{"https://elastic.example.com:9200"},
 				Index:       "",
@@ -70,8 +72,11 @@ func TestConfig(t *testing.T) {
 				Discovery: DiscoverySettings{
 					OnStart: true,
 				},
-				Flush: FlushSettings{
-					Bytes: 10485760,
+				BatcherConfig: exporterbatcher.Config{
+					Enabled:       true,
+					FlushTimeout:  5 * time.Second,
+					MinSizeConfig: exporterbatcher.MinSizeConfig{MinSizeItems: 100},
+					MaxSizeConfig: exporterbatcher.MaxSizeConfig{MaxSizeItems: 200},
 				},
 				Retry: RetrySettings{
 					Enabled:         true,
@@ -90,16 +95,17 @@ func TestConfig(t *testing.T) {
 					PrefixSeparator: "-",
 					DateFormat:      "%Y.%m.%d",
 				},
+				NumWorkers: 1,
 			},
 		},
 		{
 			id:         component.NewIDWithName(metadata.Type, "log"),
 			configFile: "config.yaml",
 			expected: &Config{
 				QueueSettings: exporterhelper.QueueSettings{
-					Enabled:      true,
-					NumConsumers: exporterhelper.NewDefaultQueueSettings().NumConsumers,
-					QueueSize:    exporterhelper.NewDefaultQueueSettings().QueueSize,
+					Enabled:      false,
+					NumConsumers: 100,
+					QueueSize:    exporterqueue.NewDefaultConfig().QueueSize,
 				},
 				Endpoints:   []string{"http://localhost:9200"},
 				Index:       "",
@@ -120,8 +126,11 @@ func TestConfig(t *testing.T) {
 				Discovery: DiscoverySettings{
 					OnStart: true,
 				},
-				Flush: FlushSettings{
-					Bytes: 10485760,
+				BatcherConfig: exporterbatcher.Config{
+					Enabled:       true,
+					FlushTimeout:  5 * time.Second,
+					MinSizeConfig: exporterbatcher.MinSizeConfig{MinSizeItems: 100},
+					MaxSizeConfig: exporterbatcher.MaxSizeConfig{MaxSizeItems: 200},
 				},
 				Retry: RetrySettings{
 					Enabled:         true,
@@ -140,6 +149,7 @@ func TestConfig(t *testing.T) {
 					PrefixSeparator: "-",
 					DateFormat:      "%Y.%m.%d",
 				},
+				NumWorkers: 1,
 			},
 		},
 		{