[testbed] - Add scenarios to handle large files

testbed/testbed/validator.go

@@ -8,10 +8,12 @@
 	"log"
 	"reflect"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 )
@@ -561,3 +563,73 @@
 func traceIDAndSpanIDToString(traceID pcommon.TraceID, spanID pcommon.SpanID) string {
 	return fmt.Sprintf("%s-%s", traceID, spanID)
 }
+
+type CorrectnessLogTestValidator struct {
+	dataProvider DataProvider
+}
+
+func NewCorrectnessLogTestValidator(provider DataProvider) *CorrectnessLogTestValidator {
+	return &CorrectnessLogTestValidator{
+		dataProvider: provider,
+	}
+}
+
+func (c *CorrectnessLogTestValidator) Validate(tc *TestCase) {
+	if dataProvider, ok := c.dataProvider.(*perfTestDataProvider); ok {
+		logsReceived := tc.MockBackend.ReceivedLogs
+
+		idsSent := make([][2]string, 0)
+		idsReceived := make([][2]string, 0)
+
+		for batch := 0; batch < int(dataProvider.traceIDSequence.Load()); batch++ {
+			for idx := 0; idx < dataProvider.options.ItemsPerBatch; idx++ {
+				idsSent = append(idsSent, [2]string{"batch_" + strconv.Itoa(batch), "item_" + strconv.Itoa(idx)})
+			}
+		}
+		for _, log := range logsReceived {
+			for i := 0; i < log.ResourceLogs().Len(); i++ {
+				for j := 0; j < log.ResourceLogs().At(i).ScopeLogs().Len(); j++ {
+					s := log.ResourceLogs().At(i).ScopeLogs().At(j)
+					for k := 0; k < s.LogRecords().Len(); k++ {
+						logRecord := s.LogRecords().At(k)
+						batchIndex, ok := logRecord.Attributes().Get("batch_index")
+						require.True(tc.t, ok, "batch_index missing from attributes; use perfDataProvider")
+						itemIndex, ok := logRecord.Attributes().Get("item_index")
+						require.True(tc.t, ok, "item_index missing from attributes; use perfDataProvider")
+
+						idsReceived = append(idsReceived, [2]string{batchIndex.Str(), itemIndex.Str()})
+					}
+				}
+			}
+		}
+
+		assert.ElementsMatch(tc.t, idsSent, idsReceived)
+	}
+}
+
+func (v *CorrectnessLogTestValidator) RecordResults(tc *TestCase) {
+	rc := tc.agentProc.GetTotalConsumption()
+
+	var result string
+	if tc.t.Failed() {
+		result = "FAIL"
+	} else {
+		result = "PASS"
+	}
+
+	// Remove "Test" prefix from test name.
+	testName := tc.t.Name()[4:]
+
+	tc.resultsSummary.Add(tc.t.Name(), &PerformanceTestResult{
+		testName:          testName,
+		result:            result,
+		receivedSpanCount: tc.MockBackend.DataItemsReceived(),
+		sentSpanCount:     tc.LoadGenerator.DataItemsSent(),
+		duration:          time.Since(tc.startTime),
+		cpuPercentageAvg:  rc.CPUPercentAvg,
+		cpuPercentageMax:  rc.CPUPercentMax,
+		ramMibAvg:         rc.RAMMiBAvg,
+		ramMibMax:         rc.RAMMiBMax,
+		errorCause:        tc.errorCause,
+	})
+}

testbed/tests/log_test.go

@@ -7,12 +7,20 @@
 package tests
 
 import (
+	"context"
+	"fmt"
+	"path"
+	"path/filepath"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/testutil"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/testbed/datareceivers"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/testbed/datasenders"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/testbed/testbed"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/plog"
 )
 
 func TestLog10kDPS(t *testing.T) {
@@ -233,3 +241,116 @@ func TestLogOtlpSendingQueue(t *testing.T) {
 	})
 
 }
+
+func TestLogLargeFiles(t *testing.T) {
+	tests := []struct {
+		name         string
+		sender       testbed.DataSender
+		receiver     testbed.DataReceiver
+		loadOptions  testbed.LoadOptions
+		sleepSeconds int
+	}{
+		{
+			name:     "filelog-largefiles-2Gb-lifetime",
+			sender:   datasenders.NewFileLogWriter(),
+			receiver: testbed.NewOTLPDataReceiver(testutil.GetAvailablePort(t)),
+			loadOptions: testbed.LoadOptions{
+				DataItemsPerSecond: 200000,
+				ItemsPerBatch:      1,
+				Parallel:           100,
+			},
+			sleepSeconds: 100,
+		},
+		{
+			name:     "filelog-largefiles-6GB-lifetime",
+			sender:   datasenders.NewFileLogWriter(),
+			receiver: testbed.NewOTLPDataReceiver(testutil.GetAvailablePort(t)),
+			loadOptions: testbed.LoadOptions{
+				DataItemsPerSecond: 330000,
+				ItemsPerBatch:      10,
+				Parallel:           10,
+			},
+			sleepSeconds: 200,
+		},
+		{
+			name:     "filelog-largefiles-50MB/sec",
+			sender:   datasenders.NewFileLogWriter(),
+			receiver: testbed.NewOTLPDataReceiver(testutil.GetAvailablePort(t)),
+			loadOptions: testbed.LoadOptions{
+				DataItemsPerSecond: 400000,
+				ItemsPerBatch:      100,
+				Parallel:           1,
+			},
+			sleepSeconds: 100,
+		},
+	}
+	processors := map[string]string{
+		"batch": `
+  batch:
+`}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ScenarioLong(
+				t,
+				test.sender,
+				test.receiver,
+				test.loadOptions,
+				performanceResultsSummary,
+				test.sleepSeconds,
+				processors,
+			)
+		})
+	}
+}
+
+func TestLargeFileOnce(t *testing.T) {
+	processors := map[string]string{
+		"batch": `
+  batch:
+`,
+	}
+	resultDir, err := filepath.Abs(path.Join("results", t.Name()))
+	require.NoError(t, err)
+	sender := datasenders.NewFileLogWriter()
+	receiver := testbed.NewOTLPDataReceiver(testutil.GetAvailablePort(t))
+	loadOptions := testbed.LoadOptions{
+		DataItemsPerSecond: 1,
+		ItemsPerBatch:      10000000,
+		Parallel:           1,
+	}
+
+	// Write data at once, before starting up the collector
+	dataProvider := testbed.NewPerfTestDataProvider(loadOptions)
+	dataItemsGenerated := atomic.Uint64{}
+	dataProvider.SetLoadGeneratorCounters(&dataItemsGenerated)
+	ld, _ := dataProvider.GenerateLogs()
+
+	m := &plog.ProtoMarshaler{}
+	fmt.Println(m.LogsSize(ld))
+	require.NoError(t, sender.ConsumeLogs(context.Background(), ld))
+	agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
+
+	configStr := createConfigYaml(t, sender, receiver, resultDir, processors, nil)
+	configCleanup, err := agentProc.PrepareConfig(configStr)
+	require.NoError(t, err)
+	defer configCleanup()
+
+	tc := testbed.NewTestCase(
+		t,
+		dataProvider,
+		sender,
+		receiver,
+		agentProc,
+		&testbed.CorrectnessLogTestValidator{},
+		performanceResultsSummary,
+	)
+	defer tc.Stop()
+
+	tc.StartBackend()
+	tc.StartAgent()
+
+	tc.WaitForN(func() bool { return dataItemsGenerated.Load() == tc.MockBackend.DataItemsReceived() }, 20*time.Second, "all logs received")
+
+	tc.StopAgent()
+	tc.ValidateData()
+}

testbed/tests/scenarios.go

@@ -501,6 +501,53 @@ func ScenarioSendingQueuesNotFull(
 	tc.ValidateData()
 }
 
+func ScenarioLong(
+	t *testing.T,
+	sender testbed.DataSender,
+	receiver testbed.DataReceiver,
+	loadOptions testbed.LoadOptions,
+	resultsSummary testbed.TestResultsSummary,
+	sleepTime int,
+	processors map[string]string,
+) {
+	resultDir, err := filepath.Abs(path.Join("results", t.Name()))
+	require.NoError(t, err)
+
+	agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
+
+	configStr := createConfigYaml(t, sender, receiver, resultDir, processors, nil)
+	configCleanup, err := agentProc.PrepareConfig(configStr)
+	require.NoError(t, err)
+	defer configCleanup()
+	dataProvider := testbed.NewPerfTestDataProvider(loadOptions)
+	tc := testbed.NewTestCase(
+		t,
+		dataProvider,
+		sender,
+		receiver,
+		agentProc,
+		&testbed.CorrectnessLogTestValidator{},
+		resultsSummary,
+	)
+	defer tc.Stop()
+
+	tc.StartBackend()
+	tc.StartAgent()
+
+	tc.StartLoad(loadOptions)
+
+	tc.WaitFor(func() bool { return tc.LoadGenerator.DataItemsSent() > 0 }, "load generator started")
+
+	tc.Sleep(time.Second * time.Duration(sleepTime))
+
+	tc.StopLoad()
+
+	tc.WaitFor(func() bool { return tc.LoadGenerator.DataItemsSent() == tc.MockBackend.DataItemsReceived() }, "all spans received")
+
+	tc.StopAgent()
+	tc.ValidateData()
+}
+
 func constructLoadOptions(test TestCase) testbed.LoadOptions {
 	options := testbed.LoadOptions{DataItemsPerSecond: 1000, ItemsPerBatch: 10}
 	options.Attributes = make(map[string]string)