From b48337a7c83a28aafb40492eda6274103b271689 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mustafa=20Sencer=20=C3=96zcan?= <32759850+mustafasencer@users.noreply.github.com> Date: Tue, 17 Jun 2025 16:28:51 +0200 Subject: [PATCH] feat: introduce ttl cache for bleve indices (#106842) --- pkg/setting/setting.go | 1 + pkg/setting/setting_unified_storage.go | 1 + pkg/storage/unified/resource/search.go | 4 +++ pkg/storage/unified/search/bleve.go | 48 ++++++++++++++++---------- pkg/storage/unified/search/options.go | 1 + 5 files changed, 37 insertions(+), 18 deletions(-) diff --git a/pkg/setting/setting.go b/pkg/setting/setting.go index 6e4baaefe1b..a45780dfdab 100644 --- a/pkg/setting/setting.go +++ b/pkg/setting/setting.go @@ -561,6 +561,7 @@ type Cfg struct { IndexFileThreshold int IndexMinCount int IndexRebuildInterval time.Duration + IndexCacheTTL time.Duration EnableSharding bool MemberlistBindAddr string MemberlistAdvertiseAddr string diff --git a/pkg/setting/setting_unified_storage.go b/pkg/setting/setting_unified_storage.go index 8f07a5cc1eb..03768a73777 100644 --- a/pkg/setting/setting_unified_storage.go +++ b/pkg/setting/setting_unified_storage.go @@ -66,6 +66,7 @@ func (cfg *Cfg) setUnifiedStorageConfig() { cfg.IndexMinCount = section.Key("index_min_count").MustInt(1) // default to 24 hours because usage insights summarizes the data every 24 hours cfg.IndexRebuildInterval = section.Key("index_rebuild_interval").MustDuration(24 * time.Hour) + cfg.IndexCacheTTL = section.Key("index_cache_ttl").MustDuration(10 * time.Minute) cfg.SprinklesApiServer = section.Key("sprinkles_api_server").String() cfg.SprinklesApiServerPageLimit = section.Key("sprinkles_api_server_page_limit").MustInt(100) cfg.CACertPath = section.Key("ca_cert_path").String() diff --git a/pkg/storage/unified/resource/search.go b/pkg/storage/unified/resource/search.go index b6ecfe0bd72..94837f1c180 100644 --- a/pkg/storage/unified/resource/search.go +++ b/pkg/storage/unified/resource/search.go @@ -36,6 +36,10 @@ func (s *NamespacedResource) Valid() bool { return s.Namespace != "" && s.Group != "" && s.Resource != "" } +func (s *NamespacedResource) String() string { + return fmt.Sprintf("%s/%s/%s", s.Namespace, s.Group, s.Resource) +} + type IndexAction int const ( diff --git a/pkg/storage/unified/search/bleve.go b/pkg/storage/unified/search/bleve.go index d44cd0ecc78..a73b0249fce 100644 --- a/pkg/storage/unified/search/bleve.go +++ b/pkg/storage/unified/search/bleve.go @@ -11,7 +11,6 @@ import ( "slices" "strconv" "strings" - "sync" "time" "github.com/blevesearch/bleve/v2" @@ -32,11 +31,17 @@ import ( authlib "github.com/grafana/authlib/types" "github.com/grafana/grafana/pkg/apimachinery/utils" + "github.com/grafana/grafana/pkg/infra/localcache" "github.com/grafana/grafana/pkg/infra/log" "github.com/grafana/grafana/pkg/storage/unified/resource" ) -const tracingPrexfixBleve = "unified_search.bleve." +const ( + // tracingPrexfixBleve is the prefix used for tracing spans in the Bleve backend + tracingPrexfixBleve = "unified_search.bleve." + // Default index cache cleanup TTL is 1 minute + indexCacheCleanupInterval = time.Minute +) var _ resource.SearchBackend = &bleveBackend{} var _ resource.ResourceIndex = &bleveIndex{} @@ -51,6 +56,9 @@ type BleveOptions struct { // How big should a batch get before flushing // ?? not totally sure the units BatchSize int + + // Index cache TTL for bleve indices + IndexCacheTTL time.Duration } type bleveBackend struct { @@ -59,9 +67,7 @@ type bleveBackend struct { opts BleveOptions start time.Time - // cache info - cache map[resource.NamespacedResource]*bleveIndex - cacheMu sync.RWMutex + cache *localcache.CacheService features featuremgmt.FeatureToggles indexMetrics *resource.BleveIndexMetrics @@ -82,7 +88,7 @@ func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, features featuremgm bleveBackend := &bleveBackend{ log: slog.Default().With("logger", "bleve-backend"), tracer: tracer, - cache: make(map[resource.NamespacedResource]*bleveIndex), + cache: localcache.New(opts.IndexCacheTTL, indexCacheCleanupInterval), opts: opts, start: time.Now(), features: features, @@ -96,14 +102,16 @@ func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, features featuremgm // This will return nil if the key does not exist func (b *bleveBackend) GetIndex(ctx context.Context, key resource.NamespacedResource) (resource.ResourceIndex, error) { - b.cacheMu.RLock() - defer b.cacheMu.RUnlock() - - idx, ok := b.cache[key] - if ok { - return idx, nil + val, ok := b.cache.Get(key.String()) + if !ok { + return nil, nil } - return nil, nil + + idx, ok := val.(*bleveIndex) + if !ok { + return nil, fmt.Errorf("cache item is not a bleve index: %s", key.String()) + } + return idx, nil } // updateIndexSizeMetric sets the total size of all file-based indices metric. @@ -247,9 +255,7 @@ func (b *bleveBackend) BuildIndex(ctx context.Context, } } - b.cacheMu.Lock() - b.cache[key] = idx - b.cacheMu.Unlock() + b.cache.SetDefault(key.String(), idx) return idx, nil } @@ -293,8 +299,14 @@ func isValidPath(path, safeDir string) bool { // TotalDocs returns the total number of documents across all indices func (b *bleveBackend) TotalDocs() int64 { var totalDocs int64 - for _, v := range b.cache { - c, err := v.index.DocCount() + for _, v := range b.cache.Items() { + idx, ok := v.Object.(*bleveIndex) + if !ok { + b.log.Warn("cache item is not a bleve index", "key", v.Object) + continue + } + + c, err := idx.index.DocCount() if err != nil { continue } diff --git a/pkg/storage/unified/search/options.go b/pkg/storage/unified/search/options.go index 4fd15ac2679..abc58779bae 100644 --- a/pkg/storage/unified/search/options.go +++ b/pkg/storage/unified/search/options.go @@ -26,6 +26,7 @@ func NewSearchOptions(features featuremgmt.FeatureToggles, cfg *setting.Cfg, tra 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 }, tracer, features, indexMetrics) if err != nil {