[exporterhelper] Use configoptional.Optional for exporterhelper QueueBatchConfig

Author: mx-psi

exporter/debugexporter/config.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 )
@@ -33,7 +34,7 @@ type Config struct {
 	// UseInternalLogger defines whether the exporter sends the output to the collector's internal logger.
 	UseInternalLogger bool `mapstructure:"use_internal_logger"`
 
-	QueueConfig exporterhelper.QueueBatchConfig `mapstructure:"sending_queue"`
+	QueueConfig configoptional.Optional[exporterhelper.QueueBatchConfig] `mapstructure:"sending_queue"`
 
 	// prevent unkeyed literal initialization
 	_ struct{}

exporter/debugexporter/config_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/confmap"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
@@ -24,8 +25,6 @@ func TestUnmarshalDefaultConfig(t *testing.T) {
 }
 
 func TestUnmarshalConfig(t *testing.T) {
-	queueCfg := exporterhelper.NewDefaultQueueConfig()
-	queueCfg.Enabled = false
 	tests := []struct {
 		filename    string
 		cfg         *Config
@@ -37,7 +36,7 @@ func TestUnmarshalConfig(t *testing.T) {
 				Verbosity:          configtelemetry.LevelDetailed,
 				SamplingInitial:    10,
 				SamplingThereafter: 50,
-				QueueConfig:        queueCfg,
+				QueueConfig:        configoptional.None[exporterhelper.QueueBatchConfig](),
 			},
 		},
 		{

exporter/debugexporter/exporter_test.go

@@ -14,6 +14,7 @@ import (
 
 	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/exporter/debugexporter/internal/metadata"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/collector/exporter/exportertest"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
@@ -111,15 +112,17 @@ func createTestCases() []testCase {
 			name: "default config",
 			config: func() *Config {
 				c := createDefaultConfig().(*Config)
-				c.QueueConfig.QueueSize = 10
+				c.QueueConfig = exporterhelper.NewDefaultQueueConfig()
+				c.QueueConfig.Get().QueueSize = 10
 				return c
 			}(),
 		},
 		{
 			name: "don't use internal logger",
 			config: func() *Config {
 				cfg := createDefaultConfig().(*Config)
-				cfg.QueueConfig.QueueSize = 10
+				cfg.QueueConfig = exporterhelper.NewDefaultQueueConfig()
+				cfg.QueueConfig.Get().QueueSize = 10
 				cfg.UseInternalLogger = false
 				return cfg
 			}(),

exporter/debugexporter/factory.go

@@ -11,6 +11,7 @@ import (
 	"go.uber.org/zap/zapcore"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter"
@@ -42,15 +43,12 @@ func NewFactory() exporter.Factory {
 }
 
 func createDefaultConfig() component.Config {
-	queueCfg := exporterhelper.NewDefaultQueueConfig()
-	queueCfg.Enabled = false
-
 	return &Config{
 		Verbosity:          configtelemetry.LevelBasic,
 		SamplingInitial:    defaultSamplingInitial,
 		SamplingThereafter: defaultSamplingThereafter,
 		UseInternalLogger:  true,
-		QueueConfig:        queueCfg,
+		QueueConfig:        configoptional.None[exporterhelper.QueueBatchConfig](),
 	}
 }
 

exporter/debugexporter/go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/collector/component v1.45.0
 	go.opentelemetry.io/collector/component/componenttest v0.139.0
+	go.opentelemetry.io/collector/config/configoptional v1.45.0
 	go.opentelemetry.io/collector/config/configtelemetry v0.139.0
 	go.opentelemetry.io/collector/confmap v1.45.0
 	go.opentelemetry.io/collector/consumer v1.45.0
@@ -43,7 +44,6 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/collector/client v1.45.0 // indirect
-	go.opentelemetry.io/collector/config/configoptional v1.45.0 // indirect
 	go.opentelemetry.io/collector/config/configretry v1.45.0 // indirect
 	go.opentelemetry.io/collector/confmap/xconfmap v0.139.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.139.0 // indirect

exporter/exporterhelper/internal/base_exporter.go

@@ -11,6 +11,7 @@ import (
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter"
@@ -47,7 +48,7 @@ type BaseExporter struct {
 	retryCfg   configretry.BackOffConfig
 
 	queueBatchSettings queuebatch.Settings[request.Request]
-	queueCfg           queuebatch.Config
+	queueCfg           configoptional.Optional[queuebatch.Config]
 }
 
 func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, pusher sender.SendFunc[request.Request], options ...Option) (*BaseExporter, error) {
@@ -82,19 +83,19 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, pusher sende
 		return nil, err
 	}
 
-	if be.queueCfg.Batch.HasValue() {
+	if be.queueCfg.HasValue() && be.queueCfg.Get().Batch.HasValue() {
 		// Batcher mutates the data.
 		be.ConsumerOptions = append(be.ConsumerOptions, consumer.WithCapabilities(consumer.Capabilities{MutatesData: true}))
 	}
 
-	if be.queueCfg.Enabled {
+	if be.queueCfg.HasValue() {
 		qSet := queuebatch.AllSettings[request.Request]{
 			Settings:  be.queueBatchSettings,
 			Signal:    signal,
 			ID:        set.ID,
 			Telemetry: set.TelemetrySettings,
 		}
-		be.QueueSender, err = NewQueueSender(qSet, be.queueCfg, be.ExportFailureMessage, be.firstSender)
+		be.QueueSender, err = NewQueueSender(qSet, *be.queueCfg.Get(), be.ExportFailureMessage, be.firstSender)
 		if err != nil {
 			return nil, err
 		}
@@ -191,7 +192,7 @@ func WithRetry(config configretry.BackOffConfig) Option {
 // WithQueue overrides the default queuebatch.Config for an exporter.
 // The default queuebatch.Config is to disable queueing.
 // This option cannot be used with the new exporter helpers New[Traces|Metrics|Logs]RequestExporter.
-func WithQueue(cfg queuebatch.Config) Option {
+func WithQueue(cfg configoptional.Optional[queuebatch.Config]) Option {
 	return func(o *BaseExporter) error {
 		if o.queueBatchSettings.Encoding == nil {
 			return errors.New("WithQueue option is not available for the new request exporters, use WithQueueBatch instead")
@@ -204,13 +205,13 @@ func WithQueue(cfg queuebatch.Config) Option {
 // This option should be used with the new exporter helpers New[Traces|Metrics|Logs]RequestExporter.
 // 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.
-func WithQueueBatch(cfg queuebatch.Config, set queuebatch.Settings[request.Request]) Option {
+func WithQueueBatch(cfg configoptional.Optional[queuebatch.Config], set queuebatch.Settings[request.Request]) Option {
 	return func(o *BaseExporter) error {
-		if !cfg.Enabled {
+		if !cfg.HasValue() {
 			o.ExportFailureMessage += " Try enabling sending_queue to survive temporary failures."
 			return nil
 		}
-		if cfg.StorageID != nil && set.Encoding == nil {
+		if cfg.Get().StorageID != nil && set.Encoding == nil {
 			return errors.New("`Settings.Encoding` must not be nil when persistent queue is enabled")
 		}
 		o.queueBatchSettings = set

exporter/exporterhelper/internal/base_exporter_test.go

@@ -15,6 +15,7 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
@@ -54,7 +55,7 @@ func TestQueueOptionsWithRequestExporter(t *testing.T) {
 
 	qCfg := NewDefaultQueueConfig()
 	storageID := component.NewID(component.MustNewType("test"))
-	qCfg.StorageID = &storageID
+	qCfg.Get().StorageID = &storageID
 	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
 		WithQueueBatchSettings(newFakeQueueBatch()),
 		WithRetry(configretry.NewDefaultBackOffConfig()),
@@ -69,7 +70,7 @@ func TestBaseExporterLogging(t *testing.T) {
 	rCfg := configretry.NewDefaultBackOffConfig()
 	rCfg.Enabled = false
 	qCfg := NewDefaultQueueConfig()
-	qCfg.WaitForResult = true
+	qCfg.Get().WaitForResult = true
 	bs, err := NewBaseExporter(set, pipeline.SignalMetrics, errExport,
 		WithQueueBatchSettings(newFakeQueueBatch()),
 		WithQueue(qCfg),
@@ -98,19 +99,15 @@ func TestQueueRetryWithDisabledQueue(t *testing.T) {
 			queueOptions: []Option{
 				WithQueueBatchSettings(newFakeQueueBatch()),
 				func() Option {
-					qs := NewDefaultQueueConfig()
-					qs.Enabled = false
-					return WithQueue(qs)
+					return WithQueue(configoptional.None[queuebatch.Config]())
 				}(),
 			},
 		},
 		{
 			name: "WithRequestQueue",
 			queueOptions: []Option{
 				func() Option {
-					qs := NewDefaultQueueConfig()
-					qs.Enabled = false
-					return WithQueueBatch(qs, newFakeQueueBatch())
+					return WithQueueBatch(configoptional.None[queuebatch.Config](), newFakeQueueBatch())
 				}(),
 			},
 		},

exporter/exporterhelper/internal/queue_sender.go

@@ -17,9 +17,8 @@ import (
 
 // NewDefaultQueueConfig returns the default config for queuebatch.Config.
 // By default, the queue stores 1000 requests of telemetry and is non-blocking when full.
-func NewDefaultQueueConfig() queuebatch.Config {
-	return queuebatch.Config{
-		Enabled:      true,
+func NewDefaultQueueConfig() configoptional.Optional[queuebatch.Config] {
+	return configoptional.Some(queuebatch.Config{
 		Sizer:        request.SizerTypeRequests,
 		NumConsumers: 10,
 		// By default, batches are 8192 spans, for a total of up to 8 million spans in the queue
@@ -32,7 +31,7 @@ func NewDefaultQueueConfig() queuebatch.Config {
 			Sizer:        request.SizerTypeItems,
 			MinSize:      8192,
 		}),
-	}
+	})
 }
 
 func NewQueueSender(

exporter/exporterhelper/internal/queue_sender_test.go

@@ -15,6 +15,7 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
@@ -31,8 +32,9 @@ func TestNewQueueSenderFailedRequestDropped(t *testing.T) {
 	}
 	logger, observed := observer.New(zap.ErrorLevel)
 	qSet.Telemetry.Logger = zap.New(logger)
+	qCfg := NewDefaultQueueConfig()
 	be, err := NewQueueSender(
-		qSet, NewDefaultQueueConfig(), "", sender.NewSender(func(context.Context, request.Request) error { return errors.New("some error") }))
+		qSet, *qCfg.Get(), "", sender.NewSender(func(context.Context, request.Request) error { return errors.New("some error") }))
 	require.NoError(t, err)
 
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
@@ -46,14 +48,14 @@ func TestQueueConfig_Validate(t *testing.T) {
 	qCfg := NewDefaultQueueConfig()
 	require.NoError(t, qCfg.Validate())
 
-	qCfg.NumConsumers = 0
+	qCfg.Get().NumConsumers = 0
 	require.EqualError(t, qCfg.Validate(), "`num_consumers` must be positive")
 
 	qCfg = NewDefaultQueueConfig()
-	qCfg.QueueSize = 0
+	qCfg.Get().QueueSize = 0
 	require.EqualError(t, qCfg.Validate(), "`queue_size` must be positive")
 
 	// Confirm Validate doesn't return error with invalid config when feature is disabled
-	qCfg.Enabled = false
+	qCfg = configoptional.None[queuebatch.Config]()
 	assert.NoError(t, qCfg.Validate())
 }

exporter/exporterhelper/internal/queuebatch/config.go

@@ -16,9 +16,6 @@ import (
 
 // Config defines configuration for queueing and batching incoming requests.
 type Config struct {
-	// Enabled indicates whether to not enqueue and batch before exporting.
-	Enabled bool `mapstructure:"enabled"`
-
 	// WaitForResult determines if incoming requests are blocked until the request is processed or not.
 	// Currently, this option is not available when persistent queue is configured using the storage configuration.
 	WaitForResult bool `mapstructure:"wait_for_result"`
@@ -66,10 +63,6 @@ func (cfg *Config) Unmarshal(conf *confmap.Conf) error {
 
 // Validate checks if the Config is valid
 func (cfg *Config) Validate() error {
-	if !cfg.Enabled {
-		return nil
-	}
-
 	if cfg.NumConsumers <= 0 {
 		return errors.New("`num_consumers` must be positive")
 	}