Error on non monotonic bounds (#744)

* Error on non monotonic bounds

* changelog

* add issue #

* feedback

Author: mackjmr

.chloggen/mackjmr_err-non-monotonic-bounds.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component (e.g. pkg/quantile)
+component: pkg/quantile
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Return an error on non monotonic bounds for histogram metrics. Previously, this would create a panic.
+
+# The PR related to this change
+issues: [744]
+
+# (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:

pkg/otlp/metrics/histograms_test.go

@@ -6,12 +6,16 @@
 package metrics
 
 import (
+	"context"
+	"os"
 	"testing"
 
 	"github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes"
+	"github.com/DataDog/opentelemetry-mapping-go/pkg/quantile"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zaptest/observer"
@@ -146,6 +150,28 @@ func TestDeltaHistogramTranslatorOptions(t *testing.T) {
 	}
 }
 
+func TestNonMonotonicCount(t *testing.T) {
+	// Unmarshal OTLP data.
+	otlpbytes, err := os.ReadFile("testdata/otlpdata/histogram/simple-delta-non-monotonic-bound.json")
+	require.NoError(t, err, "failed to read OTLP file %q", "testdata/otlpdata/histogram/simple-delta-non-monotonic-bound.json")
+
+	var unmarshaler pmetric.JSONUnmarshaler
+	otlpdata, err := unmarshaler.UnmarshalMetrics(otlpbytes)
+	require.NoError(t, err, "failed to unmarshal OTLP data from file %q", "testdata/otlpdata/histogram/simple-delta-non-monotonic-bound.json")
+	// Map metrics using translator.
+	consumer := NewTestConsumer()
+	options := []TranslatorOption{WithOriginProduct(OriginProductDatadogAgent)}
+	set := componenttest.NewNopTelemetrySettings()
+	attributesTranslator, err := attributes.NewTranslator(set)
+	assert.NoError(t, err)
+
+	translator, err := NewTranslator(set, attributesTranslator, options...)
+	assert.NoError(t, err)
+
+	_, err = translator.MapMetrics(context.Background(), otlpdata, &consumer, nil)
+	assert.EqualError(t, err, quantile.ErrNonMonotonicBoundaries)
+}
+
 func TestCumulativeHistogramTranslatorOptions(t *testing.T) {
 	tests := []struct {
 		name     string

pkg/otlp/metrics/metrics_translator.go

@@ -298,7 +298,7 @@ func (t *Translator) getSketchBuckets(
 	p pmetric.HistogramDataPoint,
 	histInfo histogramInfo,
 	delta bool,
-) {
+) error {
 	startTs := uint64(p.StartTimestamp())
 	ts := uint64(p.Timestamp())
 	as := &quantile.Agent{}
@@ -360,10 +360,17 @@ func (t *Translator) getSketchBuckets(
 		var nonZeroBucket bool
 		if delta {
 			nonZeroBucket = count > 0
-			as.InsertInterpolate(lowerBound, upperBound, uint(count))
+			err := as.InsertInterpolate(lowerBound, upperBound, uint(count))
+			if err != nil {
+				return err
+			}
 		} else if dx, ok := t.prevPts.Diff(bucketDims, startTs, ts, float64(count)); ok {
 			nonZeroBucket = dx > 0
-			as.InsertInterpolate(lowerBound, upperBound, uint(dx))
+			err := as.InsertInterpolate(lowerBound, upperBound, uint(dx))
+			if err != nil {
+				return err
+			}
+
 		}
 
 		if nonZeroBucket {
@@ -413,6 +420,7 @@ func (t *Translator) getSketchBuckets(
 
 		consumer.ConsumeSketch(ctx, pointDims, ts, 0, sketch)
 	}
+	return nil
 }
 
 func (t *Translator) getLegacyBuckets(
@@ -462,7 +470,7 @@ func (t *Translator) mapHistogramMetrics(
 	dims *Dimensions,
 	slice pmetric.HistogramDataPointSlice,
 	delta bool,
-) {
+) error {
 	for i := 0; i < slice.Len(); i++ {
 		p := slice.At(i)
 		if p.Flags().NoRecordedValue() {
@@ -532,9 +540,13 @@ func (t *Translator) mapHistogramMetrics(
 		case HistogramModeCounters:
 			t.getLegacyBuckets(ctx, consumer, pointDims, p, delta)
 		case HistogramModeDistributions:
-			t.getSketchBuckets(ctx, consumer, pointDims, p, histInfo, delta)
+			err := t.getSketchBuckets(ctx, consumer, pointDims, p, histInfo, delta)
+			if err != nil {
+				return err
+			}
 		}
 	}
+	return nil
 }
 
 // formatFloat formats a float number as close as possible to what
@@ -825,12 +837,18 @@ func (t *Translator) MapMetrics(ctx context.Context, md pmetric.Metrics, consume
 					renameMetrics(md)
 				}
 
-				t.mapToDDFormat(ctx, md, consumer, additionalTags, host, scopeName, rattrs)
+				err := t.mapToDDFormat(ctx, md, consumer, additionalTags, host, scopeName, rattrs)
+				if err != nil {
+					return metadata, err
+				}
 			}
 
 			for k := 0; k < newMetrics.Len(); k++ {
 				md := newMetrics.At(k)
-				t.mapToDDFormat(ctx, md, consumer, additionalTags, host, scopeName, rattrs)
+				err := t.mapToDDFormat(ctx, md, consumer, additionalTags, host, scopeName, rattrs)
+				if err != nil {
+					return metadata, err
+				}
 			}
 		}
 
@@ -851,7 +869,7 @@ func (t *Translator) MapMetrics(ctx context.Context, md pmetric.Metrics, consume
 	return metadata, nil
 }
 
-func (t *Translator) mapToDDFormat(ctx context.Context, md pmetric.Metric, consumer Consumer, additionalTags []string, host string, scopeName string, rattrs pcommon.Map) {
+func (t *Translator) mapToDDFormat(ctx context.Context, md pmetric.Metric, consumer Consumer, additionalTags []string, host string, scopeName string, rattrs pcommon.Map) error {
 	baseDims := &Dimensions{
 		name:                md.Name(),
 		tags:                additionalTags,
@@ -889,7 +907,10 @@ func (t *Translator) mapToDDFormat(ctx context.Context, md pmetric.Metric, consu
 		switch md.Histogram().AggregationTemporality() {
 		case pmetric.AggregationTemporalityCumulative, pmetric.AggregationTemporalityDelta:
 			delta := md.Histogram().AggregationTemporality() == pmetric.AggregationTemporalityDelta
-			t.mapHistogramMetrics(ctx, consumer, baseDims, md.Histogram().DataPoints(), delta)
+			err := t.mapHistogramMetrics(ctx, consumer, baseDims, md.Histogram().DataPoints(), delta)
+			if err != nil {
+				return err
+			}
 		default: // pmetric.AggregationTemporalityUnspecified or any other not supported type
 			t.logger.Debug("Unknown or unsupported aggregation temporality",
 				zap.String("metric name", md.Name()),
@@ -912,4 +933,5 @@ func (t *Translator) mapToDDFormat(ctx context.Context, md pmetric.Metric, consu
 	default: // pmetric.MetricDataTypeNone or any other not supported type
 		t.logger.Debug("Unknown or unsupported metric type", zap.String(metricName, md.Name()), zap.Any("data type", md.Type()))
 	}
+	return nil
 }

pkg/otlp/metrics/testdata/otlpdata/histogram/simple-delta-non-monotonic-bound.json

@@ -0,0 +1,54 @@
+{
+  "resourceMetrics": [
+    {
+      "resource": {
+        "attributes": [
+          {
+            "key": "host.name",
+            "value": {
+              "stringValue": "hostname"
+            }
+          }
+        ]
+      },
+      "scopeMetrics": [
+        {
+          "scope": {},
+          "metrics": [
+            {
+              "name": "doubleHist.test",
+              "histogram": {
+                "dataPoints": [
+                  {
+                    "attributes": [
+                      {
+                        "key": "attribute_tag",
+                        "value": {
+                          "stringValue": "attribute_value"
+                        }
+                      }
+                    ],
+                    "timeUnixNano": "1667560641226420924",
+                    "count": "20",
+                    "sum": 3.141592653589793,
+                    "bucketCounts": [
+                      "2",
+                      "18"
+                    ],
+                    "explicitBounds": [
+                      1,
+                      0.1
+                    ],
+                    "min": -100,
+                    "max": 100
+                  }
+                ],
+                "aggregationTemporality": 1
+              }
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}

pkg/quantile/agent.go

@@ -5,11 +5,16 @@
 
 package quantile
 
+import "errors"
+
 const (
 	agentBufCap = 512
 )
 
-var agentConfig = Default()
+var (
+	agentConfig               = Default()
+	ErrNonMonotonicBoundaries = "explicit bucket histogram: non-monotonic boundaries"
+)
 
 // An Agent sketch is an insert optimized version of the sketch for use in the
 // datadog-agent.
@@ -83,14 +88,17 @@ func (a *Agent) Insert(v float64, sampleRate float64) {
 }
 
 // InsertInterpolate linearly interpolates a count from the given lower to upper bounds
-func (a *Agent) InsertInterpolate(lower float64, upper float64, count uint) {
+func (a *Agent) InsertInterpolate(lower float64, upper float64, count uint) error {
 	keys := make([]Key, 0)
 	for k := agentConfig.key(lower); k <= agentConfig.key(upper); k++ {
 		keys = append(keys, k)
 	}
 	whatsLeft := int(count)
 	distance := upper - lower
 	startIdx := 0
+	if len(keys) == 0 {
+		return errors.New(ErrNonMonotonicBoundaries)
+	}
 	lowerB := agentConfig.binLow(keys[startIdx])
 	endIdx := 1
 	var remainder float64
@@ -126,4 +134,5 @@ func (a *Agent) InsertInterpolate(lower float64, upper float64, count uint) {
 		a.CountBuf = append(a.CountBuf, KeyCount{k: keys[startIdx], n: uint(whatsLeft)})
 	}
 	a.flush()
+	return nil
 }