[Exporterhelper Exporter] Enable metadata population in exporterhelper when sending_queue is enabled

.chloggen/exporterhelper_merge_metadata.yaml

@@ -0,0 +1,28 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: pkg/exporterhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Add `metadata_keys` configuration to `sending_queue` to partition batches by client metadata"
+
+# One or more tracking issues or pull requests related to the change
+issues: [14139]
+
+# (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: |
+  The `metadata_keys` configuration option is now available in the `sending_queue` section for all exporters
+  using queue batch functionality. When specified, batches are partitioned based on the values of the listed metadata keys, allowing separate batching per metadata partition. This feature
+  is automatically configured when using `exporterhelper.WithQueue()`.
+
+# 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, api]

exporter/exporterhelper/internal/base_exporter.go

@@ -202,6 +202,8 @@ func WithQueue(cfg queuebatch.Config) Option {
 
 // WithQueueBatch enables queueing for an exporter.
 // This option should be used with the new exporter helpers New[Traces|Metrics|Logs]RequestExporter.
+// If cfg.MetadataKeys is set, it will automatically configure the partitioner and merge function
+// to partition batches based on the specified metadata keys.
 // 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 {
@@ -213,6 +215,17 @@ func WithQueueBatch(cfg queuebatch.Config, set queuebatch.Settings[request.Reque
 		if cfg.StorageID != nil && set.Encoding == nil {
 			return errors.New("`Settings.Encoding` must not be nil when persistent queue is enabled")
 		}
+		// Automatically configure partitioner if MetadataKeys is set
+		if len(cfg.MetadataKeys) > 0 {
+			if set.Partitioner != nil {
+				return errors.New("cannot use metadata_keys when a custom partitioner is already configured")
+			}
+			if set.MergeCtx != nil {
+				return errors.New("cannot use metadata_keys when a custom merge function is already configured")
+			}
+			set.Partitioner = queuebatch.NewMetadataKeysPartitioner(cfg.MetadataKeys)
+			set.MergeCtx = queuebatch.NewMetadataKeysMergeCtx(cfg.MetadataKeys)
+		}
 		o.queueBatchSettings = set
 		o.queueCfg = cfg
 		return nil

exporter/exporterhelper/internal/base_exporter_test.go

@@ -88,6 +88,115 @@ func TestBaseExporterLogging(t *testing.T) {
 	require.NoError(t, bs.Shutdown(context.Background()))
 }
 
+func TestWithQueue_MetadataKeys(t *testing.T) {
+	t.Run("with MetadataKeys - configures partitioner and merge function", func(t *testing.T) {
+		qCfg := NewDefaultQueueConfig()
+		qCfg.MetadataKeys = []string{"key1", "key2"}
+		qCfg.Enabled = true
+
+		be, err := NewBaseExporter(
+			exportertest.NewNopSettings(exportertest.NopType),
+			pipeline.SignalMetrics,
+			noopExport,
+			WithQueueBatchSettings(newFakeQueueBatch()),
+			WithQueue(qCfg),
+		)
+		require.NoError(t, err)
+		assert.NotNil(t, be)
+
+		// Verify partitioner and merge function are configured
+		assert.NotNil(t, be.queueBatchSettings.Partitioner, "Partitioner should be set when MetadataKeys is provided")
+		assert.NotNil(t, be.queueBatchSettings.MergeCtx, "MergeCtx should be set when MetadataKeys is provided")
+	})
+
+	t.Run("without MetadataKeys - does not configure partitioner", func(t *testing.T) {
+		qCfg := NewDefaultQueueConfig()
+		qCfg.MetadataKeys = []string{}
+		qCfg.Enabled = true
+
+		be, err := NewBaseExporter(
+			exportertest.NewNopSettings(exportertest.NopType),
+			pipeline.SignalMetrics,
+			noopExport,
+			WithQueueBatchSettings(newFakeQueueBatch()),
+			WithQueue(qCfg),
+		)
+		require.NoError(t, err)
+		assert.NotNil(t, be)
+
+		// Verify partitioner and merge function are NOT configured
+		assert.Nil(t, be.queueBatchSettings.Partitioner, "Partitioner should not be set when MetadataKeys is empty")
+		assert.Nil(t, be.queueBatchSettings.MergeCtx, "MergeCtx should not be set when MetadataKeys is empty")
+	})
+
+	t.Run("with nil MetadataKeys - does not configure partitioner", func(t *testing.T) {
+		qCfg := NewDefaultQueueConfig()
+		qCfg.MetadataKeys = nil
+		qCfg.Enabled = true
+
+		be, err := NewBaseExporter(
+			exportertest.NewNopSettings(exportertest.NopType),
+			pipeline.SignalMetrics,
+			noopExport,
+			WithQueueBatchSettings(newFakeQueueBatch()),
+			WithQueue(qCfg),
+		)
+		require.NoError(t, err)
+		assert.NotNil(t, be)
+
+		// Verify partitioner and merge function are NOT configured
+		assert.Nil(t, be.queueBatchSettings.Partitioner, "Partitioner should not be set when MetadataKeys is nil")
+		assert.Nil(t, be.queueBatchSettings.MergeCtx, "MergeCtx should not be set when MetadataKeys is nil")
+	})
+
+	t.Run("error when custom partitioner already set and metadata_keys used", func(t *testing.T) {
+		qCfg := NewDefaultQueueConfig()
+		qCfg.MetadataKeys = []string{"key1", "key2"}
+		qCfg.Enabled = true
+
+		// Set up queue batch settings with a custom partitioner already configured
+		customSettings := newFakeQueueBatch()
+		customPartitioner := queuebatch.NewPartitioner(
+			func(context.Context, request.Request) string {
+				return "custom"
+			},
+		)
+		customSettings.Partitioner = customPartitioner
+
+		_, err := NewBaseExporter(
+			exportertest.NewNopSettings(exportertest.NopType),
+			pipeline.SignalMetrics,
+			noopExport,
+			WithQueueBatchSettings(customSettings),
+			WithQueue(qCfg),
+		)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "cannot use metadata_keys when a custom partitioner is already configured")
+	})
+
+	t.Run("error when custom merge function already set and metadata_keys used", func(t *testing.T) {
+		qCfg := NewDefaultQueueConfig()
+		qCfg.MetadataKeys = []string{"key1", "key2"}
+		qCfg.Enabled = true
+
+		// Set up queue batch settings with a custom merge function already configured
+		customSettings := newFakeQueueBatch()
+		customSettings.MergeCtx = func(context.Context, context.Context) context.Context {
+			return context.Background()
+		}
+
+		_, err := NewBaseExporter(
+			exportertest.NewNopSettings(exportertest.NopType),
+			pipeline.SignalMetrics,
+			noopExport,
+			WithQueueBatchSettings(customSettings),
+			WithQueue(qCfg),
+		)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "cannot use metadata_keys when a custom merge function is already configured")
+	})
+}
+
 func TestQueueRetryWithDisabledQueue(t *testing.T) {
 	tests := []struct {
 		name         string

exporter/exporterhelper/internal/queuebatch/config.go

@@ -6,6 +6,7 @@ package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhel
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"go.opentelemetry.io/collector/component"
@@ -46,6 +47,16 @@ type Config struct {
 
 	// BatchConfig it configures how the requests are consumed from the queue and batch together during consumption.
 	Batch configoptional.Optional[BatchConfig] `mapstructure:"batch"`
+
+	// MetadataKeys is a list of client.Metadata keys that will be used to partition
+	// the data into batches. If this setting is empty, a single batcher instance
+	// will be used. When this setting is not empty, one batcher will be used per
+	// distinct combination of values for the listed metadata keys.
+	//
+	// Empty value and unset metadata are treated as distinct cases.
+	//
+	// Entries are case-insensitive. Duplicated entries will trigger a validation error.
+	MetadataKeys []string `mapstructure:"metadata_keys"`
 }
 
 func (cfg *Config) Unmarshal(conf *confmap.Conf) error {
@@ -90,6 +101,16 @@ func (cfg *Config) Validate() error {
 		}
 	}
 
+	// Validate metadata_keys for duplicates (case-insensitive)
+	uniq := map[string]bool{}
+	for _, k := range cfg.MetadataKeys {
+		l := strings.ToLower(k)
+		if _, has := uniq[l]; has {
+			return fmt.Errorf("duplicate entry in metadata_keys: %q (case-insensitive)", l)
+		}
+		uniq[l] = true
+	}
+
 	return nil
 }
 

exporter/exporterhelper/internal/queuebatch/config_test.go

@@ -56,6 +56,65 @@ func TestConfig_Validate(t *testing.T) {
 	assert.NoError(t, xconfmap.Validate(cfg))
 }
 
+func TestConfig_Validate_MetadataKeys(t *testing.T) {
+	t.Run("no duplicates - valid", func(t *testing.T) {
+		cfg := newTestConfig()
+		cfg.MetadataKeys = []string{"key1", "key2", "key3"}
+		require.NoError(t, xconfmap.Validate(cfg))
+	})
+
+	t.Run("duplicate keys same case - invalid", func(t *testing.T) {
+		cfg := newTestConfig()
+		cfg.MetadataKeys = []string{"key1", "key2", "key1"}
+		err := xconfmap.Validate(cfg)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "duplicate entry in metadata_keys")
+		assert.Contains(t, err.Error(), "key1")
+		assert.Contains(t, err.Error(), "case-insensitive")
+	})
+
+	t.Run("duplicate keys different case - invalid", func(t *testing.T) {
+		cfg := newTestConfig()
+		cfg.MetadataKeys = []string{"key1", "KEY1", "key2"}
+		err := xconfmap.Validate(cfg)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "duplicate entry in metadata_keys")
+		assert.Contains(t, err.Error(), "key1")
+		assert.Contains(t, err.Error(), "case-insensitive")
+	})
+
+	t.Run("duplicate keys mixed case - invalid", func(t *testing.T) {
+		cfg := newTestConfig()
+		cfg.MetadataKeys = []string{"Key1", "kEy1", "key2"}
+		err := xconfmap.Validate(cfg)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "duplicate entry in metadata_keys")
+		assert.Contains(t, err.Error(), "key1")
+		assert.Contains(t, err.Error(), "case-insensitive")
+	})
+
+	t.Run("empty metadata_keys - valid", func(t *testing.T) {
+		cfg := newTestConfig()
+		cfg.MetadataKeys = []string{}
+		require.NoError(t, xconfmap.Validate(cfg))
+	})
+
+	t.Run("nil metadata_keys - valid", func(t *testing.T) {
+		cfg := newTestConfig()
+		cfg.MetadataKeys = nil
+		require.NoError(t, xconfmap.Validate(cfg))
+	})
+
+	t.Run("multiple duplicates - reports first duplicate", func(t *testing.T) {
+		cfg := newTestConfig()
+		cfg.MetadataKeys = []string{"key1", "key2", "key1", "key2"}
+		err := xconfmap.Validate(cfg)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "duplicate entry in metadata_keys")
+		assert.Contains(t, err.Error(), "key1")
+	})
+}
+
 func TestBatchConfig_Validate(t *testing.T) {
 	cfg := newTestBatchConfig()
 	require.NoError(t, xconfmap.Validate(cfg))

exporter/exporterhelper/internal/queuebatch/metadata_partitioner.go

@@ -0,0 +1,88 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"slices"
+
+	"go.opentelemetry.io/collector/client"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
+)
+
+// metadataKeysPartitioner partitions requests based on client metadata keys.
+type metadataKeysPartitioner struct {
+	keys []string
+}
+
+// NewMetadataKeysPartitioner creates a new partitioner that partitions requests
+// based on the specified metadata keys. If keys is empty, returns nil.
+func NewMetadataKeysPartitioner(keys []string) Partitioner[request.Request] {
+	if len(keys) == 0 {
+		return nil
+	}
+	return &metadataKeysPartitioner{keys: keys}
+}
+
+// GetKey returns a partition key based on the metadata keys configured.
+func (p *metadataKeysPartitioner) GetKey(
+	ctx context.Context,
+	_ request.Request,
+) string {
+	var kb bytes.Buffer
+	meta := client.FromContext(ctx).Metadata
+
+	var afterFirst bool
+	for _, k := range p.keys {
+		if values := meta.Get(k); len(values) != 0 {
+			if afterFirst {
+				kb.WriteByte(0)
+			}
+			kb.WriteString(k)
+			afterFirst = true
+			for _, val := range values {
+				kb.WriteByte(0)
+				kb.WriteString(val)
+			}
+		}
+	}
+	return kb.String()
+}
+
+// NewMetadataKeysMergeCtx creates a merge function for contexts that merges
+// metadata based on the specified keys. If keys is empty, returns nil.
+func NewMetadataKeysMergeCtx(keys []string) func(context.Context, context.Context) context.Context {
+	if len(keys) == 0 {
+		return nil
+	}
+	return func(ctx1, ctx2 context.Context) context.Context {
+		m1 := client.FromContext(ctx1).Metadata
+		m2 := client.FromContext(ctx2).Metadata
+
+		m := make(map[string][]string, len(keys))
+		for _, key := range keys {
+			v1 := m1.Get(key)
+			v2 := m2.Get(key)
+			if len(v1) == 0 && len(v2) == 0 {
+				continue
+			}
+
+			// Since the mergeCtx is based on partition key, we MUST have the same
+			// partition key-values in both the metadata. If they are not same then
+			// fail fast and dramatically.
+			if !slices.Equal(v1, v2) {
+				panic(fmt.Errorf(
+					"unexpected client metadata found when merging context for key %s", key,
+				))
+			}
+			m[key] = v1
+		}
+		return client.NewContext(
+			context.Background(),
+			client.Info{Metadata: client.NewMetadata(m)},
+		)
+	}
+}