testbed: add and adopt WithEnvVar for child process (#30491)

**Description:**
Adding a feature - These changes add a new `WithEnvVar`
`ChildProcessOption` to be able to influence the child process
environment without acting on the current environment. They also move
the `GOMAXPROCS=2` setting added in
dd8e010
to each invoking test because though being helpful to address
#27429,
the constraint doesn't seem applicable to the helper directly. Limiting
the utility as a whole instead of specifying in the test context cannot
be easily worked around and is interfering w/ some load testing efforts.

Author: rmfitzpatrick

.chloggen/testbedmaxprocs.yaml

@@ -0,0 +1,27 @@
+# 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. filelogreceiver)
+component: testbed
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds and adopts new WithEnvVar child process option, moving GOMAXPROCS=2 to initializations
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [30491]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# 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: [api]

testbed/testbed/child_process_collector.go

@@ -13,6 +13,7 @@ import (
 	"path"
 	"path/filepath"
 	"runtime"
+	"sort"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -41,6 +42,9 @@ type childProcessCollector struct {
 	// Command to execute
 	cmd *exec.Cmd
 
+	// additional env vars (os.Environ() populated by default)
+	additionalEnv map[string]string
+
 	// Various starting/stopping flags
 	isStarted  bool
 	stopOnce   sync.Once
@@ -85,7 +89,7 @@ type ChildProcessOption func(*childProcessCollector)
 
 // NewChildProcessCollector creates a new OtelcolRunner as a child process on the same machine executing the test.
 func NewChildProcessCollector(options ...ChildProcessOption) OtelcolRunner {
-	col := &childProcessCollector{}
+	col := &childProcessCollector{additionalEnv: map[string]string{}}
 
 	for _, option := range options {
 		option(col)
@@ -101,6 +105,13 @@ func WithAgentExePath(exePath string) ChildProcessOption {
 	}
 }
 
+// WithEnvVar sets an additional environment variable for the process
+func WithEnvVar(k, v string) ChildProcessOption {
+	return func(cpc *childProcessCollector) {
+		cpc.additionalEnv[k] = v
+	}
+}
+
 func (cp *childProcessCollector) PrepareConfig(configStr string) (configCleanup func(), err error) {
 	configCleanup = func() {
 		// NoOp
@@ -204,9 +215,17 @@ func (cp *childProcessCollector) Start(params StartParams) error {
 	}
 	// #nosec
 	cp.cmd = exec.Command(exePath, args...)
-	cp.cmd.Env = append(os.Environ(),
-		"GOMAXPROCS=2",
-	)
+	cp.cmd.Env = os.Environ()
+
+	// update env deterministically
+	var additionalEnvVars []string
+	for k := range cp.additionalEnv {
+		additionalEnvVars = append(additionalEnvVars, k)
+	}
+	sort.Strings(additionalEnvVars)
+	for _, k := range additionalEnvVars {
+		cp.cmd.Env = append(cp.cmd.Env, fmt.Sprintf("%s=%s", k, cp.additionalEnv[k]))
+	}
 
 	// Capture standard output and standard error.
 	cp.cmd.Stdout = logFile

testbed/testbed/child_process_collector_test.go

@@ -4,6 +4,7 @@
 package testbed // import "github.com/open-telemetry/opentelemetry-collector-contrib/testbed/testbed"
 
 import (
+	"os"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -18,3 +19,31 @@ func TestAgentExeOption(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, path, cpc.agentExePath)
 }
+
+func TestAgentEnvVarOption(t *testing.T) {
+	col := NewChildProcessCollector(
+		WithAgentExePath("not-real"),
+		WithEnvVar("var-one", "var-one-value"),
+		WithEnvVar("var-two", "var-two-value"),
+		WithEnvVar("var-two", "actual-var-two-value"),
+	)
+
+	cpc, ok := col.(*childProcessCollector)
+	require.True(t, ok)
+
+	expeectedEnvVarMap := map[string]string{
+		"var-one": "var-one-value",
+		"var-two": "actual-var-two-value",
+	}
+	require.Equal(t, expeectedEnvVarMap, cpc.additionalEnv)
+
+	// results from `not-real` not being found but contents unrelated to this test
+	require.Error(t, col.Start(
+		StartParams{
+			CmdArgs:     []string{"--config", "to-prevent-builder"},
+			LogFilePath: "/dev/null",
+		},
+	))
+	expectedEnvVars := append(os.Environ(), "var-one=var-one-value", "var-two=actual-var-two-value")
+	require.Equal(t, expectedEnvVars, cpc.cmd.Env)
+}

testbed/tests/e2e_test.go

@@ -29,7 +29,8 @@ func TestIdleMode(t *testing.T) {
 	sender := testbed.NewOTLPTraceDataSender(testbed.DefaultHost, testbed.GetAvailablePort(t))
 	receiver := testbed.NewOTLPDataReceiver(testbed.GetAvailablePort(t))
 	cfg := createConfigYaml(t, sender, receiver, resultDir, nil, nil)
-	cp := testbed.NewChildProcessCollector()
+	cp := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
+
 	cleanup, err := cp.PrepareConfig(cfg)
 	require.NoError(t, err)
 	t.Cleanup(cleanup)
@@ -77,7 +78,7 @@ func TestBallastMemory(t *testing.T) {
 			ballastCfg := createConfigYaml(
 				t, sender, receiver, resultDir, nil,
 				map[string]string{"memory_ballast": fmt.Sprintf(ballastConfig, test.ballastSize)})
-			cp := testbed.NewChildProcessCollector()
+			cp := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 			cleanup, err := cp.PrepareConfig(ballastCfg)
 			require.NoError(t, err)
 			t.Cleanup(cleanup)

testbed/tests/metric_test.go

@@ -109,7 +109,7 @@ func TestMetricsFromFile(t *testing.T) {
 		// ItemsPerBatch is based on the data from the file.
 		ItemsPerBatch: dataProvider.ItemsPerBatch,
 	}
-	agentProc := testbed.NewChildProcessCollector()
+	agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 
 	sender := testbed.NewOTLPMetricDataSender(testbed.DefaultHost, testbed.GetAvailablePort(t))
 	receiver := testbed.NewOTLPDataReceiver(testbed.GetAvailablePort(t))

testbed/tests/resource_processor_test.go

@@ -107,7 +107,7 @@ func TestMetricResourceProcessor(t *testing.T) {
 			resultDir, err := filepath.Abs(filepath.Join("results", t.Name()))
 			require.NoError(t, err)
 
-			agentProc := testbed.NewChildProcessCollector()
+			agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 			processors := map[string]string{
 				"resource": test.resourceProcessorConfig,
 			}

testbed/tests/scenarios.go

@@ -143,7 +143,7 @@ func Scenario10kItemsPerSecond(
 		ItemsPerBatch:      100,
 		Parallel:           1,
 	}
-	agentProc := testbed.NewChildProcessCollector()
+	agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 
 	configStr := createConfigYaml(t, sender, receiver, resultDir, processors, extensions)
 	configCleanup, err := agentProc.PrepareConfig(configStr)
@@ -201,7 +201,7 @@ func Scenario10kItemsPerSecondAlternateBackend(
 		ItemsPerBatch:      100,
 		Parallel:           1,
 	}
-	agentProc := testbed.NewChildProcessCollector()
+	agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 
 	configStr := createConfigYaml(t, sender, receiver, resultDir, processors, extensions)
 	fmt.Println(configStr)
@@ -268,7 +268,7 @@ func Scenario1kSPSWithAttrs(t *testing.T, args []string, tests []TestCase, proce
 
 			options := constructLoadOptions(test)
 
-			agentProc := testbed.NewChildProcessCollector()
+			agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 
 			// Prepare results dir.
 			resultDir, err := filepath.Abs(path.Join("results", t.Name()))
@@ -334,7 +334,7 @@ func ScenarioTestTraceNoBackend10kSPS(
 	require.NoError(t, err)
 
 	options := testbed.LoadOptions{DataItemsPerSecond: 10000, ItemsPerBatch: 10}
-	agentProc := testbed.NewChildProcessCollector()
+	agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 	configStr := createConfigYaml(t, sender, receiver, resultDir, configuration.Processor, nil)
 	configCleanup, err := agentProc.PrepareConfig(configStr)
 	require.NoError(t, err)
@@ -378,7 +378,7 @@ func ScenarioSendingQueuesFull(
 	resultDir, err := filepath.Abs(path.Join("results", t.Name()))
 	require.NoError(t, err)
 
-	agentProc := testbed.NewChildProcessCollector()
+	agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 
 	configStr := createConfigYaml(t, sender, receiver, resultDir, processors, extensions)
 	configCleanup, err := agentProc.PrepareConfig(configStr)
@@ -460,7 +460,7 @@ func ScenarioSendingQueuesNotFull(
 	resultDir, err := filepath.Abs(path.Join("results", t.Name()))
 	require.NoError(t, err)
 
-	agentProc := testbed.NewChildProcessCollector()
+	agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 
 	configStr := createConfigYaml(t, sender, receiver, resultDir, processors, extensions)
 	configCleanup, err := agentProc.PrepareConfig(configStr)

testbed/tests/trace_test.go

@@ -445,7 +445,7 @@ func TestTraceAttributesProcessor(t *testing.T) {
 `,
 			}
 
-			agentProc := testbed.NewChildProcessCollector()
+			agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 			configStr := createConfigYaml(t, test.sender, test.receiver, resultDir, processors, nil)
 			configCleanup, err := agentProc.PrepareConfig(configStr)
 			require.NoError(t, err)
@@ -531,7 +531,7 @@ func TestTraceAttributesProcessorJaegerGRPC(t *testing.T) {
 `,
 	}
 
-	agentProc := testbed.NewChildProcessCollector()
+	agentProc := testbed.NewChildProcessCollector(testbed.WithEnvVar("GOMAXPROCS", "2"))
 	configStr := createConfigYaml(t, sender, receiver, resultDir, processors, nil)
 	configCleanup, err := agentProc.PrepareConfig(configStr)
 	require.NoError(t, err)