[receiver/statsdreceiver] allow changing percentiles for summary types

.chloggen/statsd-percentiles.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: statsdreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Allow configuring summary percentiles"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [33701]
+
+# (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]

receiver/statsdreceiver/README.md

@@ -40,7 +40,7 @@ The Following settings are optional:
 `"statsd_type"` specifies received Statsd data type. Possible values for this setting are `"timing"`, `"timer"`, `"histogram"` and `"distribution"`.
 
 `"observer_type"` specifies OTLP data type to convert to. We support `"gauge"`, `"summary"`, and `"histogram"`. For `"gauge"`, it does not perform any aggregation.
-For `"summary`, the statsD receiver will aggregate to one OTLP summary metric for one metric description (the same metric name with the same tags). It will send percentile 0, 10, 50, 90, 95, 100 to the downstream.  The `"histogram"` setting selects an [auto-scaling exponential histogram configured with only a maximum size](https://github.com/lightstep/go-expohisto#readme), as shown in the example below.
+For `"summary`, the statsD receiver will aggregate to one OTLP summary metric for one metric description (the same metric name with the same tags). By default, it will send percentile 0, 10, 50, 90, 95, 100 to the downstream.  The `"histogram"` setting selects an [auto-scaling exponential histogram configured with only a maximum size](https://github.com/lightstep/go-expohisto#readme), as shown in the example below.
 TODO: Add a new option to use a smoothed summary like Prometheus: https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/3261 
 
 Example:
@@ -61,9 +61,9 @@ receivers:
         histogram: 
           max_size: 100
       - statsd_type: "distribution"
-        observer_type: "histogram"
-        histogram: 
-          max_size: 50    
+        observer_type: "summary"
+        summary: 
+          percentiles: [0, 10, 50, 90, 95, 100]
 ```
 
 The full list of settings exposed for this receiver are documented [here](./config.go)
@@ -168,4 +168,4 @@ echo "test.metric:42|c|#myKey:myVal" | nc -w 1 -u -4 localhost 8125;
 echo "test.metric:42|c|#myKey:myVal" | nc -w 1 -u -6 localhost 8125;
 ```
 
-Which sends a UDP packet using both IPV4 and IPV6, which is needed because the receiver's UDP server only accepts one or the other.
+Which sends a UDP packet using both IPV4 and IPV6, which is needed because the receiver's UDP server only accepts one or the other.

receiver/statsdreceiver/config.go

@@ -73,6 +73,16 @@ func (c *Config) Validate() error {
 				errs = multierr.Append(errs, fmt.Errorf("histogram configuration requires observer_type: histogram"))
 			}
 		}
+		if len(eachMap.Summary.Percentiles) != 0 {
+			for _, percentile := range eachMap.Summary.Percentiles {
+				if percentile > 100 || percentile < 0 {
+					errs = multierr.Append(errs, fmt.Errorf("summary percentiles out of [0, 100] range: %v", percentile))
+				}
+			}
+			if eachMap.ObserverType != protocol.SummaryObserver {
+				errs = multierr.Append(errs, fmt.Errorf("summary configuration requires observer_type: summary"))
+			}
+		}
 	}
 
 	if TimerHistogramMappingMissingObjectName {

receiver/statsdreceiver/config_test.go

@@ -56,9 +56,9 @@ func TestLoadConfig(t *testing.T) {
 					},
 					{
 						StatsdType:   "distribution",
-						ObserverType: "histogram",
-						Histogram: protocol.HistogramConfig{
-							MaxSize: 170,
+						ObserverType: "summary",
+						Summary: protocol.SummaryConfig{
+							Percentiles: []float64{0, 10, 50, 90, 95, 100},
 						},
 					},
 				},
@@ -94,6 +94,7 @@ func TestValidate(t *testing.T) {
 		statsdTypeNotSupportErr        = "statsd_type is not a supported mapping for histogram and timing metrics: %s"
 		observerTypeNotSupportErr      = "observer_type is not supported for histogram and timing metrics: %s"
 		invalidHistogramErr            = "histogram configuration requires observer_type: histogram"
+		invalidSummaryErr              = "summary configuration requires observer_type: summary"
 	)
 
 	tests := []test{
@@ -163,6 +164,22 @@ func TestValidate(t *testing.T) {
 			},
 			expectedErr: invalidHistogramErr,
 		},
+		{
+			name: "invalidSummary",
+			cfg: &Config{
+				AggregationInterval: 20 * time.Second,
+				TimerHistogramMapping: []protocol.TimerHistogramMapping{
+					{
+						StatsdType:   "timing",
+						ObserverType: "gauge",
+						Summary: protocol.SummaryConfig{
+							Percentiles: []float64{1},
+						},
+					},
+				},
+			},
+			expectedErr: invalidSummaryErr,
+		},
 		{
 			name: "negativeAggregationInterval",
 			cfg: &Config{

receiver/statsdreceiver/internal/protocol/statsd_parser.go

@@ -61,15 +61,21 @@ type TimerHistogramMapping struct {
 	StatsdType   TypeName        `mapstructure:"statsd_type"`
 	ObserverType ObserverType    `mapstructure:"observer_type"`
 	Histogram    HistogramConfig `mapstructure:"histogram"`
+	Summary      SummaryConfig   `mapstructure:"summary"`
 }
 
 type HistogramConfig struct {
 	MaxSize int32 `mapstructure:"max_size"`
 }
 
+type SummaryConfig struct {
+	Percentiles []float64 `mapstructure:"percentiles"`
+}
+
 type ObserverCategory struct {
-	method          ObserverType
-	histogramConfig structure.Config
+	method             ObserverType
+	histogramConfig    structure.Config
+	summaryPercentiles []float64
 }
 
 var defaultObserverCategory = ObserverCategory{
@@ -113,8 +119,9 @@ type sampleValue struct {
 }
 
 type summaryMetric struct {
-	points  []float64
-	weights []float64
+	points      []float64
+	weights     []float64
+	percentiles []float64
 }
 
 type histogramStructure = structure.Histogram[float64]
@@ -173,9 +180,11 @@ func (p *StatsDParser) Initialize(enableMetricType bool, enableSimpleTags bool,
 		case HistogramTypeName, DistributionTypeName:
 			p.histogramEvents.method = eachMap.ObserverType
 			p.histogramEvents.histogramConfig = expoHistogramConfig(eachMap.Histogram)
+			p.histogramEvents.summaryPercentiles = eachMap.Summary.Percentiles
 		case TimingTypeName, TimingAltTypeName:
 			p.timerEvents.method = eachMap.ObserverType
 			p.timerEvents.histogramConfig = expoHistogramConfig(eachMap.Histogram)
+			p.timerEvents.summaryPercentiles = eachMap.Summary.Percentiles
 		case CounterTypeName, GaugeTypeName:
 		}
 	}
@@ -218,13 +227,16 @@ func (p *StatsDParser) GetMetrics() []BatchMetrics {
 		for desc, summaryMetric := range instrument.summaries {
 			ilm := rm.ScopeMetrics().AppendEmpty()
 			p.setVersionAndNameScope(ilm.Scope())
-
+			percentiles := summaryMetric.percentiles
+			if len(summaryMetric.percentiles) == 0 {
+				percentiles = statsDDefaultPercentiles
+			}
 			buildSummaryMetric(
 				desc,
 				summaryMetric,
 				p.lastIntervalTime,
 				now,
-				statsDDefaultPercentiles,
+				percentiles,
 				ilm,
 			)
 		}
@@ -318,13 +330,15 @@ func (p *StatsDParser) Aggregate(line string, addr net.Addr) error {
 			raw := parsedMetric.sampleValue()
 			if existing, ok := instrument.summaries[parsedMetric.description]; !ok {
 				instrument.summaries[parsedMetric.description] = summaryMetric{
-					points:  []float64{raw.value},
-					weights: []float64{raw.count},
+					points:      []float64{raw.value},
+					weights:     []float64{raw.count},
+					percentiles: category.summaryPercentiles,
 				}
 			} else {
 				instrument.summaries[parsedMetric.description] = summaryMetric{
-					points:  append(existing.points, raw.value),
-					weights: append(existing.weights, raw.count),
+					points:      append(existing.points, raw.value),
+					weights:     append(existing.weights, raw.count),
+					percentiles: category.summaryPercentiles,
 				}
 			}
 		case HistogramObserver:

receiver/statsdreceiver/internal/protocol/statsd_parser_test.go

@@ -1394,13 +1394,15 @@ func TestStatsDParser_AggregateTimerWithSummary(t *testing.T) {
 			expectedSummaries: map[statsDMetricDescription]summaryMetric{
 				testDescription("statsdTestMetric1", "h",
 					[]string{"mykey"}, []string{"myvalue"}): {
-					points:  []float64{1, 1, 10, 20},
-					weights: []float64{1, 1, 1, 1},
+					points:      []float64{1, 1, 10, 20},
+					weights:     []float64{1, 1, 1, 1},
+					percentiles: []float64{0, 95, 99},
 				},
 				testDescription("statsdTestMetric2", "h",
 					[]string{"mykey"}, []string{"myvalue"}): {
-					points:  []float64{2, 5, 10},
-					weights: []float64{1, 1, 1},
+					points:      []float64{2, 5, 10},
+					weights:     []float64{1, 1, 1},
+					percentiles: []float64{0, 95, 99},
 				},
 			},
 		},
@@ -1415,8 +1417,9 @@ func TestStatsDParser_AggregateTimerWithSummary(t *testing.T) {
 			expectedSummaries: map[statsDMetricDescription]summaryMetric{
 				testDescription("statsdTestMetric1", "h",
 					[]string{"mykey"}, []string{"myvalue"}): {
-					points:  []float64{300, 100, 300, 200},
-					weights: []float64{10, 20, 10, 100},
+					points:      []float64{300, 100, 300, 200},
+					weights:     []float64{10, 20, 10, 100},
+					percentiles: []float64{0, 95, 99},
 				},
 			},
 		},
@@ -1434,13 +1437,15 @@ func TestStatsDParser_AggregateTimerWithSummary(t *testing.T) {
 			expectedSummaries: map[statsDMetricDescription]summaryMetric{
 				testDescription("statsdTestMetric1", "d",
 					[]string{"mykey"}, []string{"myvalue"}): {
-					points:  []float64{1, 1, 10, 20},
-					weights: []float64{1, 1, 1, 1},
+					points:      []float64{1, 1, 10, 20},
+					weights:     []float64{1, 1, 1, 1},
+					percentiles: []float64{0, 95, 99},
 				},
 				testDescription("statsdTestMetric2", "d",
 					[]string{"mykey"}, []string{"myvalue"}): {
-					points:  []float64{2, 5, 10},
-					weights: []float64{1, 1, 1},
+					points:      []float64{2, 5, 10},
+					weights:     []float64{1, 1, 1},
+					percentiles: []float64{0, 95, 99},
 				},
 			},
 		},
@@ -1449,7 +1454,7 @@ func TestStatsDParser_AggregateTimerWithSummary(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			var err error
 			p := &StatsDParser{}
-			assert.NoError(t, p.Initialize(false, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "summary"}, {StatsdType: "histogram", ObserverType: "summary"}}))
+			assert.NoError(t, p.Initialize(false, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "summary"}, {StatsdType: "histogram", ObserverType: "summary", Summary: SummaryConfig{Percentiles: []float64{0, 95, 99}}}}))
 			addr, _ := net.ResolveUDPAddr("udp", "1.2.3.4:5678")
 			addrKey := newNetAddr(addr)
 			for _, line := range tt.input {

receiver/statsdreceiver/testdata/config.yaml

@@ -12,6 +12,6 @@ statsd/receiver_settings:
       histogram:
         max_size: 170
     - statsd_type: "distribution"
-      observer_type: "histogram"
-      histogram:
-        max_size: 170
+      observer_type: "summary"
+      summary:
+        percentiles: [0, 10, 50, 90, 95, 100]