Add connection pooling support to OTLP exporter for high-throughput scenarios

This enhancement adds a connection_pool_size configuration option to the OTLP
exporter, enabling multiple gRPC connections with round-robin load balancing.

Key changes:
- Add connection_pool_size config parameter (default: 0, uses 1 connection)
- Implement round-robin load balancing across multiple connections
- Support for 1-256 concurrent gRPC connections
- Backward compatible: default behavior unchanged

This resolves performance issues in high-throughput environments (10K+ spans/sec)
and high-latency network scenarios where a single gRPC connection becomes a
bottleneck.

Also fixes unrelated service.go issue per contributor feedback on PR open-telemetry#14342.

Author: Arunodoy18

.chloggen/otlp-exporter-connection-pool.yaml

@@ -0,0 +1,29 @@
+# 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. receiver/otlp)
+component: exporter/otlp
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Add `connection_pool_size` configuration option to support multiple gRPC connections for high-throughput scenarios"
+
+# One or more tracking issues or pull requests related to the change
+issues: []
+
+# (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 OTLP exporter now supports connection pooling through the `connection_pool_size` configuration option.
+  This helps improve performance in high-throughput environments (10K+ spans/sec) and high-latency scenarios
+  by distributing load across multiple gRPC connections using round-robin load balancing.
+  Default value is 1 (single connection) for backward compatibility. Recommended value for high-throughput: 4-8.
+
+# 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]

exporter/otlpexporter/README.md

@@ -33,6 +33,14 @@ If a scheme of `https` is used then client transport security is enabled and ove
 - `sending_queue`: see [Sending Queue](../exporterhelper/README.md#sending-queue) for the full set of available options.
 - `timeout` (default = 5s): Time to wait per individual attempt to send data to a backend.
 
+The following settings are optional:
+
+- `connection_pool_size` (default = 0, uses 1 connection): Number of gRPC connections to maintain in the connection pool.
+  Useful for high-throughput scenarios (10K+ spans/sec) or high-latency network connections.
+  The exporter distributes requests across connections using round-robin load balancing.
+  Recommended value for high-throughput deployments: 4-8 connections.
+  Maximum allowed value: 256. Setting to 0 or omitting uses the default of 1 connection.
+
 Example:
 
 ```yaml
@@ -46,6 +54,14 @@ exporters:
     endpoint: otelcol2:4317
     tls:
       insecure: true
+  # High-throughput configuration with connection pooling
+  otlp/high-throughput:
+    endpoint: otelcol-gateway:4317
+    connection_pool_size: 5
+    compression: snappy
+    sending_queue:
+      num_consumers: 100
+      queue_size: 2000
 ```
 
 By default, `gzip` compression is enabled. See [compression comparison](../../config/configgrpc/README.md#compression-comparison) for details benchmark information. To disable, configure as follows:

exporter/otlpexporter/config.go

@@ -23,6 +23,11 @@ type Config struct {
 	RetryConfig   configretry.BackOffConfig                                `mapstructure:"retry_on_failure"`
 	ClientConfig  configgrpc.ClientConfig                                  `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
 
+	// ConnectionPoolSize configures the number of gRPC connections to create and maintain.
+	// A larger pool size helps with high-throughput scenarios by distributing load across multiple connections.
+	// Default is 1 (single connection). Recommended for high-throughput: 4-8 connections.
+	ConnectionPoolSize int `mapstructure:"connection_pool_size"`
+
 	// prevent unkeyed literal initialization
 	_ struct{}
 }
@@ -36,6 +41,12 @@ func (c *Config) Validate() error {
 	if endpoint := c.sanitizedEndpoint(); endpoint == "" {
 		return errors.New(`requires a non-empty "endpoint"`)
 	}
+	if c.ConnectionPoolSize < 0 {
+		return errors.New("connection_pool_size must be >= 0")
+	}
+	if c.ConnectionPoolSize > 256 {
+		return errors.New("connection_pool_size must be <= 256")
+	}
 	return nil
 }
 

exporter/otlpexporter/factory.go

@@ -42,10 +42,11 @@ func createDefaultConfig() component.Config {
 	clientCfg.BalancerName = ""
 
 	return &Config{
-		TimeoutConfig: exporterhelper.NewDefaultTimeoutConfig(),
-		RetryConfig:   configretry.NewDefaultBackOffConfig(),
-		QueueConfig:   configoptional.Some(exporterhelper.NewDefaultQueueConfig()),
-		ClientConfig:  clientCfg,
+		TimeoutConfig:      exporterhelper.NewDefaultTimeoutConfig(),
+		RetryConfig:        configretry.NewDefaultBackOffConfig(),
+		QueueConfig:        configoptional.Some(exporterhelper.NewDefaultQueueConfig()),
+		ClientConfig:       clientCfg,
+		ConnectionPoolSize: 0, // 0 means use default (1) at runtime, allowing backward compatibility
 	}
 }
 

exporter/otlpexporter/otlp.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"runtime"
+	"sync/atomic"
 
 	"go.uber.org/zap"
 	"google.golang.org/genproto/googleapis/rpc/errdetails"
@@ -35,14 +36,19 @@ type baseExporter struct {
 	// Input configuration.
 	config *Config
 
-	// gRPC clients and connection.
-	traceExporter   ptraceotlp.GRPCClient
-	metricExporter  pmetricotlp.GRPCClient
-	logExporter     plogotlp.GRPCClient
-	profileExporter pprofileotlp.GRPCClient
-	clientConn      *grpc.ClientConn
-	metadata        metadata.MD
-	callOptions     []grpc.CallOption
+	// gRPC clients and connections pool
+	// When connection pool size > 1, we maintain multiple connections and round-robin across them
+	traceExporters   []ptraceotlp.GRPCClient
+	metricExporters  []pmetricotlp.GRPCClient
+	logExporters     []plogotlp.GRPCClient
+	profileExporters []pprofileotlp.GRPCClient
+	clientConns      []*grpc.ClientConn
+	
+	// Round-robin counter for load balancing across connections
+	roundRobinCounter atomic.Uint32
+	
+	metadata    metadata.MD
+	callOptions []grpc.CallOption
 
 	settings component.TelemetrySettings
 
@@ -59,17 +65,48 @@ func newExporter(cfg component.Config, set exporter.Settings) *baseExporter {
 	return &baseExporter{config: oCfg, settings: set.TelemetrySettings, userAgent: userAgent}
 }
 
-// start actually creates the gRPC connection. The client construction is deferred till this point as this
+// start actually creates the gRPC connection(s). The client construction is deferred till this point as this
 // is the only place we get hold of Extensions which are required to construct auth round tripper.
 func (e *baseExporter) start(ctx context.Context, host component.Host) (err error) {
 	agentOpt := configgrpc.WithGrpcDialOption(grpc.WithUserAgent(e.userAgent))
-	if e.clientConn, err = e.config.ClientConfig.ToClientConn(ctx, host.GetExtensions(), e.settings, agentOpt); err != nil {
-		return err
+	
+	// Determine pool size: default to 1 if not configured or 0
+	poolSize := e.config.ConnectionPoolSize
+	if poolSize <= 0 {
+		poolSize = 1
+	}
+	
+	// Initialize slices for connection pool
+	e.clientConns = make([]*grpc.ClientConn, poolSize)
+	e.traceExporters = make([]ptraceotlp.GRPCClient, poolSize)
+	e.metricExporters = make([]pmetricotlp.GRPCClient, poolSize)
+	e.logExporters = make([]plogotlp.GRPCClient, poolSize)
+	e.profileExporters = make([]pprofileotlp.GRPCClient, poolSize)
+	
+	// Create multiple connections for the pool
+	for i := 0; i < poolSize; i++ {
+		if e.clientConns[i], err = e.config.ClientConfig.ToClientConn(ctx, host.GetExtensions(), e.settings, agentOpt); err != nil {
+			// Close any connections created before the error
+			for j := 0; j < i; j++ {
+				if e.clientConns[j] != nil {
+					_ = e.clientConns[j].Close()
+				}
+			}
+			return err
+		}
+		e.traceExporters[i] = ptraceotlp.NewGRPCClient(e.clientConns[i])
+		e.metricExporters[i] = pmetricotlp.NewGRPCClient(e.clientConns[i])
+		e.logExporters[i] = plogotlp.NewGRPCClient(e.clientConns[i])
+		e.profileExporters[i] = pprofileotlp.NewGRPCClient(e.clientConns[i])
+	}
+	
+	if poolSize > 1 {
+		e.settings.Logger.Info("OTLP exporter connection pool created",
+			zap.Int("pool_size", poolSize),
+			zap.String("endpoint", e.config.ClientConfig.Endpoint),
+		)
 	}
-	e.traceExporter = ptraceotlp.NewGRPCClient(e.clientConn)
-	e.metricExporter = pmetricotlp.NewGRPCClient(e.clientConn)
-	e.logExporter = plogotlp.NewGRPCClient(e.clientConn)
-	e.profileExporter = pprofileotlp.NewGRPCClient(e.clientConn)
+	
 	headers := map[string]string{}
 	for k, v := range e.config.ClientConfig.Headers.Iter {
 		headers[k] = string(v)
@@ -83,15 +120,30 @@ func (e *baseExporter) start(ctx context.Context, host component.Host) (err erro
 }
 
 func (e *baseExporter) shutdown(context.Context) error {
-	if e.clientConn != nil {
-		return e.clientConn.Close()
+	var lastErr error
+	for _, conn := range e.clientConns {
+		if conn != nil {
+			if err := conn.Close(); err != nil {
+				lastErr = err
+			}
+		}
 	}
-	return nil
+	return lastErr
+}
+
+// getNextExporterIndex returns the next index for round-robin load balancing across connections
+func (e *baseExporter) getNextExporterIndex() int {
+	if len(e.clientConns) == 1 {
+		return 0
+	}
+	counter := e.roundRobinCounter.Add(1)
+	return int(counter % uint32(len(e.clientConns)))
 }
 
 func (e *baseExporter) pushTraces(ctx context.Context, td ptrace.Traces) error {
+	idx := e.getNextExporterIndex()
 	req := ptraceotlp.NewExportRequestFromTraces(td)
-	resp, respErr := e.traceExporter.Export(ctx, req, e.callOptions...)
+	resp, respErr := e.traceExporters[idx].Export(ctx, req, e.callOptions...)
 	if err := processError(respErr); err != nil {
 		return err
 	}
@@ -106,8 +158,9 @@ func (e *baseExporter) pushTraces(ctx context.Context, td ptrace.Traces) error {
 }
 
 func (e *baseExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) error {
+	idx := e.getNextExporterIndex()
 	req := pmetricotlp.NewExportRequestFromMetrics(md)
-	resp, respErr := e.metricExporter.Export(ctx, req, e.callOptions...)
+	resp, respErr := e.metricExporters[idx].Export(ctx, req, e.callOptions...)
 	if err := processError(respErr); err != nil {
 		return err
 	}
@@ -122,8 +175,9 @@ func (e *baseExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) erro
 }
 
 func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
+	idx := e.getNextExporterIndex()
 	req := plogotlp.NewExportRequestFromLogs(ld)
-	resp, respErr := e.logExporter.Export(ctx, req, e.callOptions...)
+	resp, respErr := e.logExporters[idx].Export(ctx, req, e.callOptions...)
 	if err := processError(respErr); err != nil {
 		return err
 	}
@@ -138,8 +192,9 @@ func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
 }
 
 func (e *baseExporter) pushProfiles(ctx context.Context, td pprofile.Profiles) error {
+	idx := e.getNextExporterIndex()
 	req := pprofileotlp.NewExportRequestFromProfiles(td)
-	resp, respErr := e.profileExporter.Export(ctx, req, e.callOptions...)
+	resp, respErr := e.profileExporters[idx].Export(ctx, req, e.callOptions...)
 	if err := processError(respErr); err != nil {
 		return err
 	}

service/service.go

@@ -34,7 +34,6 @@ import (
 	"go.opentelemetry.io/collector/service/internal/proctelemetry"
 	"go.opentelemetry.io/collector/service/internal/status"
 	"go.opentelemetry.io/collector/service/telemetry"
-	"go.opentelemetry.io/collector/service/telemetry/otelconftelemetry"
 )
 
 // This feature gate is deprecated and will be removed in 1.40.0. Views can now be configured.
@@ -235,11 +234,8 @@ func New(ctx context.Context, set Settings, cfg Config) (_ *Service, resultErr e
 		return nil, err
 	}
 
-	// Only register process metrics if metrics telemetry is enabled
-	if telemetryCfg, ok := cfg.Telemetry.(*otelconftelemetry.Config); !ok || telemetryCfg.Metrics.Level != configtelemetry.LevelNone {
-		if err := proctelemetry.RegisterProcessMetrics(srv.telemetrySettings); err != nil {
-			return nil, fmt.Errorf("failed to register process metrics: %w", err)
-		}
+	if err := proctelemetry.RegisterProcessMetrics(srv.telemetrySettings); err != nil {
+		return nil, fmt.Errorf("failed to register process metrics: %w", err)
 	}
 	return srv, nil
 }