[pkg/translate/promremotewrite] downscale exponential histograms for prometheus

.chloggen/prometheusremotewrite-downscale-histogram.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: pkg/translator/prometheusremotewrite
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Downscale exponential histograms to fit prometheus native histograms.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [17565]
+
+# (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/translator/prometheusremotewrite/histograms.go

@@ -58,15 +58,20 @@ func addSingleExponentialHistogramDataPoint(
 // to Prometheus Native Histogram.
 func exponentialToNativeHistogram(p pmetric.ExponentialHistogramDataPoint) (prompb.Histogram, error) {
 	scale := p.Scale()
-	if scale < -4 || scale > 8 {
+	if scale < -4 {
 		return prompb.Histogram{},
 			fmt.Errorf("cannot convert exponential to native histogram."+
-				" Scale must be <= 8 and >= -4, was %d", scale)
-		// TODO: downscale to 8 if scale > 8
+				" Scale must be >= -4, was %d", scale)
 	}
 
-	pSpans, pDeltas := convertBucketsLayout(p.Positive())
-	nSpans, nDeltas := convertBucketsLayout(p.Negative())
+	var scaleDown int32
+	if scale > 8 {
+		scaleDown = scale - 8
+		scale = 8
+	}
+
+	pSpans, pDeltas := convertBucketsLayout(p.Positive(), scaleDown)
+	nSpans, nDeltas := convertBucketsLayout(p.Negative(), scaleDown)
 
 	h := prompb.Histogram{
 		Schema: scale,
@@ -104,17 +109,19 @@ func exponentialToNativeHistogram(p pmetric.ExponentialHistogramDataPoint) (prom
 // The bucket indexes conversion was adjusted, since OTel exp. histogram bucket
 // index 0 corresponds to the range (1, base] while Prometheus bucket index 0
 // to the range (base 1].
-func convertBucketsLayout(buckets pmetric.ExponentialHistogramDataPointBuckets) ([]prompb.BucketSpan, []int64) {
+//
+// scaleDown is the factor by which the buckets are scaled down. In other words 2^scaleDown buckets will be merged into one.
+func convertBucketsLayout(buckets pmetric.ExponentialHistogramDataPointBuckets, scaleDown int32) ([]prompb.BucketSpan, []int64) {
 	bucketCounts := buckets.BucketCounts()
 	if bucketCounts.Len() == 0 {
 		return nil, nil
 	}
 
 	var (
-		spans         []prompb.BucketSpan
-		deltas        []int64
-		prevCount     int64
-		nextBucketIdx int32
+		spans     []prompb.BucketSpan
+		deltas    []int64
+		count     int64
+		prevCount int64
 	)
 
 	appendDelta := func(count int64) {
@@ -123,34 +130,67 @@ func convertBucketsLayout(buckets pmetric.ExponentialHistogramDataPointBuckets)
 		prevCount = count
 	}
 
-	for i := 0; i < bucketCounts.Len(); i++ {
-		count := int64(bucketCounts.At(i))
+	// Let the compiler figure out that this is const during this function by
+	// moving it into a local variable.
+	bucketsCount := bucketCounts.Len()
+
+	// The offset is scaled and adjusted by 1 as described above.
+	bucketIdx := buckets.Offset()>>scaleDown + 1
+	spans = append(spans, prompb.BucketSpan{
+		Offset: bucketIdx,
+		Length: 0,
+	})
+
+	for i := 0; i < bucketsCount; i++ {
+		// The offset is scaled and adjusted by 1 as described above.
+		nextBucketIdx := (int32(i)+buckets.Offset())>>scaleDown + 1
+		if bucketIdx == nextBucketIdx { // We have not collected enough buckets to merge yet.
+			count += int64(bucketCounts.At(i))
+			continue
+		}
 		if count == 0 {
+			count = int64(bucketCounts.At(i))
 			continue
 		}
 
-		// The offset is adjusted by 1 as described above.
-		bucketIdx := int32(i) + buckets.Offset() + 1
-		delta := bucketIdx - nextBucketIdx
-		if i == 0 || delta > 2 {
-			// We have to create a new span, either because we are
-			// at the very beginning, or because we have found a gap
+		gap := nextBucketIdx - bucketIdx - 1
+		if gap > 2 {
+			// We have to create a new span, because we have found a gap
 			// of more than two buckets. The constant 2 is copied from the logic in
 			// https://github.com/prometheus/client_golang/blob/27f0506d6ebbb117b6b697d0552ee5be2502c5f2/prometheus/histogram.go#L1296
 			spans = append(spans, prompb.BucketSpan{
-				Offset: delta,
+				Offset: gap,
 				Length: 0,
 			})
 		} else {
 			// We have found a small gap (or no gap at all).
 			// Insert empty buckets as needed.
-			for j := int32(0); j < delta; j++ {
+			for j := int32(0); j < gap; j++ {
 				appendDelta(0)
 			}
 		}
 		appendDelta(count)
-		nextBucketIdx = bucketIdx + 1
+		count = int64(bucketCounts.At(i))
+		bucketIdx = nextBucketIdx
+	}
+	// Need to use the last item's index. The offset is scaled and adjusted by 1 as described above.
+	gap := (int32(bucketsCount)+buckets.Offset()-1)>>scaleDown + 1 - bucketIdx
+	if gap > 2 {
+		// We have to create a new span, because we have found a gap
+		// of more than two buckets. The constant 2 is copied from the logic in
+		// https://github.com/prometheus/client_golang/blob/27f0506d6ebbb117b6b697d0552ee5be2502c5f2/prometheus/histogram.go#L1296
+		spans = append(spans, prompb.BucketSpan{
+			Offset: gap,
+			Length: 0,
+		})
+	} else {
+		// We have found a small gap (or no gap at all).
+		// Insert empty buckets as needed.
+		for j := int32(0); j < gap; j++ {
+			appendDelta(0)
+		}
 	}
+	appendDelta(count)
 
 	return spans, deltas
 }