Add option to set Counter to be monotonic

receiver/statsdreceiver/README.md

@@ -21,6 +21,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.
 
+- `is_monotonic_counter` (default value is false): Set all counter-type metrics the statsd receiver received as monotonic.
+
 - `timer_histogram_mapping:`(default value is below): Specify what OTLP type to convert received timing/histogram data to.
 
 
@@ -39,6 +41,7 @@ receivers:
  endpoint: "localhost:8127"
  aggregation_interval: 70s
  enable_metric_type: true
+ is_monotonic_counter: false
  timer_histogram_mapping:
  - statsd_type: "histogram"
  observer_type: "gauge"
@@ -115,6 +118,7 @@ receivers:
  endpoint: "localhost:8125" # default
  aggregation_interval: 60s # default
  enable_metric_type: false # default
+ is_monotonic_counter: false # default
  timer_histogram_mapping:
  - statsd_type: "histogram"
  observer_type: "gauge"

receiver/statsdreceiver/config.go

@@ -31,6 +31,7 @@ type Config struct {
  NetAddr confignet.NetAddr `mapstructure:",squash"`
  AggregationInterval time.Duration `mapstructure:"aggregation_interval"`
  EnableMetricType bool `mapstructure:"enable_metric_type"`
+ IsMonotonicCounter bool `mapstructure:"is_monotonic_counter"`
  TimerHistogramMapping []protocol.TimerHistogramMapping `mapstructure:"timer_histogram_mapping"`
 }
 

receiver/statsdreceiver/factory.go

@@ -34,6 +34,7 @@ const (
  defaultTransport = "udp"
  defaultAggregationInterval = 60 * time.Second
  defaultEnableMetricType = false
+ defaultIsMonotonicCounter = false
 )
 
 var (
@@ -58,6 +59,7 @@ func createDefaultConfig() config.Receiver {
  },
  AggregationInterval: defaultAggregationInterval,
  EnableMetricType: defaultEnableMetricType,
+ IsMonotonicCounter: defaultIsMonotonicCounter,
  TimerHistogramMapping: defaultTimerHistogramMapping,
  }
 }

receiver/statsdreceiver/protocol/metric_translator.go

@@ -21,15 +21,20 @@ import (
  "go.opentelemetry.io/collector/model/pdata"
 )
 
-func buildCounterMetric(parsedMetric statsDMetric, timeNow time.Time) pdata.InstrumentationLibraryMetrics {
+func buildCounterMetric(parsedMetric statsDMetric, isMonotonicCounter bool, timeNow time.Time) pdata.InstrumentationLibraryMetrics {
  ilm := pdata.NewInstrumentationLibraryMetrics()
  nm := ilm.Metrics().AppendEmpty()
  nm.SetName(parsedMetric.description.name)
  if parsedMetric.unit != "" {
  nm.SetUnit(parsedMetric.unit)
  }
  nm.SetDataType(pdata.MetricDataTypeIntSum)
+
  nm.IntSum().SetAggregationTemporality(pdata.AggregationTemporalityDelta)
+ if isMonotonicCounter {
+ nm.IntSum().SetAggregationTemporality(pdata.AggregationTemporalityCumulative)
+ }
+
  nm.IntSum().SetIsMonotonic(true)
 
  dp := nm.IntSum().DataPoints().AppendEmpty()

receiver/statsdreceiver/protocol/parser.go

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

receiver/statsdreceiver/protocol/statsd_parser.go

@@ -54,6 +54,7 @@ type StatsDParser struct {
  summaries map[statsDMetricdescription]summaryMetric
  timersAndDistributions []pdata.InstrumentationLibraryMetrics
  enableMetricType bool
+ isMonotonicCounter bool
  observeTimer string
  observeHistogram string
 }
@@ -84,13 +85,14 @@ type statsDMetricdescription struct {
  labels attribute.Distinct
 }
 
-func (p *StatsDParser) Initialize(enableMetricType bool, sendTimerHistogram []TimerHistogramMapping) error {
+func (p *StatsDParser) Initialize(enableMetricType bool, isMonotonicCounter bool, sendTimerHistogram []TimerHistogramMapping) error {
  p.gauges = make(map[statsDMetricdescription]pdata.InstrumentationLibraryMetrics)
  p.counters = make(map[statsDMetricdescription]pdata.InstrumentationLibraryMetrics)
  p.timersAndDistributions = make([]pdata.InstrumentationLibraryMetrics, 0)
  p.summaries = make(map[statsDMetricdescription]summaryMetric)
 
  p.enableMetricType = enableMetricType
+ p.isMonotonicCounter = isMonotonicCounter
  for _, eachMap := range sendTimerHistogram {
  switch eachMap.StatsdType {
  case "histogram":
@@ -159,11 +161,11 @@ func (p *StatsDParser) Aggregate(line string) error {
  case statsdCounter:
  _, ok := p.counters[parsedMetric.description]
  if !ok {
- p.counters[parsedMetric.description] = buildCounterMetric(parsedMetric, timeNowFunc())
+ p.counters[parsedMetric.description] = buildCounterMetric(parsedMetric, p.isMonotonicCounter, timeNowFunc())
  } else {
  savedValue := p.counters[parsedMetric.description].Metrics().At(0).IntSum().DataPoints().At(0).Value()
  parsedMetric.intvalue = parsedMetric.intvalue + savedValue
- p.counters[parsedMetric.description] = buildCounterMetric(parsedMetric, timeNowFunc())
+ p.counters[parsedMetric.description] = buildCounterMetric(parsedMetric, p.isMonotonicCounter, timeNowFunc())
  }
 
  case statsdHistogram:

receiver/statsdreceiver/receiver.go

@@ -89,7 +89,7 @@ func (r *statsdReceiver) Start(ctx context.Context, host component.Host) error {
  ctx, r.cancel = context.WithCancel(ctx)
  var transferChan = make(chan string, 10)
  ticker := time.NewTicker(r.config.AggregationInterval)
- r.parser.Initialize(r.config.EnableMetricType, r.config.TimerHistogramMapping)
+ r.parser.Initialize(r.config.EnableMetricType, r.config.IsMonotonicCounter, r.config.TimerHistogramMapping)
  go func() {
  if err := r.server.ListenAndServe(r.parser, r.nextConsumer, r.reporter, transferChan); err != nil {
  host.ReportFatalError(err)