Add TTL eviction for histogram vectors

prometheus/histogram.go

@@ -511,6 +511,8 @@ type HistogramOpts struct {
 type HistogramVecOpts struct {
 	HistogramOpts
 
+	MetricVecOpts
+
 	// VariableLabels are used to partition the metric vector by the given set
 	// of labels. Each label value will be constrained with the optional Constraint
 	// function, if provided.
@@ -761,20 +763,24 @@ type histogram struct {
 
 	// afterFunc is for testing purposes, by default it's time.AfterFunc.
 	afterFunc func(time.Duration, func()) *time.Timer
+
+	lastActiveUnixNano atomic.Int64
 }
 
 func (h *histogram) Desc() *Desc {
 	return h.desc
 }
 
 func (h *histogram) Observe(v float64) {
+	h.setLastActive(h.now())
 	h.observe(v, h.findBucket(v))
 }
 
 // ObserveWithExemplar should not be called in a high-frequency setting
 // for a native histogram with configured exemplars. For this case,
 // the implementation isn't lock-free and might suffer from lock contention.
 func (h *histogram) ObserveWithExemplar(v float64, e Labels) {
+	h.setLastActive(h.now())
 	i := h.findBucket(v)
 	h.observe(v, i)
 	h.updateExemplar(v, i, e)
@@ -865,6 +871,18 @@ func (h *histogram) Write(out *dto.Metric) error {
 	return nil
 }
 
+func (h *histogram) setLastActive(t time.Time) {
+	h.lastActiveUnixNano.Store(t.UnixNano())
+}
+
+func (h *histogram) isExpired(now time.Time, idleTTL time.Duration) bool {
+	lastActiveUnixNano := h.lastActiveUnixNano.Load()
+	if lastActiveUnixNano == 0 {
+		return false
+	}
+	return now.Sub(time.Unix(0, lastActiveUnixNano)) > idleTTL
+}
+
 // findBucket returns the index of the bucket for the provided value, or
 // len(h.upperBounds) for the +Inf bucket.
 func (h *histogram) findBucket(v float64) int {
@@ -1189,6 +1207,9 @@ func NewHistogramVec(opts HistogramOpts, labelNames []string) *HistogramVec {
 
 // NewHistogramVec creates a new HistogramVec based on the provided HistogramVecOpts.
 func (v2) NewHistogramVec(opts HistogramVecOpts) *HistogramVec {
+	if opts.MetricVecOpts.now == nil {
+		opts.MetricVecOpts.now = opts.HistogramOpts.now
+	}
 	desc := V2.NewDesc(
 		BuildFQName(opts.Namespace, opts.Subsystem, opts.Name),
 		opts.Help,
@@ -1197,9 +1218,9 @@ func (v2) NewHistogramVec(opts HistogramVecOpts) *HistogramVec {
 		WithUnit(opts.Unit),
 	)
 	return &HistogramVec{
-		MetricVec: NewMetricVec(desc, func(lvs ...string) Metric {
+		MetricVec: NewMetricVecWithOpts(desc, func(lvs ...string) Metric {
 			return newHistogram(desc, opts.HistogramOpts, lvs...)
-		}),
+		}, opts.MetricVecOpts),
 	}
 }
 

prometheus/histogram_test.go

@@ -1277,6 +1277,56 @@
 	expectCTsForMetricVecValues(t, histogramVec.MetricVec, dto.MetricType_HISTOGRAM, expected)
 }
 
+func TestHistogramVecTTLPrunesIdleMetricsOnAccessAndCollect(t *testing.T) {
+	now := time.Unix(1, 0)
+	histogramVec := V2.NewHistogramVec(HistogramVecOpts{
+		HistogramOpts: HistogramOpts{
+			Name:    "test",
+			Help:    "test help",
+			Buckets: []float64{1, 2, 3, 4},
+			now:     func() time.Time { return now },
+		},
+		MetricVecOpts: MetricVecOpts{
+			IdleTTL: 50 * time.Millisecond,
+		},
+		VariableLabels: ConstrainableLabels{{Name: "label"}},
+	})
+
+	histogramVec.WithLabelValues("stale").Observe(1)
+	if got := len(histogramVec.metrics); got != 1 {
+		t.Fatalf("expected 1 metric after creation, got %d", got)
+	}
+
+	now = now.Add(100 * time.Millisecond)
+	histogramVec.WithLabelValues("fresh").Observe(2)
+	if got := len(histogramVec.metrics); got != 1 {
+		t.Fatalf("expected idle metric to be pruned on access, got %d metrics", got)
+	}
+	if _, ok := histogramVec.metrics[hashLabelValuesForTest(t, histogramVec, "fresh")]; !ok {
+		t.Fatalf("expected fresh metric to remain after pruning")
+	}
+
+	now = now.Add(100 * time.Millisecond)
+	ch := make(chan Metric, 10)
+	histogramVec.Collect(ch)
+	close(ch)
+	if got := len(histogramVec.metrics); got != 0 {
+		t.Fatalf("expected collect to prune idle metrics, got %d metrics", got)
+	}
+	if got := len(ch); got != 0 {
+		t.Fatalf("expected no active metrics after TTL eviction, got %d", got)
+	}
+}
+
+func hashLabelValuesForTest(t *testing.T, vec *HistogramVec, label string) uint64 {
+	t.Helper()
+	h, err := vec.hashLabelValues([]string{label})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return h
+}
+
 func TestNewConstHistogramWithCreatedTimestamp(t *testing.T) {
 	metricDesc := NewDesc(
 		"sample_value",

prometheus/vec.go

@@ -16,6 +16,7 @@ package prometheus
 import (
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/prometheus/common/model"
 )
@@ -43,13 +44,33 @@ type MetricVec struct {
 	hashAddByte func(h uint64, b byte) uint64
 }
 
+// MetricVecOpts bundles optional behavior that applies to metric vectors.
+type MetricVecOpts struct {
+	// IdleTTL evicts metrics that have not been observed for the configured
+	// duration. A zero value disables eviction.
+	IdleTTL time.Duration
+
+	// now is for testing purposes, by default it's time.Now.
+	now func() time.Time
+}
+
 // NewMetricVec returns an initialized metricVec.
 func NewMetricVec(desc *Desc, newMetric func(lvs ...string) Metric) *MetricVec {
+	return NewMetricVecWithOpts(desc, newMetric, MetricVecOpts{})
+}
+
+// NewMetricVecWithOpts returns an initialized metricVec with optional behavior.
+func NewMetricVecWithOpts(desc *Desc, newMetric func(lvs ...string) Metric, opts MetricVecOpts) *MetricVec {
+	if opts.now == nil {
+		opts.now = time.Now
+	}
 	return &MetricVec{
 		metricMap: &metricMap{
 			metrics:   map[uint64][]metricWithLabelValues{},
 			desc:      desc,
 			newMetric: newMetric,
+			idleTTL:   opts.IdleTTL,
+			now:       opts.now,
 		},
 		hashAdd:     hashAdd,
 		hashAddByte: hashAddByte,
@@ -321,6 +342,8 @@ type metricMap struct {
 	metrics   map[uint64][]metricWithLabelValues
 	desc      *Desc
 	newMetric func(labelValues ...string) Metric
+	idleTTL   time.Duration
+	now       func() time.Time
 }
 
 // Describe implements Collector. It will send exactly one Desc to the provided
@@ -331,6 +354,17 @@ func (m *metricMap) Describe(ch chan<- *Desc) {
 
 // Collect implements Collector.
 func (m *metricMap) Collect(ch chan<- Metric) {
+	if m.idleTTL > 0 {
+		m.mtx.Lock()
+		m.pruneExpiredLocked(m.now())
+		metrics := m.collectMetricsLocked()
+		m.mtx.Unlock()
+		for _, metric := range metrics {
+			ch <- metric
+		}
+		return
+	}
+
 	m.mtx.RLock()
 	defer m.mtx.RUnlock()
 
@@ -491,6 +525,20 @@ func matchPartialLabels(desc *Desc, values []string, labels Labels, curry []curr
 func (m *metricMap) getOrCreateMetricWithLabelValues(
 	hash uint64, lvs []string, curry []curriedLabelValue,
 ) Metric {
+	if m.idleTTL > 0 {
+		m.mtx.Lock()
+		m.pruneExpiredLocked(m.now())
+		metric, ok := m.getMetricWithHashAndLabelValues(hash, lvs, curry)
+		if !ok {
+			inlinedLVs := inlineLabelValues(lvs, curry)
+			metric = m.newMetric(inlinedLVs...)
+			m.markMetricActive(metric, m.now())
+			m.metrics[hash] = append(m.metrics[hash], metricWithLabelValues{values: inlinedLVs, metric: metric})
+		}
+		m.mtx.Unlock()
+		return metric
+	}
+
 	m.mtx.RLock()
 	metric, ok := m.getMetricWithHashAndLabelValues(hash, lvs, curry)
 	m.mtx.RUnlock()
@@ -504,6 +552,7 @@ func (m *metricMap) getOrCreateMetricWithLabelValues(
 	if !ok {
 		inlinedLVs := inlineLabelValues(lvs, curry)
 		metric = m.newMetric(inlinedLVs...)
+		m.markMetricActive(metric, m.now())
 		m.metrics[hash] = append(m.metrics[hash], metricWithLabelValues{values: inlinedLVs, metric: metric})
 	}
 	return metric
@@ -516,6 +565,20 @@ func (m *metricMap) getOrCreateMetricWithLabelValues(
 func (m *metricMap) getOrCreateMetricWithLabels(
 	hash uint64, labels Labels, curry []curriedLabelValue,
 ) Metric {
+	if m.idleTTL > 0 {
+		m.mtx.Lock()
+		m.pruneExpiredLocked(m.now())
+		metric, ok := m.getMetricWithHashAndLabels(hash, labels, curry)
+		if !ok {
+			lvs := extractLabelValues(m.desc, labels, curry)
+			metric = m.newMetric(lvs...)
+			m.markMetricActive(metric, m.now())
+			m.metrics[hash] = append(m.metrics[hash], metricWithLabelValues{values: lvs, metric: metric})
+		}
+		m.mtx.Unlock()
+		return metric
+	}
+
 	m.mtx.RLock()
 	metric, ok := m.getMetricWithHashAndLabels(hash, labels, curry)
 	m.mtx.RUnlock()
@@ -529,11 +592,56 @@ func (m *metricMap) getOrCreateMetricWithLabels(
 	if !ok {
 		lvs := extractLabelValues(m.desc, labels, curry)
 		metric = m.newMetric(lvs...)
+		m.markMetricActive(metric, m.now())
 		m.metrics[hash] = append(m.metrics[hash], metricWithLabelValues{values: lvs, metric: metric})
 	}
 	return metric
 }
 
+func (m *metricMap) collectMetricsLocked() []Metric {
+	metrics := make([]Metric, 0, len(m.metrics))
+	for _, bucket := range m.metrics {
+		for _, metric := range bucket {
+			metrics = append(metrics, metric.metric)
+		}
+	}
+	return metrics
+}
+
+func (m *metricMap) pruneExpiredLocked(now time.Time) {
+	if m.idleTTL <= 0 {
+		return
+	}
+	for h, bucket := range m.metrics {
+		kept := bucket[:0]
+		for _, metric := range bucket {
+			if active, ok := metric.metric.(lastActiveMetric); ok && !active.isExpired(now, m.idleTTL) {
+				kept = append(kept, metric)
+				continue
+			}
+			if _, ok := metric.metric.(lastActiveMetric); !ok {
+				kept = append(kept, metric)
+			}
+		}
+		if len(kept) == 0 {
+			delete(m.metrics, h)
+			continue
+		}
+		m.metrics[h] = kept
+	}
+}
+
+func (m *metricMap) markMetricActive(metric Metric, now time.Time) {
+	if active, ok := metric.(lastActiveMetric); ok {
+		active.setLastActive(now)
+	}
+}
+
+type lastActiveMetric interface {
+	setLastActive(time.Time)
+	isExpired(time.Time, time.Duration) bool
+}
+
 // getMetricWithHashAndLabelValues gets a metric while handling possible
 // collisions in the hash space. Must be called while holding the read mutex.
 func (m *metricMap) getMetricWithHashAndLabelValues(