From 2ab79f26047f00faabab46b899edee490d4655a6 Mon Sep 17 00:00:00 2001 From: Emmanuel T Odeke Date: Sun, 18 Jul 2021 22:57:14 -0700 Subject: [PATCH] receiver/prometheus: add metricGroup.toNumberDataPoint pdata conversion Implements metricGroupPdata toNumberDataPoint and added unit tests as well as equivalence tests to ensure the migration will render the same results. While here, added TODOs for issue #3691 which found a bug in which cumulative types weren't using the actual duration start timestamp. Given that this current change is a translation of prior logic and has parity checks, making that bug fix would complicate the PR. Updates #3137 Depends on PR #3668 Updates PR #3427 Updates #3691 --- .../internal/metricfamily.go | 2 + .../internal/otlp_metricfamily.go | 19 ++++ .../internal/otlp_metricfamily_test.go | 106 ++++++++++++++++++ 3 files changed, 127 insertions(+) diff --git a/receiver/prometheusreceiver/internal/metricfamily.go b/receiver/prometheusreceiver/internal/metricfamily.go index e8f200a46f9c..b735bb0fcfb5 100644 --- a/receiver/prometheusreceiver/internal/metricfamily.go +++ b/receiver/prometheusreceiver/internal/metricfamily.go @@ -388,6 +388,8 @@ func (mg *metricGroup) toDoubleValueTimeSeries(orderedLabelKeys []string) *metri var startTs *timestamppb.Timestamp // gauge/undefined types has no start time if mg.family.isCumulativeType() { + // TODO(@odeke-em): use the actual interval start time as reported in + // https://github.com/open-telemetry/opentelemetry-collector/issues/3691 startTs = timestampFromMs(mg.ts) } diff --git a/receiver/prometheusreceiver/internal/otlp_metricfamily.go b/receiver/prometheusreceiver/internal/otlp_metricfamily.go index 80a34a1ac2cf..c6553643c30e 100644 --- a/receiver/prometheusreceiver/internal/otlp_metricfamily.go +++ b/receiver/prometheusreceiver/internal/otlp_metricfamily.go @@ -172,6 +172,25 @@ func (mg *metricGroupPdata) toSummaryPoint(orderedLabelKeys []string, dest *pdat return true } +func (mg *metricGroupPdata) toNumberDataPoint(orderedLabelKeys []string, dest *pdata.NumberDataPointSlice) bool { + var startTsNanos pdata.Timestamp + tsNanos := pdata.Timestamp(mg.ts * 1e6) + // gauge/undefined types have no start time. + if mg.family.isCumulativeTypePdata() { + // TODO(@odeke-em): use the actual interval start time as reported in + // https://github.com/open-telemetry/opentelemetry-collector/issues/3691 + startTsNanos = tsNanos + } + + point := dest.AppendEmpty() + point.SetStartTimestamp(startTsNanos) + point.SetTimestamp(tsNanos) + point.SetValue(mg.value) + populateLabelValuesPdata(orderedLabelKeys, mg.ls, point.LabelsMap()) + + return true +} + func populateLabelValuesPdata(orderedKeys []string, ls labels.Labels, dest pdata.StringMap) { src := ls.Map() for _, key := range orderedKeys { diff --git a/receiver/prometheusreceiver/internal/otlp_metricfamily_test.go b/receiver/prometheusreceiver/internal/otlp_metricfamily_test.go index d01bd77c5b34..d942b12ec77c 100644 --- a/receiver/prometheusreceiver/internal/otlp_metricfamily_test.go +++ b/receiver/prometheusreceiver/internal/otlp_metricfamily_test.go @@ -445,3 +445,109 @@ func TestMetricGroupData_toSummaryPointEquivalence(t *testing.T) { }) } } + +func TestMetricGroupData_toNumberDataUnitTest(t *testing.T) { + type scrape struct { + at int64 + value float64 + metric string + } + tests := []struct { + name string + labels labels.Labels + scrapes []*scrape + want func() pdata.NumberDataPoint + }{ + { + name: "counter", + labels: labels.Labels{{Name: "a", Value: "A"}, {Name: "b", Value: "B"}}, + scrapes: []*scrape{ + {at: 38, value: 39.9, metric: "value"}, + }, + want: func() pdata.NumberDataPoint { + point := pdata.NewNumberDataPoint() + point.SetValue(39.9) + point.SetTimestamp(38 * 1e6) // the time in milliseconds -> nanoseconds. + point.SetStartTimestamp(38 * 1e6) + labelsMap := point.LabelsMap() + labelsMap.Insert("a", "A") + labelsMap.Insert("b", "B") + return point + }, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + mp := newMetricFamilyPdata(tt.name, mc).(*metricFamilyPdata) + for _, tv := range tt.scrapes { + require.NoError(t, mp.Add(tv.metric, tt.labels.Copy(), tv.at, tv.value)) + } + + require.Equal(t, 1, len(mp.groups), "Expecting exactly 1 groupKey") + groupKey := mp.getGroupKey(tt.labels.Copy()) + require.NotNil(t, mp.groups[groupKey], "Expecting the groupKey to have a value given key:: "+groupKey) + + ndpL := pdata.NewNumberDataPointSlice() + require.True(t, mp.groups[groupKey].toNumberDataPoint(mp.labelKeysOrdered, &ndpL)) + require.Equal(t, 1, ndpL.Len(), "Exactly one point expected") + got := ndpL.At(0) + want := tt.want() + require.Equal(t, want, got, "Expected the points to be equal") + }) + } +} + +func TestMetricGroupData_toNumberDataPointEquivalence(t *testing.T) { + type scrape struct { + at int64 + value float64 + metric string + } + tests := []struct { + name string + labels labels.Labels + scrapes []*scrape + }{ + { + name: "counter", + labels: labels.Labels{{Name: "a", Value: "A"}, {Name: "b", Value: "B"}}, + scrapes: []*scrape{ + {at: 13, value: 33.7, metric: "value"}, + }, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + mf := newMetricFamily(tt.name, mc, zap.NewNop()).(*metricFamily) + mp := newMetricFamilyPdata(tt.name, mc).(*metricFamilyPdata) + for _, tv := range tt.scrapes { + require.NoError(t, mp.Add(tv.metric, tt.labels.Copy(), tv.at, tv.value)) + require.NoError(t, mf.Add(tv.metric, tt.labels.Copy(), tv.at, tv.value)) + } + groupKey := mf.getGroupKey(tt.labels.Copy()) + ocTimeseries := mf.groups[groupKey].toDoubleValueTimeSeries(mf.labelKeysOrdered) + ddpL := pdata.NewNumberDataPointSlice() + require.True(t, mp.groups[groupKey].toNumberDataPoint(mp.labelKeysOrdered, &ddpL)) + require.Equal(t, len(ocTimeseries.Points), ddpL.Len(), "They should have the exact same number of points") + require.Equal(t, 1, ddpL.Len(), "Exactly one point expected") + ocPoint := ocTimeseries.Points[0] + pdataPoint := ddpL.At(0) + // 1. Ensure that the startTimestamps are equal. + require.Equal(t, ocTimeseries.GetStartTimestamp().AsTime(), pdataPoint.Timestamp().AsTime(), "The timestamp must be equal") + // 2. Ensure that the value is equal. + require.Equal(t, ocPoint.GetDoubleValue(), pdataPoint.Value(), "Count must be equal") + // 4. Ensure that the point's timestamp is equal to that from the OpenCensusProto data point. + require.Equal(t, ocPoint.GetTimestamp().AsTime(), pdataPoint.Timestamp().AsTime(), "Point timestamps must be equal") + // 5. Ensure that the labels all match up. + ocStringMap := pdata.NewStringMap() + for i, labelValue := range ocTimeseries.LabelValues { + ocStringMap.Insert(mf.labelKeysOrdered[i], labelValue.Value) + } + require.Equal(t, ocStringMap.Sort(), pdataPoint.LabelsMap().Sort()) + }) + } +}