search: Track number of open indexes per storage type. (#108842)

* Track number of open indexes per storage type.

* Fix tests after changing description.

pkg/storage/unified/resource/bleve_index_metrics.go

@@ -13,16 +13,15 @@ type BleveIndexMetrics struct {
 	IndexSize         prometheus.Gauge
 	IndexedKinds      *prometheus.GaugeVec
 	IndexCreationTime *prometheus.HistogramVec
-	IndexTenants      *prometheus.CounterVec
+	OpenIndexes       *prometheus.GaugeVec
 }
 
 var IndexCreationBuckets = []float64{1, 5, 10, 25, 50, 75, 100, 200, 300, 400, 500, 600, 700, 800, 900, 1000}
 
 func ProvideIndexMetrics(reg prometheus.Registerer) *BleveIndexMetrics {
-	return &BleveIndexMetrics{
+	m := &BleveIndexMetrics{
 		IndexLatency: promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
-			Namespace:                       "index_server",
+			Name:                            "index_server_index_latency_seconds",
-			Name:                            "index_latency_seconds",
 			Help:                            "Time (in seconds) until index is updated with new event",
 			Buckets:                         instrument.DefBuckets,
 			NativeHistogramBucketFactor:     1.1, // enable native histograms
@@ -30,28 +29,30 @@ func ProvideIndexMetrics(reg prometheus.Registerer) *BleveIndexMetrics {
 			NativeHistogramMinResetDuration: time.Hour,
 		}, []string{"resource"}),
 		IndexSize: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
-			Namespace: "index_server",
+			Name: "index_server_index_size",
-			Name:      "index_size",
+			Help: "Size of the index in bytes - only for file-based indices",
-			Help:      "Size of the index in bytes - only for file-based indices",
 		}),
 		IndexedKinds: promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{
-			Namespace: "index_server",
+			Name: "index_server_indexed_kinds",
-			Name:      "indexed_kinds",
+			Help: "Number of indexed documents by kind",
-			Help:      "Number of indexed documents by kind",
 		}, []string{"kind"}),
 		IndexCreationTime: promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
-			Namespace:                       "index_server",
+			Name:                            "index_server_index_creation_time_seconds",
-			Name:                            "index_creation_time_seconds",
 			Help:                            "Time (in seconds) it takes until index is created",
 			Buckets:                         IndexCreationBuckets,
 			NativeHistogramBucketFactor:     1.1, // enable native histograms
 			NativeHistogramMaxBucketNumber:  160,
 			NativeHistogramMinResetDuration: time.Hour,
 		}, []string{}),
-		IndexTenants: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+		OpenIndexes: promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{
-			Namespace: "index_server",
+			Name: "index_server_open_indexes",
-			Name:      "index_tenants",
+			Help: "Number of open indexes per storage type. An open index corresponds to single resource group.",
-			Help:      "Number of tenants in the index",
 		}, []string{"index_storage"}), // index_storage is either "file" or "memory"
 	}
+
+	// Initialize labels.
+	m.OpenIndexes.WithLabelValues("file").Set(0)
+	m.OpenIndexes.WithLabelValues("memory").Set(0)
+
+	return m
 }

pkg/storage/unified/search/bleve.go

@@ -40,6 +40,9 @@ import (
 const (
 	// tracingPrexfixBleve is the prefix used for tracing spans in the Bleve backend
 	tracingPrexfixBleve = "unified_search.bleve."
+
+	indexStorageMemory = "memory"
+	indexStorageFile   = "file"
 )
 
 var _ resource.SearchBackend = &bleveBackend{}
@@ -145,6 +148,11 @@ func (b *bleveBackend) getCachedIndex(key resource.NamespacedResource) *bleveInd
 		b.log.Error("failed to close index", "key", key, "err", err)
 	}
 	b.log.Info("index evicted from cache", "key", key)
+
+	if b.indexMetrics != nil {
+		b.indexMetrics.OpenIndexes.WithLabelValues(val.indexStorage).Dec()
+	}
+
 	return nil
 }
 
@@ -209,7 +217,10 @@ func (b *bleveBackend) BuildIndex(
 
 	resourceDir := filepath.Join(b.opts.Root, cleanFileSegment(key.Namespace), cleanFileSegment(fmt.Sprintf("%s.%s", key.Resource, key.Group)))
 
+	newIndexType := indexStorageMemory
 	if size > b.opts.FileThreshold {
+		newIndexType = indexStorageFile
+
 		// We only check for the existing file-based index if we don't already have an open index for this key.
 		// This happens on startup, or when memory-based index has expired. (We don't expire file-based indexes)
 		// If we do have an unexpired cached index already, we always build a new index from scratch.
@@ -246,29 +257,23 @@ func (b *bleveBackend) BuildIndex(
 
 			logWithDetails.Info("Building index using filesystem", "directory", indexDir)
 		}
-
-		if b.indexMetrics != nil {
-			b.indexMetrics.IndexTenants.WithLabelValues("file").Inc()
-		}
 	} else {
 		index, err = bleve.NewMemOnly(mapper)
 		if err != nil {
 			return nil, fmt.Errorf("error creating new in-memory bleve index: %w", err)
 		}
-		if b.indexMetrics != nil {
-			b.indexMetrics.IndexTenants.WithLabelValues("memory").Inc()
-		}
 		logWithDetails.Info("Building index using memory")
 	}
 
 	// Batch all the changes
 	idx := &bleveIndex{
-		key:      key,
+		key:          key,
-		index:    index,
+		index:        index,
-		fields:   fields,
+		indexStorage: newIndexType,
-		standard: resource.StandardSearchFields(),
+		fields:       fields,
-		features: b.features,
+		standard:     resource.StandardSearchFields(),
-		tracing:  b.tracer,
+		features:     b.features,
+		tracing:      b.tracer,
 	}
 
 	idx.allFields, err = getAllFields(idx.standard, fields)
@@ -305,11 +310,18 @@ func (b *bleveBackend) BuildIndex(
 
 	// If there was a previous index in the cache, close it.
 	if prev != nil {
+		if b.indexMetrics != nil {
+			b.indexMetrics.OpenIndexes.WithLabelValues(prev.indexStorage).Dec()
+		}
+
 		err := prev.index.Close()
 		if err != nil {
 			logWithDetails.Error("failed to close previous index", "key", key, "err", err)
 		}
 	}
+	if b.indexMetrics != nil {
+		b.indexMetrics.OpenIndexes.WithLabelValues(idx.indexStorage).Inc()
+	}
 
 	// Start a background task to cleanup the old index directories. If we have built a new file-based index,
 	// the new name is ignored. If we have created in-memory index and fileIndexName is empty, all old directories can be removed.
@@ -466,6 +478,10 @@ func (b *bleveBackend) closeAllIndexes() {
 	for key, idx := range b.cache {
 		_ = idx.index.Close()
 		delete(b.cache, key)
+
+		if b.indexMetrics != nil {
+			b.indexMetrics.OpenIndexes.WithLabelValues(idx.indexStorage).Dec()
+		}
 	}
 }
 
@@ -476,6 +492,8 @@ type bleveIndex struct {
 	standard resource.SearchableDocumentFields
 	fields   resource.SearchableDocumentFields
 
+	indexStorage string // memory or file, used when updating metrics
+
 	// When to expire and close the index. Zero value = no expiration.
 	// We only expire in-memory indexes.
 	expiration time.Time

pkg/storage/unified/search/bleve_test.go

@@ -1,6 +1,7 @@
 package search
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -11,6 +12,8 @@ import (
 	"time"
 
 	"github.com/blevesearch/bleve/v2"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -736,23 +739,26 @@ func Test_isPathWithinRoot(t *testing.T) {
 	}
 }
 
-func setupBleveBackend(t *testing.T, fileThreshold int, cacheTTL time.Duration, dir string) *bleveBackend {
+func setupBleveBackend(t *testing.T, fileThreshold int, cacheTTL time.Duration, dir string) (*bleveBackend, prometheus.Gatherer) {
 	if dir == "" {
 		dir = t.TempDir()
 	}
+	reg := prometheus.NewRegistry()
+	metrics := resource.ProvideIndexMetrics(reg)
+
 	backend, err := NewBleveBackend(BleveOptions{
 		Root:          dir,
 		FileThreshold: int64(fileThreshold),
 		IndexCacheTTL: cacheTTL,
-	}, tracing.NewNoopTracerService(), featuremgmt.WithFeatures(featuremgmt.FlagUnifiedStorageSearchPermissionFiltering), nil)
+	}, tracing.NewNoopTracerService(), featuremgmt.WithFeatures(featuremgmt.FlagUnifiedStorageSearchPermissionFiltering), metrics)
 	require.NoError(t, err)
 	require.NotNil(t, backend)
 	t.Cleanup(backend.closeAllIndexes)
-	return backend
+	return backend, reg
 }
 
 func TestBleveInMemoryIndexExpiration(t *testing.T) {
-	backend := setupBleveBackend(t, 5, time.Nanosecond, "")
+	backend, reg := setupBleveBackend(t, 5, time.Nanosecond, "")
 
 	ns := resource.NamespacedResource{
 		Namespace: "test",
@@ -772,10 +778,18 @@ func TestBleveInMemoryIndexExpiration(t *testing.T) {
 	// Verify that builtIndex is now closed.
 	_, err = builtIndex.DocCount(context.Background(), "")
 	require.ErrorIs(t, err, bleve.ErrorIndexClosed)
+
+	// Verify that there are no open indexes.
+	require.NoError(t, testutil.GatherAndCompare(reg, bytes.NewBufferString(`
+		# HELP index_server_open_indexes Number of open indexes per storage type. An open index corresponds to single resource group.
+		# TYPE index_server_open_indexes gauge
+		index_server_open_indexes{index_storage="memory"} 0
+		index_server_open_indexes{index_storage="file"} 0
+	`), "index_server_open_indexes"))
 }
 
 func TestBleveFileIndexExpiration(t *testing.T) {
-	backend := setupBleveBackend(t, 5, time.Nanosecond, "")
+	backend, reg := setupBleveBackend(t, 5, time.Nanosecond, "")
 
 	ns := resource.NamespacedResource{
 		Namespace: "test",
@@ -797,6 +811,13 @@ func TestBleveFileIndexExpiration(t *testing.T) {
 	cnt, err := builtIndex.DocCount(context.Background(), "")
 	require.NoError(t, err)
 	require.Equal(t, int64(1), cnt)
+
+	require.NoError(t, testutil.GatherAndCompare(reg, bytes.NewBufferString(`
+		# HELP index_server_open_indexes Number of open indexes per storage type. An open index corresponds to single resource group.
+		# TYPE index_server_open_indexes gauge
+		index_server_open_indexes{index_storage="memory"} 0
+		index_server_open_indexes{index_storage="file"} 1
+	`), "index_server_open_indexes"))
 }
 
 func TestFileIndexIsReusedOnSameSizeAndRV(t *testing.T) {
@@ -808,13 +829,30 @@ func TestFileIndexIsReusedOnSameSizeAndRV(t *testing.T) {
 
 	tmpDir := t.TempDir()
 
-	backend1 := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
+	backend1, reg1 := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
 	_, err := backend1.BuildIndex(context.Background(), ns, 10 /* file based */, 100, nil, indexTestDocs(ns, 10))
 	require.NoError(t, err)
+
+	// Verify one open index.
+	require.NoError(t, testutil.GatherAndCompare(reg1, bytes.NewBufferString(`
+		# HELP index_server_open_indexes Number of open indexes per storage type. An open index corresponds to single resource group.
+		# TYPE index_server_open_indexes gauge
+		index_server_open_indexes{index_storage="memory"} 0
+		index_server_open_indexes{index_storage="file"} 1
+	`), "index_server_open_indexes"))
+
 	backend1.closeAllIndexes()
 
+	// Verify that there are no open indexes after closeAllIndexes call.
+	require.NoError(t, testutil.GatherAndCompare(reg1, bytes.NewBufferString(`
+		# HELP index_server_open_indexes Number of open indexes per storage type. An open index corresponds to single resource group.
+		# TYPE index_server_open_indexes gauge
+		index_server_open_indexes{index_storage="memory"} 0
+		index_server_open_indexes{index_storage="file"} 0
+	`), "index_server_open_indexes"))
+
 	// We open new backend using same directory, and run indexing with same size (10) and RV (100). This should reuse existing index, and skip indexing.
-	backend2 := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
+	backend2, reg2 := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
 	idx, err := backend2.BuildIndex(context.Background(), ns, 10 /* file based */, 100, nil, indexTestDocs(ns, 1000))
 	require.NoError(t, err)
 
@@ -822,6 +860,13 @@ func TestFileIndexIsReusedOnSameSizeAndRV(t *testing.T) {
 	cnt, err := idx.DocCount(context.Background(), "")
 	require.NoError(t, err)
 	require.Equal(t, int64(10), cnt)
+
+	require.NoError(t, testutil.GatherAndCompare(reg2, bytes.NewBufferString(`
+		# HELP index_server_open_indexes Number of open indexes per storage type. An open index corresponds to single resource group.
+		# TYPE index_server_open_indexes gauge
+		index_server_open_indexes{index_storage="memory"} 0
+		index_server_open_indexes{index_storage="file"} 1
+	`), "index_server_open_indexes"))
 }
 
 func TestFileIndexIsNotReusedOnDifferentSize(t *testing.T) {
@@ -833,13 +878,13 @@ func TestFileIndexIsNotReusedOnDifferentSize(t *testing.T) {
 
 	tmpDir := t.TempDir()
 
-	backend1 := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
+	backend1, _ := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
 	_, err := backend1.BuildIndex(context.Background(), ns, 10, 100, nil, indexTestDocs(ns, 10))
 	require.NoError(t, err)
 	backend1.closeAllIndexes()
 
 	// We open new backend using same directory, but with different size. Index should be rebuilt.
-	backend2 := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
+	backend2, _ := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
 	idx, err := backend2.BuildIndex(context.Background(), ns, 100, 100, nil, indexTestDocs(ns, 100))
 	require.NoError(t, err)
 
@@ -858,13 +903,13 @@ func TestFileIndexIsNotReusedOnDifferentRV(t *testing.T) {
 
 	tmpDir := t.TempDir()
 
-	backend1 := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
+	backend1, _ := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
 	_, err := backend1.BuildIndex(context.Background(), ns, 10, 100, nil, indexTestDocs(ns, 10))
 	require.NoError(t, err)
 	backend1.closeAllIndexes()
 
 	// We open new backend using same directory, but with different RV. Index should be rebuilt.
-	backend2 := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
+	backend2, _ := setupBleveBackend(t, 5, time.Nanosecond, tmpDir)
 	idx, err := backend2.BuildIndex(context.Background(), ns, 10 /* file based */, 999999, nil, indexTestDocs(ns, 100))
 	require.NoError(t, err)
 
@@ -891,7 +936,7 @@ func TestRebuildingIndexClosesPreviousCachedIndex(t *testing.T) {
 		"file, file":           {false, false},
 	} {
 		t.Run(name, func(t *testing.T) {
-			backend := setupBleveBackend(t, 5, time.Nanosecond, "")
+			backend, reg := setupBleveBackend(t, 5, time.Nanosecond, "")
 
 			firstSize := 100
 			if testCase.firstInMemory {
@@ -900,9 +945,12 @@ func TestRebuildingIndexClosesPreviousCachedIndex(t *testing.T) {
 			firstIndex, err := backend.BuildIndex(context.Background(), ns, int64(firstSize), 100, nil, indexTestDocs(ns, firstSize))
 			require.NoError(t, err)
 
+			openInMemoryIndexes := 0
+
 			secondSize := 100
-			if testCase.firstInMemory {
+			if testCase.secondInMemory {
 				secondSize = 1
+				openInMemoryIndexes = 1
 			}
 			secondIndex, err := backend.BuildIndex(context.Background(), ns, int64(secondSize), 100, nil, indexTestDocs(ns, secondSize))
 			require.NoError(t, err)
@@ -916,6 +964,13 @@ func TestRebuildingIndexClosesPreviousCachedIndex(t *testing.T) {
 			cnt, err := secondIndex.DocCount(context.Background(), "")
 			require.NoError(t, err)
 			require.Equal(t, int64(secondSize), cnt)
+
+			require.NoError(t, testutil.GatherAndCompare(reg, bytes.NewBufferString(fmt.Sprintf(`
+				# HELP index_server_open_indexes Number of open indexes per storage type. An open index corresponds to single resource group.
+				# TYPE index_server_open_indexes gauge
+				index_server_open_indexes{index_storage="memory"} %d
+				index_server_open_indexes{index_storage="file"} %d
+			`, openInMemoryIndexes, 1-openInMemoryIndexes)), "index_server_open_indexes"))
 		})
 	}
 }
@@ -946,7 +1001,7 @@ func indexTestDocs(ns resource.NamespacedResource, docs int) func(index resource
 func TestCleanOldIndexes(t *testing.T) {
 	dir := t.TempDir()
 
-	b := setupBleveBackend(t, 5, time.Nanosecond, dir)
+	b, _ := setupBleveBackend(t, 5, time.Nanosecond, dir)
 
 	t.Run("with skip", func(t *testing.T) {
 		require.NoError(t, os.MkdirAll(filepath.Join(dir, "index-1/a"), 0750))