Small cleanups in testbed, hide unnecessary public structs (#3446)

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

testbed/correctness/metrics/correctness_test_case.go

@@ -32,7 +32,7 @@ type correctnessTestCase struct {
 	sender    testbed.DataSender
 	receiver  testbed.DataReceiver
 	harness   *testHarness
-	collector *testbed.InProcessCollector
+	collector testbed.OtelcolRunner
 }
 
 func newCorrectnessTestCase(

testbed/testbed/child_process.go renamed to testbed/testbed/child_process_collector.go

@@ -36,34 +36,9 @@ import (
 	"go.uber.org/atomic"
 )
 
-// ResourceSpec is a resource consumption specification.
-type ResourceSpec struct {
-	// Percentage of one core the process is expected to consume at most.
-	// Test is aborted and failed if consumption during
-	// ResourceCheckPeriod exceeds this number. If 0 the CPU
-	// consumption is not monitored and does not affect the test result.
-	ExpectedMaxCPU uint32
-
-	// Maximum RAM in MiB the process is expected to consume.
-	// Test is aborted and failed if consumption exceeds this number.
-	// If 0 memory consumption is not monitored and does not affect
-	// the test result.
-	ExpectedMaxRAM uint32
-
-	// Period during which CPU and RAM of the process are measured.
-	// Bigger numbers will result in more averaging of short spikes.
-	ResourceCheckPeriod time.Duration
-}
-
-// isSpecified returns true if any part of ResourceSpec is specified,
-// i.e. has non-zero value.
-func (rs *ResourceSpec) isSpecified() bool {
-	return rs != nil && (rs.ExpectedMaxCPU != 0 || rs.ExpectedMaxRAM != 0)
-}
-
-// ChildProcess implements the OtelcolRunner interface as a child process on the same machine executing
+// childProcessCollector implements the OtelcolRunner interface as a child process on the same machine executing
 // the test. The process can be monitored and the output of which will be written to a log file.
-type ChildProcess struct {
+type childProcessCollector struct {
 	// Path to agent executable. If unset the default executable in
 	// bin/otelcol_{{.GOOS}}_{{.GOARCH}} will be used.
 	// Can be set for example to use the unstable executable for a specific test.
@@ -121,21 +96,12 @@ type ChildProcess struct {
 	ramMiBMax uint32
 }
 
-type StartParams struct {
-	Name         string
-	LogFilePath  string
-	CmdArgs      []string
-	resourceSpec *ResourceSpec
-}
-
-type ResourceConsumption struct {
-	CPUPercentAvg float64
-	CPUPercentMax float64
-	RAMMiBAvg     uint32
-	RAMMiBMax     uint32
+// NewChildProcessCollector crewtes a new OtelcolRunner as a child process on the same machine executing the test.
+func NewChildProcessCollector() OtelcolRunner {
+	return &childProcessCollector{}
 }
 
-func (cp *ChildProcess) PrepareConfig(configStr string) (configCleanup func(), err error) {
+func (cp *childProcessCollector) PrepareConfig(configStr string) (configCleanup func(), err error) {
 	configCleanup = func() {
 		// NoOp
 	}
@@ -198,7 +164,7 @@ func expandExeFileName(exeName string) string {
 // logFilePath is the file path to write the standard output and standard error of
 // the process to.
 // cmdArgs is the command line arguments to pass to the process.
-func (cp *ChildProcess) Start(params StartParams) error {
+func (cp *childProcessCollector) Start(params StartParams) error {
 
 	cp.name = params.Name
 	cp.doneSignal = make(chan struct{})
@@ -275,7 +241,7 @@ func (cp *ChildProcess) Start(params StartParams) error {
 	return err
 }
 
-func (cp *ChildProcess) Stop() (stopped bool, err error) {
+func (cp *childProcessCollector) Stop() (stopped bool, err error) {
 	if !cp.isStarted || cp.isStopped {
 		return false, nil
 	}
@@ -341,7 +307,7 @@ func (cp *ChildProcess) Stop() (stopped bool, err error) {
 	return stopped, err
 }
 
-func (cp *ChildProcess) WatchResourceConsumption() error {
+func (cp *childProcessCollector) WatchResourceConsumption() error {
 	if !cp.resourceSpec.isSpecified() {
 		// Resource monitoring is not enabled.
 		return nil
@@ -388,11 +354,11 @@ func (cp *ChildProcess) WatchResourceConsumption() error {
 	}
 }
 
-func (cp *ChildProcess) GetProcessMon() *process.Process {
+func (cp *childProcessCollector) GetProcessMon() *process.Process {
 	return cp.processMon
 }
 
-func (cp *ChildProcess) fetchRAMUsage() {
+func (cp *childProcessCollector) fetchRAMUsage() {
 	// Get process memory and CPU times
 	mi, err := cp.processMon.MemoryInfo()
 	if err != nil {
@@ -415,7 +381,7 @@ func (cp *ChildProcess) fetchRAMUsage() {
 	cp.ramMiBCur.Store(ramMiBCur)
 }
 
-func (cp *ChildProcess) fetchCPUUsage() {
+func (cp *childProcessCollector) fetchCPUUsage() {
 	times, err := cp.processMon.Times()
 	if err != nil {
 		log.Printf("cannot get process times for %d: %s",
@@ -448,7 +414,7 @@ func (cp *ChildProcess) fetchCPUUsage() {
 	cp.cpuPercentX1000Cur.Store(curCPUPercentageX1000)
 }
 
-func (cp *ChildProcess) checkAllowedResourceUsage() error {
+func (cp *childProcessCollector) checkAllowedResourceUsage() error {
 	// Check if current CPU usage exceeds expected.
 	var errMsg string
 	if cp.resourceSpec.ExpectedMaxCPU != 0 && cp.cpuPercentX1000Cur.Load()/1000 > cp.resourceSpec.ExpectedMaxCPU {
@@ -472,7 +438,7 @@ func (cp *ChildProcess) checkAllowedResourceUsage() error {
 }
 
 // GetResourceConsumption returns resource consumption as a string
-func (cp *ChildProcess) GetResourceConsumption() string {
+func (cp *childProcessCollector) GetResourceConsumption() string {
 	if !cp.resourceSpec.isSpecified() {
 		// Monitoring is not enabled.
 		return ""
@@ -486,7 +452,7 @@ func (cp *ChildProcess) GetResourceConsumption() string {
 }
 
 // GetTotalConsumption returns total resource consumption since start of process
-func (cp *ChildProcess) GetTotalConsumption() *ResourceConsumption {
+func (cp *childProcessCollector) GetTotalConsumption() *ResourceConsumption {
 	rc := &ResourceConsumption{}
 
 	if cp.processMon != nil {

testbed/testbed/in_process_collector.go

@@ -0,0 +1,137 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package testbed
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/shirou/gopsutil/process"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/internal/version"
+	"go.opentelemetry.io/collector/service"
+	"go.opentelemetry.io/collector/service/parserprovider"
+)
+
+// inProcessCollector implements the OtelcolRunner interfaces running a single otelcol as a go routine within the
+// same process as the test executor.
+type inProcessCollector struct {
+	logger    *zap.Logger
+	factories component.Factories
+	configStr string
+	svc       *service.Collector
+	appDone   chan struct{}
+	stopped   bool
+}
+
+// NewInProcessCollector crewtes a new inProcessCollector using the supplied component factories.
+func NewInProcessCollector(factories component.Factories) OtelcolRunner {
+	return &inProcessCollector{
+		factories: factories,
+	}
+}
+
+func (ipp *inProcessCollector) PrepareConfig(configStr string) (configCleanup func(), err error) {
+	configCleanup = func() {
+		// NoOp
+	}
+	var logger *zap.Logger
+	logger, err = configureLogger()
+	if err != nil {
+		return configCleanup, err
+	}
+	ipp.logger = logger
+	ipp.configStr = configStr
+	return configCleanup, err
+}
+
+func (ipp *inProcessCollector) Start(args StartParams) error {
+	settings := service.CollectorSettings{
+		BuildInfo: component.BuildInfo{
+			Command: "otelcol",
+			Version: version.Version,
+		},
+		Factories:      ipp.factories,
+		ParserProvider: parserprovider.NewInMemory(strings.NewReader(ipp.configStr)),
+	}
+	var err error
+	ipp.svc, err = service.New(settings)
+	if err != nil {
+		return err
+	}
+	ipp.svc.Command().SetArgs(args.CmdArgs)
+
+	ipp.appDone = make(chan struct{})
+	go func() {
+		defer close(ipp.appDone)
+		appErr := ipp.svc.Run()
+		if appErr != nil {
+			err = appErr
+		}
+	}()
+
+	for state := range ipp.svc.GetStateChannel() {
+		switch state {
+		case service.Starting:
+			// NoOp
+		case service.Running:
+			return err
+		default:
+			err = fmt.Errorf("unable to start, otelcol state is %d", state)
+		}
+	}
+	return err
+}
+
+func (ipp *inProcessCollector) Stop() (stopped bool, err error) {
+	if !ipp.stopped {
+		ipp.stopped = true
+		ipp.svc.Shutdown()
+	}
+	<-ipp.appDone
+	stopped = ipp.stopped
+	return stopped, err
+}
+
+func (ipp *inProcessCollector) WatchResourceConsumption() error {
+	return nil
+}
+
+func (ipp *inProcessCollector) GetProcessMon() *process.Process {
+	return nil
+}
+
+func (ipp *inProcessCollector) GetTotalConsumption() *ResourceConsumption {
+	return &ResourceConsumption{
+		CPUPercentAvg: 0,
+		CPUPercentMax: 0,
+		RAMMiBAvg:     0,
+		RAMMiBMax:     0,
+	}
+}
+
+func (ipp *inProcessCollector) GetResourceConsumption() string {
+	return ""
+}
+
+func configureLogger() (*zap.Logger, error) {
+	conf := zap.NewDevelopmentConfig()
+	conf.Level.SetLevel(zapcore.InfoLevel)
+	logger, err := conf.Build()
+	return logger, err
+}

testbed/testbed/otelcol_runner_test.go renamed to testbed/testbed/in_process_collector_test.go

@@ -29,7 +29,8 @@ func TestNewInProcessPipeline(t *testing.T) {
 	assert.NoError(t, err)
 	sender := NewOTLPTraceDataSender(DefaultHost, GetAvailablePort(t))
 	receiver := NewOTLPDataReceiver(DefaultOTLPPort)
-	runner := NewInProcessCollector(factories)
+	runner, ok := NewInProcessCollector(factories).(*inProcessCollector)
+	require.True(t, ok)
 
 	format := `
 receivers:%v

testbed/testbed/options.go

@@ -17,6 +17,35 @@
 
 package testbed
 
+import (
+	"time"
+)
+
+// ResourceSpec is a resource consumption specification.
+type ResourceSpec struct {
+	// Percentage of one core the process is expected to consume at most.
+	// Test is aborted and failed if consumption during
+	// ResourceCheckPeriod exceeds this number. If 0 the CPU
+	// consumption is not monitored and does not affect the test result.
+	ExpectedMaxCPU uint32
+
+	// Maximum RAM in MiB the process is expected to consume.
+	// Test is aborted and failed if consumption exceeds this number.
+	// If 0 memory consumption is not monitored and does not affect
+	// the test result.
+	ExpectedMaxRAM uint32
+
+	// Period during which CPU and RAM of the process are measured.
+	// Bigger numbers will result in more averaging of short spikes.
+	ResourceCheckPeriod time.Duration
+}
+
+// isSpecified returns true if any part of ResourceSpec is specified,
+// i.e. has non-zero value.
+func (rs *ResourceSpec) isSpecified() bool {
+	return rs != nil && (rs.ExpectedMaxCPU != 0 || rs.ExpectedMaxRAM != 0)
+}
+
 // TestCaseOption defines a TestCase option.
 type TestCaseOption func(t *TestCase)