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

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,114 @@ 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.HistogramDataPoint
+	}{
+		{
+			name:   "histogram",
+			labels: labels.Labels{{Name: "a", Value: "A"}, {Name: "le", Value: "0.75"}, {Name: "b", Value: "B"}},
+			scrapes: []*scrape{
+				{at: 11, value: 10, metric: "histogram_count"},
+				{at: 11, value: 1004.78, metric: "histogram_sum"},
+				{at: 13, value: 33.7, metric: "value"},
+			},
+			want: func() pdata.HistogramDataPoint {
+				point := pdata.NewHistogramDataPoint()
+				point.SetCount(10)
+				point.SetSum(1004.78)
+				point.SetTimestamp(11 * 1e6) // the time in milliseconds -> nanoseconds.
+				point.SetBucketCounts([]uint64{33})
+				point.SetExplicitBounds([]float64{})
+				point.SetStartTimestamp(11 * 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)
+
+			hdpL := pdata.NewHistogramDataPointSlice()
+			require.True(t, mp.groups[groupKey].toDistributionPoint(mp.labelKeysOrdered, &hdpL))
+			require.Equal(t, 1, hdpL.Len(), "Exactly one point expected")
+			got := hdpL.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())
+		})
+	}
+}