feat: Adds statistics to debug/vars for tag value cache size

tsdb/index.go

@@ -76,6 +76,10 @@ type Index interface {
 	// Size of the index on disk, if applicable.
 	DiskSizeBytes() int64
 
+	// TagValueCacheBytes is the size of tag value cache for TSI indexes.
+	// This is only to be used with TSI.
+	TagValueCacheBytes() int64
+
 	// Bytes estimates the memory footprint of this Index, in bytes.
 	Bytes() int
 

tsdb/index/inmem/inmem.go

@@ -1040,6 +1040,10 @@ func (i *Index) SeriesIDIterator(opt query.IteratorOptions) (tsdb.SeriesIDIterat
 // DiskSizeBytes always returns zero bytes, since this is an in-memory index.
 func (i *Index) DiskSizeBytes() int64 { return 0 }
 
+// TagValueCacheBytes always returns zero bytes, since the in-memory index
+// does not use a tag value series ID cache.
+func (i *Index) TagValueCacheBytes() int64 { return 0 }
+
 // Rebuild recreates the measurement indexes to allow deleted series to be removed
 // and garbage collected.
 func (i *Index) Rebuild() {

tsdb/index/tsi1/cache.go

@@ -3,6 +3,7 @@ package tsi1
 import (
 	"container/list"
 	"sync"
+	"unsafe"
 
 	"github.com/influxdata/influxdb/tsdb"
 )
@@ -191,6 +192,29 @@ func (c *TagValueSeriesIDCache) checkEviction() {
 	}
 }
 
+// HeapSize estimates the total heap memory usage of the cache in bytes.
+func (c *TagValueSeriesIDCache) HeapSize() int {
+	c.RLock()
+	defer c.RUnlock()
+
+	size := int(unsafe.Sizeof(*c))
+	for name, mmap := range c.cache {
+		size += len(name) + int(unsafe.Sizeof(mmap))
+		for key, tkmap := range mmap {
+			size += len(key) + int(unsafe.Sizeof(tkmap))
+			for value, ele := range tkmap {
+				size += len(value) + int(unsafe.Sizeof(ele))
+				elem := ele.Value.(*seriesIDCacheElement)
+				size += len(elem.name) + len(elem.key) + len(elem.value)
+				if elem.SeriesIDSet != nil {
+					size += elem.SeriesIDSet.Bytes()
+				}
+			}
+		}
+	}
+	return size
+}
+
 // seriesIDCacheElement is an item stored within a cache.
 type seriesIDCacheElement struct {
 	name        string

tsdb/index/tsi1/index.go

@@ -159,6 +159,13 @@ type Index struct {
 
 	// Number of partitions used by the index.
 	PartitionN uint64
+
+	// Number of bytes cache is currently using.
+	// Updated periodically by CollectTagValueCacheMetrics goroutine.
+	cacheBytes int64
+
+	// Closed when the index is closing, to signal background goroutines to stop.
+	closing chan struct{}
 }
 
 func (i *Index) UniqueReferenceID() uintptr {
@@ -180,6 +187,7 @@ func NewIndex(sfile *tsdb.SeriesFile, database string, options ...IndexOption) *
 		sSketch:           hll.NewDefaultPlus(),
 		sTSketch:          hll.NewDefaultPlus(),
 		PartitionN:        DefaultPartitionN,
+		closing:           make(chan struct{}),
 	}
 
 	for _, option := range options {
@@ -260,6 +268,9 @@ func (i *Index) Open() (rErr error) {
 		return errors.New("index already open")
 	}
 
+	// Re-initialize closing channel for reopen support.
+	i.closing = make(chan struct{})
+
 	// Ensure root exists.
 	if err := os.MkdirAll(i.path, 0777); err != nil {
 		return err
@@ -303,6 +314,10 @@ func (i *Index) Open() (rErr error) {
 	// Mark opened.
 	i.opened = true
 	i.logger.Info(fmt.Sprintf("index opened with %d partitions", partitionN))
+
+	// Start background goroutine to periodically collect cache metrics.
+	i.collectTagValueCacheMetrics()
+
 	return nil
 }
 
@@ -350,6 +365,14 @@ func (i *Index) Close() error {
 
 // close closes the index without locking
 func (i *Index) close() (rErr error) {
+	// Signal background goroutines to stop.
+	select {
+	case <-i.closing:
+		// Already closed.
+	default:
+		close(i.closing)
+	}
+
 	for _, p := range i.partitions {
 		if (p != nil) && p.IsOpen() {
 			if pErr := p.Close(); pErr != nil {
@@ -1062,6 +1085,30 @@ func (i *Index) TagKeySeriesIDIterator(name, key []byte) (tsdb.SeriesIDIterator,
 	return tsdb.MergeSeriesIDIterators(a...), nil
 }
 
+// TagValueCacheBytes returns the most recently sampled heap size of the
+// tag value series ID cache, in bytes.
+func (i *Index) TagValueCacheBytes() int64 {
+	return atomic.LoadInt64(&i.cacheBytes)
+}
+
+// collectTagValueCacheMetrics starts a background goroutine that periodically
+// samples the tag value cache heap size. It exits when the index is closed.
+func (i *Index) collectTagValueCacheMetrics() {
+	const cacheTrigger = 10 * time.Minute
+	go func() {
+		ticker := time.NewTicker(cacheTrigger)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-i.closing:
+				return
+			case <-ticker.C:
+				atomic.StoreInt64(&i.cacheBytes, int64(i.tagValueCache.HeapSize()))
+			}
+		}
+	}()
+}
+
 // TagValueSeriesIDIterator returns a series iterator for a single tag value.
 func (i *Index) TagValueSeriesIDIterator(name, key, value []byte) (tsdb.SeriesIDIterator, error) {
 	// Check series ID set cache...

tsdb/store.go

@@ -45,11 +45,12 @@ var (
 
 // Statistics gathered by the store.
 const (
-	statDatabaseSeries       = "numSeries"       // number of series in a database
-	statDatabaseMeasurements = "numMeasurements" // number of measurements in a database
-	statPointsWritten        = "pointsWritten"   // number of points parsed by engines successfully
-	statValuesWritten        = "valuesWritten"   // number of values parsed by engines successfully
-	statSeriesCreated        = "seriesCreated"   // number of series created since startup
+	statDatabaseSeries       = "numSeries"          // number of series in a database
+	statDatabaseMeasurements = "numMeasurements"    // number of measurements in a database
+	statTagValueCacheBytes   = "tagValueCacheBytes" // bytes used by the tag value series ID cache
+	statPointsWritten        = "pointsWritten"      // number of points parsed by engines successfully
+	statValuesWritten        = "valuesWritten"      // number of values parsed by engines successfully
+	statSeriesCreated        = "seriesCreated"      // number of series created since startup
 )
 
 // SeriesFileDirectory is the name of the directory containing series files for
@@ -217,12 +218,28 @@ func (s *Store) Statistics(tags map[string]string) []models.Statistic {
 			continue
 		}
 
+		// Collect tag value cache bytes from unique indexes for this database.
+		var tagValueCacheBytes int64
+		s.mu.RLock()
+		uniqueIndexes := make(map[uintptr]Index)
+		for _, sh := range s.shards {
+			if sh.database == database {
+				idx := sh.index
+				uniqueIndexes[idx.UniqueReferenceID()] = idx
+			}
+		}
+		s.mu.RUnlock()
+		for _, idx := range uniqueIndexes {
+			tagValueCacheBytes += idx.TagValueCacheBytes()
+		}
+
 		statistics = append(statistics, models.Statistic{
 			Name: "database",
 			Tags: models.StatisticTags{"database": database}.Merge(tags),
 			Values: map[string]interface{}{
 				statDatabaseSeries:       sc,
 				statDatabaseMeasurements: mc,
+				statTagValueCacheBytes:   tagValueCacheBytes,
 			},
 		})
 	}