Add opt-in Vec TTL via MetricVec and TTLRegistry

prometheus/counter.go

@@ -15,6 +15,7 @@ package prometheus
 
 import (
 	"errors"
+	"fmt"
 	"math"
 	"sync/atomic"
 	"time"
@@ -70,6 +71,11 @@ type CounterVecOpts struct {
 	// of labels. Each label value will be constrained with the optional Constraint
 	// function, if provided.
 	VariableLabels ConstrainableLabels
+
+	// TTL, if greater than zero, enables per-child expiration for this vector.
+	// See MetricVecOpts.TTL for semantics, orphaned-handle behavior, and
+	// cleanup via Registry.Gather / CleanupExpired.
+	TTL time.Duration
 }
 
 // NewCounter creates a new Counter based on the provided CounterOpts.
@@ -211,15 +217,26 @@ func (v2) NewCounterVec(opts CounterVecOpts) *CounterVec {
 	if opts.now == nil {
 		opts.now = time.Now
 	}
-	return &CounterVec{
-		MetricVec: NewMetricVec(desc, func(lvs ...string) Metric {
-			if len(lvs) != len(desc.variableLabels.names) {
-				panic(makeInconsistentCardinalityError(desc.fqName, desc.variableLabels.names, lvs))
-			}
-			result := &counter{desc: desc, labelPairs: MakeLabelPairs(desc, lvs), now: opts.now}
-			result.init(result) // Init self-collection.
-			result.createdTs = timestamppb.New(opts.now())
+	if opts.TTL < 0 {
+		panic(fmt.Sprintf("invalid negative ttl: %v", opts.TTL))
+	}
+	newMetric := func(lvs ...string) Metric {
+		if len(lvs) != len(desc.variableLabels.names) {
+			panic(makeInconsistentCardinalityError(desc.fqName, desc.variableLabels.names, lvs))
+		}
+		result := &counter{desc: desc, labelPairs: MakeLabelPairs(desc, lvs), now: opts.now}
+		result.init(result) // Init self-collection.
+		result.createdTs = timestamppb.New(opts.now())
+		if opts.TTL <= 0 {
 			return result
+		}
+		return newTTLCounter(result)
+	}
+	return &CounterVec{
+		MetricVec: V2.NewMetricVec(MetricVecOpts{
+			Desc:      desc,
+			NewMetric: newMetric,
+			TTL:       opts.TTL,
 		}),
 	}
 }

prometheus/gauge.go

@@ -14,6 +14,7 @@
 package prometheus
 
 import (
+	"fmt"
 	"math"
 	"sync/atomic"
 	"time"
@@ -65,6 +66,10 @@ type GaugeVecOpts struct {
 	// of labels. Each label value will be constrained with the optional Constraint
 	// function, if provided.
 	VariableLabels ConstrainableLabels
+
+	// TTL, if greater than zero, enables per-child expiration for this vector.
+	// See MetricVecOpts.TTL for semantics.
+	TTL time.Duration
 }
 
 // NewGauge creates a new Gauge based on the provided GaugeOpts.
@@ -166,14 +171,25 @@ func (v2) NewGaugeVec(opts GaugeVecOpts) *GaugeVec {
 		opts.ConstLabels,
 		WithUnit(opts.Unit),
 	)
-	return &GaugeVec{
-		MetricVec: NewMetricVec(desc, func(lvs ...string) Metric {
-			if len(lvs) != len(desc.variableLabels.names) {
-				panic(makeInconsistentCardinalityError(desc.fqName, desc.variableLabels.names, lvs))
-			}
-			result := &gauge{desc: desc, labelPairs: MakeLabelPairs(desc, lvs)}
-			result.init(result) // Init self-collection.
+	if opts.TTL < 0 {
+		panic(fmt.Sprintf("invalid negative ttl: %v", opts.TTL))
+	}
+	newMetric := func(lvs ...string) Metric {
+		if len(lvs) != len(desc.variableLabels.names) {
+			panic(makeInconsistentCardinalityError(desc.fqName, desc.variableLabels.names, lvs))
+		}
+		result := &gauge{desc: desc, labelPairs: MakeLabelPairs(desc, lvs)}
+		result.init(result) // Init self-collection.
+		if opts.TTL <= 0 {
 			return result
+		}
+		return newTTLGauge(result)
+	}
+	return &GaugeVec{
+		MetricVec: V2.NewMetricVec(MetricVecOpts{
+			Desc:      desc,
+			NewMetric: newMetric,
+			TTL:       opts.TTL,
 		}),
 	}
 }

prometheus/histogram.go

@@ -515,6 +515,10 @@ type HistogramVecOpts struct {
 	// of labels. Each label value will be constrained with the optional Constraint
 	// function, if provided.
 	VariableLabels ConstrainableLabels
+
+	// TTL, if greater than zero, enables per-child expiration for this vector.
+	// See MetricVecOpts.TTL for semantics.
+	TTL time.Duration
 }
 
 // NewHistogram creates a new Histogram based on the provided HistogramOpts. It
@@ -970,7 +974,7 @@ func (h *histogram) maybeReset(
 	// We are using the possibly mocked h.now() rather than
 	// time.Since(h.lastResetTime) to enable testing.
 	if h.nativeHistogramMinResetDuration == 0 || // No reset configured.
-		h.resetScheduled || // Do not interfere if a reset is already scheduled.
+		h.resetScheduled || // Do not interefere if a reset is already scheduled.
 		h.now().Sub(h.lastResetTime) < h.nativeHistogramMinResetDuration {
 		return false
 	}
@@ -1057,8 +1061,8 @@ func (h *histogram) maybeWidenZeroBucket(hot, cold *histogramCounts) bool {
 	atomic.StoreUint64(&cold.nativeHistogramZeroThresholdBits, math.Float64bits(newZeroThreshold))
 	// ...and then merge the newly deleted buckets into the wider zero
 	// bucket.
-	mergeAndDeleteOrAddAndReset := func(hotBuckets, coldBuckets *sync.Map) func(k, v any) bool {
-		return func(k, v any) bool {
+	mergeAndDeleteOrAddAndReset := func(hotBuckets, coldBuckets *sync.Map) func(k, v interface{}) bool {
+		return func(k, v interface{}) bool {
 			key := k.(int)
 			bucket := v.(*int64)
 			if key == smallestKey {
@@ -1111,8 +1115,8 @@ func (h *histogram) doubleBucketWidth(hot, cold *histogramCounts) {
 	// ...adjust the schema in the cold counts, too...
 	atomic.StoreInt32(&cold.nativeHistogramSchema, coldSchema)
 	// ...and then merge the cold buckets into the wider hot buckets.
-	merge := func(hotBuckets *sync.Map) func(k, v any) bool {
-		return func(k, v any) bool {
+	merge := func(hotBuckets *sync.Map) func(k, v interface{}) bool {
+		return func(k, v interface{}) bool {
 			key := k.(int)
 			bucket := v.(*int64)
 			// Adjust key to match the bucket to merge into.
@@ -1196,9 +1200,21 @@ func (v2) NewHistogramVec(opts HistogramVecOpts) *HistogramVec {
 		opts.ConstLabels,
 		WithUnit(opts.Unit),
 	)
+	if opts.TTL < 0 {
+		panic(fmt.Sprintf("invalid negative ttl: %v", opts.TTL))
+	}
+	newMetric := func(lvs ...string) Metric {
+		h := newHistogram(desc, opts.HistogramOpts, lvs...)
+		if opts.TTL <= 0 {
+			return h
+		}
+		return newTTLHistogram(h)
+	}
 	return &HistogramVec{
-		MetricVec: NewMetricVec(desc, func(lvs ...string) Metric {
-			return newHistogram(desc, opts.HistogramOpts, lvs...)
+		MetricVec: V2.NewMetricVec(MetricVecOpts{
+			Desc:      desc,
+			NewMetric: newMetric,
+			TTL:       opts.TTL,
 		}),
 	}
 }
@@ -1481,7 +1497,7 @@ func pickSchema(bucketFactor float64) int32 {
 
 func makeBuckets(buckets *sync.Map) ([]*dto.BucketSpan, []int64) {
 	var ii []int
-	buckets.Range(func(k, v any) bool {
+	buckets.Range(func(k, v interface{}) bool {
 		ii = append(ii, k.(int))
 		return true
 	})
@@ -1558,8 +1574,8 @@ func addToBucket(buckets *sync.Map, key int, increment int64) bool {
 // according to the buckets ranged through. It then resets all buckets ranged
 // through to 0 (but leaves them in place so that they don't need to get
 // recreated on the next scrape).
-func addAndReset(hotBuckets *sync.Map, bucketNumber *uint32) func(k, v any) bool {
-	return func(k, v any) bool {
+func addAndReset(hotBuckets *sync.Map, bucketNumber *uint32) func(k, v interface{}) bool {
+	return func(k, v interface{}) bool {
 		bucket := v.(*int64)
 		if addToBucket(hotBuckets, k.(int), atomic.LoadInt64(bucket)) {
 			atomic.AddUint32(bucketNumber, 1)
@@ -1570,15 +1586,15 @@ func addAndReset(hotBuckets *sync.Map, bucketNumber *uint32) func(k, v any) bool
 }
 
 func deleteSyncMap(m *sync.Map) {
-	m.Range(func(k, v any) bool {
+	m.Range(func(k, v interface{}) bool {
 		m.Delete(k)
 		return true
 	})
 }
 
 func findSmallestKey(m *sync.Map) int {
 	result := math.MaxInt32
-	m.Range(func(k, v any) bool {
+	m.Range(func(k, v interface{}) bool {
 		key := k.(int)
 		if key < result {
 			result = key

prometheus/registry.go

@@ -432,6 +432,10 @@ func (r *Registry) MustGather() []*dto.MetricFamily {
 }
 
 // Gather implements Gatherer.
+//
+// Before Collect, Gather calls CleanupExpired on registered collectors that
+// implement ExpiredCleaner and have TTL enabled, so expired Vec children can be
+// reclaimed on scrape without touching non-TTL collectors.
 func (r *Registry) Gather() ([]*dto.MetricFamily, error) {
 	r.mtx.RLock()
 
@@ -476,8 +480,14 @@ func (r *Registry) Gather() ([]*dto.MetricFamily, error) {
 		for {
 			select {
 			case collector := <-checkedCollectors:
+				if cleaner, ok := collector.(ttlEnabledCollector); ok && cleaner.ttlEnabled() {
+					cleaner.CleanupExpired()
+				}
 				safeErrs.Append((safeCollect(collector, checkedMetricChan)))
 			case collector := <-uncheckedCollectors:
+				if cleaner, ok := collector.(ttlEnabledCollector); ok && cleaner.ttlEnabled() {
+					cleaner.CleanupExpired()
+				}
 				safeErrs.Append(safeCollect(collector, uncheckedMetricChan))
 			default:
 				return
@@ -641,12 +651,10 @@ func WriteToTextfile(filename string, g Gatherer) error {
 
 	mfs, err := g.Gather()
 	if err != nil {
-		tmp.Close()
 		return err
 	}
 	for _, mf := range mfs {
 		if _, err := expfmt.MetricFamilyToText(tmp, mf); err != nil {
-			tmp.Close()
 			return err
 		}
 	}

prometheus/summary.go

@@ -164,6 +164,10 @@ type SummaryVecOpts struct {
 	// of labels. Each label value will be constrained with the optional Constraint
 	// function, if provided.
 	VariableLabels ConstrainableLabels
+
+	// TTL, if greater than zero, enables per-child expiration for this vector.
+	// See MetricVecOpts.TTL for semantics.
+	TTL time.Duration
 }
 
 // Problem with the sliding-window decay algorithm... The Merge method of
@@ -577,16 +581,28 @@ func (v2) NewSummaryVec(opts SummaryVecOpts) *SummaryVec {
 			panic(errQuantileLabelNotAllowed)
 		}
 	}
+	if opts.TTL < 0 {
+		panic(fmt.Sprintf("invalid negative ttl: %v", opts.TTL))
+	}
 	desc := V2.NewDesc(
 		BuildFQName(opts.Namespace, opts.Subsystem, opts.Name),
 		opts.Help,
 		opts.VariableLabels,
 		opts.ConstLabels,
 		WithUnit(opts.Unit),
 	)
+	newMetric := func(lvs ...string) Metric {
+		s := newSummary(desc, opts.SummaryOpts, lvs...)
+		if opts.TTL <= 0 {
+			return s
+		}
+		return newTTLSummary(s)
+	}
 	return &SummaryVec{
-		MetricVec: NewMetricVec(desc, func(lvs ...string) Metric {
-			return newSummary(desc, opts.SummaryOpts, lvs...)
+		MetricVec: V2.NewMetricVec(MetricVecOpts{
+			Desc:      desc,
+			NewMetric: newMetric,
+			TTL:       opts.TTL,
 		}),
 	}
 }