Merge branch 'main' into exphist_prwreceiver

Author: dashpole

.chloggen/signalfxexporter-splunkcloud-default-urls.yaml

@@ -0,0 +1,7 @@
+change_type: breaking
+component: exporter/signalfx
+note: Default `api_url` and `ingest_url` values derived from `realm` now use `*.observability.splunkcloud.com` instead of `*.signalfx.com`.
+issues: [47670]
+subtext: |
+  Explicit `api_url` and `ingest_url` settings are unchanged. Update network allowlists if they targeted only `*.signalfx.com`.
+change_logs: [user]

.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/signalfxexporter/README.md

@@ -33,12 +33,12 @@ The following configuration options are required:
   - `api_url` (no default): Destination to which [properties and
     tags](https://help.splunk.com/en/splunk-observability-cloud/data-tools/metric-finder-and-metadata-catalogue)
     are sent. If `realm` is set, this option is derived and will be
-    `https://api.{realm}.signalfx.com`. If a value is explicitly set, the
+    `https://api.{realm}.observability.splunkcloud.com`. If a value is explicitly set, the
     value of `realm` will not be used in determining `api_url`. The explicit
     value will be used instead.
   - `ingest_url` (no default): Destination where SignalFx metrics are sent. If
     `realm` is set, this option is derived and will be
-    `https://ingest.{realm}.signalfx.com`. If a value is
+    `https://ingest.{realm}.observability.splunkcloud.com`. If a value is
     explicitly set, the value of `realm` will not be used in determining
     `ingest_url`. The explicit value will be used instead. The exporter will 
     automatically append the appropriate path: "/v2/datapoint" for metrics, 
@@ -168,11 +168,11 @@ One of `realm` and `api_url` are required.
 - `access_token` (required, no default): The access token is the authentication token
   provided by SignalFx.
 - `realm` (no default): SignalFx realm where the data will be received.
-- `api_url` (default = `https://api.{realm}.signalfx.com/`): Destination to which correlation updates
+- `api_url` (default = `https://api.{realm}.observability.splunkcloud.com/`): Destination to which correlation updates
    are sent. If a value is explicitly set, the value of `realm` will not be used in determining `api_url`.
    The explicit value will be used instead.
 - `correlation` Contains options controlling the syncing of service and environment properties onto dimensions.
-  - `endpoint` (required, default = `api_url` or `https://api.{realm}.signalfx.com/`): This is the base URL for API requests (e.g. `https://api.us0.signalfx.com`).
+  - `endpoint` (required, default = `api_url` or `https://api.{realm}.observability.splunkcloud.com/`): This is the base URL for API requests (e.g. `https://api.us0.observability.splunkcloud.com`).
   - `timeout` (default = 5s): Is the timeout for every attempt to send data to the backend.
   - `stale_service_timeout` (default = 5 minutes): How long to wait after a span's service name is last seen before uncorrelating it.
   - `max_requests` (default = 20): Max HTTP requests to be made in parallel.

exporter/signalfxexporter/config.go

@@ -185,7 +185,7 @@ func (cfg *Config) getMetricTranslator(done chan struct{}) (*translation.MetricT
 func (cfg *Config) getIngestURL() (*url.URL, error) {
 	strURL := cfg.IngestURL
 	if cfg.IngestURL == "" {
-		strURL = fmt.Sprintf("https://ingest.%s.signalfx.com", cfg.Realm)
+		strURL = fmt.Sprintf("https://ingest.%s.observability.splunkcloud.com", cfg.Realm)
 	}
 
 	ingestURL, err := url.Parse(strURL)
@@ -198,7 +198,7 @@ func (cfg *Config) getIngestURL() (*url.URL, error) {
 func (cfg *Config) getAPIURL() (*url.URL, error) {
 	strURL := cfg.APIURL
 	if cfg.APIURL == "" {
-		strURL = fmt.Sprintf("https://api.%s.signalfx.com", cfg.Realm)
+		strURL = fmt.Sprintf("https://api.%s.observability.splunkcloud.com", cfg.Realm)
 	}
 
 	apiURL, err := url.Parse(strURL)

exporter/signalfxexporter/config_test.go

@@ -339,19 +339,19 @@ func TestConfigGetIngestURL(t *testing.T) {
 			},
 			want: &url.URL{
 				Scheme: "https",
-				Host:   "ingest.us0.signalfx.com",
+				Host:   "ingest.us0.observability.splunkcloud.com",
 				Path:   "",
 			},
 		},
 		{
 			name: "Test URL overrides",
 			cfg: &Config{
 				Realm:     "us0",
-				IngestURL: "https://ingest.us1.signalfx.com/",
+				IngestURL: "https://ingest.us1.observability.splunkcloud.com/",
 			},
 			want: &url.URL{
 				Scheme: "https",
-				Host:   "ingest.us1.signalfx.com",
+				Host:   "ingest.us1.observability.splunkcloud.com",
 				Path:   "/",
 			},
 		},
@@ -390,18 +390,18 @@ func TestConfigGetAPIURL(t *testing.T) {
 			},
 			want: &url.URL{
 				Scheme: "https",
-				Host:   "api.us0.signalfx.com",
+				Host:   "api.us0.observability.splunkcloud.com",
 			},
 		},
 		{
 			name: "Test URL overrides",
 			cfg: &Config{
 				Realm:  "us0",
-				APIURL: "https://api.us1.signalfx.com/",
+				APIURL: "https://api.us1.observability.splunkcloud.com/",
 			},
 			want: &url.URL{
 				Scheme: "https",
-				Host:   "api.us1.signalfx.com",
+				Host:   "api.us1.observability.splunkcloud.com",
 				Path:   "/",
 			},
 		},
@@ -439,14 +439,14 @@ func TestConfigValidateErrors(t *testing.T) {
 			name: "Test empty realm and API URL",
 			cfg: &Config{
 				AccessToken: "access_token",
-				IngestURL:   "https://ingest.us1.signalfx.com/",
+				IngestURL:   "https://ingest.us1.observability.splunkcloud.com/",
 			},
 		},
 		{
 			name: "Test empty realm and Ingest URL",
 			cfg: &Config{
 				AccessToken: "access_token",
-				APIURL:      "https://api.us1.signalfx.com/",
+				APIURL:      "https://api.us1.observability.splunkcloud.com/",
 			},
 		},
 		{

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.")