[receiver/loadgen] Add optional performance optimization to reuse pdata struct

loadgen/cmd/otelbench/config.yaml

@@ -1,5 +1,6 @@
 receivers:
   loadgen:
+    perf_reuse_pdata: true
   nop:
 
 exporters:

receiver/loadgenreceiver/config.go

@@ -40,6 +40,12 @@ type Config struct {
 	// As requests are synchronous, when concurrency is N, there will be N in-flight requests.
 	// This is similar to the `agent_replicas` config in apmsoak.
 	Concurrency int `mapstructure:"concurrency"`
+
+	// PerfReusePdata enables reusing pdata data structures to reduce allocation and GC pressure on the loadgenreceiver.
+	// This optimization is not compatible with fanoutconsumer, i.e. in pipelines where there are more than 1 consumer,
+	// as fanoutconsumer will mark the pdata struct as read only and cannot be reused.
+	// See https://github.com/open-telemetry/opentelemetry-collector/blob/461a3558086a03ab13ea121d12e28e185a1c79b0/internal/fanoutconsumer/logs.go#L70
+	PerfReusePdata bool `mapstructure:"perf_reuse_pdata"`
 }
 
 type MetricsConfig struct {

receiver/loadgenreceiver/logs.go

@@ -106,13 +106,21 @@ func (ar *logsGenerator) Start(ctx context.Context, _ component.Host) error {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
+			var next plog.Logs
+			if ar.cfg.PerfReusePdata {
+				// per-worker temporary container to avoid allocs
+				next = plog.NewLogs()
+			}
 			for {
 				select {
 				case <-startCtx.Done():
 					return
 				default:
 				}
-				next, err := ar.nextLogs()
+				if !ar.cfg.PerfReusePdata {
+					next = plog.NewLogs()
+				}
+				err := ar.nextLogs(next)
 				if errors.Is(err, list.ErrLoopLimitReached) {
 					return
 				}
@@ -147,13 +155,12 @@ func (ar *logsGenerator) Shutdown(context.Context) error {
 	return nil
 }
 
-func (ar *logsGenerator) nextLogs() (plog.Logs, error) {
+func (ar *logsGenerator) nextLogs(next plog.Logs) error {
 	now := pcommon.NewTimestampFromTime(time.Now())
 
-	next := plog.NewLogs()
 	sample, err := ar.samples.Next()
 	if err != nil {
-		return sample, err
+		return err
 	}
 	sample.CopyTo(next)
 
@@ -167,5 +174,5 @@ func (ar *logsGenerator) nextLogs() (plog.Logs, error) {
 		}
 	}
 
-	return next, nil
+	return nil
 }

receiver/loadgenreceiver/logs_bench_test.go

@@ -19,6 +19,7 @@ package loadgenreceiver // import "github.com/elastic/opentelemetry-collector-co
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -30,22 +31,27 @@ import (
 )
 
 func BenchmarkLogsGenerator(b *testing.B) {
-	doneCh := make(chan Stats)
-	cfg := createDefaultReceiverConfig(doneCh, nil, nil)
-	cfg.(*Config).Logs.MaxReplay = b.N
-	r, _ := createLogsReceiver(context.Background(), receiver.Settings{
-		ID: component.ID{},
-		TelemetrySettings: component.TelemetrySettings{
-			Logger: zap.NewNop(),
-		},
-		BuildInfo: component.BuildInfo{},
-	}, cfg, consumertest.NewNop())
-	b.ResetTimer()
-	err := r.Start(context.Background(), componenttest.NewNopHost())
-	require.NoError(b, err)
-	defer func() {
-		err := r.Shutdown(context.Background())
-		require.NoError(b, err)
-	}()
-	<-doneCh
+	for _, perfReusePdata := range []bool{false, true} {
+		b.Run(fmt.Sprintf("perfReusePdata=%v", perfReusePdata), func(b *testing.B) {
+			doneCh := make(chan Stats)
+			cfg := createDefaultReceiverConfig(doneCh, nil, nil)
+			cfg.(*Config).Logs.MaxReplay = b.N
+			cfg.(*Config).PerfReusePdata = perfReusePdata
+			r, _ := createLogsReceiver(context.Background(), receiver.Settings{
+				ID: component.ID{},
+				TelemetrySettings: component.TelemetrySettings{
+					Logger: zap.NewNop(),
+				},
+				BuildInfo: component.BuildInfo{},
+			}, cfg, consumertest.NewNop())
+			b.ResetTimer()
+			err := r.Start(context.Background(), componenttest.NewNopHost())
+			require.NoError(b, err)
+			defer func() {
+				err := r.Shutdown(context.Background())
+				require.NoError(b, err)
+			}()
+			<-doneCh
+		})
+	}
 }

receiver/loadgenreceiver/metrics.go

@@ -106,13 +106,21 @@ func (ar *metricsGenerator) Start(ctx context.Context, _ component.Host) error {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
+			var next pmetric.Metrics
+			if ar.cfg.PerfReusePdata {
+				// per-worker temporary container to avoid allocs
+				next = pmetric.NewMetrics()
+			}
 			for {
 				select {
 				case <-startCtx.Done():
 					return
 				default:
 				}
-				next, err := ar.nextMetrics()
+				if !ar.cfg.PerfReusePdata {
+					next = pmetric.NewMetrics()
+				}
+				err := ar.nextMetrics(next)
 				if errors.Is(err, list.ErrLoopLimitReached) {
 					return
 				}
@@ -147,13 +155,12 @@ func (ar *metricsGenerator) Shutdown(context.Context) error {
 	return nil
 }
 
-func (ar *metricsGenerator) nextMetrics() (pmetric.Metrics, error) {
+func (ar *metricsGenerator) nextMetrics(next pmetric.Metrics) error {
 	now := pcommon.NewTimestampFromTime(time.Now())
 
-	next := pmetric.NewMetrics()
 	sample, err := ar.samples.Next()
 	if err != nil {
-		return sample, err
+		return err
 	}
 	sample.CopyTo(next)
 
@@ -199,5 +206,5 @@ func (ar *metricsGenerator) nextMetrics() (pmetric.Metrics, error) {
 		}
 	}
 
-	return next, nil
+	return nil
 }

receiver/loadgenreceiver/metrics_bench_test.go

@@ -19,6 +19,7 @@ package loadgenreceiver // import "github.com/elastic/opentelemetry-collector-co
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -30,22 +31,27 @@ import (
 )
 
 func BenchmarkMetricsGenerator(b *testing.B) {
-	doneCh := make(chan Stats)
-	cfg := createDefaultReceiverConfig(nil, doneCh, nil)
-	cfg.(*Config).Metrics.MaxReplay = b.N
-	r, _ := createMetricsReceiver(context.Background(), receiver.Settings{
-		ID: component.ID{},
-		TelemetrySettings: component.TelemetrySettings{
-			Logger: zap.NewNop(),
-		},
-		BuildInfo: component.BuildInfo{},
-	}, cfg, consumertest.NewNop())
-	b.ResetTimer()
-	err := r.Start(context.Background(), componenttest.NewNopHost())
-	require.NoError(b, err)
-	defer func() {
-		err := r.Shutdown(context.Background())
-		require.NoError(b, err)
-	}()
-	<-doneCh
+	for _, perfReusePdata := range []bool{false, true} {
+		b.Run(fmt.Sprintf("perfReusePdata=%v", perfReusePdata), func(b *testing.B) {
+			doneCh := make(chan Stats)
+			cfg := createDefaultReceiverConfig(nil, doneCh, nil)
+			cfg.(*Config).Metrics.MaxReplay = b.N
+			cfg.(*Config).PerfReusePdata = perfReusePdata
+			r, _ := createMetricsReceiver(context.Background(), receiver.Settings{
+				ID: component.ID{},
+				TelemetrySettings: component.TelemetrySettings{
+					Logger: zap.NewNop(),
+				},
+				BuildInfo: component.BuildInfo{},
+			}, cfg, consumertest.NewNop())
+			b.ResetTimer()
+			err := r.Start(context.Background(), componenttest.NewNopHost())
+			require.NoError(b, err)
+			defer func() {
+				err := r.Shutdown(context.Background())
+				require.NoError(b, err)
+			}()
+			<-doneCh
+		})
+	}
 }

receiver/loadgenreceiver/traces.go

@@ -108,13 +108,21 @@ func (ar *tracesGenerator) Start(ctx context.Context, _ component.Host) error {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
+			var next ptrace.Traces
+			if ar.cfg.PerfReusePdata {
+				// per-worker temporary container to avoid allocs
+				next = ptrace.NewTraces()
+			}
 			for {
 				select {
 				case <-startCtx.Done():
 					return
 				default:
 				}
-				next, err := ar.nextTraces()
+				if !ar.cfg.PerfReusePdata {
+					next = ptrace.NewTraces()
+				}
+				err := ar.nextTraces(next)
 				if errors.Is(err, list.ErrLoopLimitReached) {
 					return
 				}
@@ -149,11 +157,10 @@ func (ar *tracesGenerator) Shutdown(context.Context) error {
 	return nil
 }
 
-func (ar *tracesGenerator) nextTraces() (ptrace.Traces, error) {
-	next := ptrace.NewTraces()
+func (ar *tracesGenerator) nextTraces(next ptrace.Traces) error {
 	sample, err := ar.samples.Next()
 	if err != nil {
-		return sample, err
+		return err
 	}
 	sample.CopyTo(next)
 
@@ -171,5 +178,5 @@ func (ar *tracesGenerator) nextTraces() (ptrace.Traces, error) {
 		}
 	}
 
-	return next, nil
+	return nil
 }

receiver/loadgenreceiver/traces_bench_test.go

@@ -19,6 +19,7 @@ package loadgenreceiver // import "github.com/elastic/opentelemetry-collector-co
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -30,22 +31,27 @@ import (
 )
 
 func BenchmarkTracesGenerator(b *testing.B) {
-	doneCh := make(chan Stats)
-	cfg := createDefaultReceiverConfig(nil, nil, doneCh)
-	cfg.(*Config).Traces.MaxReplay = b.N
-	r, _ := createTracesReceiver(context.Background(), receiver.Settings{
-		ID: component.ID{},
-		TelemetrySettings: component.TelemetrySettings{
-			Logger: zap.NewNop(),
-		},
-		BuildInfo: component.BuildInfo{},
-	}, cfg, consumertest.NewNop())
-	b.ResetTimer()
-	err := r.Start(context.Background(), componenttest.NewNopHost())
-	require.NoError(b, err)
-	defer func() {
-		err := r.Shutdown(context.Background())
-		require.NoError(b, err)
-	}()
-	<-doneCh
+	for _, perfReusePdata := range []bool{false, true} {
+		b.Run(fmt.Sprintf("perfReusePdata=%v", perfReusePdata), func(b *testing.B) {
+			doneCh := make(chan Stats)
+			cfg := createDefaultReceiverConfig(nil, nil, doneCh)
+			cfg.(*Config).Traces.MaxReplay = b.N
+			cfg.(*Config).PerfReusePdata = perfReusePdata
+			r, _ := createTracesReceiver(context.Background(), receiver.Settings{
+				ID: component.ID{},
+				TelemetrySettings: component.TelemetrySettings{
+					Logger: zap.NewNop(),
+				},
+				BuildInfo: component.BuildInfo{},
+			}, cfg, consumertest.NewNop())
+			b.ResetTimer()
+			err := r.Start(context.Background(), componenttest.NewNopHost())
+			require.NoError(b, err)
+			defer func() {
+				err := r.Shutdown(context.Background())
+				require.NoError(b, err)
+			}()
+			<-doneCh
+		})
+	}
 }