[statsdreceiver]add aggregation for StatsD receiver (#1670)

* add aggregation for StatsD receiver, support metrics types: gauge and counter

* revise readme for StatsD receiver

* Use Distinct in OTel-Go SDK as the description field for statsDMetric

Author: gavindoudou

receiver/statsdreceiver/README.md

@@ -1,6 +1,6 @@
 # StatsD Receiver
 
-StatsD receiver for ingesting StatsD messages into the OpenTelemetry Collector.
+StatsD receiver for ingesting StatsD messages(https://github.com/statsd/statsd/blob/master/docs/metric_types.md) into the OpenTelemetry Collector.
 
 Supported pipeline types: metrics
 
@@ -12,27 +12,46 @@ The following settings are required:
 
 - `endpoint` (default = `localhost:8125`): Address and port to listen on.
 
+
+The Following settings are optional:
+
+- `aggregation_interval: 70s`(default value is 60s): The aggregation time that the receiver aggregates the metrics (similar to the flush interval in StatsD server)
+
 Example:
 
 ```yaml
 receivers:
   statsd:
   statsd/2:
     endpoint: "localhost:8127"
+    aggregation_interval: 70s
 ```
 
 The full list of settings exposed for this receiver are documented [here](./config.go)
 with detailed sample configurations [here](./testdata/config.yaml).
 
 ## Aggregation
 
-Currently the `statsdreceiver` is not providing any aggregation. There are
-ideas such as the [Metrics Transform Processor
-Proposal](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/332)
-that intend to enable control over Metric aggregation in a processor.
-
-An alternative will be to implement some simple aggregation in this receiver.
-
+Aggregation is done in statsD receiver. The default aggregation interval is 60s. The receiver only aggregates the metrics with the same metric name, metric type, label keys and label values. After each aggregation interval, the receiver will send all metrics (after aggregation) in this aggregation interval to the following workflow.
+
+It supports:
+
+Gauge(transferred to double):
+- statsdTestMetric1:500|g|#mykey:myvalue
+statsdTestMetric1:400|g|#mykey:myvalue
+(get the latest value: 400)
+- statsdTestMetric1:500|g|#mykey:myvalue
+statsdTestMetric1:+2|g|#mykey:myvalue
+statsdTestMetric1:-1|g|#mykey:myvalue
+(get the value after calculation: 501)
+
+Counter(transferred to int):
+- statsdTestMetric1:3000|c|#mykey:myvalue
+statsdTestMetric1:4000|c|#mykey:myvalue
+(get the value after incrementation: 7000)
+- statsdTestMetric1:3000|c|#mykey:myvalue
+statsdTestMetric1:20|c|@0.8|#mykey:myvalue
+(get the value after incrementation with sample rate: 3000+20/0.8=3025)
 ## Metrics
 
 General format is:
@@ -43,14 +62,15 @@ General format is:
 
 `<name>:<value>|c|@<sample-rate>|#<tag1-key>:<tag1-value>`
 
+it supports sample rate
+
 ### Gauge
 
 `<name>:<value>|g|@<sample-rate>|#<tag1-key>:<tag1-value>`
 
 ### Timer
 
-`<name>:<value>|ms|@<sample-rate>|#<tag1-key>:<tag1-value>`
-
+TODO: add support for timer
 
 ## Testing
 
@@ -60,6 +80,7 @@ General format is:
 receivers:
   statsd:
     endpoint: "localhost:8125" # default
+    aggregation_interval: 60s  # default
 
 exporters:
   file:

receiver/statsdreceiver/config.go

@@ -15,6 +15,8 @@
 package statsdreceiver
 
 import (
+	"time"
+
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/config/confignet"
 )
@@ -23,4 +25,5 @@ import (
 type Config struct {
 	configmodels.ReceiverSettings `mapstructure:",squash"`
 	NetAddr                       confignet.NetAddr `mapstructure:",squash"`
+	AggregationInterval           time.Duration     `mapstructure:"aggregation_interval"`
 }

receiver/statsdreceiver/config_test.go

@@ -17,6 +17,7 @@ package statsdreceiver
 import (
 	"path"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -54,5 +55,6 @@ func TestLoadConfig(t *testing.T) {
 			Endpoint:  "localhost:12345",
 			Transport: "custom_transport",
 		},
+		AggregationInterval: 70 * time.Second,
 	}, r1)
 }

receiver/statsdreceiver/factory.go

@@ -16,6 +16,7 @@ package statsdreceiver
 
 import (
 	"context"
+	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configmodels"
@@ -26,9 +27,10 @@ import (
 
 const (
 	// The value of "type" key in configuration.
-	typeStr             = "statsd"
-	defaultBindEndpoint = "localhost:8125"
-	defaultTransport    = "udp"
+	typeStr                    = "statsd"
+	defaultBindEndpoint        = "localhost:8125"
+	defaultTransport           = "udp"
+	defaultAggregationInterval = 60 * time.Second
 )
 
 // NewFactory creates a factory for the StatsD receiver.
@@ -50,6 +52,7 @@ func createDefaultConfig() configmodels.Receiver {
 			Endpoint:  defaultBindEndpoint,
 			Transport: defaultTransport,
 		},
+		AggregationInterval: defaultAggregationInterval,
 	}
 }
 

receiver/statsdreceiver/factory_test.go

@@ -41,3 +41,15 @@ func TestCreateReceiver(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, tReceiver, "receiver creation failed")
 }
+
+func TestCreateMetricsReceiverWithNilConsumer(t *testing.T) {
+	receiver, err := createMetricsReceiver(
+		context.Background(),
+		component.ReceiverCreateParams{Logger: zap.NewNop()},
+		createDefaultConfig(),
+		nil,
+	)
+
+	assert.Error(t, err, "nil consumer")
+	assert.Nil(t, receiver)
+}

receiver/statsdreceiver/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/stretchr/testify v1.6.1
 	go.opencensus.io v0.22.5
 	go.opentelemetry.io/collector v0.16.1-0.20201207152538-326931de8c32
+	go.opentelemetry.io/otel v0.13.0
 	go.uber.org/zap v1.16.0
 	google.golang.org/protobuf v1.25.0
 )

receiver/statsdreceiver/protocol/parser.go

@@ -20,5 +20,7 @@ import (
 
 // Parser is something that can map input StatsD strings to OTLP Metric representations.
 type Parser interface {
-	Parse(in string) (*metricspb.Metric, error)
+	Initialize() error
+	GetMetrics() []*metricspb.Metric
+	Aggregate(line string) error
 }