Expose number of entries in refcache as a metric.

pkg/ingester/ingester_v2.go

@@ -140,6 +140,7 @@ type TSDBState struct {
 	appenderAddDuration    prometheus.Histogram
 	appenderCommitDuration prometheus.Histogram
 	refCachePurgeDuration  prometheus.Histogram
+	refCacheEntriesPerUser *prometheus.GaugeVec
 }
 
 func newTSDBState(bucketClient objstore.Bucket, registerer prometheus.Registerer) TSDBState {
@@ -179,6 +180,10 @@ func newTSDBState(bucketClient objstore.Bucket, registerer prometheus.Registerer
 			Help:    "The total time it takes to purge the TSDB series reference cache for a single tenant.",
 			Buckets: prometheus.DefBuckets,
 		}),
+		refCacheEntriesPerUser: promauto.With(registerer).NewGaugeVec(prometheus.GaugeOpts{
+			Name: "cortex_ingester_tsdb_refcache_user_entries",
+			Help: "Number of entries per user in RefCache -- this roughly corresponds to active series per user.",
+		}, []string{"user"}),
 	}
 }
 
@@ -353,8 +358,9 @@ func (i *Ingester) updateLoop(ctx context.Context) error {
 				}
 
 				startTime := time.Now()
-				userDB.refCache.Purge(startTime.Add(-cortex_tsdb.DefaultRefCacheTTL))
+				entries := userDB.refCache.Purge(startTime.Add(-cortex_tsdb.DefaultRefCacheTTL))
 				i.TSDBState.refCachePurgeDuration.Observe(time.Since(startTime).Seconds())
+				i.TSDBState.refCacheEntriesPerUser.WithLabelValues(userID).Set(float64(entries))
 			}
 		case <-ctx.Done():
 			return nil

pkg/storage/tsdb/ref_cache.go

@@ -74,11 +74,14 @@ func (c *RefCache) SetRef(now time.Time, series labels.Labels, ref uint64) {
 }
 
 // Purge removes expired entries from the cache. This function should be called
-// periodically to avoid memory leaks.
-func (c *RefCache) Purge(keepUntil time.Time) {
+// periodically to avoid memory leaks. Purge returns number of entries
+// after purging.
+func (c *RefCache) Purge(keepUntil time.Time) int {
+	entries := 0
 	for s := 0; s < numRefCacheStripes; s++ {
-		c.stripes[s].purge(keepUntil)
+		entries += c.stripes[s].purge(keepUntil)
 	}
+	return entries
 }
 
 func (s *refCacheStripe) ref(now time.Time, series labels.Labels, fp model.Fingerprint) (uint64, bool) {
@@ -123,18 +126,22 @@ func (s *refCacheStripe) setRef(now time.Time, series labels.Labels, fp model.Fi
 	s.refs[fp] = append(s.refs[fp], refCacheEntry)
 }
 
-func (s *refCacheStripe) purge(keepUntil time.Time) {
+// Purge returns number of entries after purging.
+func (s *refCacheStripe) purge(keepUntil time.Time) int {
 	s.refsMu.Lock()
 	defer s.refsMu.Unlock()
 
 	keepUntilNanos := keepUntil.UnixNano()
 
+	result := 0
 	for fp, entries := range s.refs {
 		// Since we do expect very few fingerprint collisions, we
 		// have an optimized implementation for the common case.
 		if len(entries) == 1 {
 			if entries[0].touchedAt.Load() < keepUntilNanos {
 				delete(s.refs, fp)
+			} else {
+				result++
 			}
 
 			continue
@@ -154,7 +161,9 @@ func (s *refCacheStripe) purge(keepUntil time.Time) {
 		if len(entries) == 0 {
 			delete(s.refs, fp)
 		} else {
+			result += len(entries)
 			s.refs[fp] = entries
 		}
 	}
+	return result
 }

pkg/storage/tsdb/ref_cache_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestRefCache_GetAndSetReferences(t *testing.T) {
@@ -82,7 +83,13 @@ func TestRefCache_Purge(t *testing.T) {
 			c.SetRef(now.Add(time.Duration(-i)*time.Second), series[i], uint64(i))
 		}
 
-		c.Purge(now.Add(time.Duration(-ttl) * time.Second))
+		entries := c.Purge(now.Add(time.Duration(-ttl) * time.Second))
+		expEntries := ttl + 1
+		if expEntries > len(series) {
+			expEntries = len(series)
+		}
+
+		require.Equal(t, entries, expEntries)
 
 		// Check retained and purged entries
 		for i := 0; i <= ttl && i < len(series); i++ {