Skip to content

Commit 2b794c6

Browse files
authored
Add helpers for OTLP histogram -> DD Sketch (#41967)
### What does this PR do? Copy the helpers for OTLP histogram -> DD Sketch translation from dd-go (created in [this PR](DataDog/dd-go#198508)) to the opentelemetry-mapping-go library. A followup PR on dd-go will import from here. ### Motivation ### Describe how you validated your changes ### Additional Notes
1 parent e012fb7 commit 2b794c6

4 files changed

Lines changed: 544 additions & 13 deletions

File tree

Lines changed: 191 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,191 @@
1+
// Unless explicitly stated otherwise all files in this repository are licensed
2+
// under the Apache License Version 2.0.
3+
// This product includes software developed at Datadog (https://www.datadoghq.com/).
4+
// Copyright 2022-present Datadog, Inc.
5+
6+
package metrics
7+
8+
import (
9+
"fmt"
10+
"math"
11+
"time"
12+
13+
"go.opentelemetry.io/collector/pdata/pcommon"
14+
"go.opentelemetry.io/collector/pdata/pmetric"
15+
16+
"github.com/DataDog/sketches-go/ddsketch"
17+
"github.com/DataDog/sketches-go/ddsketch/mapping"
18+
"github.com/DataDog/sketches-go/ddsketch/store"
19+
)
20+
21+
// These methods will be imported in dd-go to convert OTLP to DD trace metrics - see https://github.com/DataDog/dd-go/pull/198508
22+
23+
// getTimeUnitScaleToNanos returns the scaling factor to convert the given unit to nanoseconds
24+
func getTimeUnitScaleToNanos(unit string) float64 {
25+
switch unit {
26+
case "ns":
27+
return float64(time.Nanosecond)
28+
case "us", "μs":
29+
return float64(time.Microsecond)
30+
case "ms":
31+
return float64(time.Millisecond)
32+
case "s":
33+
return float64(time.Second)
34+
case "min":
35+
return float64(time.Minute)
36+
case "h":
37+
return float64(time.Hour)
38+
default:
39+
// If unit is unknown, assume seconds (common for duration metrics)
40+
return float64(time.Second)
41+
}
42+
}
43+
44+
// getBounds returns the lower and upper bounds for a histogram bucket
45+
func getBounds(explicitBounds pcommon.Float64Slice, idx int) (lowerBound float64, upperBound float64) {
46+
// See https://github.com/open-telemetry/opentelemetry-proto/blob/v0.10.0/opentelemetry/proto/metrics/v1/metrics.proto#L427-L439
47+
lowerBound = math.Inf(-1)
48+
upperBound = math.Inf(1)
49+
if idx > 0 {
50+
lowerBound = explicitBounds.At(idx - 1)
51+
}
52+
if idx < explicitBounds.Len() {
53+
upperBound = explicitBounds.At(idx)
54+
}
55+
return
56+
}
57+
58+
// CreateDDSketchFromHistogramOfDuration creates a DDSketch from regular histogram data point
59+
func CreateDDSketchFromHistogramOfDuration(dp pmetric.HistogramDataPoint, unit string) (*ddsketch.DDSketch, error) {
60+
relativeAccuracy := 0.01 // 1% relative accuracy
61+
maxNumBins := 2048
62+
newSketch, err := ddsketch.LogCollapsingLowestDenseDDSketch(relativeAccuracy, maxNumBins)
63+
if err != nil {
64+
return nil, err
65+
}
66+
67+
bucketCounts := dp.BucketCounts()
68+
explicitBounds := dp.ExplicitBounds()
69+
70+
// Get scaling factor to convert unit to nanoseconds
71+
scaleToNanos := getTimeUnitScaleToNanos(unit)
72+
73+
// Find first and last bucket indices with count > 0
74+
lowestBucketIndex := -1
75+
highestBucketIndex := -1
76+
for j := 0; j < bucketCounts.Len(); j++ {
77+
count := bucketCounts.At(j)
78+
if count > 0 {
79+
if lowestBucketIndex == -1 {
80+
lowestBucketIndex = j
81+
}
82+
highestBucketIndex = j
83+
}
84+
}
85+
86+
hasMin := dp.HasMin()
87+
hasMax := dp.HasMax()
88+
minNanoseconds := dp.Min() * scaleToNanos
89+
maxNanoseconds := dp.Max() * scaleToNanos
90+
91+
for j := 0; j < bucketCounts.Len(); j++ {
92+
lowerBound, upperBound := getBounds(explicitBounds, j)
93+
94+
if math.IsInf(upperBound, 1) {
95+
upperBound = lowerBound
96+
} else if math.IsInf(lowerBound, -1) {
97+
lowerBound = upperBound
98+
}
99+
100+
count := bucketCounts.At(j)
101+
102+
if count > 0 {
103+
insertionPoint := 0.0
104+
adjustedCount := float64(count)
105+
midpoint := (lowerBound + upperBound) / 2 * scaleToNanos
106+
// Determine insertion point based on bucket position
107+
if j == lowestBucketIndex && j == highestBucketIndex {
108+
// Special case: min and max are in the same bucket
109+
if hasMin && hasMax {
110+
insertionPoint = (minNanoseconds + maxNanoseconds) / 2
111+
}
112+
} else if j == lowestBucketIndex {
113+
// Bottom bucket: insert at min value
114+
if hasMin {
115+
insertionPoint = minNanoseconds
116+
}
117+
} else if j == highestBucketIndex {
118+
// Top bucket: insert at max value
119+
if hasMax {
120+
insertionPoint = maxNanoseconds
121+
}
122+
}
123+
124+
if insertionPoint == 0.0 {
125+
insertionPoint = midpoint
126+
}
127+
128+
err := newSketch.AddWithCount(insertionPoint, adjustedCount)
129+
if err != nil {
130+
return nil, fmt.Errorf("failed to add value to DDSketch: %w", err)
131+
}
132+
}
133+
}
134+
135+
return newSketch, nil
136+
}
137+
138+
func toStoreFromExponentialBucketsWithUnitScale(b pmetric.ExponentialHistogramDataPointBuckets, mapping *mapping.LogarithmicMapping, base float64, scaleToNanos float64) store.Store {
139+
offset := b.Offset()
140+
bucketCounts := b.BucketCounts()
141+
142+
store := store.NewDenseStore()
143+
for j := 0; j < bucketCounts.Len(); j++ {
144+
bucketIndex := j + int(offset)
145+
count := bucketCounts.At(j)
146+
147+
if count > 0 {
148+
// Calculate the actual bucket boundary value
149+
bucketValue := math.Pow(base, float64(bucketIndex))
150+
151+
// Scale the bucket value to nanoseconds
152+
scaledValue := bucketValue * scaleToNanos
153+
154+
// Convert back to the index in the nanosecond space
155+
// Using the same gamma since we're keeping the same precision
156+
scaledIndex := mapping.Index(scaledValue)
157+
store.AddWithCount(scaledIndex, float64(count))
158+
}
159+
}
160+
return store
161+
}
162+
163+
// CreateDDSketchFromExponentialHistogramOfDuration creates a DDSketch from exponential histogram data point
164+
func CreateDDSketchFromExponentialHistogramOfDuration(p pmetric.ExponentialHistogramDataPoint, unit string) (*ddsketch.DDSketch, error) {
165+
// Create the DDSketch stores
166+
scaleToNanos := getTimeUnitScaleToNanos(unit)
167+
168+
// Create the DDSketch mapping that corresponds to the ExponentialHistogram settings
169+
gammaWithOnePercentAccuracy := 1.01 / 0.99
170+
gamma := math.Pow(2, math.Pow(2, float64(-p.Scale())))
171+
gamma = math.Min(gamma, gammaWithOnePercentAccuracy)
172+
indexOffset := math.Log(scaleToNanos)
173+
mapping, err := mapping.NewLogarithmicMappingWithGamma(gamma, indexOffset)
174+
if err != nil {
175+
return nil, fmt.Errorf("couldn't create LogarithmicMapping for DDSketch: %w", err)
176+
}
177+
178+
// Calculate the base for the exponential histogram
179+
base := math.Pow(2, math.Pow(2, float64(-p.Scale())))
180+
positiveStore := toStoreFromExponentialBucketsWithUnitScale(p.Positive(), mapping, base, scaleToNanos)
181+
negativeStore := toStoreFromExponentialBucketsWithUnitScale(p.Negative(), mapping, base, scaleToNanos)
182+
183+
// Create DDSketch with the above mapping and stores
184+
sketch := ddsketch.NewDDSketch(mapping, positiveStore, negativeStore)
185+
err = sketch.AddWithCount(0, float64(p.ZeroCount()))
186+
if err != nil {
187+
return nil, fmt.Errorf("failed to add ZeroCount to DDSketch: %w", err)
188+
}
189+
190+
return sketch, nil
191+
}

0 commit comments

Comments
 (0)