add native histogram exemplar support

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

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.2.0
 	github.com/davecgh/go-spew v1.1.1
 	github.com/json-iterator/go v1.1.12
-	github.com/prometheus/client_model v0.5.0
+	github.com/prometheus/client_model v0.6.0
 	github.com/prometheus/common v0.46.0
 	github.com/prometheus/procfs v0.12.0
 	golang.org/x/sys v0.16.0

go.sum

@@ -31,8 +31,8 @@ github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRW
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/prometheus/client_model v0.5.0 h1:VQw1hfvPvk3Uv6Qf29VrPF32JB6rtbgI6cYPYQjL0Qw=
-github.com/prometheus/client_model v0.5.0/go.mod h1:dTiFglRmd66nLR9Pv9f0mZi7B7fk5Pm3gvsjB5tr+kI=
+github.com/prometheus/client_model v0.6.0 h1:k1v3CzpSRUTrKMppY35TLwPvxHqBu0bYgxZzqGIgaos=
+github.com/prometheus/client_model v0.6.0/go.mod h1:NTQHnmxFpouOD0DpvP4XujX3CdOAGQPoaGhyTchlyt8=
 github.com/prometheus/common v0.46.0 h1:doXzt5ybi1HBKpsZOL0sSkaNHJJqkyfEWZGGqqScV0Y=
 github.com/prometheus/common v0.46.0/go.mod h1:Tp0qkxpb9Jsg54QMe+EAmqXkSV7Evdy1BTn+g2pa/hQ=
 github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=

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
@@ -539,6 +541,8 @@ func newHistogram(desc *Desc, opts HistogramOpts, labelValues ...string) Histogr
 		nativeHistogramMaxBuckets:       opts.NativeHistogramMaxBucketNumber,
 		nativeHistogramMaxZeroThreshold: opts.NativeHistogramMaxZeroThreshold,
 		nativeHistogramMinResetDuration: opts.NativeHistogramMinResetDuration,
+		nativeHistogramMaxExemplarCount: opts.NativeHistogramMaxExemplarCount,
+		nativeHistogramExemplarTTL:      opts.NativeHistogramExemplarTTL,
 		lastResetTime:                   opts.now(),
 		now:                             opts.now,
 		afterFunc:                       opts.afterFunc,
@@ -556,6 +560,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 +737,10 @@ type histogram struct {
 
 	// afterFunc is for testing purposes, by default it's time.AfterFunc.
 	afterFunc func(time.Duration, func()) *time.Timer
+
+	nativeExemplars                 *nativeExemplars
+	nativeHistogramMaxExemplarCount uint32
+	nativeHistogramExemplarTTL      time.Duration
 }
 
 func (h *histogram) Desc() *Desc {
@@ -821,6 +830,11 @@ func (h *histogram) Write(out *dto.Metric) error {
 				Length: proto.Uint32(0),
 			}}
 		}
+
+		his.Exemplars = make([]*dto.Exemplar, 0)
+		for e := range h.nativeExemplars.exemplars {
+			his.Exemplars = append(his.Exemplars, e)
+		}
 	}
 	addAndResetCounts(hotCounts, coldCounts)
 	return nil
@@ -1102,6 +1116,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 +1593,152 @@ func addAndResetCounts(hot, cold *histogramCounts) {
 	atomic.AddUint64(&hot.nativeHistogramZeroBucket, atomic.LoadUint64(&cold.nativeHistogramZeroBucket))
 	atomic.StoreUint64(&cold.nativeHistogramZeroBucket, 0)
 }
+
+type nativeExemplars struct {
+	exemplars                       map[*dto.Exemplar]*totalItem
+	nativeHistogramExemplarTTL      time.Duration
+	nativeHistogramMaxExemplarCount uint32
+	timeList                        *timeItem
+	logarithmList                   *logarithmItem
+}
+
+type totalItem struct {
+	timeItem      *timeItem
+	logarithmItem *logarithmItem
+}
+
+func newNativeExemplars(ttl time.Duration, count uint32) *nativeExemplars {
+	return &nativeExemplars{
+		nativeHistogramExemplarTTL:      ttl,
+		nativeHistogramMaxExemplarCount: count,
+		exemplars:                       make(map[*dto.Exemplar]*totalItem),
+		timeList:                        &timeItem{},
+		logarithmList:                   &logarithmItem{},
+	}
+}
+
+func (n *nativeExemplars) addExemplar(e *dto.Exemplar) {
+	var de *dto.Exemplar
+	if len(n.exemplars) >= int(n.nativeHistogramMaxExemplarCount) {
+		de = n.timeList.pickExemplar(e)
+		if time.Since(de.Timestamp.AsTime()) < n.nativeHistogramExemplarTTL {
+			de = n.logarithmList.pickExemplar(e)
+		}
+	}
+
+	if de != nil {
+		n.timeList.removeExemplar(n.exemplars[de].timeItem)
+		n.logarithmList.removeExemplar(n.exemplars[de].logarithmItem)
+
+		delete(n.exemplars, de)
+	}
+
+	t := &totalItem{}
+	t.timeItem = n.timeList.addExemplar(e)
+	t.logarithmItem = n.logarithmList.addExemplar(e)
+	n.exemplars[e] = t
+}
+
+type timeItem struct {
+	e         *dto.Exemplar
+	timestamp time.Time
+	next      *timeItem
+	prev      *timeItem
+}
+
+func (h *timeItem) addExemplar(e *dto.Exemplar) (t *timeItem) {
+	t = &timeItem{
+		e:         e,
+		timestamp: e.GetTimestamp().AsTime(),
+	}
+
+	// if the list is empty
+	i := h.next
+	prev := h
+	for i != nil {
+		if i.timestamp.After(t.timestamp) {
+			break
+		}
+		prev = i
+		i = i.next
+	}
+
+	if i != nil {
+		t.next = i
+		i.prev = t
+	}
+	t.prev = prev
+	prev.next = t
+	return t
+}
+
+func (h *timeItem) pickExemplar(e *dto.Exemplar) (p *dto.Exemplar) {
+	return h.next.e
+}
+
+func (h *timeItem) removeExemplar(d *timeItem) {
+	prev := d.prev
+	next := d.next
+	prev.next = next
+	if next != nil {
+		next.prev = prev
+	}
+}
+
+type logarithmItem struct {
+	e         *dto.Exemplar
+	logarithm float64
+	next      *logarithmItem
+	prev      *logarithmItem
+}
+
+func (h *logarithmItem) addExemplar(e *dto.Exemplar) (l *logarithmItem) {
+	l = &logarithmItem{
+		e:         e,
+		logarithm: math.Log2(e.GetValue()),
+	}
+
+	// if the list is empty
+	i := h.next
+	prev := h
+	for i != nil {
+		if l.logarithm < i.logarithm {
+			break
+		}
+		prev = i
+		i = i.next
+	}
+
+	if i != nil {
+		l.next = i
+		i.prev = l
+	}
+	l.prev = prev
+	prev.next = l
+	return l
+}
+
+func (h *logarithmItem) pickExemplar(e *dto.Exemplar) (p *dto.Exemplar) {
+	var minDiff float64 = -1
+	var logarithm = math.Log(e.GetValue())
+	i := h.next
+	for i != nil {
+		diff := math.Abs(i.logarithm - logarithm)
+		if minDiff < 0 || diff < minDiff {
+			p = i.e
+			minDiff = diff
+		}
+		i = i.next
+	}
+
+	return p
+}
+
+func (h *logarithmItem) removeExemplar(d *logarithmItem) {
+	prev := d.prev
+	next := d.next
+	prev.next = next
+	if next != nil {
+		next.prev = prev
+	}
+}

prometheus/histogram_test.go

@@ -1271,3 +1271,93 @@ 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: {},
+		4: {},
+		5: {},
+	}
+
+	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: {},
+		4: {},
+		5: {},
+	}
+	for e := range histogram.nativeExemplars.exemplars {
+		if _, ok := expectedValues[e.GetValue()]; !ok {
+			t.Errorf("the value is not in expected value")
+		}
+	}
+
+}