Use configoptional for exporterhelper

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

.chloggen/fix-todo-optional.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: breaking
+
+# 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: Use configoptional for optional fields in exporterhelper
+
+# One or more tracking issues or pull requests related to the change
+issues: [13345]
+
+# (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/debugexporter/go.mod

@@ -42,6 +42,7 @@ 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.35.0 // indirect
+	go.opentelemetry.io/collector/config/configoptional v0.129.0 // indirect
 	go.opentelemetry.io/collector/config/configretry v1.35.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.129.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror/xconsumererror v0.129.0 // indirect
@@ -131,3 +132,5 @@ replace go.opentelemetry.io/collector/internal/telemetry => ../../internal/telem
 replace go.opentelemetry.io/collector/client => ../../client
 
 replace go.opentelemetry.io/collector/pdata/xpdata => ../../pdata/xpdata
+
+replace go.opentelemetry.io/collector/config/configoptional => ../../config/configoptional

exporter/exporterhelper/internal/base_exporter.go

@@ -82,7 +82,7 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, pusher sende
 		return nil, err
 	}
 
-	if be.queueCfg.Batch != nil {
+	if be.queueCfg.Batch.HasValue() {
 		// Batcher mutates the data.
 		be.ConsumerOptions = append(be.ConsumerOptions, consumer.WithCapabilities(consumer.Capabilities{MutatesData: true}))
 	}
@@ -213,7 +213,7 @@ func WithQueueBatch(cfg queuebatch.Config, set QueueBatchSettings[request.Reques
 			o.ExportFailureMessage += " Try enabling sending_queue to survive temporary failures."
 			return nil
 		}
-		if cfg.StorageID != nil && set.Encoding == nil {
+		if cfg.StorageID.HasValue() && set.Encoding == nil {
 			return errors.New("`QueueBatchSettings.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/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
@@ -52,8 +53,7 @@ func TestQueueOptionsWithRequestExporter(t *testing.T) {
 	require.Error(t, err)
 
 	qCfg := NewDefaultQueueConfig()
-	storageID := component.NewID(component.MustNewType("test"))
-	qCfg.StorageID = &storageID
+	qCfg.StorageID = configoptional.Some(component.MustNewID("test"))
 	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
 		WithQueueBatchSettings(newFakeQueueBatch()),
 		WithRetry(configretry.NewDefaultBackOffConfig()),

exporter/exporterhelper/internal/queue/persistent_queue.go

@@ -102,7 +102,7 @@ func newPersistentQueue[T any](set Settings[T]) readableQueue[T] {
 		activeSizer:     set.activeSizer(),
 		itemsSizer:      set.ItemsSizer,
 		bytesSizer:      set.BytesSizer,
-		storageID:       *set.StorageID,
+		storageID:       *set.StorageID.Get(),
 		id:              set.ID,
 		signal:          set.Signal,
 		blockOnOverflow: set.BlockOnOverflow,

exporter/exporterhelper/internal/queue/persistent_queue_test.go

@@ -20,6 +20,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/experr"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/hosttest"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
@@ -231,8 +232,7 @@ func newSettings(sizerType request.SizerType, capacity int64) Settings[int64] {
 
 func newSettingsWithStorage(sizerType request.SizerType, capacity int64) Settings[int64] {
 	set := newSettings(sizerType, capacity)
-	storageID := component.ID{}
-	set.StorageID = &storageID
+	set.StorageID = configoptional.Some(component.ID{})
 	return set
 }
 
@@ -513,8 +513,7 @@ func TestInvalidStorageExtensionType(t *testing.T) {
 }
 
 func TestPersistentQueue_StopAfterBadStart(t *testing.T) {
-	storageID := component.ID{}
-	pq := newPersistentQueue[int64](Settings[int64]{StorageID: &storageID})
+	pq := newPersistentQueue[int64](Settings[int64]{StorageID: configoptional.Some(component.ID{})})
 	// verify that stopping a un-start/started w/error queue does not panic
 	assert.NoError(t, pq.Shutdown(context.Background()))
 }

exporter/exporterhelper/internal/queue/queue.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/pipeline"
 )
@@ -63,7 +64,7 @@ type Settings[T any] struct {
 	WaitForResult   bool
 	BlockOnOverflow bool
 	Signal          pipeline.Signal
-	StorageID       *component.ID
+	StorageID       configoptional.Optional[component.ID]
 	Encoding        Encoding[T]
 	ID              component.ID
 	Telemetry       component.TelemetrySettings
@@ -82,7 +83,7 @@ func (set *Settings[T]) activeSizer() request.Sizer[T] {
 
 func NewQueue[T any](set Settings[T], next ConsumeFunc[T]) (Queue[T], error) {
 	// Configure memory queue or persistent based on the config.
-	if set.StorageID == nil {
+	if !set.StorageID.HasValue() {
 		return newAsyncQueue(newMemoryQueue[T](set), set.NumConsumers, next), nil
 	}
 	if set.ItemsSizer == nil {

exporter/exporterhelper/internal/queue_sender.go

@@ -34,8 +34,6 @@ func NewDefaultQueueConfig() queuebatch.Config {
 		// This default is probably still too high, and may be adjusted further down in a future release
 		QueueSize:       1_000,
 		BlockOnOverflow: false,
-		StorageID:       nil,
-		Batch:           nil,
 	}
 }
 

exporter/exporterhelper/internal/queuebatch/batcher.go

@@ -7,6 +7,7 @@ import (
 	"context"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queue"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
@@ -26,14 +27,14 @@ type batcherSettings[T any] struct {
 	maxWorkers  int
 }
 
-func NewBatcher(cfg *BatchConfig, set batcherSettings[request.Request]) Batcher[request.Request] {
-	if cfg == nil {
+func NewBatcher(cfg configoptional.Optional[BatchConfig], set batcherSettings[request.Request]) Batcher[request.Request] {
+	if !cfg.HasValue() {
 		return newDisabledBatcher[request.Request](set.next)
 	}
 
 	if set.partitioner == nil {
-		return newPartitionBatcher(*cfg, set.sizerType, set.sizer, newWorkerPool(set.maxWorkers), set.next)
+		return newPartitionBatcher(*cfg.Get(), set.sizerType, set.sizer, newWorkerPool(set.maxWorkers), set.next)
 	}
 
-	return newMultiBatcher(*cfg, set.sizerType, set.sizer, newWorkerPool(set.maxWorkers), set.partitioner, set.next)
+	return newMultiBatcher(*cfg.Get(), set.sizerType, set.sizer, newWorkerPool(set.maxWorkers), set.partitioner, set.next)
 }

exporter/exporterhelper/internal/queuebatch/config.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 )
 
@@ -31,10 +32,9 @@ type Config struct {
 	// If true, the component will wait for space; otherwise, operations will immediately return a retryable error.
 	BlockOnOverflow bool `mapstructure:"block_on_overflow"`
 
-	// StorageID if not empty, enables the persistent storage and uses the component specified
+	// StorageID, if not empty, enables the persistent storage and uses the component specified
 	// as a storage extension for the persistent queue.
-	// TODO: This will be changed to Optional when available.
-	StorageID *component.ID `mapstructure:"storage"`
+	StorageID configoptional.Optional[component.ID] `mapstructure:"storage"`
 
 	// NumConsumers is the maximum number of concurrent consumers from the queue.
 	// This applies across all different optional configurations from above (e.g. wait_for_result, blockOnOverflow, persistent, etc.).
@@ -43,8 +43,7 @@ type Config struct {
 	NumConsumers int `mapstructure:"num_consumers"`
 
 	// BatchConfig it configures how the requests are consumed from the queue and batch together during consumption.
-	// TODO: This will be changed to Optional when available.
-	Batch *BatchConfig `mapstructure:"batch"`
+	Batch configoptional.Optional[BatchConfig] `mapstructure:"batch"`
 }
 
 // Validate checks if the Config is valid
@@ -62,18 +61,18 @@ func (cfg *Config) Validate() error {
 	}
 
 	// Only support request sizer for persistent queue at this moment.
-	if cfg.StorageID != nil && cfg.WaitForResult {
+	if cfg.StorageID.HasValue() && cfg.WaitForResult {
 		return errors.New("`wait_for_result` is not supported with a persistent queue configured with `storage`")
 	}
 
-	if cfg.Batch != nil {
+	if cfg.Batch.HasValue() {
 		// Only support items or bytes sizer for batch at this moment.
 		if cfg.Sizer != request.SizerTypeItems && cfg.Sizer != request.SizerTypeBytes {
 			return errors.New("`batch` supports only `items` or `bytes` sizer")
 		}
 
 		// Avoid situations where the queue is not able to hold any data.
-		if cfg.Batch.MinSize > cfg.QueueSize {
+		if cfg.Batch.Get().MinSize > cfg.QueueSize {
 			return errors.New("`min_size` must be less than or equal to `queue_size`")
 		}
 	}