[cmd/telemetrygen] Use exporter per worker for better metrics throughput

.chloggen/telemetrygen-add-exporter-per-worker.yaml

@@ -0,0 +1,25 @@
+# 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: cmd/telemetrygen
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Use exporter per worker for better metrics throughput
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [26709]
+
+# (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: [user]

cmd/telemetrygen/internal/metrics/metrics.go

@@ -52,26 +52,19 @@ func Start(cfg *Config) error {
 		httpExpOpt = append(httpExpOpt, otlpmetrichttp.WithHeaders(cfg.Headers))
 	}
 
-	var exp sdkmetric.Exporter
-	if cfg.UseHTTP {
-		logger.Info("starting HTTP exporter")
-		exp, err = otlpmetrichttp.New(context.Background(), httpExpOpt...)
-	} else {
-		logger.Info("starting gRPC exporter")
-		exp, err = otlpmetricgrpc.New(context.Background(), grpcExpOpt...)
-	}
-
-	if err != nil {
-		return fmt.Errorf("failed to obtain OTLP exporter: %w", err)
-	}
-	defer func() {
-		logger.Info("stopping the exporter")
-		if tempError := exp.Shutdown(context.Background()); tempError != nil {
-			logger.Error("failed to stop the exporter", zap.Error(tempError))
+	expFunc := func() (sdkmetric.Exporter, error) {
+		var exp sdkmetric.Exporter
+		if cfg.UseHTTP {
+			logger.Info("starting HTTP exporter")
+			exp, err = otlpmetrichttp.New(context.Background(), httpExpOpt...)
+		} else {
+			logger.Info("starting gRPC exporter")
+			exp, err = otlpmetricgrpc.New(context.Background(), grpcExpOpt...)
 		}
-	}()
+		return exp, err
+	}
 
-	if err = Run(cfg, exp, logger); err != nil {
+	if err = Run(cfg, expFunc, logger); err != nil {
 		logger.Error("failed to stop the exporter", zap.Error(err))
 		return err
 	}
@@ -80,7 +73,7 @@ func Start(cfg *Config) error {
 }
 
 // Run executes the test scenario.
-func Run(c *Config, exp sdkmetric.Exporter, logger *zap.Logger) error {
+func Run(c *Config, exp func() (sdkmetric.Exporter, error), logger *zap.Logger) error {
 	if c.TotalDuration > 0 {
 		c.NumMetrics = 0
 	} else if c.NumMetrics <= 0 {

cmd/telemetrygen/internal/metrics/worker.go

@@ -27,9 +27,22 @@ type worker struct {
 	index          int             // worker index
 }
 
-func (w worker) simulateMetrics(res *resource.Resource, exporter sdkmetric.Exporter) {
+func (w worker) simulateMetrics(res *resource.Resource, exporterFunc func() (sdkmetric.Exporter, error)) {
 	limiter := rate.NewLimiter(w.limitPerSecond, 1)
 
+	exporter, err := exporterFunc()
+	if err != nil {
+		w.logger.Error("failed to create the exporter", zap.Error(err))
+		return
+	}
+
+	defer func() {
+		w.logger.Info("stopping the exporter")
+		if tempError := exporter.Shutdown(context.Background()); tempError != nil {
+			w.logger.Error("failed to stop the exporter", zap.Error(tempError))
+		}
+	}()
+
 	var i int64
 	for w.running.Load() {
 		var metrics []metricdata.Metrics

cmd/telemetrygen/internal/metrics/worker_test.go

@@ -51,16 +51,19 @@ func TestFixedNumberOfMetrics(t *testing.T) {
 		MetricType: metricTypeSum,
 	}
 
-	exp := &mockExporter{}
+	m := &mockExporter{}
+	expFunc := func() (sdkmetric.Exporter, error) {
+		return m, nil
+	}
 
 	// test
 	logger, _ := zap.NewDevelopment()
-	require.NoError(t, Run(cfg, exp, logger))
+	require.NoError(t, Run(cfg, expFunc, logger))
 
 	time.Sleep(1 * time.Second)
 
 	// verify
-	require.Len(t, exp.rms, 5)
+	require.Len(t, m.rms, 5)
 }
 
 func TestRateOfMetrics(t *testing.T) {
@@ -72,16 +75,20 @@ func TestRateOfMetrics(t *testing.T) {
 		},
 		MetricType: metricTypeSum,
 	}
-	exp := &mockExporter{}
+
+	m := &mockExporter{}
+	expFunc := func() (sdkmetric.Exporter, error) {
+		return m, nil
+	}
 
 	// test
-	require.NoError(t, Run(cfg, exp, zap.NewNop()))
+	require.NoError(t, Run(cfg, expFunc, zap.NewNop()))
 
 	// verify
 	// the minimum acceptable number of metrics for the rate of 10/sec for half a second
-	assert.True(t, len(exp.rms) >= 6, "there should have been more than 6 metrics, had %d", len(exp.rms))
-	// the maximum acceptable number of metrics for the rate of 10/sec for half a second
-	assert.True(t, len(exp.rms) <= 20, "there should have been less than 20 metrics, had %d", len(exp.rms))
+	assert.True(t, len(m.rms) >= 6, "there should have been more than 6 metrics, had %d", len(m.rms))
+	// // the maximum acceptable number of metrics for the rate of 10/sec for half a second
+	assert.True(t, len(m.rms) <= 20, "there should have been less than 20 metrics, had %d", len(m.rms))
 }
 
 func TestUnthrottled(t *testing.T) {
@@ -92,11 +99,16 @@ func TestUnthrottled(t *testing.T) {
 		},
 		MetricType: metricTypeSum,
 	}
-	exp := &mockExporter{}
+
+	m := &mockExporter{}
+	expFunc := func() (sdkmetric.Exporter, error) {
+
+		return m, nil
+	}
 
 	// test
 	logger, _ := zap.NewDevelopment()
-	require.NoError(t, Run(cfg, exp, logger))
+	require.NoError(t, Run(cfg, expFunc, logger))
 
-	assert.True(t, len(exp.rms) > 100, "there should have been more than 100 metrics, had %d", len(exp.rms))
+	assert.True(t, len(m.rms) > 100, "there should have been more than 100 metrics, had %d", len(m.rms))
 }