Deprecate BatcherConfig and WithBatcher in favor of the new QueueBatchConfig (#12748)

Signed-off-by: Alex Boten <[email protected]>
Signed-off-by: Bogdan Drutu <[email protected]>
Co-authored-by: Alex Boten <[email protected]>

Author: bogdandrutu

.chloggen/codeboten_deprecate-batcher-config.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: deprecation
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: otlpexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Mark BatcherConfig as deprecated, use `sending_queue::batch` instead
+
+# One or more tracking issues or pull requests related to the change
+issues: [12726]
+
+# (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:
+
+# 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]

.chloggen/deprecate-batcher-config.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: deprecation
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: exporterhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Deprecate BatcherConfig and WithBatcher in favor of the new QueueBatchConfig.
+
+# One or more tracking issues or pull requests related to the change
+issues: [12748]
+
+# (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:
+
+# 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: [api]

exporter/exporterbatcher/config.go

@@ -8,7 +8,7 @@ import (
 )
 
 // Deprecated: [v0.123.0] use exporterhelper.BatcherConfig
-type Config = exporterhelper.BatcherConfig
+type Config = exporterhelper.BatcherConfig //nolint:staticcheck
 
 // Deprecated: [v0.123.0] use exporterhelper.SizeConfig
 type SizeConfig = exporterhelper.SizeConfig
@@ -26,4 +26,4 @@ var SizerTypeItems = exporterhelper.RequestSizerTypeItems
 var SizerTypeBytes = exporterhelper.RequestSizerTypeBytes
 
 // Deprecated: [v0.123.0] use exporterhelper.NewDefaultBatcherConfig
-var NewDefaultConfig = exporterhelper.NewDefaultBatcherConfig
+var NewDefaultConfig = exporterhelper.NewDefaultBatcherConfig //nolint:staticcheck

exporter/exporterhelper/internal/queue_sender.go

@@ -125,14 +125,14 @@ type SizeConfig struct {
 }
 
 func (c *BatcherConfig) Validate() error {
+	if !c.Enabled {
+		return nil
+	}
+
 	if c.FlushTimeout <= 0 {
 		return errors.New("`flush_timeout` must be greater than zero")
 	}
 
-	return nil
-}
-
-func (c SizeConfig) Validate() error {
 	if c.Sizer != request.SizerTypeItems {
 		return fmt.Errorf("unsupported sizer type: %q", c.Sizer)
 	}

exporter/exporterhelper/internal/queue_sender_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -71,31 +72,47 @@ func TestBatcherConfig_Validate(t *testing.T) {
 }
 
 func TestSizeConfig_Validate(t *testing.T) {
-	cfg := SizeConfig{
-		Sizer:   request.SizerTypeBytes,
-		MaxSize: 10,
-		MinSize: 100,
+	cfg := BatcherConfig{
+		Enabled:      true,
+		FlushTimeout: 200 * time.Millisecond,
+		SizeConfig: SizeConfig{
+			Sizer:   request.SizerTypeBytes,
+			MinSize: 100,
+			MaxSize: 1000,
+		},
 	}
 	require.EqualError(t, cfg.Validate(), "unsupported sizer type: {\"bytes\"}")
 
-	cfg = SizeConfig{
-		Sizer:   request.SizerTypeItems,
-		MaxSize: -100,
-		MinSize: 100,
+	cfg = BatcherConfig{
+		Enabled:      true,
+		FlushTimeout: 200 * time.Millisecond,
+		SizeConfig: SizeConfig{
+			Sizer:   request.SizerTypeItems,
+			MinSize: 100,
+			MaxSize: -1000,
+		},
 	}
 	require.EqualError(t, cfg.Validate(), "`max_size` must be greater than or equal to zero")
 
-	cfg = SizeConfig{
-		Sizer:   request.SizerTypeItems,
-		MaxSize: 100,
-		MinSize: -100,
+	cfg = BatcherConfig{
+		Enabled:      true,
+		FlushTimeout: 200 * time.Millisecond,
+		SizeConfig: SizeConfig{
+			Sizer:   request.SizerTypeItems,
+			MinSize: -100,
+			MaxSize: 1000,
+		},
 	}
 	require.EqualError(t, cfg.Validate(), "`min_size` must be greater than or equal to zero")
 
-	cfg = SizeConfig{
-		Sizer:   request.SizerTypeItems,
-		MaxSize: 100,
-		MinSize: 200,
+	cfg = BatcherConfig{
+		Enabled:      true,
+		FlushTimeout: 200 * time.Millisecond,
+		SizeConfig: SizeConfig{
+			Sizer:   request.SizerTypeItems,
+			MinSize: 1000,
+			MaxSize: 100,
+		},
 	}
 	require.EqualError(t, cfg.Validate(), "`max_size` must be greater than or equal to mix_size")
 }

exporter/exporterhelper/queue_batch.go

@@ -23,18 +23,17 @@ func WithQueue(config QueueBatchConfig) Option {
 	return internal.WithQueue(config)
 }
 
-// WithBatcher enables batching for an exporter based on custom request types.
-// For now, it can be used only with the New[Traces|Metrics|Logs]RequestExporter exporter helpers and
-// WithRequestBatchFuncs provided.
-// This API 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.
+// Deprecated: [v0.123.0] use WithQueueBatch.
 func WithBatcher(cfg BatcherConfig) Option {
 	return internal.WithBatcher(cfg)
 }
 
 // QueueBatchConfig defines configuration for queueing and batching for the exporter.
 type QueueBatchConfig = queuebatch.Config
 
+// BatchConfig defines a configuration for batching requests based on a timeout and a minimum number of items.
+type BatchConfig = queuebatch.BatchConfig
+
 // QueueBatchEncoding defines the encoding to be used if persistent queue is configured.
 // Duplicate definition with queuebatch.Encoding since aliasing generics is not supported by default.
 type QueueBatchEncoding[T any] interface {
@@ -63,14 +62,13 @@ func WithQueueBatch(cfg QueueBatchConfig, set QueueBatchSettings) Option {
 // By default, the queue stores 1000 items of telemetry and is non-blocking when full.
 var NewDefaultQueueConfig = internal.NewDefaultQueueConfig
 
-// BatcherConfig defines a configuration for batching requests based on a timeout and a minimum number of items.
-// Experimental: This API 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.
+// Deprecated: [v0.123.0] use WithQueueBatch.
 type BatcherConfig = internal.BatcherConfig
 
 // SizeConfig sets the size limits for a batch.
 // Experimental: This API 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.
 type SizeConfig = internal.SizeConfig
 
+// Deprecated: [v0.123.0] use WithQueueBatch.
 var NewDefaultBatcherConfig = internal.NewDefaultBatcherConfig

exporter/otlpexporter/config.go

@@ -14,6 +14,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/configretry"
+	"go.opentelemetry.io/collector/confmap"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 )
 
@@ -26,7 +27,26 @@ type Config struct {
 
 	// 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 exporterhelper.BatcherConfig `mapstructure:"batcher"`
+	//
+	// Deprecated: [v0.123.0] batch configuration moving to queue configuration.
+	BatcherConfig exporterhelper.BatcherConfig `mapstructure:"batcher"` //nolint:staticcheck
+
+	// remove at the same time as BatcherConfig
+	hasBatcher bool
+}
+
+func (c *Config) Unmarshal(conf *confmap.Conf) error {
+	if conf.IsSet("batcher") {
+		c.BatcherConfig = exporterhelper.NewDefaultBatcherConfig() //nolint:staticcheck
+		c.BatcherConfig.Enabled = false
+		c.hasBatcher = true
+	}
+
+	if err := conf.Unmarshal(c); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 func (c *Config) Validate() error {

exporter/otlpexporter/config_test.go

@@ -52,17 +52,13 @@ func TestUnmarshalConfig(t *testing.T) {
 			},
 			QueueConfig: exporterhelper.QueueBatchConfig{
 				Enabled:      true,
-				Sizer:        exporterhelper.RequestSizerTypeRequests,
+				Sizer:        exporterhelper.RequestSizerTypeItems,
 				NumConsumers: 2,
-				QueueSize:    10,
-			},
-			BatcherConfig: exporterhelper.BatcherConfig{
-				Enabled:      true,
-				FlushTimeout: 200 * time.Millisecond,
-				SizeConfig: exporterhelper.SizeConfig{
-					Sizer:   exporterhelper.RequestSizerTypeItems,
-					MinSize: 1000,
-					MaxSize: 10000,
+				QueueSize:    100000,
+				Batch: &exporterhelper.BatchConfig{
+					FlushTimeout: 200 * time.Millisecond,
+					MinSize:      1000,
+					MaxSize:      10000,
 				},
 			},
 			ClientConfig: configgrpc.ClientConfig{

exporter/otlpexporter/factory.go

@@ -31,9 +31,6 @@ func NewFactory() exporter.Factory {
 }
 
 func createDefaultConfig() component.Config {
-	batcherCfg := exporterhelper.NewDefaultBatcherConfig()
-	batcherCfg.Enabled = false
-
 	clientCfg := *configgrpc.NewDefaultClientConfig()
 	// Default to gzip compression
 	clientCfg.Compression = configcompression.TypeGzip
@@ -47,7 +44,6 @@ func createDefaultConfig() component.Config {
 		TimeoutConfig: exporterhelper.NewDefaultTimeoutConfig(),
 		RetryConfig:   configretry.NewDefaultBackOffConfig(),
 		QueueConfig:   exporterhelper.NewDefaultQueueConfig(),
-		BatcherConfig: batcherCfg,
 		ClientConfig:  clientCfg,
 	}
 }
@@ -65,7 +61,7 @@ func createTraces(
 		exporterhelper.WithTimeout(oCfg.TimeoutConfig),
 		exporterhelper.WithRetry(oCfg.RetryConfig),
 		exporterhelper.WithQueue(oCfg.QueueConfig),
-		exporterhelper.WithBatcher(oCfg.BatcherConfig),
+		exporterhelper.WithBatcher(oCfg.BatcherConfig), //nolint:staticcheck
 		exporterhelper.WithStart(oce.start),
 		exporterhelper.WithShutdown(oce.shutdown),
 	)
@@ -84,7 +80,7 @@ func createMetrics(
 		exporterhelper.WithTimeout(oCfg.TimeoutConfig),
 		exporterhelper.WithRetry(oCfg.RetryConfig),
 		exporterhelper.WithQueue(oCfg.QueueConfig),
-		exporterhelper.WithBatcher(oCfg.BatcherConfig),
+		exporterhelper.WithBatcher(oCfg.BatcherConfig), //nolint:staticcheck
 		exporterhelper.WithStart(oce.start),
 		exporterhelper.WithShutdown(oce.shutdown),
 	)
@@ -103,7 +99,7 @@ func createLogs(
 		exporterhelper.WithTimeout(oCfg.TimeoutConfig),
 		exporterhelper.WithRetry(oCfg.RetryConfig),
 		exporterhelper.WithQueue(oCfg.QueueConfig),
-		exporterhelper.WithBatcher(oCfg.BatcherConfig),
+		exporterhelper.WithBatcher(oCfg.BatcherConfig), //nolint:staticcheck
 		exporterhelper.WithStart(oce.start),
 		exporterhelper.WithShutdown(oce.shutdown),
 	)
@@ -122,7 +118,7 @@ func createProfilesExporter(
 		exporterhelper.WithTimeout(oCfg.TimeoutConfig),
 		exporterhelper.WithRetry(oCfg.RetryConfig),
 		exporterhelper.WithQueue(oCfg.QueueConfig),
-		exporterhelper.WithBatcher(oCfg.BatcherConfig),
+		exporterhelper.WithBatcher(oCfg.BatcherConfig), //nolint:staticcheck
 		exporterhelper.WithStart(oce.start),
 		exporterhelper.WithShutdown(oce.shutdown),
 	)

exporter/otlpexporter/otlp.go

@@ -53,6 +53,10 @@ type baseExporter struct {
 func newExporter(cfg component.Config, set exporter.Settings) *baseExporter {
 	oCfg := cfg.(*Config)
 
+	if oCfg.hasBatcher {
+		set.TelemetrySettings.Logger.Warn("using deprecated field: batcher")
+	}
+
 	userAgent := fmt.Sprintf("%s/%s (%s/%s)",
 		set.BuildInfo.Description, set.BuildInfo.Version, runtime.GOOS, runtime.GOARCH)
 

exporter/otlpexporter/testdata/config.yaml

@@ -5,21 +5,20 @@ tls:
 timeout: 10s
 sending_queue:
   enabled: true
+  sizer: "items"
   num_consumers: 2
-  queue_size: 10
+  queue_size: 100000
+  batch:
+    flush_timeout: 200ms
+    min_size: 1000
+    max_size: 10000
 retry_on_failure:
   enabled: true
   initial_interval: 10s
   randomization_factor: 0.7
   multiplier: 1.3
   max_interval: 60s
   max_elapsed_time: 10m
-batcher:
-  enabled: true
-  flush_timeout: 200ms
-  sizer: "items"
-  min_size: 1000
-  max_size: 10000
 auth:
   authenticator: nop
 headers: