diff --git a/pkg/storage/unified/resource/bleve_index_metrics.go b/pkg/storage/unified/resource/bleve_index_metrics.go index 310886666cc..73a18d4fabe 100644 --- a/pkg/storage/unified/resource/bleve_index_metrics.go +++ b/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_latency_seconds", + Name: "index_server_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_size", - Help: "Size of the index in bytes - only for file-based indices", + Name: "index_server_index_size", + Help: "Size of the index in bytes - only for file-based indices", }), IndexedKinds: promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{ - Namespace: "index_server", - Name: "indexed_kinds", - Help: "Number of indexed documents by kind", + Name: "index_server_indexed_kinds", + Help: "Number of indexed documents by kind", }, []string{"kind"}), IndexCreationTime: promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{ - Namespace: "index_server", - Name: "index_creation_time_seconds", + Name: "index_server_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{ - Namespace: "index_server", - Name: "index_tenants", - Help: "Number of tenants in the index", + OpenIndexes: promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{ + Name: "index_server_open_indexes", + Help: "Number of open indexes per storage type. An open index corresponds to single resource group.", }, []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 } diff --git a/pkg/storage/unified/search/bleve.go b/pkg/storage/unified/search/bleve.go index a2415c528b0..6956af234c7 100644 --- a/pkg/storage/unified/search/bleve.go +++ b/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, - index: index, - fields: fields, - standard: resource.StandardSearchFields(), - features: b.features, - tracing: b.tracer, + key: key, + index: index, + indexStorage: newIndexType, + fields: fields, + standard: resource.StandardSearchFields(), + 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 diff --git a/pkg/storage/unified/search/bleve_test.go b/pkg/storage/unified/search/bleve_test.go index b6e2bd05611..b33ef468e32 100644 --- a/pkg/storage/unified/search/bleve_test.go +++ b/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))