Skip to content

Commit

Permalink
[receiver/statsd] Add Option to Aggregate on IP/Host (open-telemetry#…
Browse files Browse the repository at this point in the history
…34851)

**Description:** <Describe what has changed.>

The `statsdreceiver` only aggregates metrics on `protocol+host+ip`, this
leads to issues or inconsistencies when dealing with clients that
constantly switch tcp/udp ports. To address the issue, this PR adds a
configuration option `enableIPOnlyAggregation` that allows the use to
specify if they want to aggregate on the `IP` instead of `IP+Port`.

For example:

_otel_config.yaml:_

```yaml
receivers:
  statsd:
    endpoint: "0.0.0.0:8125"
    enable_metric_type: true
    is_monotonic_counter: false
    aggregation_interval: 10s
    enable_ip_only_aggregation: true # <-- enable ip only aggregation
    timer_histogram_mapping:
      - statsd_type: "timing"
        observer_type: "histogram"
        histogram: 
          max_size: 50  

exporters:
  debug:
    verbosity: detailed

service:
  pipelines:
    metrics:
      receivers:
        - statsd
      exporters:
        - debug
```

_run:_
```sh
STATSD_HOST="localhost"
STATSD_PORT=8125

for port in {10000..10010}; do
  echo -n "my.metric:1|c" | nc -w 1 -u -p $port ${STATSD_HOST} ${STATSD_PORT}
  echo "Sent from port $port"
done
```

_result:_
```
2024-08-26T23:36:00.224+0200    info    ResourceMetrics #0
Resource SchemaURL: 
ScopeMetrics #0
ScopeMetrics SchemaURL: 
InstrumentationScope otelcol/statsdreceiver 0.103.0-dev
Metric #0
Descriptor:
     -> Name: -n my.metric
     -> Description: 
     -> Unit: 
     -> DataType: Sum
     -> IsMonotonic: false
     -> AggregationTemporality: Delta
NumberDataPoints #0
Data point attributes:
     -> metric_type: Str(counter)
StartTimestamp: 2024-08-26 21:35:50.223101 +0000 UTC
Timestamp: 2024-08-26 21:36:00.224252 +0000 UTC
Value: 7
        {"kind": "exporter", "data_type": "metrics", "name": "debug"}
2024-08-26T23:36:10.224+0200    info    MetricsExporter {"kind": "exporter", "data_type": "metrics", "name": "debug", "resource metrics": 1, "metrics": 1, "data points": 1}
2024-08-26T23:36:10.224+0200    info    ResourceMetrics #0
Resource SchemaURL: 
ScopeMetrics #0
ScopeMetrics SchemaURL: 
InstrumentationScope otelcol/statsdreceiver 0.103.0-dev
Metric #0
Descriptor:
     -> Name: -n my.metric
     -> Description: 
     -> Unit: 
     -> DataType: Sum
     -> IsMonotonic: false
     -> AggregationTemporality: Delta
NumberDataPoints #0
Data point attributes:
     -> metric_type: Str(counter)
StartTimestamp: 2024-08-26 21:36:00.224252 +0000 UTC
Timestamp: 2024-08-26 21:36:10.224607 +0000 UTC
Value: 4
        {"kind": "exporter", "data_type": "metrics", "name": "debug"}
```

Instead of generating 11 metrics for each port that was used to send,
only 2 metrics are blocks are returned, who's values total 11.

![2024-08-26 23 44
15](https://github.com/user-attachments/assets/6b8a89d1-186e-4257-9c82-90c5f9d14f98)


**Link to tracking Issue:** open-telemetry#23809

**Testing:** 

- [x] Added unit tests

**Documentation:** <Describe the documentation added.>

- [x] Added information to the statsdreceiver `README.md` describing the
option.

---------

Co-authored-by: Povilas Versockas <povilas.versockas@coralogix.com>
  • Loading branch information
2 people authored and jriguera committed Oct 4, 2024
1 parent 48168ea commit 7fce809
Show file tree
Hide file tree
Showing 7 changed files with 112 additions and 26 deletions.
28 changes: 28 additions & 0 deletions .chloggen/es-337.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
# 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: |
Add support for aggregating on Host/IP.
# 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]
2 changes: 2 additions & 0 deletions receiver/statsdreceiver/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
13 changes: 7 additions & 6 deletions receiver/statsdreceiver/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
2 changes: 1 addition & 1 deletion receiver/statsdreceiver/internal/protocol/parser.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down
34 changes: 25 additions & 9 deletions receiver/statsdreceiver/internal/protocol/statsd_parser.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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}
}
58 changes: 48 additions & 10 deletions receiver/statsdreceiver/internal/protocol/statsd_parser_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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] {
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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 {
Expand All @@ -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",
Expand All @@ -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(
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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"},
Expand Down Expand Up @@ -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)
Expand All @@ -1927,3 +1927,41 @@ 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:3|c", testAddr02))
require.Len(t, p.instrumentsByAddress, 1)

for k := range p.instrumentsByAddress {
assert.Equal(t, "1.2.3.4", k.String)
assert.Equal(t, "udp", k.Network)
}
metrics := p.GetMetrics()
require.Len(t, metrics, 1)

value := metrics[0].Metrics.
ResourceMetrics().At(0).
ScopeMetrics().At(0).
Metrics().At(0).Sum().DataPoints().At(0).IntValue()

assert.Equal(t, int64(4), value)

}
1 change: 1 addition & 0 deletions receiver/statsdreceiver/receiver.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down

0 comments on commit 7fce809

Please sign in to comment.