WIP

receiver/loadgenreceiver/config.go

@@ -30,6 +30,9 @@ type Config struct {
 	Metrics MetricsConfig `mapstructure:"metrics"`
 	Logs    LogsConfig    `mapstructure:"logs"`
 	Traces  TracesConfig  `mapstructure:"traces"`
+
+	// NumWorker is the number of workers to share the load
+	NumWorkers int `mapstructure:"num_workers"`
 }
 
 type MetricsConfig struct {

receiver/loadgenreceiver/factory.go

@@ -39,6 +39,7 @@ func createDefaultReceiverConfig(logsDone, metricsDone, tracesDone chan Stats) c
 		Traces: TracesConfig{
 			doneCh: tracesDone,
 		},
+		NumWorkers: 1,
 	}
 }
 

receiver/loadgenreceiver/internal/loopinglist.go → receiver/loadgenreceiver/internal/loopinglist/list.go

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package internal // import "github.com/elastic/opentelemetry-collector-components/receiver/loadgenreceiver/internal"
+package loopinglist // import "github.com/elastic/opentelemetry-collector-components/receiver/loadgenreceiver/internal/loopinglist"
 
 type LoopingList[T any] struct {
 	items   []T

receiver/loadgenreceiver/logs.go

@@ -32,7 +32,7 @@ import (
 	"go.opentelemetry.io/collector/receiver"
 	"go.uber.org/zap"
 
-	"github.com/elastic/opentelemetry-collector-components/receiver/loadgenreceiver/internal"
+	"github.com/elastic/opentelemetry-collector-components/receiver/loadgenreceiver/internal/loopinglist"
 )
 
 //go:embed testdata/logs.jsonl
@@ -42,7 +42,7 @@ type logsGenerator struct {
 	cfg    *Config
 	logger *zap.Logger
 
-	samples internal.LoopingList[plog.Logs]
+	samples loopinglist.LoopingList[plog.Logs]
 
 	stats Stats
 
@@ -89,7 +89,7 @@ func createLogsReceiver(
 		cfg:      genConfig,
 		logger:   set.Logger,
 		consumer: consumer,
-		samples:  internal.NewLoopingList(samples),
+		samples:  loopinglist.NewLoopingList(samples),
 	}, nil
 }
 

receiver/loadgenreceiver/metrics.go

@@ -32,7 +32,7 @@ import (
 	"go.opentelemetry.io/collector/receiver"
 	"go.uber.org/zap"
 
-	"github.com/elastic/opentelemetry-collector-components/receiver/loadgenreceiver/internal"
+	"github.com/elastic/opentelemetry-collector-components/receiver/loadgenreceiver/internal/loopinglist"
 )
 
 //go:embed testdata/metrics.jsonl
@@ -42,7 +42,7 @@ type metricsGenerator struct {
 	cfg    *Config
 	logger *zap.Logger
 
-	samples internal.LoopingList[pmetric.Metrics]
+	samples loopinglist.LoopingList[pmetric.Metrics]
 
 	stats Stats
 
@@ -89,7 +89,7 @@ func createMetricsReceiver(
 		cfg:      genConfig,
 		logger:   set.Logger,
 		consumer: consumer,
-		samples:  internal.NewLoopingList(samples),
+		samples:  loopinglist.NewLoopingList(samples),
 	}, nil
 }
 

receiver/loadgenreceiver/traces.go

@@ -22,7 +22,9 @@ import (
 	"bytes"
 	"context"
 	_ "embed"
+	"errors"
 	"os"
+	"sync"
 	"time"
 
 	"go.opentelemetry.io/collector/component"
@@ -32,7 +34,7 @@ import (
 	"go.opentelemetry.io/collector/receiver"
 	"go.uber.org/zap"
 
-	"github.com/elastic/opentelemetry-collector-components/receiver/loadgenreceiver/internal"
+	"github.com/elastic/opentelemetry-collector-components/receiver/loadgenreceiver/internal/loopinglist"
 )
 
 const maxScannerBufSize = 1024 * 1024
@@ -44,9 +46,10 @@ type tracesGenerator struct {
 	cfg    *Config
 	logger *zap.Logger
 
-	samples internal.LoopingList[ptrace.Traces]
+	samples loopinglist.BoundedLoopingList[ptrace.Traces]
 
-	stats Stats
+	stats   Stats
+	statsMu sync.Mutex
 
 	consumer consumer.Traces
 
@@ -91,36 +94,50 @@ func createTracesReceiver(
 		cfg:      genConfig,
 		logger:   set.Logger,
 		consumer: consumer,
-		samples:  internal.NewLoopingList(samples),
+		samples:  loopinglist.NewBoundedLoopingList(samples, genConfig.Traces.MaxReplay),
 	}, nil
 }
 
 func (ar *tracesGenerator) Start(ctx context.Context, _ component.Host) error {
 	startCtx, cancelFn := context.WithCancel(ctx)
 	ar.cancelFn = cancelFn
 
-	go func() {
-		for {
-			select {
-			case <-startCtx.Done():
-				return
-			default:
-			}
-			m := ar.nextTraces()
-			if err := ar.consumer.ConsumeTraces(startCtx, m); err != nil {
-				ar.logger.Error(err.Error())
-				ar.stats.FailedRequests++
-				ar.stats.FailedSpans += m.SpanCount()
-			} else {
-				ar.stats.Requests++
-				ar.stats.Spans += m.SpanCount()
-			}
-			if ar.isDone() {
-				if ar.cfg.Traces.doneCh != nil {
-					ar.cfg.Traces.doneCh <- ar.stats
+	wg := sync.WaitGroup{}
+
+	for i := 0; i < ar.cfg.NumWorkers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-startCtx.Done():
+					return
+				default:
 				}
-				return
+				m, err := ar.nextTraces()
+				if errors.Is(err, loopinglist.ErrLoopLimitReached) {
+					return
+				}
+				if err := ar.consumer.ConsumeTraces(startCtx, m); err != nil {
+					ar.logger.Error(err.Error())
+					ar.statsMu.Lock()
+					ar.stats.FailedRequests++
+					ar.stats.FailedSpans += m.SpanCount()
+					ar.statsMu.Unlock()
+				} else {
+					ar.statsMu.Lock()
+					ar.stats.Requests++
+					ar.stats.Spans += m.SpanCount()
+					ar.statsMu.Unlock()
+				}
+
 			}
+		}()
+	}
+	go func() {
+		wg.Wait()
+		if ar.cfg.Traces.doneCh != nil {
+			ar.cfg.Traces.doneCh <- ar.stats
 		}
 	}()
 	return nil
@@ -133,9 +150,13 @@ func (ar *tracesGenerator) Shutdown(context.Context) error {
 	return nil
 }
 
-func (ar *tracesGenerator) nextTraces() ptrace.Traces {
+func (ar *tracesGenerator) nextTraces() (ptrace.Traces, error) {
 	nextLogs := ptrace.NewTraces()
-	ar.samples.Next().CopyTo(nextLogs)
+	s, err := ar.samples.Next()
+	if err != nil {
+		return s, err
+	}
+	s.CopyTo(nextLogs)
 
 	rm := nextLogs.ResourceSpans()
 	for i := 0; i < rm.Len(); i++ {
@@ -151,9 +172,5 @@ func (ar *tracesGenerator) nextTraces() ptrace.Traces {
 		}
 	}
 
-	return nextLogs
-}
-
-func (ar *tracesGenerator) isDone() bool {
-	return ar.cfg.Traces.MaxReplay > 0 && ar.samples.LoopCount() >= ar.cfg.Traces.MaxReplay
+	return nextLogs, nil
 }

receiver/loadgenreceiver/traces_test.go

@@ -20,6 +20,7 @@ package loadgenreceiver // import "github.com/elastic/opentelemetry-collector-co
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -32,26 +33,30 @@ import (
 
 func TestTracesGenerator_doneCh(t *testing.T) {
 	const maxReplay = 2
-	doneCh := make(chan Stats)
-	sink := &consumertest.TracesSink{}
-	r, _ := createTracesReceiver(context.Background(), receiver.Settings{
-		ID: component.ID{},
-		TelemetrySettings: component.TelemetrySettings{
-			Logger: zap.NewNop(),
-		},
-		BuildInfo: component.BuildInfo{},
-	}, &Config{Traces: TracesConfig{
-		MaxReplay: maxReplay,
-		doneCh:    doneCh,
-	}}, sink)
-	err := r.Start(context.Background(), componenttest.NewNopHost())
-	assert.NoError(t, err)
-	defer func() {
-		assert.NoError(t, r.Shutdown(context.Background()))
-	}()
-	stats := <-doneCh
-	want := maxReplay * bytes.Count(demoTraces, []byte("\n"))
-	assert.Equal(t, want, stats.Requests)
-	assert.Equal(t, want, len(sink.AllTraces()))
-	assert.Equal(t, sink.SpanCount(), stats.Spans)
+	for _, workers := range []int{1, 2} {
+		t.Run(fmt.Sprintf("workers=%d", workers), func(t *testing.T) {
+			doneCh := make(chan Stats)
+			sink := &consumertest.TracesSink{}
+			cfg := createDefaultReceiverConfig(nil, nil, doneCh)
+			cfg.(*Config).Traces.MaxReplay = maxReplay
+			cfg.(*Config).NumWorkers = workers
+			r, _ := createTracesReceiver(context.Background(), receiver.Settings{
+				ID: component.ID{},
+				TelemetrySettings: component.TelemetrySettings{
+					Logger: zap.NewNop(),
+				},
+				BuildInfo: component.BuildInfo{},
+			}, cfg, sink)
+			err := r.Start(context.Background(), componenttest.NewNopHost())
+			assert.NoError(t, err)
+			defer func() {
+				assert.NoError(t, r.Shutdown(context.Background()))
+			}()
+			stats := <-doneCh
+			want := maxReplay * bytes.Count(demoTraces, []byte("\n"))
+			assert.Equal(t, want, stats.Requests)
+			assert.Equal(t, want, len(sink.AllTraces()))
+			assert.Equal(t, sink.SpanCount(), stats.Spans)
+		})
+	}
 }