public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Introduced options to ignore old indexes based on build time and build version. (#111168)

Browse Source
This commit is contained in:
Peter Štibraný
2025-09-17 10:59:59 +02:00
committed by GitHub
parent 1a035094e5
commit 48250a1605
5 changed files with 234 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/setting/setting.go
+2
View File
@@ -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
pkg/setting/setting_unified_storage.go
+3
View File
@@ -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("")
}
pkg/storage/unified/search/bleve.go
+66 -5
View File
@@ -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
pkg/storage/unified/search/bleve_test.go
+145 -91
View File
@@ -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{
pkg/storage/unified/search/options.go
+18 -5
View File
@@ -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 {