Allow passing a timeout to the testbed senders

.chloggen/testbed-options-timeout.yaml

@@ -0,0 +1,27 @@
+# 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. receiver/filelog)
+component: testbed
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add functional options support to `NewOTLPTraceDataSender`, `NewOTLPMetricDataSender`, and `NewOTLPLogsDataSender`, with a `WithTimeout` option to override the exporter's default per-send timeout.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [47811]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# 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]

testbed/testbed/senders.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"time"
 
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configcompression"
@@ -210,8 +211,25 @@ func (olds *otlpHTTPLogsDataSender) Start() error {
 	return exp.Start(context.Background(), componenttest.NewNopHost())
 }
 
+// OTLPDataSenderOption is a functional option for OTLP gRPC data senders.
+type OTLPDataSenderOption interface {
+	apply(*otlpDataSender)
+}
+
+type timeoutOption struct{ timeout time.Duration }
+
+func (o timeoutOption) apply(s *otlpDataSender) { s.timeout = o.timeout }
+
+// WithTimeout overrides the exporter's default per-send timeout. WAN benchmarks
+// and slow backends need a larger value. A value of zero is ignored; the
+// exporter's default applies.
+func WithTimeout(d time.Duration) OTLPDataSenderOption {
+	return timeoutOption{timeout: d}
+}
+
 type otlpDataSender struct {
 	DataSenderBase
+	timeout time.Duration
 }
 
 func (ods *otlpDataSender) fillConfig(cfg *otlpexporter.Config) *otlpexporter.Config {
@@ -223,6 +241,9 @@ func (ods *otlpDataSender) fillConfig(cfg *otlpexporter.Config) *otlpexporter.Co
 	cfg.ClientConfig.TLS = configtls.ClientConfig{
 		Insecure: true,
 	}
+	if ods.timeout > 0 {
+		cfg.TimeoutConfig.Timeout = ods.timeout
+	}
 	return cfg
 }
 
@@ -246,15 +267,19 @@ type otlpTraceDataSender struct {
 }
 
 // NewOTLPTraceDataSender creates a new TraceDataSender for OTLP traces exporter.
-func NewOTLPTraceDataSender(host string, port int) TraceDataSender {
-	return &otlpTraceDataSender{
+func NewOTLPTraceDataSender(host string, port int, opts ...OTLPDataSenderOption) TraceDataSender {
+	s := &otlpTraceDataSender{
 		otlpDataSender: otlpDataSender{
 			DataSenderBase: DataSenderBase{
 				Port: port,
 				Host: host,
 			},
 		},
 	}
+	for _, opt := range opts {
+		opt.apply(&s.otlpDataSender)
+	}
+	return s
 }
 
 func (ote *otlpTraceDataSender) Start() error {
@@ -280,15 +305,19 @@ type otlpMetricsDataSender struct {
 
 // NewOTLPMetricDataSender creates a new OTLP metric exporter sender that will send
 // to the specified port after Start is called.
-func NewOTLPMetricDataSender(host string, port int) MetricDataSender {
-	return &otlpMetricsDataSender{
+func NewOTLPMetricDataSender(host string, port int, opts ...OTLPDataSenderOption) MetricDataSender {
+	s := &otlpMetricsDataSender{
 		otlpDataSender: otlpDataSender{
 			DataSenderBase: DataSenderBase{
 				Port: port,
 				Host: host,
 			},
 		},
 	}
+	for _, opt := range opts {
+		opt.apply(&s.otlpDataSender)
+	}
+	return s
 }
 
 func (ome *otlpMetricsDataSender) Start() error {
@@ -314,15 +343,19 @@ type otlpLogsDataSender struct {
 
 // NewOTLPLogsDataSender creates a new OTLP logs exporter sender that will send
 // to the specified port after Start is called.
-func NewOTLPLogsDataSender(host string, port int) LogDataSender {
-	return &otlpLogsDataSender{
+func NewOTLPLogsDataSender(host string, port int, opts ...OTLPDataSenderOption) LogDataSender {
+	s := &otlpLogsDataSender{
 		otlpDataSender: otlpDataSender{
 			DataSenderBase: DataSenderBase{
 				Port: port,
 				Host: host,
 			},
 		},
 	}
+	for _, opt := range opts {
+		opt.apply(&s.otlpDataSender)
+	}
+	return s
 }
 
 func (olds *otlpLogsDataSender) Start() error {

testbed/testbed/senders_test.go

@@ -0,0 +1,95 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package testbed
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/exporter/otlpexporter"
+)
+
+func TestOTLPDataSenderConstructors(t *testing.T) {
+	tests := []struct {
+		name        string
+		opts        []OTLPDataSenderOption
+		wantTimeout time.Duration
+	}{
+		{
+			name:        "no options leaves timeout at zero",
+			opts:        nil,
+			wantTimeout: 0,
+		},
+		{
+			name:        "WithTimeout sets timeout",
+			opts:        []OTLPDataSenderOption{WithTimeout(30 * time.Second)},
+			wantTimeout: 30 * time.Second,
+		},
+		{
+			name:        "WithTimeout(0) is a no-op",
+			opts:        []OTLPDataSenderOption{WithTimeout(0)},
+			wantTimeout: 0,
+		},
+		{
+			name:        "last WithTimeout wins",
+			opts:        []OTLPDataSenderOption{WithTimeout(10 * time.Second), WithTimeout(20 * time.Second)},
+			wantTimeout: 20 * time.Second,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Run("traces", func(t *testing.T) {
+				s, ok := NewOTLPTraceDataSender("localhost", 4317, tt.opts...).(*otlpTraceDataSender)
+				require.True(t, ok)
+				assert.Equal(t, tt.wantTimeout, s.timeout)
+			})
+			t.Run("metrics", func(t *testing.T) {
+				s, ok := NewOTLPMetricDataSender("localhost", 4317, tt.opts...).(*otlpMetricsDataSender)
+				require.True(t, ok)
+				assert.Equal(t, tt.wantTimeout, s.timeout)
+			})
+			t.Run("logs", func(t *testing.T) {
+				s, ok := NewOTLPLogsDataSender("localhost", 4317, tt.opts...).(*otlpLogsDataSender)
+				require.True(t, ok)
+				assert.Equal(t, tt.wantTimeout, s.timeout)
+			})
+		})
+	}
+}
+
+func TestOTLPDataSenderFillConfigTimeout(t *testing.T) {
+	factory := otlpexporter.NewFactory()
+	defaultTimeout := factory.CreateDefaultConfig().(*otlpexporter.Config).TimeoutConfig.Timeout
+
+	tests := []struct {
+		name        string
+		timeout     time.Duration
+		wantTimeout time.Duration
+	}{
+		{
+			name:        "zero timeout preserves exporter default",
+			timeout:     0,
+			wantTimeout: defaultTimeout,
+		},
+		{
+			name:        "non-zero timeout overrides exporter default",
+			timeout:     30 * time.Second,
+			wantTimeout: 30 * time.Second,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &otlpDataSender{
+				DataSenderBase: DataSenderBase{Host: "localhost", Port: 4317},
+				timeout:        tt.timeout,
+			}
+			cfg := s.fillConfig(factory.CreateDefaultConfig().(*otlpexporter.Config))
+			assert.Equal(t, tt.wantTimeout, cfg.TimeoutConfig.Timeout)
+		})
+	}
+}