From 818aebb06a6dab841076b84f1d081e15364e56c5 Mon Sep 17 00:00:00 2001 From: Zubi-fix Date: Sun, 12 Jul 2026 12:29:55 +0100 Subject: [PATCH] Add TTL eviction for histogram vectors Signed-off-by: Zubi-fix --- prometheus/histogram.go | 25 +++++++- prometheus/histogram_test.go | 50 ++++++++++++++++ prometheus/vec.go | 108 +++++++++++++++++++++++++++++++++++ 3 files changed, 181 insertions(+), 2 deletions(-) diff --git a/prometheus/histogram.go b/prometheus/histogram.go index 88bae3b32..5f3023ba9 100644 --- a/prometheus/histogram.go +++ b/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,6 +763,8 @@ 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 { @@ -768,6 +772,7 @@ func (h *histogram) Desc() *Desc { } func (h *histogram) Observe(v float64) { + h.setLastActive(h.now()) h.observe(v, h.findBucket(v)) } @@ -775,6 +780,7 @@ func (h *histogram) Observe(v float64) { // 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), } } diff --git a/prometheus/histogram_test.go b/prometheus/histogram_test.go index 402df8ae3..91fcdb29d 100644 --- a/prometheus/histogram_test.go +++ b/prometheus/histogram_test.go @@ -1277,6 +1277,56 @@ func TestHistogramVecCreatedTimestampWithDeletes(t *testing.T) { 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", diff --git a/prometheus/vec.go b/prometheus/vec.go index 121d2a963..aad601962 100644 --- a/prometheus/vec.go +++ b/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(