[receiver/statsd] Add Option to Aggregate on IP/Host

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

@@ -30,6 +30,8 @@ The Following settings are optional:
 
 - `enable_metric_type: true`(default value is false): Enable the statsd receiver to be able to emit the metric type(gauge, counter, timer(in the future), histogram(in the future)) as a label.
 
+- `enable_ip_only_aggregation` (default value is false): Enables metric aggregation on `Client+IP` only. Normally, aggregation is performed on `Client+IP+Port`. This setting is useful when the client sends metrics from a random ports or the receiver should aggregate metrics from the same client but different ports.
+
 - `enable_simple_tags: true`(default value is false): Enable parsing tags that do not have a value, e.g. `#mykey` instead of `#mykey:myvalue`. DogStatsD supports such tagging.
 
 - `is_monotonic_counter` (default value is false): Set all counter-type metrics the statsd receiver received as monotonic.

receiver/statsdreceiver/config.go

@@ -16,12 +16,13 @@ import (
 
 // Config defines configuration for StatsD receiver.
 type Config struct {
-	NetAddr               confignet.AddrConfig             `mapstructure:",squash"`
-	AggregationInterval   time.Duration                    `mapstructure:"aggregation_interval"`
-	EnableMetricType      bool                             `mapstructure:"enable_metric_type"`
-	EnableSimpleTags      bool                             `mapstructure:"enable_simple_tags"`
-	IsMonotonicCounter    bool                             `mapstructure:"is_monotonic_counter"`
-	TimerHistogramMapping []protocol.TimerHistogramMapping `mapstructure:"timer_histogram_mapping"`
+	NetAddr                 confignet.AddrConfig             `mapstructure:",squash"`
+	AggregationInterval     time.Duration                    `mapstructure:"aggregation_interval"`
+	EnableIPOnlyAggregation bool                             `mapstructure:"enable_ip_only_aggregation"`
+	EnableMetricType        bool                             `mapstructure:"enable_metric_type"`
+	EnableSimpleTags        bool                             `mapstructure:"enable_simple_tags"`
+	IsMonotonicCounter      bool                             `mapstructure:"is_monotonic_counter"`
+	TimerHistogramMapping   []protocol.TimerHistogramMapping `mapstructure:"timer_histogram_mapping"`
 }
 
 func (c *Config) Validate() error {

receiver/statsdreceiver/internal/protocol/parser.go

@@ -12,7 +12,7 @@ import (
 
 // Parser is something that can map input StatsD strings to OTLP Metric representations.
 type Parser interface {
-	Initialize(enableMetricType bool, enableSimpleTags bool, isMonotonicCounter bool, sendTimerHistogram []TimerHistogramMapping) error
+	Initialize(enableMetricType bool, enableSimpleTags bool, isMonotonicCounter bool, enableIPOnlyAggregation bool, sendTimerHistogram []TimerHistogramMapping) error
 	GetMetrics() []BatchMetrics
 	Aggregate(line string, addr net.Addr) error
 }

receiver/statsdreceiver/internal/protocol/statsd_parser.go

@@ -84,14 +84,15 @@ var defaultObserverCategory = ObserverCategory{
 
 // StatsDParser supports the Parse method for parsing StatsD messages with Tags.
 type StatsDParser struct {
-	instrumentsByAddress map[netAddr]*instruments
-	enableMetricType     bool
-	enableSimpleTags     bool
-	isMonotonicCounter   bool
-	timerEvents          ObserverCategory
-	histogramEvents      ObserverCategory
-	lastIntervalTime     time.Time
-	BuildInfo            component.BuildInfo
+	instrumentsByAddress    map[netAddr]*instruments
+	enableMetricType        bool
+	enableSimpleTags        bool
+	isMonotonicCounter      bool
+	enableIpOnlyAggregation bool
+	timerEvents             ObserverCategory
+	histogramEvents         ObserverCategory
+	lastIntervalTime        time.Time
+	BuildInfo               component.BuildInfo
 }
 
 type instruments struct {
@@ -166,14 +167,16 @@ func (p *StatsDParser) resetState(when time.Time) {
 	p.instrumentsByAddress = make(map[netAddr]*instruments)
 }
 
-func (p *StatsDParser) Initialize(enableMetricType bool, enableSimpleTags bool, isMonotonicCounter bool, sendTimerHistogram []TimerHistogramMapping) error {
+func (p *StatsDParser) Initialize(enableMetricType bool, enableSimpleTags bool, isMonotonicCounter bool, enableIpOnlyAggregation bool, sendTimerHistogram []TimerHistogramMapping) error {
 	p.resetState(timeNowFunc())
 
 	p.histogramEvents = defaultObserverCategory
 	p.timerEvents = defaultObserverCategory
 	p.enableMetricType = enableMetricType
 	p.enableSimpleTags = enableSimpleTags
 	p.isMonotonicCounter = isMonotonicCounter
+	p.enableIpOnlyAggregation = enableIpOnlyAggregation
+
 	// Note: validation occurs in ("../".Config).validate()
 	for _, eachMap := range sendTimerHistogram {
 		switch eachMap.StatsdType {
@@ -292,6 +295,10 @@ func (p *StatsDParser) Aggregate(line string, addr net.Addr) error {
 	}
 
 	addrKey := newNetAddr(addr)
+	if p.enableIpOnlyAggregation {
+		addrKey = newIPOnlyNetAddr(addr)
+	}
+
 	instrument, ok := p.instrumentsByAddress[addrKey]
 	if !ok {
 		instrument = newInstruments(addr)
@@ -494,3 +501,12 @@ type netAddr struct {
 func newNetAddr(addr net.Addr) netAddr {
 	return netAddr{addr.Network(), addr.String()}
 }
+
+func newIPOnlyNetAddr(addr net.Addr) netAddr {
+	host, _, err := net.SplitHostPort(addr.String())
+	if err != nil {
+		// if there is an error, use the original address
+		return netAddr{addr.Network(), addr.String()}
+	}
+	return netAddr{addr.Network(), host}
+}

receiver/statsdreceiver/internal/protocol/statsd_parser_test.go

@@ -1049,7 +1049,7 @@ func TestStatsDParser_Aggregate(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: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
+			assert.NoError(t, p.Initialize(false, false, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
 			p.lastIntervalTime = time.Unix(611, 0)
 			addr, _ := net.ResolveUDPAddr("udp", "1.2.3.4:5678")
 			addrKey := newNetAddr(addr)
@@ -1158,7 +1158,7 @@ func TestStatsDParser_AggregateByAddress(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			p := &StatsDParser{}
-			assert.NoError(t, p.Initialize(true, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
+			assert.NoError(t, p.Initialize(true, false, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
 			p.lastIntervalTime = time.Unix(611, 0)
 			for i, addr := range tt.addresses {
 				for _, line := range tt.input[i] {
@@ -1266,7 +1266,7 @@ func TestStatsDParser_AggregateWithMetricType(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			var err error
 			p := &StatsDParser{}
-			assert.NoError(t, p.Initialize(true, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
+			assert.NoError(t, p.Initialize(true, false, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
 			p.lastIntervalTime = time.Unix(611, 0)
 			addr, _ := net.ResolveUDPAddr("udp", "1.2.3.4:5678")
 			addrKey := newNetAddr(addr)
@@ -1336,7 +1336,7 @@ func TestStatsDParser_AggregateWithIsMonotonicCounter(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			var err error
 			p := &StatsDParser{}
-			assert.NoError(t, p.Initialize(false, false, true, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
+			assert.NoError(t, p.Initialize(false, false, true, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
 			p.lastIntervalTime = time.Unix(611, 0)
 			addr, _ := net.ResolveUDPAddr("udp", "1.2.3.4:5678")
 			addrKey := newNetAddr(addr)
@@ -1463,7 +1463,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", Summary: SummaryConfig{Percentiles: []float64{0, 95, 99}}}}))
+			assert.NoError(t, p.Initialize(false, 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 {
@@ -1480,7 +1480,7 @@ func TestStatsDParser_AggregateTimerWithSummary(t *testing.T) {
 
 func TestStatsDParser_Initialize(t *testing.T) {
 	p := &StatsDParser{}
-	assert.NoError(t, p.Initialize(true, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
+	assert.NoError(t, p.Initialize(true, false, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
 	teststatsdDMetricdescription := statsDMetricDescription{
 		name:       "test",
 		metricType: "g",
@@ -1499,7 +1499,7 @@ func TestStatsDParser_Initialize(t *testing.T) {
 
 func TestStatsDParser_GetMetricsWithMetricType(t *testing.T) {
 	p := &StatsDParser{}
-	assert.NoError(t, p.Initialize(true, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
+	assert.NoError(t, p.Initialize(true, false, false, false, []TimerHistogramMapping{{StatsdType: "timer", ObserverType: "gauge"}, {StatsdType: "histogram", ObserverType: "gauge"}}))
 	instrument := newInstruments(nil)
 	instrument.gauges[testDescription("statsdTestMetric1", "g",
 		[]string{"mykey", "metric_type"}, []string{"myvalue", "gauge"})] = buildGaugeMetric(
@@ -1612,7 +1612,7 @@ func TestStatsDParser_Mappings(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			p := &StatsDParser{}
 
-			assert.NoError(t, p.Initialize(false, false, false, tc.mapping))
+			assert.NoError(t, p.Initialize(false, false, false, false, tc.mapping))
 
 			addr, _ := net.ResolveUDPAddr("udp", "1.2.3.4:5678")
 			assert.NoError(t, p.Aggregate("H:10|h", addr))
@@ -1646,7 +1646,7 @@ func TestStatsDParser_ScopeIsIncluded(t *testing.T) {
 	}
 	testAddress, _ := net.ResolveUDPAddr("udp", "1.2.3.4:5678")
 
-	err := p.Initialize(true, false, false,
+	err := p.Initialize(true, false, false, false,
 		[]TimerHistogramMapping{
 			{StatsdType: "timer", ObserverType: "summary"},
 			{StatsdType: "histogram", ObserverType: "histogram"},
@@ -1916,7 +1916,7 @@ func TestStatsDParser_AggregateTimerWithHistogram(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			var err error
 			p := &StatsDParser{}
-			assert.NoError(t, p.Initialize(false, false, false, tt.mapping))
+			assert.NoError(t, p.Initialize(false, false, false, false, tt.mapping))
 			addr, _ := net.ResolveUDPAddr("udp", "1.2.3.4:5678")
 			for _, line := range tt.input {
 				err = p.Aggregate(line, addr)
@@ -1927,3 +1927,32 @@ func TestStatsDParser_AggregateTimerWithHistogram(t *testing.T) {
 		})
 	}
 }
+
+func TestStatsDParser_IPOnlyAggregation(t *testing.T) {
+	const devVersion = "dev-0.0.1"
+	p := &StatsDParser{
+		BuildInfo: component.BuildInfo{
+			Version: devVersion,
+		},
+	}
+	testAddr01, _ := net.ResolveUDPAddr("udp", "1.2.3.4:5678")
+	testAddr02, _ := net.ResolveUDPAddr("udp", "1.2.3.4:8765")
+
+	err := p.Initialize(true, false, false, true,
+		[]TimerHistogramMapping{
+			{StatsdType: "timer", ObserverType: "summary"},
+			{StatsdType: "histogram", ObserverType: "histogram"},
+		},
+	)
+
+	require.NoError(t, err)
+	require.NoError(t, p.Aggregate("test.metric:1|c", testAddr01))
+	require.NoError(t, p.Aggregate("test.metric:2|g", testAddr02))
+	require.Equal(t, 1, len(p.instrumentsByAddress))
+	for k := range p.instrumentsByAddress {
+		assert.Equal(t, "1.2.3.4", k.String)
+		assert.Equal(t, "udp", k.Network)
+	}
+	metrics := p.GetMetrics()
+	require.Equal(t, 1, len(metrics))
+}

receiver/statsdreceiver/receiver.go

@@ -107,6 +107,7 @@ func (r *statsdReceiver) Start(ctx context.Context, host component.Host) error {
 		r.config.EnableMetricType,
 		r.config.EnableSimpleTags,
 		r.config.IsMonotonicCounter,
+		r.config.EnableIPOnlyAggregation,
 		r.config.TimerHistogramMapping,
 	)
 	if err != nil {