Add MaxRequestSize option to OTLP exporters (#8157)

Per https://github.com/open-telemetry/opentelemetry-proto/pull/782

- Introduce WithMaxRequestSize option for OTLP exporters to set a limit
on the size of serialized export requests.
- Implement logic in the HTTP and gRPC clients to check the request
size against the configured maximum before compression and sending.
- The default configuration is that the maximum request size is 32 MiB.

internal/shared/otlp/otlpmetric/oconf/options.go.tmpl

@@ -35,6 +35,9 @@ const (
 	// DefaultMetricsPath is a default URL path for endpoint that
 	// receives metrics.
 	DefaultMetricsPath string = "/v1/metrics"
+	// DefaultMaxRequestSize is the default maximum size of a serialized export
+	// request, before compression.
+	DefaultMaxRequestSize int = 32 * 1024 * 1024
 	// DefaultBackoff is a default base backoff time used in the
 	// exponential backoff strategy.
 	DefaultBackoff time.Duration = 300 * time.Millisecond
@@ -49,13 +52,14 @@ type (
 	HTTPTransportProxyFunc func(*http.Request) (*url.URL, error)
 
 	SignalConfig struct {
-		Endpoint    string
-		Insecure    bool
-		TLSCfg      *tls.Config
-		Headers     map[string]string
-		Compression Compression
-		Timeout     time.Duration
-		URLPath     string
+		Endpoint       string
+		Insecure       bool
+		TLSCfg         *tls.Config
+		Headers        map[string]string
+		Compression    Compression
+		MaxRequestSize int
+		Timeout        time.Duration
+		URLPath        string
 
 		TemporalitySelector metric.TemporalitySelector
 		AggregationSelector metric.AggregationSelector
@@ -87,10 +91,11 @@ type (
 func NewHTTPConfig(opts ...HTTPOption) Config {
 	cfg := Config{
 		Metrics: SignalConfig{
-			Endpoint:    fmt.Sprintf("%s:%d", DefaultCollectorHost, DefaultCollectorHTTPPort),
-			URLPath:     DefaultMetricsPath,
-			Compression: NoCompression,
-			Timeout:     DefaultTimeout,
+			Endpoint:       fmt.Sprintf("%s:%d", DefaultCollectorHost, DefaultCollectorHTTPPort),
+			URLPath:        DefaultMetricsPath,
+			Compression:    NoCompression,
+			MaxRequestSize: DefaultMaxRequestSize,
+			Timeout:        DefaultTimeout,
 
 			TemporalitySelector: metric.DefaultTemporalitySelector,
 			AggregationSelector: metric.DefaultAggregationSelector,
@@ -123,10 +128,11 @@ func cleanPath(urlPath string, defaultPath string) string {
 func NewGRPCConfig(opts ...GRPCOption) Config {
 	cfg := Config{
 		Metrics: SignalConfig{
-			Endpoint:    fmt.Sprintf("%s:%d", DefaultCollectorHost, DefaultCollectorGRPCPort),
-			URLPath:     DefaultMetricsPath,
-			Compression: NoCompression,
-			Timeout:     DefaultTimeout,
+			Endpoint:       fmt.Sprintf("%s:%d", DefaultCollectorHost, DefaultCollectorGRPCPort),
+			URLPath:        DefaultMetricsPath,
+			Compression:    NoCompression,
+			MaxRequestSize: DefaultMaxRequestSize,
+			Timeout:        DefaultTimeout,
 
 			TemporalitySelector: metric.DefaultTemporalitySelector,
 			AggregationSelector: metric.DefaultAggregationSelector,
@@ -354,6 +360,13 @@ func WithTimeout(duration time.Duration) GenericOption {
 	})
 }
 
+func WithMaxRequestSize(size int) GenericOption {
+	return newGenericOption(func(cfg Config) Config {
+		cfg.Metrics.MaxRequestSize = size
+		return cfg
+	})
+}
+
 func WithTemporalitySelector(selector metric.TemporalitySelector) GenericOption {
 	return newGenericOption(func(cfg Config) Config {
 		cfg.Metrics.TemporalitySelector = selector

internal/shared/otlp/otlpmetric/oconf/options_test.go.tmpl

@@ -79,9 +79,19 @@ func TestConfigs(t *testing.T) {
 				}
 				assert.Equal(t, NoCompression, c.Metrics.Compression)
 				assert.Equal(t, map[string]string(nil), c.Metrics.Headers)
+				assert.Equal(t, DefaultMaxRequestSize, c.Metrics.MaxRequestSize)
 				assert.Equal(t, 10*time.Second, c.Metrics.Timeout)
 			},
 		},
+		{
+			name: "Test With Max Request Size",
+			opts: []GenericOption{
+				WithMaxRequestSize(1),
+			},
+			asserts: func(t *testing.T, c *Config, grpcOption bool) {
+				assert.Equal(t, 1, c.Metrics.MaxRequestSize)
+			},
+		},
 
 		// Endpoint Tests
 		{

internal/shared/otlp/otlptrace/otlpconfig/options.go.tmpl

@@ -31,6 +31,9 @@ const (
 	// DefaultTracesPath is a default URL path for endpoint that
 	// receives spans.
 	DefaultTracesPath string = "/v1/traces"
+	// DefaultMaxRequestSize is the default maximum size of a serialized export
+	// request, before compression.
+	DefaultMaxRequestSize int = 32 * 1024 * 1024
 	// DefaultTimeout is a default max waiting time for the backend to process
 	// each span batch.
 	DefaultTimeout time.Duration = 10 * time.Second
@@ -42,13 +45,14 @@ type (
 	HTTPTransportProxyFunc func(*http.Request) (*url.URL, error)
 
 	SignalConfig struct {
-		Endpoint    string
-		Insecure    bool
-		TLSCfg      *tls.Config
-		Headers     map[string]string
-		Compression Compression
-		Timeout     time.Duration
-		URLPath     string
+		Endpoint       string
+		Insecure       bool
+		TLSCfg         *tls.Config
+		Headers        map[string]string
+		Compression    Compression
+		MaxRequestSize int
+		Timeout        time.Duration
+		URLPath        string
 
 		// gRPC configurations
 		GRPCCredentials credentials.TransportCredentials
@@ -77,10 +81,11 @@ type (
 func NewHTTPConfig(opts ...HTTPOption) Config {
 	cfg := Config{
 		Traces: SignalConfig{
-			Endpoint:    fmt.Sprintf("%s:%d", DefaultCollectorHost, DefaultCollectorHTTPPort),
-			URLPath:     DefaultTracesPath,
-			Compression: NoCompression,
-			Timeout:     DefaultTimeout,
+			Endpoint:       fmt.Sprintf("%s:%d", DefaultCollectorHost, DefaultCollectorHTTPPort),
+			URLPath:        DefaultTracesPath,
+			Compression:    NoCompression,
+			MaxRequestSize: DefaultMaxRequestSize,
+			Timeout:        DefaultTimeout,
 		},
 		RetryConfig: retry.DefaultConfig,
 	}
@@ -111,10 +116,11 @@ func NewGRPCConfig(opts ...GRPCOption) Config {
 	userAgent := "OTel OTLP Exporter Go/" + otlptrace.Version()
 	cfg := Config{
 		Traces: SignalConfig{
-			Endpoint:    fmt.Sprintf("%s:%d", DefaultCollectorHost, DefaultCollectorGRPCPort),
-			URLPath:     DefaultTracesPath,
-			Compression: NoCompression,
-			Timeout:     DefaultTimeout,
+			Endpoint:       fmt.Sprintf("%s:%d", DefaultCollectorHost, DefaultCollectorGRPCPort),
+			URLPath:        DefaultTracesPath,
+			Compression:    NoCompression,
+			MaxRequestSize: DefaultMaxRequestSize,
+			Timeout:        DefaultTimeout,
 		},
 		RetryConfig: retry.DefaultConfig,
 		DialOptions: []grpc.DialOption{grpc.WithUserAgent(userAgent)},
@@ -345,6 +351,13 @@ func WithTimeout(duration time.Duration) GenericOption {
 	})
 }
 
+func WithMaxRequestSize(size int) GenericOption {
+	return newGenericOption(func(cfg Config) Config {
+		cfg.Traces.MaxRequestSize = size
+		return cfg
+	})
+}
+
 func WithProxy(pf HTTPTransportProxyFunc) GenericOption {
 	return newGenericOption(func(cfg Config) Config {
 		cfg.Traces.Proxy = pf

internal/shared/otlp/otlptrace/otlpconfig/options_test.go.tmpl

@@ -77,9 +77,19 @@ func TestConfigs(t *testing.T) {
 				}
 				assert.Equal(t, NoCompression, c.Traces.Compression)
 				assert.Equal(t, map[string]string(nil), c.Traces.Headers)
+				assert.Equal(t, DefaultMaxRequestSize, c.Traces.MaxRequestSize)
 				assert.Equal(t, 10*time.Second, c.Traces.Timeout)
 			},
 		},
+		{
+			name: "Test With Max Request Size",
+			opts: []GenericOption{
+				WithMaxRequestSize(1),
+			},
+			asserts: func(t *testing.T, c *Config, grpcOption bool) {
+				assert.Equal(t, 1, c.Traces.MaxRequestSize)
+			},
+		},
 
 		// Endpoint Tests
 		{