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

…prometheus (open-telemetry#24026)

**Description:** 

Implement down-scaling of exponential histograms to Prometheus native
histograms in Prometheus remote writer.
Configuration of down-scaling is TBD.

**Link to tracking Issue:**

Fixes: open-telemetry#17565 

**Testing:**

Unit tests.

**Documentation:**

TBD

---------

Signed-off-by: György Krajcsovits <[email protected]>
Co-authored-by: Ruslan Kovalov <[email protected]>
Co-authored-by: Juraci Paixão Kröhling <[email protected]>
Co-authored-by: Anthony Mirabella <[email protected]>

.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 if necessary.
+
+# 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.
+ numBuckets := 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 < numBuckets; 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(numBuckets)+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
 }