[exporter/splunkhec] Fix HEC routing partitioning when batcher is enabled (#47725)

Fixes #47695

- Adds `batchperresourceattr.WithMetadataInjection()` to all three
`New*BatchPerResource*` calls so each sub-batch carries its HEC token
and index as `client.Metadata` in the context
- Adds `hecQueueSettings()` to ensure `Batch.Partition.MetadataKeys`
always contains the HEC routing keys when the user enables batching,
preventing the batcher from merging requests across different routing
partitions

Author: dmitryax

.chloggen/splunkhec-batcher-partitioning.yaml

@@ -0,0 +1,5 @@
+change_type: bug_fix
+component: exporter/splunk_hec
+note: Fix HEC routing partitioning broken when the exporterhelper batcher is enabled
+issues: [47695]
+change_logs: [user]

exporter/splunkhecexporter/client_test.go

@@ -2163,6 +2163,108 @@ func TestPushLogsRetryableFailureMultipleResources(t *testing.T) {
 	assert.Equal(t, logs, expectedErr.Data())
 }
 
+// runBatchedLogExport is a variant of runLogExport that enables the
+// exporterhelper batcher and collects all requests received until n have
+// arrived or a 10-second timeout.  Shutdown is used to force-flush pending
+// batcher items before waiting.
+func runBatchedLogExport(t *testing.T, cfg *Config, ld plog.Logs, expectedBatchesNum int) []receivedRequest {
+	t.Helper()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	cfg.Endpoint = "http://" + listener.Addr().String() + "/services/collector"
+	cfg.Token = "1234-1234"
+
+	// Buffered channel so the async handler goroutines never block.
+	rr := make(chan receivedRequest, 64)
+	capture := capturingData{testing: t, receivedRequest: rr, statusCode: 200, checkCompression: !cfg.DisableCompression}
+	s := &http.Server{
+		Handler:           &capture,
+		ReadHeaderTimeout: 20 * time.Second,
+	}
+	go func() {
+		if e := s.Serve(listener); e != http.ErrServerClosed {
+			assert.NoError(t, e)
+		}
+	}()
+	defer s.Close()
+
+	params := exportertest.NewNopSettings(metadata.Type)
+	exp, err := NewFactory().CreateLogs(t.Context(), params, cfg)
+	require.NoError(t, err)
+	require.NoError(t, exp.Start(t.Context(), componenttest.NewNopHost()))
+
+	require.NoError(t, exp.ConsumeLogs(t.Context(), ld))
+
+	// Shutdown flushes all pending batcher items before returning.
+	require.NoError(t, exp.Shutdown(t.Context()))
+
+	// Collect exactly expectedBatchesNum requests, with a generous timeout to
+	// account for the async handler goroutine scheduling.
+	var requests []receivedRequest
+	deadline := time.After(10 * time.Second)
+	for len(requests) < expectedBatchesNum {
+		select {
+		case req := <-rr:
+			requests = append(requests, req)
+		case <-deadline:
+			require.Len(t, requests, expectedBatchesNum, "timed out waiting for HTTP requests")
+			return requests
+		}
+	}
+	return requests
+}
+
+// TestBatcherPartitionsByHecToken verifies that when the exporterhelper batcher
+// is enabled, log resources with different HEC tokens are sent as separate HTTP
+// requests (different Authorization headers), while resources sharing the same
+// token are batched into a single request.
+func TestBatcherPartitionsByHecToken(t *testing.T) {
+	makeCfg := func() *Config {
+		cfg := createDefaultConfig().(*Config)
+		cfg.QueueSettings.Get().Batch = configoptional.Some(exporterhelper.BatchConfig{
+			FlushTimeout: 200 * time.Millisecond,
+			Sizer:        exporterhelper.RequestSizerTypeItems,
+			MinSize:      2,
+		})
+		return cfg
+	}
+
+	t.Run("different tokens produce separate HTTP requests", func(t *testing.T) {
+		ld := plog.NewLogs()
+
+		rl0 := ld.ResourceLogs().AppendEmpty()
+		rl0.Resource().Attributes().PutStr(splunk.HecTokenLabel, "token-A")
+		rl0.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty().Body().SetStr("log from A")
+
+		rl1 := ld.ResourceLogs().AppendEmpty()
+		rl1.Resource().Attributes().PutStr(splunk.HecTokenLabel, "token-B")
+		rl1.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty().Body().SetStr("log from B")
+
+		requests := runBatchedLogExport(t, makeCfg(), ld, 2)
+
+		authHeaders := []string{
+			requests[0].headers.Get("Authorization"),
+			requests[1].headers.Get("Authorization"),
+		}
+		sort.Strings(authHeaders)
+		assert.Equal(t, []string{"Splunk token-A", "Splunk token-B"}, authHeaders)
+	})
+
+	t.Run("same token resources are batched into one request", func(t *testing.T) {
+		ld := plog.NewLogs()
+		for range 2 {
+			rl := ld.ResourceLogs().AppendEmpty()
+			rl.Resource().Attributes().PutStr(splunk.HecTokenLabel, "token-same")
+			rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty().Body().SetStr("log")
+		}
+
+		requests := runBatchedLogExport(t, makeCfg(), ld, 1)
+		assert.Equal(t, "Splunk token-same", requests[0].headers.Get("Authorization"))
+	})
+}
+
 // validateCompressedContains validates that GZipped `got` contains `expected` strings
 func validateCompressedContains(t *testing.T, expected []string, got []byte) {
 	z, err := gzip.NewReader(bytes.NewReader(got))

exporter/splunkhecexporter/factory.go

@@ -5,6 +5,7 @@ package splunkhecexporter // import "github.com/open-telemetry/opentelemetry-col
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	"go.opentelemetry.io/collector/component"
@@ -113,7 +114,7 @@ func createTracesExporter(
 		// explicitly disable since we rely on http.Client timeout logic.
 		exporterhelper.WithTimeout(exporterhelper.TimeoutConfig{Timeout: 0}),
 		exporterhelper.WithRetry(cfg.BackOffConfig),
-		exporterhelper.WithQueue(cfg.QueueSettings),
+		exporterhelper.WithQueue(hecQueueSettings(cfg.QueueSettings)),
 		exporterhelper.WithStart(c.start),
 		exporterhelper.WithShutdown(c.stop),
 	)
@@ -123,7 +124,11 @@ func createTracesExporter(
 
 	wrapped := &baseTracesExporter{
 		Component: e,
-		Traces:    batchperresourceattr.NewMultiBatchPerResourceTraces([]string{splunk.HecTokenLabel, splunk.DefaultIndexLabel}, e),
+		Traces: batchperresourceattr.NewMultiBatchPerResourceTraces(
+			[]string{splunk.HecTokenLabel, splunk.DefaultIndexLabel},
+			e,
+			batchperresourceattr.WithMetadataInjection(),
+		),
 	}
 
 	return wrapped, nil
@@ -146,7 +151,7 @@ func createMetricsExporter(
 		// explicitly disable since we rely on http.Client timeout logic.
 		exporterhelper.WithTimeout(exporterhelper.TimeoutConfig{Timeout: 0}),
 		exporterhelper.WithRetry(cfg.BackOffConfig),
-		exporterhelper.WithQueue(cfg.QueueSettings),
+		exporterhelper.WithQueue(hecQueueSettings(cfg.QueueSettings)),
 		exporterhelper.WithStart(c.start),
 		exporterhelper.WithShutdown(c.stop),
 	)
@@ -156,7 +161,11 @@ func createMetricsExporter(
 
 	wrapped := &baseMetricsExporter{
 		Component: e,
-		Metrics:   batchperresourceattr.NewMultiBatchPerResourceMetrics([]string{splunk.HecTokenLabel, splunk.DefaultIndexLabel}, e),
+		Metrics: batchperresourceattr.NewMultiBatchPerResourceMetrics(
+			[]string{splunk.HecTokenLabel, splunk.DefaultIndexLabel},
+			e,
+			batchperresourceattr.WithMetadataInjection(),
+		),
 	}
 
 	return wrapped, nil
@@ -179,7 +188,7 @@ func createLogsExporter(
 		// explicitly disable since we rely on http.Client timeout logic.
 		exporterhelper.WithTimeout(exporterhelper.TimeoutConfig{Timeout: 0}),
 		exporterhelper.WithRetry(cfg.BackOffConfig),
-		exporterhelper.WithQueue(cfg.QueueSettings),
+		exporterhelper.WithQueue(hecQueueSettings(cfg.QueueSettings)),
 		exporterhelper.WithStart(c.start),
 		exporterhelper.WithShutdown(c.stop),
 	)
@@ -189,17 +198,50 @@ func createLogsExporter(
 
 	wrapped := &baseLogsExporter{
 		Component: logsExporter,
-		Logs: batchperresourceattr.NewMultiBatchPerResourceLogs([]string{splunk.HecTokenLabel, splunk.DefaultIndexLabel}, &perScopeBatcher{
-			logsEnabled:      cfg.LogDataEnabled,
-			profilingEnabled: cfg.ProfilingDataEnabled,
-			logger:           set.Logger,
-			next:             logsExporter,
-		}),
+		Logs: batchperresourceattr.NewMultiBatchPerResourceLogs(
+			[]string{splunk.HecTokenLabel, splunk.DefaultIndexLabel},
+			&perScopeBatcher{
+				logsEnabled:      cfg.LogDataEnabled,
+				profilingEnabled: cfg.ProfilingDataEnabled,
+				logger:           set.Logger,
+				next:             logsExporter,
+			},
+			batchperresourceattr.WithMetadataInjection(),
+		),
 	}
 
 	return wrapped, nil
 }
 
+// hecRequiredMetadataKeys are the context metadata keys the exporterhelper
+// batcher must use to partition requests so different HEC routing targets
+// are never merged into the same batch.
+var hecRequiredMetadataKeys = []string{splunk.HecTokenLabel, splunk.DefaultIndexLabel}
+
+// hecQueueSettings returns a copy of qs with hecRequiredMetadataKeys
+// guaranteed to be present in Batch.Partition.MetadataKeys. Keys already
+// set by the user are preserved; required keys are appended only when absent
+// (case-insensitive). Returns qs unchanged when batching is not configured.
+func hecQueueSettings(qs configoptional.Optional[exporterhelper.QueueBatchConfig]) configoptional.Optional[exporterhelper.QueueBatchConfig] {
+	if !qs.HasValue() || !qs.Get().Batch.HasValue() {
+		return qs
+	}
+	qCopy := *qs.Get()
+	bCopy := *qCopy.Batch.Get()
+
+	existing := make(map[string]bool, len(bCopy.Partition.MetadataKeys))
+	for _, k := range bCopy.Partition.MetadataKeys {
+		existing[strings.ToLower(k)] = true
+	}
+	for _, k := range hecRequiredMetadataKeys {
+		if !existing[strings.ToLower(k)] {
+			bCopy.Partition.MetadataKeys = append(bCopy.Partition.MetadataKeys, k)
+		}
+	}
+	qCopy.Batch = configoptional.Some(bCopy)
+	return configoptional.Some(qCopy)
+}
+
 func showDeprecationWarnings(cfg *Config, logger *zap.Logger) {
 	if cfg.DeprecatedBatcher.isSet {
 		logger.Warn("The 'batcher' field is deprecated and will be removed in a future release. Use 'sending_queue::batch' instead.")

exporter/splunkhecexporter/factory_test.go

@@ -16,6 +16,7 @@ import (
 	"go.opentelemetry.io/collector/exporter/exportertest"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/splunkhecexporter/internal/metadata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/splunk"
 )
 
 func TestCreateDefaultConfig(t *testing.T) {
@@ -129,3 +130,69 @@ func TestFactory_EnabledBatchingMakesExporterMutable(t *testing.T) {
 	require.NoError(t, err)
 	assert.True(t, le.Capabilities().MutatesData)
 }
+
+func TestHecQueueSettings(t *testing.T) {
+	t.Run("no queue settings returned unchanged", func(t *testing.T) {
+		out := hecQueueSettings(configoptional.None[exporterhelper.QueueBatchConfig]())
+		assert.False(t, out.HasValue())
+	})
+
+	t.Run("queue without batch returned unchanged", func(t *testing.T) {
+		qs := configoptional.Some(exporterhelper.QueueBatchConfig{NumConsumers: 2, QueueSize: 100})
+		out := hecQueueSettings(qs)
+		require.True(t, out.HasValue())
+		assert.False(t, out.Get().Batch.HasValue())
+	})
+
+	someBatch := func(keys []string) exporterhelper.QueueBatchConfig {
+		cfg := exporterhelper.NewDefaultQueueConfig()
+		batch := exporterhelper.BatchConfig{
+			FlushTimeout: 200 * time.Millisecond,
+			Sizer:        exporterhelper.RequestSizerTypeItems,
+			MinSize:      8192,
+		}
+		batch.Partition.MetadataKeys = keys
+		cfg.Batch = configoptional.Some(batch)
+		return cfg
+	}
+
+	t.Run("required keys added when missing", func(t *testing.T) {
+		out := hecQueueSettings(configoptional.Some(someBatch(nil)))
+		keys := out.Get().Batch.Get().Partition.MetadataKeys
+		assert.Contains(t, keys, splunk.HecTokenLabel)
+		assert.Contains(t, keys, splunk.DefaultIndexLabel)
+	})
+
+	t.Run("user keys preserved and required keys appended", func(t *testing.T) {
+		out := hecQueueSettings(configoptional.Some(someBatch([]string{"custom_key"})))
+		keys := out.Get().Batch.Get().Partition.MetadataKeys
+		assert.Contains(t, keys, "custom_key")
+		assert.Contains(t, keys, splunk.HecTokenLabel)
+		assert.Contains(t, keys, splunk.DefaultIndexLabel)
+	})
+
+	t.Run("no duplicates when required keys already present", func(t *testing.T) {
+		out := hecQueueSettings(configoptional.Some(someBatch([]string{splunk.HecTokenLabel, splunk.DefaultIndexLabel})))
+		keys := out.Get().Batch.Get().Partition.MetadataKeys
+		count := 0
+		for _, k := range keys {
+			if k == splunk.HecTokenLabel || k == splunk.DefaultIndexLabel {
+				count++
+			}
+		}
+		assert.Equal(t, 2, count)
+	})
+
+	t.Run("case-insensitive deduplication", func(t *testing.T) {
+		out := hecQueueSettings(configoptional.Some(someBatch([]string{"Com.Splunk.Hec.Access_Token", "COM.SPLUNK.INDEX"})))
+		keys := out.Get().Batch.Get().Partition.MetadataKeys
+		assert.Len(t, keys, 2)
+	})
+
+	t.Run("original config not mutated", func(t *testing.T) {
+		orig := someBatch(nil)
+		origKeys := orig.Batch.Get().Partition.MetadataKeys
+		_ = hecQueueSettings(configoptional.Some(orig))
+		assert.Equal(t, origKeys, orig.Batch.Get().Partition.MetadataKeys)
+	})
+}