feat(telemetrygen): added support for delta temporality

.chloggen/nickange_telemetrygen_metrics_support-delta-temporality.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: "Add support for `temporalityType` flag in telemetrygen.  Supported values (delta or cumulative)"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38073]
+
+# (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: []

cmd/telemetrygen/pkg/metrics/config.go

@@ -7,18 +7,20 @@ import (
 	"fmt"
 
 	"github.com/spf13/pflag"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/common"
 )
 
 // Config describes the test scenario.
 type Config struct {
 	common.Config
-	NumMetrics int
-	MetricName string
-	MetricType MetricType
-	SpanID     string
-	TraceID    string
+	NumMetrics      int
+	MetricName      string
+	MetricType      MetricType
+	TemporalityType TemporalityType
+	SpanID          string
+	TraceID         string
 }
 
 // NewConfig creates a new Config with default values.
@@ -34,11 +36,13 @@ func (c *Config) Flags(fs *pflag.FlagSet) {
 
 	fs.StringVar(&c.HTTPPath, "otlp-http-url-path", c.HTTPPath, "Which URL path to write to")
 
-	fs.Var(&c.MetricType, "metric-type", "Metric type enum. must be one of 'Gauge' or 'Sum'")
 	fs.IntVar(&c.NumMetrics, "metrics", c.NumMetrics, "Number of metrics to generate in each worker (ignored if duration is provided)")
 
 	fs.StringVar(&c.TraceID, "trace-id", c.TraceID, "TraceID to use as exemplar")
 	fs.StringVar(&c.SpanID, "span-id", c.SpanID, "SpanID to use as exemplar")
+
+	fs.Var(&c.MetricType, "metric-type", "Metric type enum. must be one of 'Gauge' or 'Sum'")
+	fs.Var(&c.TemporalityType, "temporality-type", "Temporality for Sum and Histogram metrics. Must be one of 'delta' or 'cumulative'")
 }
 
 // SetDefaults sets the default values for the configuration
@@ -49,9 +53,12 @@ func (c *Config) SetDefaults() {
 	c.HTTPPath = "/v1/metrics"
 	c.NumMetrics = 1
 
+	c.MetricName = "gen"
 	// Use Gauge as default metric type.
 	c.MetricType = MetricTypeGauge
-	c.MetricName = "gen"
+	// Use cumulative temporality as default.
+	c.TemporalityType = TemporalityType(metricdata.CumulativeTemporality)
+
 	c.TraceID = ""
 	c.SpanID = ""
 }

cmd/telemetrygen/pkg/metrics/metrics.go

@@ -66,16 +66,17 @@ func run(c *Config, expF exporterFunc, logger *zap.Logger) error {
 	for i := 0; i < c.WorkerCount; i++ {
 		wg.Add(1)
 		w := worker{
-			numMetrics:     c.NumMetrics,
-			metricName:     c.MetricName,
-			metricType:     c.MetricType,
-			exemplars:      exemplarsFromConfig(c),
-			limitPerSecond: limit,
-			totalDuration:  c.TotalDuration,
-			running:        running,
-			wg:             &wg,
-			logger:         logger.With(zap.Int("worker", i)),
-			index:          i,
+			numMetrics:      c.NumMetrics,
+			metricName:      c.MetricName,
+			metricType:      c.MetricType,
+			temporalityType: c.TemporalityType,
+			exemplars:       exemplarsFromConfig(c),
+			limitPerSecond:  limit,
+			totalDuration:   c.TotalDuration,
+			running:         running,
+			wg:              &wg,
+			logger:          logger.With(zap.Int("worker", i)),
+			index:           i,
 		}
 		exp, err := expF()
 		if err != nil {

cmd/telemetrygen/pkg/metrics/temporality_type.go

@@ -0,0 +1,35 @@
+package metrics
+
+import (
+	"fmt"
+
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+)
+
+type TemporalityType metricdata.Temporality
+
+func (t *TemporalityType) Set(v string) error {
+	switch v {
+	case "delta":
+		*t = TemporalityType(metricdata.DeltaTemporality)
+		return nil
+	case "cumulative":
+		*t = TemporalityType(metricdata.CumulativeTemporality)
+		return nil
+	default:
+		return fmt.Errorf(`temporality must be one of "delta" or "cumulative"`)
+	}
+}
+
+func (t *TemporalityType) String() string {
+	return string(metricdata.Temporality(*t))
+}
+
+func (t *TemporalityType) Type() string {
+	return "temporality"
+}
+
+// AsTemporality converts the TemporalityType to metricdata.Temporality
+func (t TemporalityType) AsTemporality() metricdata.Temporality {
+	return metricdata.Temporality(t)
+}

cmd/telemetrygen/pkg/metrics/worker.go

@@ -18,16 +18,17 @@ import (
 )
 
 type worker struct {
-	running        *atomic.Bool                 // pointer to shared flag that indicates it's time to stop the test
-	metricName     string                       // name of metric to generate
-	metricType     MetricType                   // type of metric to generate
-	exemplars      []metricdata.Exemplar[int64] // exemplars to attach to the metric
-	numMetrics     int                          // how many metrics the worker has to generate (only when duration==0)
-	totalDuration  time.Duration                // how long to run the test for (overrides `numMetrics`)
-	limitPerSecond rate.Limit                   // how many metrics per second to generate
-	wg             *sync.WaitGroup              // notify when done
-	logger         *zap.Logger                  // logger
-	index          int                          // worker index
+	running         *atomic.Bool                 // pointer to shared flag that indicates it's time to stop the test
+	metricName      string                       // name of metric to generate
+	metricType      MetricType                   // type of metric to generate
+	temporalityType TemporalityType              // Temporality type to use
+	exemplars       []metricdata.Exemplar[int64] // exemplars to attach to the metric
+	numMetrics      int                          // how many metrics the worker has to generate (only when duration==0)
+	totalDuration   time.Duration                // how long to run the test for (overrides `numMetrics`)
+	limitPerSecond  rate.Limit                   // how many metrics per second to generate
+	wg              *sync.WaitGroup              // notify when done
+	logger          *zap.Logger                  // logger
+	index           int                          // worker index
 }
 
 var histogramBucketSamples = []struct {
@@ -103,7 +104,7 @@ func (w worker) simulateMetrics(res *resource.Resource, exporter sdkmetric.Expor
 				Name: w.metricName,
 				Data: metricdata.Sum[int64]{
 					IsMonotonic: true,
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: w.temporalityType.AsTemporality(),
 					DataPoints: []metricdata.DataPoint[int64]{
 						{
 							StartTime:  time.Now().Add(-1 * time.Second),
@@ -122,7 +123,7 @@ func (w worker) simulateMetrics(res *resource.Resource, exporter sdkmetric.Expor
 			metrics = append(metrics, metricdata.Metrics{
 				Name: w.metricName,
 				Data: metricdata.Histogram[int64]{
-					Temporality: metricdata.CumulativeTemporality,
+					Temporality: w.temporalityType.AsTemporality(),
 					DataPoints: []metricdata.HistogramDataPoint[int64]{
 						{
 							StartTime:  time.Now().Add(-1 * time.Second),

cmd/telemetrygen/pkg/metrics/worker_test.go

@@ -50,6 +50,21 @@ func (m *mockExporter) Shutdown(_ context.Context) error {
 	return nil
 }
 
+func checkMetricTemporality(t *testing.T, ms metricdata.Metrics, metricType MetricType, expectedTemporality metricdata.Temporality) {
+	switch metricType {
+	case MetricTypeSum:
+		sumData, ok := ms.Data.(metricdata.Sum[int64])
+		require.True(t, ok, "expected Sum data type")
+		assert.Equal(t, expectedTemporality, sumData.Temporality)
+	case MetricTypeHistogram:
+		histogramData, ok := ms.Data.(metricdata.Histogram[int64])
+		require.True(t, ok, "expected Histogram data type")
+		assert.Equal(t, expectedTemporality, histogramData.Temporality)
+	default:
+		t.Fatalf("unsupported metric type: %v", metricType)
+	}
+}
+
 func TestFixedNumberOfMetrics(t *testing.T) {
 	// arrange
 	cfg := &Config{
@@ -98,6 +113,72 @@ func TestRateOfMetrics(t *testing.T) {
 	assert.LessOrEqual(t, len(m.rms), 20, "there should have been less than 20 metrics, had %d", len(m.rms))
 }
 
+func TestMetricsWithTemporality(t *testing.T) {
+	tests := []struct {
+		name                string
+		metricType          MetricType
+		temporalityType     TemporalityType
+		expectedTemporality metricdata.Temporality
+	}{
+		{
+			name:                "Sum: delta temporality",
+			metricType:          MetricTypeSum,
+			temporalityType:     TemporalityType(metricdata.DeltaTemporality),
+			expectedTemporality: metricdata.DeltaTemporality,
+		},
+		{
+			name:                "Sum: cumulative temporality",
+			metricType:          MetricTypeSum,
+			temporalityType:     TemporalityType(metricdata.CumulativeTemporality),
+			expectedTemporality: metricdata.CumulativeTemporality,
+		},
+		{
+			name:                "Histogram: delta temporality",
+			metricType:          MetricTypeHistogram,
+			temporalityType:     TemporalityType(metricdata.DeltaTemporality),
+			expectedTemporality: metricdata.DeltaTemporality,
+		},
+		{
+			name:                "Histogram: cumulative temporality",
+			metricType:          MetricTypeHistogram,
+			temporalityType:     TemporalityType(metricdata.CumulativeTemporality),
+			expectedTemporality: metricdata.CumulativeTemporality,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// arrange
+			cfg := &Config{
+				Config: common.Config{
+					WorkerCount: 1,
+				},
+				NumMetrics:      1,
+				MetricName:      "test",
+				MetricType:      tt.metricType,
+				TemporalityType: tt.temporalityType,
+			}
+			m := &mockExporter{}
+			expFunc := func() (sdkmetric.Exporter, error) {
+				return m, nil
+			}
+
+			// act
+			logger, _ := zap.NewDevelopment()
+			require.NoError(t, run(cfg, expFunc, logger))
+
+			time.Sleep(1 * time.Second)
+
+			// assert
+			require.Len(t, m.rms, 1)
+			ms := m.rms[0].ScopeMetrics[0].Metrics[0]
+			assert.Equal(t, "test", ms.Name)
+
+			checkMetricTemporality(t, ms, tt.metricType, tt.expectedTemporality)
+		})
+	}
+}
+
 func TestUnthrottled(t *testing.T) {
 	// arrange
 	cfg := &Config{