From 48250a16057b064258f473e26aea348ca63def39 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20=C5=A0tibran=C3=BD?= Date: Wed, 17 Sep 2025 10:59:59 +0200 Subject: [PATCH] Introduced options to ignore old indexes based on build time and build version. (#111168) --- pkg/setting/setting.go | 2 + pkg/setting/setting_unified_storage.go | 3 + pkg/storage/unified/search/bleve.go | 71 ++++++- pkg/storage/unified/search/bleve_test.go | 236 ++++++++++++++--------- pkg/storage/unified/search/options.go | 23 ++- 5 files changed, 234 insertions(+), 101 deletions(-) diff --git a/pkg/setting/setting.go b/pkg/setting/setting.go index 55b5d7da60d..63c7fa212cb 100644 --- a/pkg/setting/setting.go +++ b/pkg/setting/setting.go @@ -579,6 +579,8 @@ type Cfg struct { IndexMinCount int IndexRebuildInterval time.Duration IndexCacheTTL time.Duration + MaxFileIndexAge time.Duration // Max age of file-based indexes. Index older than this will not be reused between restarts. + MinFileIndexBuildVersion string // Minimum version of Grafana that built the file-based index. If index was built with older Grafana, it will not be reused between restarts. EnableSharding bool QOSEnabled bool QOSNumberWorker int diff --git a/pkg/setting/setting_unified_storage.go b/pkg/setting/setting_unified_storage.go index 3227e843bea..921fb2f24a5 100644 --- a/pkg/setting/setting_unified_storage.go +++ b/pkg/setting/setting_unified_storage.go @@ -76,4 +76,7 @@ func (cfg *Cfg) setUnifiedStorageConfig() { cfg.SprinklesApiServerPageLimit = section.Key("sprinkles_api_server_page_limit").MustInt(10000) cfg.CACertPath = section.Key("ca_cert_path").String() cfg.HttpsSkipVerify = section.Key("https_skip_verify").MustBool(false) + + cfg.MaxFileIndexAge = section.Key("max_file_index_age").MustDuration(0) + cfg.MinFileIndexBuildVersion = section.Key("min_file_index_build_version").MustString("") } diff --git a/pkg/storage/unified/search/bleve.go b/pkg/storage/unified/search/bleve.go index 002da763f67..40b2229f751 100644 --- a/pkg/storage/unified/search/bleve.go +++ b/pkg/storage/unified/search/bleve.go @@ -16,6 +16,7 @@ import ( "sync" "time" + "github.com/Masterminds/semver" "github.com/blevesearch/bleve/v2" "github.com/blevesearch/bleve/v2/analysis/analyzer/keyword" "github.com/blevesearch/bleve/v2/analysis/analyzer/standard" @@ -72,6 +73,9 @@ type BleveOptions struct { BuildVersion string + MaxFileIndexAge time.Duration // Maximum age of file-based index that can be reused. Ignored if zero. + MinBuildVersion *semver.Version // Minimum build version for reusing file-based indexes. Ignored if nil. + Logger *slog.Logger } @@ -104,6 +108,14 @@ func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, indexMetrics *resou return nil, fmt.Errorf("bleve root is configured against a file (not folder)") } + if opts.BuildVersion != "" { + // Don't allow storing invalid versions to the index. + _, err := semver.NewVersion(opts.BuildVersion) + if err != nil { + return nil, fmt.Errorf("cannot parse build version %s: %w", opts.BuildVersion, err) + } + } + log := opts.Logger if log == nil { log = slog.Default().With("logger", "bleve-backend") @@ -241,6 +253,24 @@ type IndexBuildInfo struct { BuildVersion string `json:"build_version"` // Grafana version used when building the index } +func (bi IndexBuildInfo) GetBuildTime() time.Time { + if bi.BuildTime == 0 { + return time.Time{} + } + return time.Unix(bi.BuildTime, 0) +} + +func (bi IndexBuildInfo) GetBuildVersion() *semver.Version { + if bi.BuildVersion == "" { + return nil + } + v, err := semver.NewVersion(bi.BuildVersion) + if err != nil { + return nil + } + return v +} + // BuildIndex builds an index from scratch or retrieves it from the filesystem. // If built successfully, the new index replaces the old index in the cache (if there was any). // Existing index in the file system is reused, if it exists, and if size indicates that we should use file-based index, and rebuild is not true. @@ -317,7 +347,11 @@ func (b *bleveBackend) BuildIndex( // 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. if cachedIndex == nil && !rebuild { - index, fileIndexName, indexRV = b.findPreviousFileBasedIndex(resourceDir) + minBuildTime := time.Time{} + if b.opts.MaxFileIndexAge > 0 { + minBuildTime = time.Now().Add(-b.opts.MaxFileIndexAge) + } + index, fileIndexName, indexRV = b.findPreviousFileBasedIndex(resourceDir, minBuildTime, b.opts.MinBuildVersion) } if index != nil { @@ -535,7 +569,7 @@ func formatIndexName(now time.Time) string { return now.Format("20060102-150405") } -func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) (bleve.Index, string, int64) { +func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string, minBuildTime time.Time, minBuildVersion *semver.Version) (bleve.Index, string, int64) { entries, err := os.ReadDir(resourceDir) if err != nil { return nil, "", 0 @@ -557,12 +591,35 @@ func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) (bleve.Ind indexRV, err := getRV(idx) if err != nil { b.log.Error("error getting rv from index", "indexDir", indexDir, "err", err) - if !errors.Is(err, bleve.ErrorIndexClosed) { - _ = idx.Close() - } + _ = idx.Close() continue } + buildInfo, err := getBuildInfo(idx) + if err != nil { + b.log.Error("error getting build info from index", "indexDir", indexDir, "err", err) + _ = idx.Close() + continue + } + + if !minBuildTime.IsZero() { + bt := buildInfo.GetBuildTime() + if bt.IsZero() || bt.Before(minBuildTime) { + b.log.Debug("index build time is before minBuildTime, not reusing the index", "indexDir", indexDir, "indexBuildTime", bt, "minBuildTime", minBuildTime) + _ = idx.Close() + continue + } + } + + if minBuildVersion != nil { + bv := buildInfo.GetBuildVersion() + if bv == nil || bv.Compare(minBuildVersion) < 0 { + b.log.Debug("index build version is before minBuildVersion, not reusing the index", "indexDir", indexDir, "indexBuildVersion", bv, "minBuildVersion", minBuildVersion) + _ = idx.Close() + continue + } + } + return idx, indexName, indexRV } @@ -727,6 +784,10 @@ func getBuildInfo(index bleve.Index) (IndexBuildInfo, error) { return IndexBuildInfo{}, err } + if len(raw) == 0 { + return IndexBuildInfo{}, nil + } + res := IndexBuildInfo{} err = json.Unmarshal(raw, &res) return res, err diff --git a/pkg/storage/unified/search/bleve_test.go b/pkg/storage/unified/search/bleve_test.go index 2aecb6a20d8..34d694e75e9 100644 --- a/pkg/storage/unified/search/bleve_test.go +++ b/pkg/storage/unified/search/bleve_test.go @@ -14,6 +14,7 @@ import ( "testing" "time" + "github.com/Masterminds/semver" "github.com/blevesearch/bleve/v2" authlib "github.com/grafana/authlib/types" "github.com/prometheus/client_golang/prometheus" @@ -756,28 +757,74 @@ func Test_isPathWithinRoot(t *testing.T) { } } -const buildVersion = "1.2.3-456" +const ( + buildVersion = "12.3.45-789" + defaultFileThreshold = 5 + defaultIndexCacheTTL = 1 * time.Minute +) -func setupBleveBackend(t *testing.T, fileThreshold int, cacheTTL time.Duration, dir string) (*bleveBackend, prometheus.Gatherer) { - if dir == "" { - dir = t.TempDir() - } +func setupBleveBackend(t *testing.T, options ...setupOption) (*bleveBackend, prometheus.Gatherer) { reg := prometheus.NewRegistry() metrics := resource.ProvideIndexMetrics(reg) - backend, err := NewBleveBackend(BleveOptions{ - Root: dir, - FileThreshold: int64(fileThreshold), - IndexCacheTTL: cacheTTL, + opts := BleveOptions{ + FileThreshold: defaultFileThreshold, + IndexCacheTTL: defaultIndexCacheTTL, Logger: slog.New(logtest.NewNopHandler(t)), BuildVersion: buildVersion, - }, tracing.NewNoopTracerService(), metrics) + } + for _, opt := range options { + opt(&opts) + } + if opts.Root == "" { + opts.Root = t.TempDir() + } + + backend, err := NewBleveBackend(opts, tracing.NewNoopTracerService(), metrics) require.NoError(t, err) require.NotNil(t, backend) t.Cleanup(backend.CloseAllIndexes) return backend, reg } +type setupOption func(options *BleveOptions) + +func withIndexCacheTTL(ttl time.Duration) setupOption { + return func(options *BleveOptions) { + options.IndexCacheTTL = ttl + } +} + +func withFileThreshold(threshold int) setupOption { + return func(options *BleveOptions) { + options.FileThreshold = int64(threshold) + } +} + +func withRootDir(root string) setupOption { + return func(options *BleveOptions) { + options.Root = root + } +} + +func withBuildVersion(version string) setupOption { + return func(options *BleveOptions) { + options.BuildVersion = version + } +} + +func withMinBuildVersion(version *semver.Version) setupOption { + return func(options *BleveOptions) { + options.MinBuildVersion = version + } +} + +func withMaxFileIndexAge(maxAge time.Duration) setupOption { + return func(options *BleveOptions) { + options.MaxFileIndexAge = maxAge + } +} + func TestBuildIndexExpiration(t *testing.T) { ns := resource.NamespacedResource{ Namespace: "test", @@ -786,7 +833,7 @@ func TestBuildIndexExpiration(t *testing.T) { } t.Run("memory based indexes should expire", func(t *testing.T) { - backend, reg := setupBleveBackend(t, 5, time.Nanosecond, "") + backend, reg := setupBleveBackend(t, withIndexCacheTTL(time.Nanosecond)) builtIndex, err := backend.BuildIndex(context.Background(), ns, 1 /* below FileThreshold */, nil, "test", indexTestDocs(ns, 1, 100), nil, false) require.NoError(t, err) @@ -806,7 +853,7 @@ func TestBuildIndexExpiration(t *testing.T) { }) t.Run("file based indexes should NOT expire", func(t *testing.T) { - backend, reg := setupBleveBackend(t, 5, time.Nanosecond, "") + backend, reg := setupBleveBackend(t, withIndexCacheTTL(time.Nanosecond)) // size=100 is above FileThreshold, this will be file-based index builtIndex, err := backend.BuildIndex(context.Background(), ns, 100, nil, "test", indexTestDocs(ns, 1, 100), nil, false) @@ -840,7 +887,7 @@ func TestCloseAllIndexes(t *testing.T) { } tmpDir := t.TempDir() - backend1, reg := setupBleveBackend(t, 5, time.Nanosecond, tmpDir) + backend1, reg := setupBleveBackend(t, withRootDir(tmpDir)) _, err := backend1.BuildIndex(context.Background(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false) require.NoError(t, err) _, err = backend1.BuildIndex(context.Background(), ns2, 1 /* memory based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false) @@ -861,86 +908,84 @@ func TestBuildIndex(t *testing.T) { Resource: "resource", } - type RV string - const ( - RVLessThan RV = "less" - RVBiggerThan RV = "more" - RVSame RV = "same" - ) + const alwaysRebuildDueToAge = 1 * time.Nanosecond + const neverRebuildDueToAge = 1 * time.Hour + for _, rebuild := range []bool{false, true} { - for _, sameSize := range []bool{false, true} { - for _, documentRV := range []RV{RVLessThan, RVSame, RVBiggerThan} { - shouldRebuild := rebuild - - testName := "" - if shouldRebuild { - testName += "should NOT reuse index " - } else { - testName += "should reuse index " - } - - if sameSize { - testName += "on same size " - } else { - testName += "on different size " - } - - switch documentRV { - case RVLessThan: - testName += "and documentRV < indexRV " - case RVBiggerThan: - testName += "and documentRV > indexRV " - case RVSame: - testName += "and documentRV = indexRV " - } - - if rebuild { - testName += "when rebuild is true " - } else { - testName += "when rebuild is false " - } - - t.Run(testName, func(t *testing.T) { - tmpDir := t.TempDir() - - var size int64 = 10 - var rv int64 = 100 - backend1, _ := createBleveBackendAndIndex(t, tmpDir, ns, size, rv, 10, rebuild) - backend1.CloseAllIndexes() - - if !sameSize { - size = 11 + for _, version := range []string{"", "12.5.123"} { + for _, minBuildVersion := range []*semver.Version{nil, semver.MustParse("12.0.0"), semver.MustParse("13.0.0")} { + for _, maxIndexAge := range []time.Duration{0, alwaysRebuildDueToAge, neverRebuildDueToAge} { + shouldRebuild := rebuild + if minBuildVersion != nil { + shouldRebuild = shouldRebuild || version == "" || minBuildVersion.GreaterThan(semver.MustParse(version)) } - switch documentRV { - case RVBiggerThan: - rv = 101 - case RVLessThan: - rv = 99 - case RVSame: + if maxIndexAge > 0 { + shouldRebuild = shouldRebuild || maxIndexAge == alwaysRebuildDueToAge } - backend2, idx := createBleveBackendAndIndex(t, tmpDir, ns, size, rv, 1000, rebuild) - cnt, err := idx.DocCount(context.Background(), "") - require.NoError(t, err) + testName := "" if shouldRebuild { - require.Equal(t, int64(1000), cnt, "Index has been not rebuilt") + testName += "should REBUILD index" } else { - require.Equal(t, int64(10), cnt, "Index has not been reused") + testName += "should REUSE index" } - backend2.CloseAllIndexes() - }) + + if rebuild { + testName += " when rebuild is true" + } else { + testName += " when rebuild is false" + } + + if version != "" { + testName += " build version is " + version + } else { + testName += " build version is empty" + } + + if minBuildVersion != nil { + testName += " min build version is " + minBuildVersion.String() + } else { + testName += " min build version is nil" + } + + testName += " max index age is " + maxIndexAge.String() + + t.Run(testName, func(t *testing.T) { + tmpDir := t.TempDir() + + const ( + firstIndexDocsCount = 10 + secondIndexDocsCount = 1000 + ) + + { + backend, _ := setupBleveBackend(t, withFileThreshold(5), withRootDir(tmpDir), withBuildVersion(version)) + _, err := backend.BuildIndex(context.Background(), ns, firstIndexDocsCount, nil, "test", indexTestDocs(ns, firstIndexDocsCount, 100), nil, rebuild) + require.NoError(t, err) + backend.CloseAllIndexes() + } + + // Make sure we pass at least 1 nanosecond (alwaysRebuildDueToAge) to ensure that the index needs to be rebuild. + time.Sleep(1 * time.Millisecond) + + newBackend, _ := setupBleveBackend(t, withFileThreshold(5), withRootDir(tmpDir), withBuildVersion(version), withMinBuildVersion(minBuildVersion), withMaxFileIndexAge(maxIndexAge)) + idx, err := newBackend.BuildIndex(context.Background(), ns, secondIndexDocsCount, nil, "test", indexTestDocs(ns, secondIndexDocsCount, 100), nil, rebuild) + require.NoError(t, err) + + cnt, err := idx.DocCount(context.Background(), "") + require.NoError(t, err) + if shouldRebuild { + require.Equal(t, int64(secondIndexDocsCount), cnt, "Index has been not rebuilt") + } else { + require.Equal(t, int64(firstIndexDocsCount), cnt, "Index has not been reused") + } + }) + } } } } } -func createBleveBackendAndIndex(t *testing.T, tmpDir string, ns resource.NamespacedResource, size, rv int64, docCount int, rebuild bool) (*bleveBackend, resource.ResourceIndex) { - backend, _ := setupBleveBackend(t, 5, time.Nanosecond, tmpDir) - idx, err := backend.BuildIndex(context.Background(), ns, size /* file based */, nil, "test", indexTestDocs(ns, docCount, rv), nil, rebuild) - require.NoError(t, err) - return backend, idx -} - func TestRebuildingIndexClosesPreviousCachedIndex(t *testing.T) { ns := resource.NamespacedResource{ Namespace: "test", @@ -958,7 +1003,7 @@ func TestRebuildingIndexClosesPreviousCachedIndex(t *testing.T) { "file, file": {false, false}, } { t.Run(name, func(t *testing.T) { - backend, reg := setupBleveBackend(t, 5, time.Nanosecond, "") + backend, reg := setupBleveBackend(t, withIndexCacheTTL(time.Nanosecond)) firstSize := 100 if testCase.firstInMemory { @@ -1080,7 +1125,7 @@ func updateTestDocs(ns resource.NamespacedResource, docs int) resource.UpdateFn func TestCleanOldIndexes(t *testing.T) { dir := t.TempDir() - b, _ := setupBleveBackend(t, 5, time.Nanosecond, dir) + b, _ := setupBleveBackend(t, withRootDir(dir)) t.Run("with skip", func(t *testing.T) { require.NoError(t, os.MkdirAll(filepath.Join(dir, "index-1/a"), 0750)) @@ -1116,7 +1161,7 @@ func TestBleveIndexWithFailures(t *testing.T) { } func testBleveIndexWithFailures(t *testing.T, fileBased bool) { - backend, _ := setupBleveBackend(t, 5, time.Nanosecond, "") + backend, _ := setupBleveBackend(t) ns := resource.NamespacedResource{ Namespace: "test", @@ -1146,8 +1191,8 @@ func TestIndexUpdate(t *testing.T) { Resource: "resource", } - be, _ := setupBleveBackend(t, 5, 1*time.Minute, "") - idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false) + be, _ := setupBleveBackend(t) + idx, err := be.BuildIndex(t.Context(), ns, defaultFileThreshold*2 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false) require.NoError(t, err) resp := searchTitle(t, idx, "gen", 10, ns) @@ -1177,7 +1222,7 @@ func TestConcurrentIndexUpdateAndBuildIndex(t *testing.T) { Resource: "resource", } - be, _ := setupBleveBackend(t, 5, 1*time.Minute, "") + be, _ := setupBleveBackend(t) updaterFn := func(context context.Context, index resource.ResourceIndex, sinceRV int64) (newRV int64, updatedDocs int, _ error) { var items []*resource.BulkIndexItem @@ -1223,7 +1268,7 @@ func TestConcurrentIndexUpdateSearchAndRebuild(t *testing.T) { Resource: "resource", } - be, _ := setupBleveBackend(t, 5, 1*time.Minute, "") + be, _ := setupBleveBackend(t) _, err := be.BuildIndex(t.Context(), ns, 10, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false) require.NoError(t, err) @@ -1309,7 +1354,7 @@ func TestConcurrentIndexUpdateAndSearch(t *testing.T) { Resource: "resource", } - be, _ := setupBleveBackend(t, 5, 1*time.Minute, "") + be, _ := setupBleveBackend(t) idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false) require.NoError(t, err) @@ -1368,7 +1413,7 @@ func TestIndexUpdateWithErrors(t *testing.T) { Resource: "resource", } - be, _ := setupBleveBackend(t, 5, 1*time.Minute, "") + be, _ := setupBleveBackend(t) updateErr := fmt.Errorf("failed to update index") updaterFn := func(context context.Context, index resource.ResourceIndex, sinceRV int64) (newRV int64, updatedDocs int, _ error) { @@ -1408,7 +1453,7 @@ func TestIndexBuildInfo(t *testing.T) { Resource: "resource", } - be, _ := setupBleveBackend(t, 100, 1*time.Minute, "") + be, _ := setupBleveBackend(t, withFileThreshold(100)) index, err := be.BuildIndex(t.Context(), ns, 10, nil, "test", indexTestDocs(ns, 10, 100), nil, false) require.NoError(t, err) @@ -1419,6 +1464,15 @@ func TestIndexBuildInfo(t *testing.T) { require.InDelta(t, float64(time.Now().Unix()), buildInfo.BuildTime, 30) // allow 30 seconds of drift } +func TestInvalidBuildVersion(t *testing.T) { + opts := BleveOptions{ + Root: t.TempDir(), + BuildVersion: "invalid", + } + _, err := NewBleveBackend(opts, tracing.NewNoopTracerService(), nil) + require.ErrorContains(t, err, "cannot parse build version") +} + func searchTitle(t *testing.T, idx resource.ResourceIndex, query string, limit int, ns resource.NamespacedResource) *resourcepb.ResourceSearchResponse { resp, err := idx.Search(t.Context(), nil, &resourcepb.ResourceSearchRequest{ Options: &resourcepb.ListOptions{ diff --git a/pkg/storage/unified/search/options.go b/pkg/storage/unified/search/options.go index 893debf6bfc..a66a22e83b2 100644 --- a/pkg/storage/unified/search/options.go +++ b/pkg/storage/unified/search/options.go @@ -4,6 +4,7 @@ import ( "os" "path/filepath" + "github.com/Masterminds/semver" "go.opentelemetry.io/otel/trace" "github.com/grafana/grafana/pkg/services/featuremgmt" @@ -24,12 +25,24 @@ func NewSearchOptions(features featuremgmt.FeatureToggles, cfg *setting.Cfg, tra return resource.SearchOptions{}, err } + var minVersion *semver.Version + if cfg.MinFileIndexBuildVersion != "" { + v, err := semver.NewVersion(cfg.MinFileIndexBuildVersion) + if err != nil { + cfg.Logger.Error("Failed to parse min_file_index_build_version, ignoring it.", "version", cfg.MinFileIndexBuildVersion, "err", err) + } else { + minVersion = v + } + } + bleve, err := NewBleveBackend(BleveOptions{ - Root: root, - FileThreshold: int64(cfg.IndexFileThreshold), // fewer than X items will use a memory index - BatchSize: cfg.IndexMaxBatchSize, // This is the batch size for how many objects to add to the index at once - IndexCacheTTL: cfg.IndexCacheTTL, // How long to keep the index cache in memory - BuildVersion: cfg.BuildVersion, + Root: root, + FileThreshold: int64(cfg.IndexFileThreshold), // fewer than X items will use a memory index + BatchSize: cfg.IndexMaxBatchSize, // This is the batch size for how many objects to add to the index at once + IndexCacheTTL: cfg.IndexCacheTTL, // How long to keep the index cache in memory + BuildVersion: cfg.BuildVersion, + MaxFileIndexAge: cfg.MaxFileIndexAge, + MinBuildVersion: minVersion, }, tracer, indexMetrics) if err != nil {