CBG-4135: new stat for rev cache capacity (#7049)

* CBG-4135: new stat for rev cache capacity

* updated based on review

* fix linter

* lint again

base/stats.go

@@ -85,6 +85,7 @@ const (
 	StatAddedVersion3dot1dot3dot1 = "3.1.3.1"
 	StatAddedVersion3dot1dot4     = "3.1.4"
 	StatAddedVersion3dot2dot0     = "3.2.0"
+	StatAddedVersion3dot2dot1     = "3.2.1"
 	StatAddedVersion3dot3dot0     = "3.3.0"
 
 	StatDeprecatedVersionNotDeprecated = ""
@@ -425,6 +426,8 @@ type CacheStats struct {
 	NumSkippedSeqs *SgwIntStat `json:"num_skipped_seqs"`
 	// The total number of pending sequences. These are out-of-sequence entries waiting to be cached.
 	PendingSeqLen *SgwIntStat `json:"pending_seq_len"`
+	// Total number of items in the rev cache
+	RevisionCacheNumItems *SgwIntStat `json:"revision_cache_num_items"`
 	// The total number of revision cache bypass operations performed.
 	RevisionCacheBypass *SgwIntStat `json:"rev_cache_bypass"`
 	// The total number of revision cache hits.
@@ -1320,6 +1323,10 @@ func (d *DbStats) initCacheStats() error {
 	if err != nil {
 		return err
 	}
+	resUtil.RevisionCacheNumItems, err = NewIntStat(SubsystemCacheKey, "revision_cache_num_items", StatUnitNoUnits, RevCacheNumItemsDesc, StatAddedVersion3dot2dot1, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.GaugeValue, 0)
+	if err != nil {
+		return err
+	}
 	resUtil.RevisionCacheBypass, err = NewIntStat(SubsystemCacheKey, "rev_cache_bypass", StatUnitNoUnits, RevCacheBypassDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.GaugeValue, 0)
 	if err != nil {
 		return err
@@ -1377,6 +1384,7 @@ func (d *DbStats) unregisterCacheStats() {
 	prometheus.Unregister(d.CacheStats.SkippedSeqCap)
 	prometheus.Unregister(d.CacheStats.NumCurrentSeqsSkipped)
 	prometheus.Unregister(d.CacheStats.PendingSeqLen)
+	prometheus.Unregister(d.CacheStats.RevisionCacheNumItems)
 	prometheus.Unregister(d.CacheStats.RevisionCacheBypass)
 	prometheus.Unregister(d.CacheStats.RevisionCacheHits)
 	prometheus.Unregister(d.CacheStats.RevisionCacheMisses)

base/stats_descriptions.go

@@ -119,6 +119,8 @@ const (
 
 	PendingSeqLengthDesc = "The total number of pending sequences. These are out-of-sequence entries waiting to be cached."
 
+	RevCacheNumItemsDesc = "The total number of items in the revision cache."
+
 	RevCacheBypassDesc = "The total number of revision cache bypass operations performed."
 
 	RevCacheHitsDesc = "The total number of revision cache hits. This metric can be used to calculate the ratio of revision cache hits: " +

db/database_test.go

@@ -402,8 +402,8 @@ func TestGetRemovedAsUser(t *testing.T) {
 	// Manually remove the temporary backup doc from the bucket
 	// Manually flush the rev cache
 	// After expiry from the rev cache and removal of doc backup, try again
-	cacheHitCounter, cacheMissCounter := db.DatabaseContext.DbStats.Cache().RevisionCacheHits, db.DatabaseContext.DbStats.Cache().RevisionCacheMisses
-	collection.dbCtx.revisionCache = NewShardedLRURevisionCache(DefaultRevisionCacheShardCount, DefaultRevisionCacheSize, backingStoreMap, cacheHitCounter, cacheMissCounter)
+	cacheHitCounter, cacheMissCounter, cacheNumItems := db.DatabaseContext.DbStats.Cache().RevisionCacheHits, db.DatabaseContext.DbStats.Cache().RevisionCacheMisses, db.DatabaseContext.DbStats.Cache().RevisionCacheNumItems
+	collection.dbCtx.revisionCache = NewShardedLRURevisionCache(DefaultRevisionCacheShardCount, DefaultRevisionCacheSize, backingStoreMap, cacheHitCounter, cacheMissCounter, cacheNumItems)
 	err = collection.PurgeOldRevisionJSON(ctx, "doc1", rev2id)
 	assert.NoError(t, err, "Purge old revision JSON")
 
@@ -754,8 +754,8 @@ func TestGetRemoved(t *testing.T) {
 	// Manually remove the temporary backup doc from the bucket
 	// Manually flush the rev cache
 	// After expiry from the rev cache and removal of doc backup, try again
-	cacheHitCounter, cacheMissCounter := db.DatabaseContext.DbStats.Cache().RevisionCacheHits, db.DatabaseContext.DbStats.Cache().RevisionCacheMisses
-	collection.dbCtx.revisionCache = NewShardedLRURevisionCache(DefaultRevisionCacheShardCount, DefaultRevisionCacheSize, backingStoreMap, cacheHitCounter, cacheMissCounter)
+	cacheHitCounter, cacheMissCounter, cacheNumItems := db.DatabaseContext.DbStats.Cache().RevisionCacheHits, db.DatabaseContext.DbStats.Cache().RevisionCacheMisses, db.DatabaseContext.DbStats.Cache().RevisionCacheNumItems
+	collection.dbCtx.revisionCache = NewShardedLRURevisionCache(DefaultRevisionCacheShardCount, DefaultRevisionCacheSize, backingStoreMap, cacheHitCounter, cacheMissCounter, cacheNumItems)
 	err = collection.PurgeOldRevisionJSON(ctx, "doc1", rev2id)
 	assert.NoError(t, err, "Purge old revision JSON")
 
@@ -823,8 +823,8 @@ func TestGetRemovedAndDeleted(t *testing.T) {
 	// Manually remove the temporary backup doc from the bucket
 	// Manually flush the rev cache
 	// After expiry from the rev cache and removal of doc backup, try again
-	cacheHitCounter, cacheMissCounter := db.DatabaseContext.DbStats.Cache().RevisionCacheHits, db.DatabaseContext.DbStats.Cache().RevisionCacheMisses
-	collection.dbCtx.revisionCache = NewShardedLRURevisionCache(DefaultRevisionCacheShardCount, DefaultRevisionCacheSize, backingStoreMap, cacheHitCounter, cacheMissCounter)
+	cacheHitCounter, cacheMissCounter, cacheNumItems := db.DatabaseContext.DbStats.Cache().RevisionCacheHits, db.DatabaseContext.DbStats.Cache().RevisionCacheMisses, db.DatabaseContext.DbStats.Cache().RevisionCacheNumItems
+	collection.dbCtx.revisionCache = NewShardedLRURevisionCache(DefaultRevisionCacheShardCount, DefaultRevisionCacheSize, backingStoreMap, cacheHitCounter, cacheMissCounter, cacheNumItems)
 	err = collection.PurgeOldRevisionJSON(ctx, "doc1", rev2id)
 	assert.NoError(t, err, "Purge old revision JSON")
 

db/revision_cache_interface.go

@@ -28,12 +28,10 @@ const (
 // RevisionCache is an interface that can be used to fetch a DocumentRevision for a Doc ID and Rev ID pair.
 type RevisionCache interface {
 	// Get returns the given revision, and stores if not already cached.
-	// When includeBody=true, the returned DocumentRevision will include a mutable shallow copy of the marshaled body.
 	// When includeDelta=true, the returned DocumentRevision will include delta - requires additional locking during retrieval.
 	Get(ctx context.Context, docID, revID string, collectionID uint32, includeDelta bool) (DocumentRevision, error)
 
 	// GetActive returns the current revision for the given doc ID, and stores if not already cached.
-	// When includeBody=true, the returned DocumentRevision will include a mutable shallow copy of the marshaled body.
 	GetActive(ctx context.Context, docID string, collectionID uint32) (docRev DocumentRevision, err error)
 
 	// Peek returns the given revision if present in the cache
@@ -77,12 +75,13 @@ func NewRevisionCache(cacheOptions *RevisionCacheOptions, backingStores map[uint
 
 	cacheHitStat := cacheStats.RevisionCacheHits
 	cacheMissStat := cacheStats.RevisionCacheMisses
+	cacheNumItemsStat := cacheStats.RevisionCacheNumItems
 
 	if cacheOptions.ShardCount > 1 {
-		return NewShardedLRURevisionCache(cacheOptions.ShardCount, cacheOptions.Size, backingStores, cacheHitStat, cacheMissStat)
+		return NewShardedLRURevisionCache(cacheOptions.ShardCount, cacheOptions.Size, backingStores, cacheHitStat, cacheMissStat, cacheNumItemsStat)
 	}
 
-	return NewLRURevisionCache(cacheOptions.Size, backingStores, cacheHitStat, cacheMissStat)
+	return NewLRURevisionCache(cacheOptions.Size, backingStores, cacheHitStat, cacheMissStat, cacheNumItemsStat)
 }
 
 type RevisionCacheOptions struct {

db/revision_cache_lru.go

@@ -26,13 +26,13 @@ type ShardedLRURevisionCache struct {
 }
 
 // Creates a sharded revision cache with the given capacity and an optional loader function.
-func NewShardedLRURevisionCache(shardCount uint16, capacity uint32, backingStores map[uint32]RevisionCacheBackingStore, cacheHitStat, cacheMissStat *base.SgwIntStat) *ShardedLRURevisionCache {
+func NewShardedLRURevisionCache(shardCount uint16, capacity uint32, backingStores map[uint32]RevisionCacheBackingStore, cacheHitStat, cacheMissStat, cacheNumItemsStat *base.SgwIntStat) *ShardedLRURevisionCache {
 
 	caches := make([]*LRURevisionCache, shardCount)
 	// Add 10% to per-shared cache capacity to ensure overall capacity is reached under non-ideal shard hashing
 	perCacheCapacity := 1.1 * float32(capacity) / float32(shardCount)
 	for i := 0; i < int(shardCount); i++ {
-		caches[i] = NewLRURevisionCache(uint32(perCacheCapacity+0.5), backingStores, cacheHitStat, cacheMissStat)
+		caches[i] = NewLRURevisionCache(uint32(perCacheCapacity+0.5), backingStores, cacheHitStat, cacheMissStat, cacheNumItemsStat)
 	}
 
 	return &ShardedLRURevisionCache{
@@ -80,6 +80,7 @@ type LRURevisionCache struct {
 	lruList       *list.List
 	cacheHits     *base.SgwIntStat
 	cacheMisses   *base.SgwIntStat
+	cacheNumItems *base.SgwIntStat
 	lock          sync.Mutex
 	capacity      uint32
 }
@@ -100,7 +101,7 @@ type revCacheValue struct {
 }
 
 // Creates a revision cache with the given capacity and an optional loader function.
-func NewLRURevisionCache(capacity uint32, backingStores map[uint32]RevisionCacheBackingStore, cacheHitStat, cacheMissStat *base.SgwIntStat) *LRURevisionCache {
+func NewLRURevisionCache(capacity uint32, backingStores map[uint32]RevisionCacheBackingStore, cacheHitStat, cacheMissStat, cacheNumItemsStat *base.SgwIntStat) *LRURevisionCache {
 
 	return &LRURevisionCache{
 		cache:         map[IDAndRev]*list.Element{},
@@ -109,6 +110,7 @@ func NewLRURevisionCache(capacity uint32, backingStores map[uint32]RevisionCache
 		backingStores: backingStores,
 		cacheHits:     cacheHitStat,
 		cacheMisses:   cacheMissStat,
+		cacheNumItems: cacheNumItemsStat,
 	}
 }
 
@@ -237,18 +239,29 @@ func (rc *LRURevisionCache) Upsert(ctx context.Context, docRev DocumentRevision,
 	key := IDAndRev{DocID: docRev.DocID, RevID: docRev.RevID, CollectionID: collectionID}
 
 	rc.lock.Lock()
+	newItem := true
 	// If element exists remove from lrulist
 	if elem := rc.cache[key]; elem != nil {
 		rc.lruList.Remove(elem)
+		newItem = false
 	}
 
 	// Add new value and overwrite existing cache key, pushing to front to maintain order
 	value := &revCacheValue{key: key}
 	rc.cache[key] = rc.lruList.PushFront(value)
+	// only increment if we are inserting new item to cache
+	if newItem {
+		rc.cacheNumItems.Add(1)
+	}
 
 	// Purge oldest item if required
+	var numItemsRemoved int
 	for len(rc.cache) > int(rc.capacity) {
 		rc.purgeOldest_()
+		numItemsRemoved++
+	}
+	if numItemsRemoved > 0 {
+		rc.cacheNumItems.Add(int64(-numItemsRemoved))
 	}
 	rc.lock.Unlock()
 
@@ -268,8 +281,15 @@ func (rc *LRURevisionCache) getValue(docID, revID string, collectionID uint32, c
 	} else if create {
 		value = &revCacheValue{key: key}
 		rc.cache[key] = rc.lruList.PushFront(value)
+		rc.cacheNumItems.Add(1)
+
+		var numItemsRemoved int
 		for len(rc.cache) > int(rc.capacity) {
 			rc.purgeOldest_()
+			numItemsRemoved++
+		}
+		if numItemsRemoved > 0 {
+			rc.cacheNumItems.Add(int64(-numItemsRemoved))
 		}
 	}
 	rc.lock.Unlock()
@@ -287,6 +307,7 @@ func (rc *LRURevisionCache) Remove(docID, revID string, collectionID uint32) {
 	}
 	rc.lruList.Remove(element)
 	delete(rc.cache, key)
+	rc.cacheNumItems.Add(-1)
 }
 
 // removeValue removes a value from the revision cache, if present and the value matches the the value. If there's an item in the revision cache with a matching docID and revID but the document is different, this item will not be removed from the rev cache.
@@ -295,6 +316,7 @@ func (rc *LRURevisionCache) removeValue(value *revCacheValue) {
 	if element := rc.cache[value.key]; element != nil && element.Value == value {
 		rc.lruList.Remove(element)
 		delete(rc.cache, value.key)
+		rc.cacheNumItems.Add(-1)
 	}
 	rc.lock.Unlock()
 }