[chore] spanmetricsprocessor: batch data points into one metric (#16102)

This helps a lot if metrics are exported as OTLP down the stack.

Signed-off-by: Bogdan Drutu <[email protected]>

Signed-off-by: Bogdan Drutu <[email protected]>

processor/spanmetricsprocessor/processor.go

@@ -299,15 +299,15 @@ func (p *processorImp) buildMetrics() (pmetric.Metrics, error) {
 // collectLatencyMetrics collects the raw latency metrics, writing the data
 // into the given instrumentation library metrics.
 func (p *processorImp) collectLatencyMetrics(ilm pmetric.ScopeMetrics) error {
+ mLatency := ilm.Metrics().AppendEmpty()
+ mLatency.SetName("latency")
+ mLatency.SetUnit("ms")
+ mLatency.SetEmptyHistogram().SetAggregationTemporality(p.config.GetAggregationTemporality())
+ dps := mLatency.Histogram().DataPoints()
+ dps.EnsureCapacity(len(p.histograms))
+ timestamp := pcommon.NewTimestampFromTime(time.Now())
  for key, hist := range p.histograms {
- mLatency := ilm.Metrics().AppendEmpty()
- mLatency.SetName("latency")
- mLatency.SetUnit("ms")
- mLatency.SetEmptyHistogram().SetAggregationTemporality(p.config.GetAggregationTemporality())
-
- timestamp := pcommon.NewTimestampFromTime(time.Now())
-
- dpLatency := mLatency.Histogram().DataPoints().AppendEmpty()
+ dpLatency := dps.AppendEmpty()
  dpLatency.SetStartTimestamp(p.startTimestamp)
  dpLatency.SetTimestamp(timestamp)
  dpLatency.ExplicitBounds().FromRaw(p.latencyBounds)
@@ -330,15 +330,17 @@ func (p *processorImp) collectLatencyMetrics(ilm pmetric.ScopeMetrics) error {
 // collectCallMetrics collects the raw call count metrics, writing the data
 // into the given instrumentation library metrics.
 func (p *processorImp) collectCallMetrics(ilm pmetric.ScopeMetrics) error {
+ mCalls := ilm.Metrics().AppendEmpty()
+ mCalls.SetName("calls_total")
+ mCalls.SetEmptySum().SetIsMonotonic(true)
+ mCalls.Sum().SetAggregationTemporality(p.config.GetAggregationTemporality())
+ dps := mCalls.Sum().DataPoints()
+ dps.EnsureCapacity(len(p.histograms))
+ timestamp := pcommon.NewTimestampFromTime(time.Now())
  for key, hist := range p.histograms {
- mCalls := ilm.Metrics().AppendEmpty()
- mCalls.SetName("calls_total")
- mCalls.SetEmptySum().SetIsMonotonic(true)
- mCalls.Sum().SetAggregationTemporality(p.config.GetAggregationTemporality())
-
- dpCalls := mCalls.Sum().DataPoints().AppendEmpty()
+ dpCalls := dps.AppendEmpty()
  dpCalls.SetStartTimestamp(p.startTimestamp)
- dpCalls.SetTimestamp(pcommon.NewTimestampFromTime(time.Now()))
+ dpCalls.SetTimestamp(timestamp)
  dpCalls.SetIntValue(int64(hist.count))
 
  dimensions, err := p.getDimensionsByMetricKey(key)

processor/spanmetricsprocessor/processor_test.go

@@ -431,8 +431,8 @@ func verifyMultipleCumulativeConsumptions() func(t testing.TB, input pmetric.Met
 // verifyConsumeMetricsInput verifies the input of the ConsumeMetrics call from this processor.
 // This is the best point to verify the computed metrics from spans are as expected.
 func verifyConsumeMetricsInput(t testing.TB, input pmetric.Metrics, expectedTemporality pmetric.AggregationTemporality, numCumulativeConsumptions int) bool {
- require.Equal(t, 6, input.MetricCount(),
- "Should be 3 for each of call count and latency. Each group of 3 metrics is made of: "+
+ require.Equal(t, 6, input.DataPointCount(),
+ "Should be 3 for each of call count and latency. Each group of 3 data points is made of: "+
  "service-a (server kind) -> service-a (client kind) -> service-b (service kind)",
  )
 
@@ -444,44 +444,32 @@ func verifyConsumeMetricsInput(t testing.TB, input pmetric.Metrics, expectedTemp
  assert.Equal(t, "spanmetricsprocessor", ilm.At(0).Scope().Name())
 
  m := ilm.At(0).Metrics()
- require.Equal(t, 6, m.Len())
+ require.Equal(t, 2, m.Len())
 
  seenMetricIDs := make(map[metricID]bool)
- mi := 0
- // The first 3 metrics are for call counts.
- for ; mi < 3; mi++ {
- assert.Equal(t, "calls_total", m.At(mi).Name())
-
- data := m.At(mi).Sum()
- assert.Equal(t, expectedTemporality, data.AggregationTemporality())
- assert.True(t, data.IsMonotonic())
-
- dps := data.DataPoints()
- require.Equal(t, 1, dps.Len())
-
- dp := dps.At(0)
+ // The first 3 data points are for call counts.
+ assert.Equal(t, "calls_total", m.At(0).Name())
+ assert.Equal(t, expectedTemporality, m.At(0).Sum().AggregationTemporality())
+ assert.True(t, m.At(0).Sum().IsMonotonic())
+ callsDps := m.At(0).Sum().DataPoints()
+ require.Equal(t, 3, callsDps.Len())
+ for dpi := 0; dpi < 3; dpi++ {
+ dp := callsDps.At(dpi)
  assert.Equal(t, int64(numCumulativeConsumptions), dp.IntValue(), "There should only be one metric per Service/operation/kind combination")
  assert.NotZero(t, dp.StartTimestamp(), "StartTimestamp should be set")
  assert.NotZero(t, dp.Timestamp(), "Timestamp should be set")
-
  verifyMetricLabels(dp, t, seenMetricIDs)
  }
 
  seenMetricIDs = make(map[metricID]bool)
- // The remaining metrics are for latency.
- for ; mi < m.Len(); mi++ {
- metric := m.At(mi)
-
- assert.Equal(t, "latency", metric.Name())
- assert.Equal(t, "ms", metric.Unit())
-
- data := metric.Histogram()
- assert.Equal(t, expectedTemporality, data.AggregationTemporality())
-
- dps := data.DataPoints()
- require.Equal(t, 1, dps.Len())
-
- dp := dps.At(0)
+ // The remaining 3 data points are for latency.
+ assert.Equal(t, "latency", m.At(1).Name())
+ assert.Equal(t, "ms", m.At(1).Unit())
+ assert.Equal(t, expectedTemporality, m.At(1).Histogram().AggregationTemporality())
+ latencyDps := m.At(1).Histogram().DataPoints()
+ require.Equal(t, 3, latencyDps.Len())
+ for dpi := 0; dpi < 3; dpi++ {
+ dp := latencyDps.At(dpi)
  assert.Equal(t, sampleLatency*float64(numCumulativeConsumptions), dp.Sum(), "Should be a 11ms latency measurement, multiplied by the number of stateful accumulations.")
  assert.NotZero(t, dp.Timestamp(), "Timestamp should be set")