receiver/prometheus: add metricGroup.toNumberDataPoint pdata conversi…

…on (#3674)

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

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)
 	}
 

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 {

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())
+		})
+	}
+}