[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/README.md

@@ -28,28 +28,44 @@ Check the [`go install` reference](https://go.dev/ref/mod#go-install) to install
 
 First, you'll need an OpenTelemetry Collector to receive the telemetry data. Follow the project's instructions for a detailed setting up guide. The following configuration file should be sufficient:
 
+config.yaml:
 ```yaml
 receivers:
   otlp:
     protocols:
       grpc:
-        endpoint: localhost:4317
+        endpoint: 0.0.0.0:4317
 
 processors:
+  batch:
 
 exporters:
   debug:
+    verbosity: detailed
 
 service:
   pipelines:
+    logs:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [debug]
+    metrics:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [debug]
     traces:
-      receivers:
-      - otlp
-      processors: []
-      exporters:
-      - debug
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [debug]
 ```
 
+Starting OpenTelemetry collector via docker:
+```
+docker run -p 4317:4317 -v $(pwd)/config.yaml:/etc/otelcol-contrib/config.yaml ghcr.io/open-telemetry/opentelemetry-collector-releases/opentelemetry-collector-contrib:0.86.0
+```
+
+Other options for running the collector are documented here https://opentelemetry.io/docs/collector/getting-started/
+
 Once the OpenTelemetry Collector instance is up and running, run `telemetrygen` for your desired telemetry:
 
 ### Traces
@@ -65,3 +81,15 @@ telemetrygen traces --otlp-insecure --traces 1
 ```
 
 Check `telemetrygen traces --help` for all the options.
+
+### Logs
+
+```console
+telemetrygen logs --duration 5s --otlp-insecure
+```
+
+### Metrics
+
+```console
+telemetrygen metrics --duration 5s --otlp-insecure
+```

cmd/telemetrygen/internal/metrics/metrics.go

@@ -29,26 +29,19 @@ func Start(cfg *Config) error {
 	}
 	logger.Info("starting the metrics generator with configuration", zap.Any("config", cfg))
 
-	var exp sdkmetric.Exporter
-	if cfg.UseHTTP {
-		logger.Info("starting HTTP exporter")
-		exp, err = otlpmetrichttp.New(context.Background(), httpExporterOptions(cfg)...)
-	} else {
-		logger.Info("starting gRPC exporter")
-		exp, err = otlpmetricgrpc.New(context.Background(), grpcExporterOptions(cfg)...)
-	}
-
-	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(), httpExporterOptions(cfg)...)
+		} else {
+			logger.Info("starting gRPC exporter")
+			exp, err = otlpmetricgrpc.New(context.Background(), grpcExporterOptions(cfg)...)
 		}
-	}()
+		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
 	}
@@ -57,7 +50,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

@@ -28,9 +28,22 @@ type worker struct {
 	index          int             // worker index
 }
 
-func (w worker) simulateMetrics(res *resource.Resource, exporter sdkmetric.Exporter, signalAttrs []attribute.KeyValue) {
+func (w worker) simulateMetrics(res *resource.Resource, exporterFunc func() (sdkmetric.Exporter, error), signalAttrs []attribute.KeyValue) {
 	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