[pkg/otlp/metrics] Add support for NoRecordedValue flag (#330)

.chloggen/mx-psi_no-recorded-flag.yaml

@@ -0,0 +1,17 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component (e.g. pkg/quantile)
+component: pkg/otlp/metrics
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Ignore metric datapoints with 'no recorded value' flag
+
+# The PR related to this change
+issues: [330]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: | 
+   - This flag is not supported by Datadog, so we just ignore these datapoints.

pkg/otlp/metrics/metrics_translator.go

@@ -151,6 +151,11 @@ func (t *Translator) mapNumberMetrics(
 
 	for i := 0; i < slice.Len(); i++ {
 		p := slice.At(i)
+		if p.Flags().NoRecordedValue() {
+			// No recorded value, skip.
+			continue
+		}
+
 		pointDims := dims.WithAttributeMap(p.Attributes())
 		var val float64
 		switch p.ValueType() {
@@ -203,6 +208,11 @@ func (t *Translator) mapNumberMonotonicMetrics(
 ) {
 	for i := 0; i < slice.Len(); i++ {
 		p := slice.At(i)
+		if p.Flags().NoRecordedValue() {
+			// No recorded value, skip.
+			continue
+		}
+
 		ts := uint64(p.Timestamp())
 		startTs := uint64(p.StartTimestamp())
 		pointDims := dims.WithAttributeMap(p.Attributes())
@@ -450,6 +460,11 @@ func (t *Translator) mapHistogramMetrics(
 ) {
 	for i := 0; i < slice.Len(); i++ {
 		p := slice.At(i)
+		if p.Flags().NoRecordedValue() {
+			// No recorded value, skip.
+			continue
+		}
+
 		startTs := uint64(p.StartTimestamp())
 		ts := uint64(p.Timestamp())
 		pointDims := dims.WithAttributeMap(p.Attributes())
@@ -554,6 +569,11 @@ func (t *Translator) mapSummaryMetrics(
 
 	for i := 0; i < slice.Len(); i++ {
 		p := slice.At(i)
+		if p.Flags().NoRecordedValue() {
+			// No recorded value, skip.
+			continue
+		}
+
 		startTs := uint64(p.StartTimestamp())
 		ts := uint64(p.Timestamp())
 		pointDims := dims.WithAttributeMap(p.Attributes())

pkg/otlp/metrics/metrics_translator_test.go

@@ -419,6 +419,40 @@ func TestMapIntMonotonicWithRebootWithinSlice(t *testing.T) {
 	})
 }
 
+// This test checks that in the case of a reboot within a NumberDataPointSlice,
+// we cache the value but we do NOT compute first value for the value at reset.
+func TestMapIntMonotonicWithNoRecordedValueWithinSlice(t *testing.T) {
+
+	buildMonotonicWithNoRecorded := func() (slice pmetric.NumberDataPointSlice) {
+		values := []int64{0, 30, 0, 40}
+		slice = pmetric.NewNumberDataPointSlice()
+		slice.EnsureCapacity(len(values))
+
+		for i, val := range values {
+			point := slice.AppendEmpty()
+			point.SetTimestamp(seconds(i * 10))
+			point.SetIntValue(val)
+		}
+
+		var flags pmetric.DataPointFlags
+		slice.At(2).SetFlags(flags.WithNoRecordedValue(true))
+		return
+	}
+
+	slice := buildMonotonicWithNoRecorded()
+	ctx := context.Background()
+	tr := newTranslator(t, zap.NewNop())
+	consumer := &mockTimeSeriesConsumer{}
+	tr.mapNumberMonotonicMetrics(ctx, consumer, exampleDims, slice)
+	assert.ElementsMatch(t,
+		consumer.metrics,
+		[]metric{
+			newCount(exampleDims, uint64(seconds(10)), 30),
+			newCount(exampleDims, uint64(seconds(30)), 10),
+		},
+	)
+}
+
 // This test checks that in the case of a reboot at the first point in a NumberDataPointSlice:
 // - diff: we cache the value AND compute first value
 // - rate: we cache the value AND don't compute first value