[cmd/telemetrygen] Export telemetrygen functions for testing

.chloggen/telemetrygen-public-api.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: telemetrygen
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Exported the API for telemetrygen for test uses. Additionally added new E2E tests and fixed race condition
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [36984]
+
+# (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]

cmd/telemetrygen/config.go

@@ -12,10 +12,10 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/logs"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/metadata"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/metrics"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/traces"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/pkg/logs"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/pkg/metrics"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/pkg/traces"
 )
 
 var (
@@ -64,13 +64,13 @@ var logsCmd = &cobra.Command{
 func init() {
 	rootCmd.AddCommand(tracesCmd, metricsCmd, logsCmd)
 
-	tracesCfg = new(traces.Config)
+	tracesCfg = traces.NewConfig()
 	tracesCfg.Flags(tracesCmd.Flags())
 
-	metricsCfg = new(metrics.Config)
+	metricsCfg = metrics.NewConfig()
 	metricsCfg.Flags(metricsCmd.Flags())
 
-	logsCfg = new(logs.Config)
+	logsCfg = logs.NewConfig()
 	logsCfg.Flags(logsCmd.Flags())
 
 	// Disabling completion command for end user
@@ -81,7 +81,7 @@ func init() {
 // Execute tries to run the input command
 func Execute() {
 	if err := rootCmd.Execute(); err != nil {
-		// TODO: Uncomment the line below when using Run instead of RunE in the xxxCmd functions
+		// TODO: Uncomment the line below when using run instead of RunE in the xxxCmd functions
 		// fmt.Fprintln(os.Stderr, err)
 		os.Exit(1)
 	}

cmd/telemetrygen/internal/common/config.go

@@ -148,40 +148,59 @@ func (c *Config) GetHeaders() map[string]string {
 
 // CommonFlags registers common config flags.
 func (c *Config) CommonFlags(fs *pflag.FlagSet) {
-	fs.IntVar(&c.WorkerCount, "workers", 1, "Number of workers (goroutines) to run")
-	fs.Float64Var(&c.Rate, "rate", 0, "Approximately how many metrics/spans/logs per second each worker should generate. Zero means no throttling.")
-	fs.DurationVar(&c.TotalDuration, "duration", 0, "For how long to run the test")
-	fs.DurationVar(&c.ReportingInterval, "interval", 1*time.Second, "Reporting interval")
+	fs.IntVar(&c.WorkerCount, "workers", c.WorkerCount, "Number of workers (goroutines) to run")
+	fs.Float64Var(&c.Rate, "rate", c.Rate, "Approximately how many metrics/spans/logs per second each worker should generate. Zero means no throttling.")
+	fs.DurationVar(&c.TotalDuration, "duration", c.TotalDuration, "For how long to run the test")
+	fs.DurationVar(&c.ReportingInterval, "interval", c.ReportingInterval, "Reporting interval")
 
-	fs.StringVar(&c.CustomEndpoint, "otlp-endpoint", "", "Destination endpoint for exporting logs, metrics and traces")
-	fs.BoolVar(&c.Insecure, "otlp-insecure", false, "Whether to enable client transport security for the exporter's grpc or http connection")
-	fs.BoolVar(&c.InsecureSkipVerify, "otlp-insecure-skip-verify", false, "Whether a client verifies the server's certificate chain and host name")
-	fs.BoolVar(&c.UseHTTP, "otlp-http", false, "Whether to use HTTP exporter rather than a gRPC one")
+	fs.StringVar(&c.CustomEndpoint, "otlp-endpoint", c.CustomEndpoint, "Destination endpoint for exporting logs, metrics and traces")
+	fs.BoolVar(&c.Insecure, "otlp-insecure", c.Insecure, "Whether to enable client transport security for the exporter's grpc or http connection")
+	fs.BoolVar(&c.InsecureSkipVerify, "otlp-insecure-skip-verify", c.InsecureSkipVerify, "Whether a client verifies the server's certificate chain and host name")
+	fs.BoolVar(&c.UseHTTP, "otlp-http", c.UseHTTP, "Whether to use HTTP exporter rather than a gRPC one")
 
 	// custom headers
-	c.Headers = make(KeyValue)
 	fs.Var(&c.Headers, "otlp-header", "Custom header to be passed along with each OTLP request. The value is expected in the format key=\"value\". "+
 		"Note you may need to escape the quotes when using the tool from a cli. "+
 		`Flag may be repeated to set multiple headers (e.g --otlp-header key1=\"value1\" --otlp-header key2=\"value2\")`)
 
 	// custom resource attributes
-	c.ResourceAttributes = make(KeyValue)
 	fs.Var(&c.ResourceAttributes, "otlp-attributes", "Custom resource attributes to use. The value is expected in the format key=\"value\". "+
 		"You can use key=true or key=false. to set boolean attribute."+
 		"Note you may need to escape the quotes when using the tool from a cli. "+
 		`Flag may be repeated to set multiple attributes (e.g --otlp-attributes key1=\"value1\" --otlp-attributes key2=\"value2\" --telemetry-attributes key3=true)`)
 
-	c.TelemetryAttributes = make(KeyValue)
 	fs.Var(&c.TelemetryAttributes, "telemetry-attributes", "Custom telemetry attributes to use. The value is expected in the format key=\"value\". "+
 		"You can use key=true or key=false. to set boolean attribute."+
 		"Note you may need to escape the quotes when using the tool from a cli. "+
 		`Flag may be repeated to set multiple attributes (e.g --telemetry-attributes key1=\"value1\" --telemetry-attributes key2=\"value2\" --telemetry-attributes key3=true)`)
 
 	// TLS CA configuration
-	fs.StringVar(&c.CaFile, "ca-cert", "", "Trusted Certificate Authority to verify server certificate")
+	fs.StringVar(&c.CaFile, "ca-cert", c.CaFile, "Trusted Certificate Authority to verify server certificate")
 
 	// mTLS configuration
-	fs.BoolVar(&c.ClientAuth.Enabled, "mtls", false, "Whether to require client authentication for mTLS")
-	fs.StringVar(&c.ClientAuth.ClientCertFile, "client-cert", "", "Client certificate file")
-	fs.StringVar(&c.ClientAuth.ClientKeyFile, "client-key", "", "Client private key file")
+	fs.BoolVar(&c.ClientAuth.Enabled, "mtls", c.ClientAuth.Enabled, "Whether to require client authentication for mTLS")
+	fs.StringVar(&c.ClientAuth.ClientCertFile, "client-cert", c.ClientAuth.ClientCertFile, "Client certificate file")
+	fs.StringVar(&c.ClientAuth.ClientKeyFile, "client-key", c.ClientAuth.ClientKeyFile, "Client private key file")
+}
+
+// SetDefaults is here to mirror the defaults for flags above,
+// This allows for us to have a single place to change the defaults
+// while exposing the API for use.
+func (c *Config) SetDefaults() {
+	c.WorkerCount = 1
+	c.Rate = 0
+	c.TotalDuration = 0
+	c.ReportingInterval = 1 * time.Second
+	c.CustomEndpoint = ""
+	c.Insecure = false
+	c.InsecureSkipVerify = false
+	c.UseHTTP = false
+	c.HTTPPath = ""
+	c.Headers = make(KeyValue)
+	c.ResourceAttributes = make(KeyValue)
+	c.TelemetryAttributes = make(KeyValue)
+	c.CaFile = ""
+	c.ClientAuth.Enabled = false
+	c.ClientAuth.ClientCertFile = ""
+	c.ClientAuth.ClientKeyFile = ""
 }

cmd/telemetrygen/internal/e2etest/go.mod

@@ -64,14 +64,21 @@ require (
 	go.opentelemetry.io/collector/pipeline v0.117.1-0.20250117002813-e970f8bb1258 // indirect
 	go.opentelemetry.io/collector/receiver v0.117.1-0.20250117002813-e970f8bb1258 // indirect
 	go.opentelemetry.io/collector/receiver/xreceiver v0.117.1-0.20250117002813-e970f8bb1258 // indirect
+	go.opentelemetry.io/collector/semconv v0.117.1-0.20250117002813-e970f8bb1258 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.56.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.56.0 // indirect
 	go.opentelemetry.io/otel v1.32.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.8.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.8.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.32.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.32.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.32.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.32.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.32.0 // indirect
+	go.opentelemetry.io/otel/log v0.8.0 // indirect
 	go.opentelemetry.io/otel/metric v1.32.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.32.0 // indirect
+	go.opentelemetry.io/otel/sdk/log v0.8.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.32.0 // indirect
 	go.opentelemetry.io/otel/trace v1.32.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.5.0 // indirect

cmd/telemetrygen/internal/e2etest/logs_test.go

@@ -0,0 +1,51 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package e2etest
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver"
+	"go.opentelemetry.io/collector/receiver/receivertest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/pkg/logs"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/testutil"
+)
+
+func TestGenerateLogs(t *testing.T) {
+	f := otlpreceiver.NewFactory()
+	sink := &consumertest.LogsSink{}
+	rCfg := f.CreateDefaultConfig()
+	endpoint := testutil.GetAvailableLocalAddress(t)
+	rCfg.(*otlpreceiver.Config).GRPC.NetAddr.Endpoint = endpoint
+	r, err := f.CreateLogs(context.Background(), receivertest.NewNopSettings(), rCfg, sink)
+	require.NoError(t, err)
+	err = r.Start(context.Background(), componenttest.NewNopHost())
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, r.Shutdown(context.Background()))
+	}()
+	cfg := logs.NewConfig()
+	cfg.WorkerCount = 10
+	cfg.Rate = 10
+	cfg.TotalDuration = 10 * time.Second
+	cfg.ReportingInterval = 10
+	cfg.CustomEndpoint = endpoint
+	cfg.Insecure = true
+	cfg.SkipSettingGRPCLogger = true
+	cfg.NumLogs = 6000
+	go func() {
+		err = logs.Start(cfg)
+		assert.NoError(t, err)
+	}()
+	require.Eventually(t, func() bool {
+		return len(sink.AllLogs()) > 0
+	}, 10*time.Second, 100*time.Millisecond)
+}

cmd/telemetrygen/internal/e2etest/metrics_test.go

@@ -0,0 +1,51 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package e2etest
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver"
+	"go.opentelemetry.io/collector/receiver/receivertest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/pkg/metrics"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/testutil"
+)
+
+func TestGenerateMetrics(t *testing.T) {
+	f := otlpreceiver.NewFactory()
+	sink := &consumertest.MetricsSink{}
+	rCfg := f.CreateDefaultConfig()
+	endpoint := testutil.GetAvailableLocalAddress(t)
+	rCfg.(*otlpreceiver.Config).GRPC.NetAddr.Endpoint = endpoint
+	r, err := f.CreateMetrics(context.Background(), receivertest.NewNopSettings(), rCfg, sink)
+	require.NoError(t, err)
+	err = r.Start(context.Background(), componenttest.NewNopHost())
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, r.Shutdown(context.Background()))
+	}()
+	cfg := metrics.NewConfig()
+	cfg.WorkerCount = 10
+	cfg.Rate = 10
+	cfg.TotalDuration = 10 * time.Second
+	cfg.ReportingInterval = 10
+	cfg.CustomEndpoint = endpoint
+	cfg.Insecure = true
+	cfg.SkipSettingGRPCLogger = true
+	cfg.NumMetrics = 6000
+	go func() {
+		err = metrics.Start(cfg)
+		assert.NoError(t, err)
+	}()
+	require.Eventually(t, func() bool {
+		return len(sink.AllMetrics()) > 0
+	}, 10*time.Second, 100*time.Millisecond)
+}

cmd/telemetrygen/internal/e2etest/e2e_test.go → cmd/telemetrygen/internal/e2etest/traces_test.go

@@ -15,8 +15,7 @@ import (
 	"go.opentelemetry.io/collector/receiver/otlpreceiver"
 	"go.opentelemetry.io/collector/receiver/receivertest"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/common"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/traces"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/pkg/traces"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/testutil"
 )
 
@@ -33,25 +32,15 @@ func TestGenerateTraces(t *testing.T) {
 	defer func() {
 		require.NoError(t, r.Shutdown(context.Background()))
 	}()
-	cfg := &traces.Config{
-		Config: common.Config{
-			WorkerCount:           10,
-			Rate:                  10,
-			TotalDuration:         10 * time.Second,
-			ReportingInterval:     10,
-			CustomEndpoint:        endpoint,
-			Insecure:              true,
-			UseHTTP:               false,
-			Headers:               nil,
-			ResourceAttributes:    nil,
-			SkipSettingGRPCLogger: true,
-		},
-		NumTraces:   6000,
-		ServiceName: "foo",
-		StatusCode:  "0",
-		LoadSize:    0,
-		Batch:       true,
-	}
+	cfg := traces.NewConfig()
+	cfg.WorkerCount = 10
+	cfg.Rate = 10
+	cfg.TotalDuration = 10 * time.Second
+	cfg.ReportingInterval = 10
+	cfg.CustomEndpoint = endpoint
+	cfg.Insecure = true
+	cfg.SkipSettingGRPCLogger = true
+	cfg.NumTraces = 6000
 	go func() {
 		err = traces.Start(cfg)
 		assert.NoError(t, err)