add native histogram exemplar support

Signed-off-by: Ziqi Zhao <zhaoziqi9146@gmail.com>

prometheus/histogram.go

@@ -472,6 +472,8 @@ type HistogramOpts struct {
 	NativeHistogramMaxBucketNumber  uint32
 	NativeHistogramMinResetDuration time.Duration
 	NativeHistogramMaxZeroThreshold float64
+	NativeHistogramMaxExemplarCount uint32
+	NativeHistogramExemplarTTL      time.Duration
 
 	// now is for testing purposes, by default it's time.Now.
 	now func() time.Time
@@ -556,6 +558,7 @@ func newHistogram(desc *Desc, opts HistogramOpts, labelValues ...string) Histogr
 			h.nativeHistogramZeroThreshold = DefNativeHistogramZeroThreshold
 		} // Leave h.nativeHistogramZeroThreshold at 0 otherwise.
 		h.nativeHistogramSchema = pickSchema(opts.NativeHistogramBucketFactor)
+		h.nativeExemplars = newNativeExemplars(opts.NativeHistogramExemplarTTL, opts.NativeHistogramMaxExemplarCount)
 	}
 	for i, upperBound := range h.upperBounds {
 		if i < len(h.upperBounds)-1 {
@@ -732,6 +735,8 @@ type histogram struct {
 
 	// afterFunc is for testing purposes, by default it's time.AfterFunc.
 	afterFunc func(time.Duration, func()) *time.Timer
+
+	nativeExemplars nativeExemplars
 }
 
 func (h *histogram) Desc() *Desc {
@@ -821,6 +826,8 @@ func (h *histogram) Write(out *dto.Metric) error {
 				Length: proto.Uint32(0),
 			}}
 		}
+
+		his.Exemplars = append(his.Exemplars, h.nativeExemplars.exemplars...)
 	}
 	addAndResetCounts(hotCounts, coldCounts)
 	return nil
@@ -1102,6 +1109,10 @@ func (h *histogram) updateExemplar(v float64, bucket int, l Labels) {
 		panic(err)
 	}
 	h.exemplars[bucket].Store(e)
+	doSparse := h.nativeHistogramSchema > math.MinInt32 && !math.IsNaN(v)
+	if doSparse {
+		h.nativeExemplars.addExemplar(e)
+	}
 }
 
 // HistogramVec is a Collector that bundles a set of Histograms that all share the
@@ -1575,3 +1586,58 @@ func addAndResetCounts(hot, cold *histogramCounts) {
 	atomic.AddUint64(&hot.nativeHistogramZeroBucket, atomic.LoadUint64(&cold.nativeHistogramZeroBucket))
 	atomic.StoreUint64(&cold.nativeHistogramZeroBucket, 0)
 }
+
+type nativeExemplars struct {
+	nativeHistogramExemplarTTL      time.Duration
+	nativeHistogramMaxExemplarCount uint32
+
+	exemplars []*dto.Exemplar
+
+	lock sync.Mutex
+}
+
+func newNativeExemplars(ttl time.Duration, count uint32) nativeExemplars {
+	return nativeExemplars{
+		nativeHistogramExemplarTTL:      ttl,
+		nativeHistogramMaxExemplarCount: count,
+		exemplars:                       make([]*dto.Exemplar, 0),
+		lock:                            sync.Mutex{},
+	}
+}
+
+func (n *nativeExemplars) addExemplar(e *dto.Exemplar) {
+	n.lock.Lock()
+	defer n.lock.Unlock()
+
+	elogarithm := math.Log(e.GetValue())
+	if len(n.exemplars) == int(n.nativeHistogramMaxExemplarCount) {
+		// check if oldestIndex is beyond TTL,
+		// if so, find the oldest exemplar, and nearest exemplar
+		oldestTimestamp := time.Now()
+		oldestIndex := -1
+		nearestValue := -1.0
+		nearestIndex := -1
+
+		for i, exemplar := range n.exemplars {
+			if exemplar.Timestamp.AsTime().Before(oldestTimestamp) {
+				oldestTimestamp = exemplar.Timestamp.AsTime()
+				oldestIndex = i
+			}
+			logarithm := math.Log(exemplar.GetValue())
+			if nearestValue == -1 || math.Abs(elogarithm-logarithm) < nearestValue {
+				fmt.Printf("gap: %f", math.Abs(elogarithm-logarithm))
+				nearestValue = math.Abs(elogarithm - logarithm)
+				nearestIndex = i
+			}
+		}
+
+		if oldestIndex != -1 && time.Since(oldestTimestamp) > n.nativeHistogramExemplarTTL {
+			n.exemplars[oldestIndex] = e
+		} else {
+			n.exemplars[nearestIndex] = e
+		}
+		return
+	}
+
+	n.exemplars = append(n.exemplars, e)
+}

prometheus/histogram_test.go

@@ -1271,3 +1271,91 @@ func TestHistogramVecCreatedTimestampWithDeletes(t *testing.T) {
 	now = now.Add(1 * time.Hour)
 	expectCTsForMetricVecValues(t, histogramVec.MetricVec, dto.MetricType_HISTOGRAM, expected)
 }
+
+func TestNativeHistogramExemplar(t *testing.T) {
+	histogram := NewHistogram(HistogramOpts{
+		Name:                            "test",
+		Help:                            "test help",
+		Buckets:                         []float64{1, 2, 3, 4},
+		NativeHistogramBucketFactor:     1.1,
+		NativeHistogramMaxExemplarCount: 3,
+		NativeHistogramExemplarTTL:      10 * time.Second,
+	}).(*histogram)
+
+	// expectedExemplars := []*dto.Exemplar{
+	// 	{
+	// 		Label: []*dto.LabelPair{
+	// 			{Name: proto.String("id"), Value: proto.String("1")},
+	// 		},
+	// 		Value: proto.Float64(1),
+	// 	},
+	// 	{
+	// 		Label: []*dto.LabelPair{
+	// 			{Name: proto.String("id"), Value: proto.String("2")},
+	// 		},
+	// 		Value: proto.Float64(3),
+	// 	},
+	// 	{
+	// 		Label: []*dto.LabelPair{
+	// 			{Name: proto.String("id"), Value: proto.String("3")},
+	// 		},
+	// 		Value: proto.Float64(5),
+	// 	},
+	// }
+
+	histogram.ObserveWithExemplar(1, Labels{"id": "1"})
+	histogram.ObserveWithExemplar(3, Labels{"id": "1"})
+	histogram.ObserveWithExemplar(5, Labels{"id": "1"})
+
+	if len(histogram.nativeExemplars.exemplars) != 3 {
+		t.Errorf("the count of exemplars is not 3")
+	}
+
+	expectedValues := map[float64]struct{}{
+		1: {},
+		3: {},
+		5: {},
+	}
+
+	for _, e := range histogram.nativeExemplars.exemplars {
+		if _, ok := expectedValues[e.GetValue()]; !ok {
+			t.Errorf("the value is not in expected value")
+		}
+	}
+
+	histogram.ObserveWithExemplar(4, Labels{"id": "1"})
+
+	if len(histogram.nativeExemplars.exemplars) != 3 {
+		t.Errorf("the count of exemplars is not 3")
+	}
+
+	expectedValues = map[float64]struct{}{
+		1: {},
+		3: {},
+		4: {},
+	}
+
+	for _, e := range histogram.nativeExemplars.exemplars {
+		if _, ok := expectedValues[e.GetValue()]; !ok {
+			t.Errorf("the value is not in expected value")
+		}
+	}
+
+	time.Sleep(10 * time.Second)
+	histogram.ObserveWithExemplar(6, Labels{"id": "1"})
+
+	if len(histogram.nativeExemplars.exemplars) != 3 {
+		t.Errorf("the count of exemplars is not 3")
+	}
+
+	expectedValues = map[float64]struct{}{
+		6: {},
+		3: {},
+		4: {},
+	}
+	for _, e := range histogram.nativeExemplars.exemplars {
+		if _, ok := expectedValues[e.GetValue()]; !ok {
+			t.Errorf("the value is not in expected value")
+		}
+	}
+}