[dogstatsd] correctly set the metric type on no aggregation pipeline …

…metrics (#13243)

* [dogstatsd] correctly set the metric type on no aggregation pipeline metrics.

* [dogstatsd] add a unit test

pkg/aggregator/no_aggregation_stream_worker.go

@@ -152,7 +152,8 @@ func (w *noAggregationStreamWorker) run() {
  serie.Points = []metrics.Point{{Ts: sample.Timestamp, Value: sample.Value}}
  serie.Tags = tagset.CompositeTagsFromSlice(w.metricBuffer.Copy())
  serie.Host = sample.Host
- // ignored when late but mimic dogstatsd traffic here anyway
+ serie.MType = sample.Mtype.ToAPIType()
+ // ignored by the intake when late but mimic dogstatsd traffic here anyway
  serie.Interval = 10
  w.seriesSink.Append(&serie)
 

pkg/dogstatsd/server_test.go

@@ -266,13 +266,26 @@ func TestUDPReceive(t *testing.T) {
  assert.Equal(t, sample.Name, "daemon2")
  demux.Reset()
 
- // Late metric
+ // Late gauge
  conn.Write([]byte("daemon:666|g|#sometag1:somevalue1,sometag2:somevalue2|T1658328888"))
  samples, timedSamples = demux.WaitForSamples(time.Second * 2)
  require.Len(t, samples, 0)
  require.Len(t, timedSamples, 1)
  sample = timedSamples[0]
  require.NotNil(t, sample)
+ assert.Equal(t, sample.Mtype, metrics.GaugeType)
+ assert.Equal(t, sample.Name, "daemon")
+ assert.Equal(t, sample.Timestamp, float64(1658328888))
+ demux.Reset()
+
+ // Late count
+ conn.Write([]byte("daemon:666|c|#sometag1:somevalue1,sometag2:somevalue2|T1658328888"))
+ samples, timedSamples = demux.WaitForSamples(time.Second * 2)
+ require.Len(t, samples, 0)
+ require.Len(t, timedSamples, 1)
+ sample = timedSamples[0]
+ require.NotNil(t, sample)
+ assert.Equal(t, sample.Mtype, metrics.CounterType)
  assert.Equal(t, sample.Name, "daemon")
  assert.Equal(t, sample.Timestamp, float64(1658328888))
  demux.Reset()
@@ -285,6 +298,7 @@ func TestUDPReceive(t *testing.T) {
  sample = timedSamples[0]
  require.NotNil(t, sample)
  assert.Equal(t, sample.Name, "daemon")
+ assert.Equal(t, sample.Mtype, metrics.GaugeType)
  assert.Equal(t, sample.Timestamp, float64(1658328888))
  sample = samples[0]
  require.NotNil(t, sample)

pkg/metrics/metric_sample.go

@@ -62,6 +62,23 @@ func (m MetricType) String() string {
  }
 }
 
+// ToAPIType returns the equivalent of MetricType in APIMetricType type.
+// APIMetricType only supports gauges, counts and rates and will default on gauges
+// for every other inputs.
+// This is used by the no-aggregation pipeline to infer an API type from a MetricType.
+func (m MetricType) ToAPIType() APIMetricType {
+ switch m {
+ case GaugeType:
+ return APIGaugeType
+ case CounterType:
+ return APICountType
+ case RateType:
+ return APIRateType
+ default:
+ return APIGaugeType
+ }
+}
+
 // MetricSampleContext allows to access a sample context data
 type MetricSampleContext interface {
  GetName() string