From 6fa6a5708a2b7766171da99cbdc9c637086bec90 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20=C5=A0tibran=C3=BD?= Date: Thu, 11 Sep 2025 10:23:03 +0200 Subject: [PATCH] Cleanup of old search functionality (#110861) * Remove support for initMinSize. Remove support for searchAfterWrite option, now it defaults to true. * Remove reference to deprecated feature toggle. * Remove feature toggle completely. * Remove code related to indexing on watch events. * Fix compilation error. * Remove unused field. --- .../src/types/featureToggles.gen.ts | 5 - pkg/services/featuremgmt/registry.go | 10 - pkg/services/featuremgmt/toggles_gen.csv | 1 - pkg/services/featuremgmt/toggles_gen.go | 4 - pkg/services/featuremgmt/toggles_gen.json | 1 + pkg/setting/setting.go | 1 - pkg/setting/setting_unified_storage.go | 1 - pkg/setting/setting_unified_storage_test.go | 5 - pkg/storage/unified/resource/bulk.go | 4 +- pkg/storage/unified/resource/search.go | 252 +++--------------- pkg/storage/unified/resource/search_queue.go | 158 ----------- .../unified/resource/search_queue_test.go | 175 ------------ pkg/storage/unified/resource/search_test.go | 209 ++------------- pkg/storage/unified/resource/server.go | 13 +- pkg/storage/unified/search/bleve.go | 36 +-- .../unified/search/bleve_search_test.go | 3 +- pkg/storage/unified/search/bleve_test.go | 179 ++++++------- pkg/storage/unified/search/options.go | 1 - pkg/storage/unified/sql/server.go | 2 +- .../unified/sql/test/benchmark_test.go | 49 ---- pkg/storage/unified/testing/benchmark.go | 92 +------ .../unified/testing/search_and_storage.go | 5 - pkg/storage/unified/testing/search_backend.go | 12 +- 23 files changed, 164 insertions(+), 1054 deletions(-) delete mode 100644 pkg/storage/unified/resource/search_queue.go delete mode 100644 pkg/storage/unified/resource/search_queue_test.go diff --git a/packages/grafana-data/src/types/featureToggles.gen.ts b/packages/grafana-data/src/types/featureToggles.gen.ts index 60b8a1edc90..8b62b52c1ec 100644 --- a/packages/grafana-data/src/types/featureToggles.gen.ts +++ b/packages/grafana-data/src/types/featureToggles.gen.ts @@ -1119,11 +1119,6 @@ export interface FeatureToggles { */ newClickhouseConfigPageDesign?: boolean; /** - * Enable experimental search-after-write guarantees to unified-storage search endpoints - * @default false - */ - unifiedStorageSearchAfterWriteExperimentalAPI?: boolean; - /** * Enables team folders functionality * @default false */ diff --git a/pkg/services/featuremgmt/registry.go b/pkg/services/featuremgmt/registry.go index 26557ab546d..802b5234bfc 100644 --- a/pkg/services/featuremgmt/registry.go +++ b/pkg/services/featuremgmt/registry.go @@ -1939,16 +1939,6 @@ var ( Owner: grafanaPartnerPluginsSquad, Expression: "false", }, - { - Name: "unifiedStorageSearchAfterWriteExperimentalAPI", - Description: "Enable experimental search-after-write guarantees to unified-storage search endpoints", - Stage: FeatureStageExperimental, - Owner: grafanaSearchAndStorageSquad, - HideFromAdminPage: true, - HideFromDocs: true, - RequiresRestart: true, - Expression: "false", - }, { Name: "teamFolders", Description: "Enables team folders functionality", diff --git a/pkg/services/featuremgmt/toggles_gen.csv b/pkg/services/featuremgmt/toggles_gen.csv index 0a356dbc465..0af8cc7c543 100644 --- a/pkg/services/featuremgmt/toggles_gen.csv +++ b/pkg/services/featuremgmt/toggles_gen.csv @@ -249,7 +249,6 @@ adhocFiltersInTooltips,GA,@grafana/datapro,false,false,true favoriteDatasources,experimental,@grafana/plugins-platform-backend,false,false,true newLogContext,experimental,@grafana/observability-logs,false,false,true newClickhouseConfigPageDesign,privatePreview,@grafana/partner-datasources,false,false,false -unifiedStorageSearchAfterWriteExperimentalAPI,experimental,@grafana/search-and-storage,false,true,false teamFolders,experimental,@grafana/grafana-search-navigate-organise,false,false,false alertingTriage,experimental,@grafana/alerting-squad,false,false,true graphiteBackendMode,privatePreview,@grafana/partner-datasources,false,false,false diff --git a/pkg/services/featuremgmt/toggles_gen.go b/pkg/services/featuremgmt/toggles_gen.go index d0a3b681f92..2b87218e5cf 100644 --- a/pkg/services/featuremgmt/toggles_gen.go +++ b/pkg/services/featuremgmt/toggles_gen.go @@ -1007,10 +1007,6 @@ const ( // Enables new design for the Clickhouse data source configuration page FlagNewClickhouseConfigPageDesign = "newClickhouseConfigPageDesign" - // FlagUnifiedStorageSearchAfterWriteExperimentalAPI - // Enable experimental search-after-write guarantees to unified-storage search endpoints - FlagUnifiedStorageSearchAfterWriteExperimentalAPI = "unifiedStorageSearchAfterWriteExperimentalAPI" - // FlagTeamFolders // Enables team folders functionality FlagTeamFolders = "teamFolders" diff --git a/pkg/services/featuremgmt/toggles_gen.json b/pkg/services/featuremgmt/toggles_gen.json index 9cec8403dc3..081bfde8a62 100644 --- a/pkg/services/featuremgmt/toggles_gen.json +++ b/pkg/services/featuremgmt/toggles_gen.json @@ -3537,6 +3537,7 @@ "name": "unifiedStorageSearchAfterWriteExperimentalAPI", "resourceVersion": "1755089543487", "creationTimestamp": "2025-08-13T14:05:15Z", + "deletionTimestamp": "2025-09-10T09:52:56Z", "annotations": { "grafana.app/updatedTimestamp": "2025-08-13 12:52:23.487521 +0000 UTC" } diff --git a/pkg/setting/setting.go b/pkg/setting/setting.go index aa37155042c..4854e821857 100644 --- a/pkg/setting/setting.go +++ b/pkg/setting/setting.go @@ -577,7 +577,6 @@ type Cfg struct { IndexMaxBatchSize int IndexFileThreshold int IndexMinCount int - IndexMaxCount int IndexRebuildInterval time.Duration IndexCacheTTL time.Duration EnableSharding bool diff --git a/pkg/setting/setting_unified_storage.go b/pkg/setting/setting_unified_storage.go index 5add861a0a1..3227e843bea 100644 --- a/pkg/setting/setting_unified_storage.go +++ b/pkg/setting/setting_unified_storage.go @@ -69,7 +69,6 @@ func (cfg *Cfg) setUnifiedStorageConfig() { cfg.InstanceID = section.Key("instance_id").String() cfg.IndexFileThreshold = section.Key("index_file_threshold").MustInt(10) cfg.IndexMinCount = section.Key("index_min_count").MustInt(1) - cfg.IndexMaxCount = section.Key("index_max_count").MustInt(0) // 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) diff --git a/pkg/setting/setting_unified_storage_test.go b/pkg/setting/setting_unified_storage_test.go index 4946939cdc1..094033a4244 100644 --- a/pkg/setting/setting_unified_storage_test.go +++ b/pkg/setting/setting_unified_storage_test.go @@ -35,9 +35,6 @@ func TestCfg_setUnifiedStorageConfig(t *testing.T) { _, err = unifiedStorageSection.NewKey("index_min_count", "5") assert.NoError(t, err) - _, err = unifiedStorageSection.NewKey("index_max_count", "1000") - assert.NoError(t, err) - cfg.setUnifiedStorageConfig() value, exists := cfg.UnifiedStorage["playlists.playlist.grafana.app"] @@ -52,7 +49,6 @@ func TestCfg_setUnifiedStorageConfig(t *testing.T) { // Test that index settings are correctly parsed assert.Equal(t, 5, cfg.IndexMinCount) - assert.Equal(t, 1000, cfg.IndexMaxCount) }) t.Run("read unified_storage configs with defaults", func(t *testing.T) { @@ -65,6 +61,5 @@ func TestCfg_setUnifiedStorageConfig(t *testing.T) { // Test that default index settings are applied assert.Equal(t, 1, cfg.IndexMinCount) - assert.Equal(t, 0, cfg.IndexMaxCount) }) } diff --git a/pkg/storage/unified/resource/bulk.go b/pkg/storage/unified/resource/bulk.go index 4bc7a9ffe7f..caa07e095a1 100644 --- a/pkg/storage/unified/resource/bulk.go +++ b/pkg/storage/unified/resource/bulk.go @@ -240,11 +240,11 @@ func (s *server) BulkProcess(stream resourcepb.BulkStore_BulkProcessServer) erro if rsp.Error == nil && s.search != nil { // Rebuild any changed indexes for _, summary := range rsp.Summary { - _, _, err := s.search.build(ctx, NamespacedResource{ + _, err := s.search.build(ctx, NamespacedResource{ Namespace: summary.Namespace, Group: summary.Group, Resource: summary.Resource, - }, summary.Count, summary.ResourceVersion, "rebuildAfterBatchLoad", true) + }, summary.Count, "rebuildAfterBatchLoad", true) if err != nil { s.log.Warn("error building search index after batch load", "err", err) rsp.Error = &resourcepb.ErrorResult{ diff --git a/pkg/storage/unified/resource/search.go b/pkg/storage/unified/resource/search.go index b842f23afbf..88444b02cee 100644 --- a/pkg/storage/unified/resource/search.go +++ b/pkg/storage/unified/resource/search.go @@ -105,13 +105,11 @@ type SearchBackend interface { ctx context.Context, key NamespacedResource, size int64, - resourceVersion int64, nonStandardFields SearchableDocumentFields, indexBuildReason string, builder BuildFn, updater UpdateFn, rebuild bool, - searchAfterWrite bool, ) (ResourceIndex, error) // TotalDocs returns the total number of documents across all indexes. @@ -122,31 +120,21 @@ const tracingPrexfixSearch = "unified_search." // This supports indexing+search regardless of implementation type searchSupport struct { - tracer trace.Tracer - log *slog.Logger - storage StorageBackend - search SearchBackend - indexMetrics *BleveIndexMetrics - access types.AccessClient - builders *builderCache - initWorkers int - initMinSize int - initMaxSize int - searchAfterWrite bool + tracer trace.Tracer + log *slog.Logger + storage StorageBackend + search SearchBackend + indexMetrics *BleveIndexMetrics + access types.AccessClient + builders *builderCache + initWorkers int + initMinSize int ring *ring.Ring ringLifecycler *ring.BasicLifecycler buildIndex singleflight.Group - // Index queue processors - indexQueueProcessorsMutex sync.Mutex - indexQueueProcessors map[string]*indexQueueProcessor - indexEventsChan chan *IndexEvent - - // testing - clientIndexEventsChan chan *IndexEvent - // periodic rebuilding of the indexes to keep usage insights up to date rebuildInterval time.Duration } @@ -156,7 +144,7 @@ var ( _ resourcepb.ManagedObjectIndexServer = (*searchSupport)(nil) ) -func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.AccessClient, blob BlobSupport, tracer trace.Tracer, indexMetrics *BleveIndexMetrics, ring *ring.Ring, ringLifecycler *ring.BasicLifecycler, searchAfterWrite bool) (support *searchSupport, err error) { +func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.AccessClient, blob BlobSupport, tracer trace.Tracer, indexMetrics *BleveIndexMetrics, ring *ring.Ring, ringLifecycler *ring.BasicLifecycler) (support *searchSupport, err error) { // No backend search support if opts.Backend == nil { return nil, nil @@ -170,22 +158,17 @@ func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.A } support = &searchSupport{ - access: access, - tracer: tracer, - storage: storage, - search: opts.Backend, - log: slog.Default().With("logger", "resource-search"), - initWorkers: opts.WorkerThreads, - initMinSize: opts.InitMinCount, - initMaxSize: opts.InitMaxCount, - searchAfterWrite: searchAfterWrite, - indexMetrics: indexMetrics, - clientIndexEventsChan: opts.IndexEventsChan, - indexEventsChan: make(chan *IndexEvent), - indexQueueProcessors: make(map[string]*indexQueueProcessor), - rebuildInterval: opts.RebuildInterval, - ring: ring, - ringLifecycler: ringLifecycler, + access: access, + tracer: tracer, + storage: storage, + search: opts.Backend, + log: slog.Default().With("logger", "resource-search"), + initWorkers: opts.WorkerThreads, + initMinSize: opts.InitMinCount, + indexMetrics: indexMetrics, + rebuildInterval: opts.RebuildInterval, + ring: ring, + ringLifecycler: ringLifecycler, } info, err := opts.Resources.GetDocumentBuilders() @@ -471,20 +454,12 @@ func (s *searchSupport) buildIndexes(ctx context.Context, rebuild bool) (int, er } totalBatchesIndexed++ - // If the count is too large, we need to set the index to empty. - // Only do this if the max size is set to a non-zero (default) value. - if s.initMaxSize > 0 && (info.Count > int64(s.initMaxSize)) { - s.log.Info("setting empty index for resource with count greater than max size", "namespace", info.Namespace, "group", info.Group, "resource", info.Resource, "count", info.Count, "maxSize", s.initMaxSize) - _, err := s.buildEmptyIndex(ctx, info.NamespacedResource, info.ResourceVersion) - return err - } - s.log.Debug("building index", "namespace", info.Namespace, "group", info.Group, "resource", info.Resource, "rebuild", rebuild) reason := "init" if rebuild { reason = "rebuild" } - _, _, err := s.build(ctx, info.NamespacedResource, info.Count, info.ResourceVersion, reason, rebuild) + _, err := s.build(ctx, info.NamespacedResource, info.Count, reason, rebuild) return err }) } @@ -498,6 +473,8 @@ func (s *searchSupport) buildIndexes(ctx context.Context, rebuild bool) (int, er } func (s *searchSupport) init(ctx context.Context) error { + origCtx := ctx + ctx, span := s.tracer.Start(ctx, tracingPrexfixSearch+"Init") defer span.End() start := time.Now().Unix() @@ -509,34 +486,10 @@ func (s *searchSupport) init(ctx context.Context) error { span.AddEvent("namespaces indexed", trace.WithAttributes(attribute.Int("namespaced_indexed", totalBatchesIndexed))) - watchctx := context.Background() // new context? - // don't start watcher when SearchAfterWrite changes are enabled - if !s.searchAfterWrite { - // Now start listening for new events - events, err := s.storage.WatchWriteEvents(watchctx) - if err != nil { - return err - } - go func() { - for { - v := <-events - - // Skip events during batch updates - if v.PreviousRV < 0 { - continue - } - - s.dispatchEvent(watchctx, v) - } - }() - - go s.monitorIndexEvents(ctx) - } - // since usage insights is not in unified storage, we need to periodically rebuild the index // to make sure these data points are up to date. if s.rebuildInterval > 0 { - go s.startPeriodicRebuild(watchctx) + go s.startPeriodicRebuild(origCtx) } end := time.Now().Unix() @@ -545,77 +498,6 @@ func (s *searchSupport) init(ctx context.Context) error { return nil } -// Async event dispatching -// This is called from the watch event loop -// It will dispatch the event to the appropriate index queue processor -func (s *searchSupport) dispatchEvent(ctx context.Context, evt *WrittenEvent) { - ctx, span := s.tracer.Start(ctx, tracingPrexfixSearch+"dispatchEvent") - defer span.End() - span.SetAttributes( - attribute.String("event_type", evt.Type.String()), - attribute.String("namespace", evt.Key.Namespace), - attribute.String("group", evt.Key.Group), - attribute.String("resource", evt.Key.Resource), - attribute.String("name", evt.Key.Name), - ) - - switch evt.Type { - case resourcepb.WatchEvent_ADDED, resourcepb.WatchEvent_MODIFIED, resourcepb.WatchEvent_DELETED: // OK - default: - s.log.Info("ignoring watch event", "type", evt.Type) - span.AddEvent("ignoring watch event", trace.WithAttributes(attribute.String("type", evt.Type.String()))) - } - - nsr := NamespacedResource{ - Namespace: evt.Key.Namespace, - Group: evt.Key.Group, - Resource: evt.Key.Resource, - } - index, err := s.getOrCreateIndex(ctx, nsr, "dispatchEvent") - if err != nil { - s.log.Warn("error getting index for watch event", "error", err) - span.RecordError(err) - return - } - // Get or create index queue processor for this index - indexQueueProcessor, err := s.getOrCreateIndexQueueProcessor(index, nsr) - if err != nil { - s.log.Error("error getting index queue processor for watch event", "error", err) - span.RecordError(err) - return - } - indexQueueProcessor.Add(evt) -} - -func (s *searchSupport) monitorIndexEvents(ctx context.Context) { - var evt *IndexEvent - for { - select { - case <-ctx.Done(): - return - case evt = <-s.indexEventsChan: - } - if evt.Err != nil { - s.log.Error("error indexing watch event", "error", evt.Err) - continue - } - _, span := s.tracer.Start(ctx, tracingPrexfixSearch+"monitorIndexEvents") - defer span.End() - // record latency from when event was created to when it was indexed - span.AddEvent("index latency", trace.WithAttributes(attribute.Float64("latency_seconds", evt.Latency.Seconds()))) - s.log.Debug("indexed new object", "resource", evt.WrittenEvent.Key.Resource, "latency_seconds", evt.Latency.Seconds(), "name", evt.WrittenEvent.Key.Name, "namespace", evt.WrittenEvent.Key.Namespace, "rv", evt.WrittenEvent.ResourceVersion) - if evt.Latency.Seconds() > 1 { - s.log.Warn("high index latency object details", "resource", evt.WrittenEvent.Key.Resource, "latency_seconds", evt.Latency.Seconds(), "name", evt.WrittenEvent.Key.Name, "namespace", evt.WrittenEvent.Key.Namespace, "rv", evt.WrittenEvent.ResourceVersion) - } - if s.indexMetrics != nil { - s.indexMetrics.IndexLatency.WithLabelValues(evt.WrittenEvent.Key.Resource).Observe(evt.Latency.Seconds()) - } - if s.clientIndexEventsChan != nil { - s.clientIndexEventsChan <- evt - } - } -} - func (s *searchSupport) startPeriodicRebuild(ctx context.Context) { ticker := time.NewTicker(s.rebuildInterval) defer ticker.Stop() @@ -700,16 +582,14 @@ func (s *searchSupport) getOrCreateIndex(ctx context.Context, key NamespacedReso } size := int64(0) - rv := int64(0) for _, stat := range stats { if stat.Namespace == key.Namespace && stat.Group == key.Group && stat.Resource == key.Resource { size = stat.Count - rv = stat.ResourceVersion break } } - idx, _, err = s.build(ctx, key, size, rv, reason, false) + idx, err = s.build(ctx, key, size, reason, false) if err != nil { return nil, fmt.Errorf("error building search index, %w", err) } @@ -730,25 +610,23 @@ func (s *searchSupport) getOrCreateIndex(ctx context.Context, key NamespacedReso } } - if s.searchAfterWrite { - span.AddEvent("Updating index") - start := time.Now() - rv, err := idx.UpdateIndex(ctx, reason) - if err != nil { - return nil, tracing.Error(span, fmt.Errorf("failed to update index to guarantee strong consistency: %w", err)) - } - elapsed := time.Since(start) - if s.indexMetrics != nil { - s.indexMetrics.SearchUpdateWaitTime.WithLabelValues(reason).Observe(elapsed.Seconds()) - } - s.log.Debug("Index updated before search", "namespace", key.Namespace, "group", key.Group, "resource", key.Resource, "reason", reason, "duration", elapsed, "rv", rv) - span.AddEvent("Index updated") + span.AddEvent("Updating index") + start := time.Now() + rv, err := idx.UpdateIndex(ctx, reason) + if err != nil { + return nil, tracing.Error(span, fmt.Errorf("failed to update index to guarantee strong consistency: %w", err)) } + elapsed := time.Since(start) + if s.indexMetrics != nil { + s.indexMetrics.SearchUpdateWaitTime.WithLabelValues(reason).Observe(elapsed.Seconds()) + } + s.log.Debug("Index updated before search", "namespace", key.Namespace, "group", key.Group, "resource", key.Resource, "reason", reason, "duration", elapsed, "rv", rv) + span.AddEvent("Index updated") return idx, nil } -func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size int64, documentStatsRV int64, indexBuildReason string, rebuild bool) (ResourceIndex, int64, error) { +func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size int64, indexBuildReason string, rebuild bool) (ResourceIndex, error) { ctx, span := s.tracer.Start(ctx, tracingPrexfixSearch+"Build") defer span.End() @@ -757,20 +635,19 @@ func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size attribute.String("group", nsr.Group), attribute.String("resource", nsr.Resource), attribute.Int64("size", size), - attribute.Int64("rv", documentStatsRV), ) logger := s.log.With("namespace", nsr.Namespace, "group", nsr.Group, "resource", nsr.Resource) builder, err := s.builders.get(ctx, nsr) if err != nil { - return nil, 0, err + return nil, err } fields := s.builders.GetFields(nsr) builderFn := func(index ResourceIndex) (int64, error) { span := trace.SpanFromContext(ctx) - span.AddEvent("building index", trace.WithAttributes(attribute.Int64("size", size), attribute.Int64("rv", documentStatsRV), attribute.String("reason", indexBuildReason))) + span.AddEvent("building index", trace.WithAttributes(attribute.Int64("size", size), attribute.String("reason", indexBuildReason))) listRV, err := s.storage.ListIterator(ctx, &resourcepb.ListRequest{ Limit: 1000000000000, // big number @@ -840,7 +717,7 @@ func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size updaterFn := func(ctx context.Context, index ResourceIndex, sinceRV int64) (int64, int, error) { span := trace.SpanFromContext(ctx) - span.AddEvent("updating index", trace.WithAttributes(attribute.Int64("sinceRV", documentStatsRV))) + span.AddEvent("updating index", trace.WithAttributes(attribute.Int64("sinceRV", sinceRV))) rv, it := s.storage.ListModifiedSince(ctx, NamespacedResource{ Group: nsr.Group, @@ -916,10 +793,10 @@ func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size return rv, docs, nil } - index, err := s.search.BuildIndex(ctx, nsr, size, documentStatsRV, fields, indexBuildReason, builderFn, updaterFn, rebuild, s.searchAfterWrite) + index, err := s.search.BuildIndex(ctx, nsr, size, fields, indexBuildReason, builderFn, updaterFn, rebuild) if err != nil { - return nil, 0, err + return nil, err } // Record the number of objects indexed for the kind/resource @@ -931,26 +808,7 @@ func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size s.indexMetrics.IndexedKinds.WithLabelValues(nsr.Resource).Add(float64(docCount)) } - // rv is the last RV we read. when watching, we must add all events since that time - return index, documentStatsRV, err -} - -// buildEmptyIndex creates an empty index without adding any documents -func (s *searchSupport) buildEmptyIndex(ctx context.Context, nsr NamespacedResource, rv int64) (ResourceIndex, error) { - ctx, span := s.tracer.Start(ctx, tracingPrexfixSearch+"BuildEmptyIndex") - defer span.End() - - fields := s.builders.GetFields(nsr) - s.log.Debug("Building empty index", "namespace", nsr.Namespace, "group", nsr.Group, "resource", nsr.Resource, "rv", rv) - - // Build an empty index by passing a builder function that doesn't add any documents - return s.search.BuildIndex(ctx, nsr, 0, rv, fields, "empty", func(index ResourceIndex) (int64, error) { - // Return the resource version without adding any documents to the index - return 0, nil - }, func(context context.Context, index ResourceIndex, sinceRV int64) (int64, int, error) { - // No update is performed. - return 0, 0, nil - }, false, s.searchAfterWrite) + return index, err } type builderCache struct { @@ -1069,28 +927,6 @@ func AsResourceKey(ns string, t string) (*resourcepb.ResourceKey, error) { return nil, fmt.Errorf("unknown resource type") } -// getOrCreateIndexQueueProcessor returns an IndexQueueProcessor for the given index -func (s *searchSupport) getOrCreateIndexQueueProcessor(index ResourceIndex, nsr NamespacedResource) (*indexQueueProcessor, error) { - s.indexQueueProcessorsMutex.Lock() - defer s.indexQueueProcessorsMutex.Unlock() - - key := fmt.Sprintf("%s/%s/%s", nsr.Namespace, nsr.Group, nsr.Resource) - if indexQueueProcessor, ok := s.indexQueueProcessors[key]; ok { - // index stored on existing processor may have been closed and rebuilt, so we need to update it - indexQueueProcessor.updateIndex(index) - return indexQueueProcessor, nil - } - - builder, err := s.builders.get(context.Background(), nsr) - if err != nil { - s.log.Error("error getting document builder", "error", err) - return nil, err - } - indexQueueProcessor := newIndexQueueProcessor(index, nsr, maxBatchSize, builder, s.indexEventsChan) - s.indexQueueProcessors[key] = indexQueueProcessor - return indexQueueProcessor, nil -} - func (s *builderCache) clearNamespacedCache(key NamespacedResource) { s.mu.Lock() defer s.mu.Unlock() diff --git a/pkg/storage/unified/resource/search_queue.go b/pkg/storage/unified/resource/search_queue.go deleted file mode 100644 index 23310caf9df..00000000000 --- a/pkg/storage/unified/resource/search_queue.go +++ /dev/null @@ -1,158 +0,0 @@ -package resource - -import ( - "context" - "sync" - "time" - - "github.com/grafana/grafana/pkg/storage/unified/resourcepb" -) - -// indexQueueProcessor manages queue-based operations for a specific index -// It is responsible for ingesting events for a single index -// It will batch events and send them to the index in a single bulk request -type indexQueueProcessor struct { - nsr NamespacedResource - queue chan *WrittenEvent - batchSize int - builder DocumentBuilder - - resChan chan *IndexEvent // Channel to send results to the caller - - indexMu sync.Mutex - index ResourceIndex - - runningMu sync.Mutex - running bool -} - -type IndexEvent struct { - WrittenEvent *WrittenEvent - Action IndexAction - IndexableDocument *IndexableDocument // empty for delete actions - Timestamp time.Time - Latency time.Duration - Err error -} - -func (b *indexQueueProcessor) updateIndex(newIndex ResourceIndex) { - b.indexMu.Lock() - defer b.indexMu.Unlock() - b.index = newIndex -} - -// newIndexQueueProcessor creates a new IndexQueueProcessor for the given index -func newIndexQueueProcessor(index ResourceIndex, nsr NamespacedResource, batchSize int, builder DocumentBuilder, resChan chan *IndexEvent) *indexQueueProcessor { - return &indexQueueProcessor{ - index: index, - nsr: nsr, - queue: make(chan *WrittenEvent, 1000), // Buffer size of 1000 events - batchSize: batchSize, - builder: builder, - resChan: resChan, - running: false, - } -} - -// Add adds an event to the queue and ensures the background processor is running -func (b *indexQueueProcessor) Add(evt *WrittenEvent) { - b.queue <- evt - - // Start the processor if it's not already running - b.runningMu.Lock() - defer b.runningMu.Unlock() - if !b.running { - b.running = true - go b.runProcessor() - } -} - -// runProcessor is the task processing the queue of written events -func (b *indexQueueProcessor) runProcessor() { - defer func() { - b.runningMu.Lock() - b.running = false - b.runningMu.Unlock() - }() - - for { - batch := make([]*WrittenEvent, 0, b.batchSize) - select { - case evt := <-b.queue: - batch = append(batch, evt) - case <-time.After(5 * time.Second): - // No events in the past few seconds, stop the processor - return - } - - prepare: - for len(batch) < b.batchSize { - select { - case evt := <-b.queue: - batch = append(batch, evt) - default: - break prepare - } - } - - b.process(batch) - } -} - -// process handles a batch of events -func (b *indexQueueProcessor) process(batch []*WrittenEvent) { - if len(batch) == 0 { - return - } - - // Create bulk request - req := &BulkIndexRequest{ - Items: make([]*BulkIndexItem, 0, len(batch)), - } - resp := make([]*IndexEvent, 0, len(batch)) - - for _, evt := range batch { - result := &IndexEvent{ - WrittenEvent: evt, - } - resp = append(resp, result) - - item := &BulkIndexItem{} - if evt.Type == resourcepb.WatchEvent_DELETED { - item.Action = ActionDelete - item.Key = evt.Key - } else { - item.Action = ActionIndex - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - doc, err := b.builder.BuildDocument(ctx, evt.Key, evt.ResourceVersion, evt.Value) - cancel() - - if err != nil { - result.Err = err - } else { - item.Doc = doc - result.IndexableDocument = doc - } - } - req.Items = append(req.Items, item) - } - - b.indexMu.Lock() - idx := b.index - b.indexMu.Unlock() - - err := idx.BulkIndex(req) - if err != nil { - for _, r := range resp { - r.Err = err - } - } - ts := time.Now() - if b.resChan != nil { - for _, r := range resp { - r.Timestamp = ts - r.Latency = time.Duration(ts.UnixMicro()-r.WrittenEvent.ResourceVersion) * time.Microsecond - b.resChan <- r - } - } -} diff --git a/pkg/storage/unified/resource/search_queue_test.go b/pkg/storage/unified/resource/search_queue_test.go deleted file mode 100644 index 2c465a6ffef..00000000000 --- a/pkg/storage/unified/resource/search_queue_test.go +++ /dev/null @@ -1,175 +0,0 @@ -package resource - -import ( - "testing" - "time" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/mock" - - "github.com/grafana/grafana/pkg/storage/unified/resourcepb" -) - -func TestNewIndexQueueProcessor(t *testing.T) { - mockIndex := &MockResourceIndex{} - mockBuilder := &MockDocumentBuilder{} - nsr := NamespacedResource{Resource: "test"} - - resChan := make(chan *IndexEvent) - - processor := newIndexQueueProcessor(mockIndex, nsr, 10, mockBuilder, resChan) - - assert.NotNil(t, processor) - assert.Equal(t, 10, processor.batchSize) - assert.NotNil(t, processor.queue) -} - -func TestIndexQueueProcessor_SingleEvent(t *testing.T) { - mockIndex := &MockResourceIndex{} - mockBuilder := &MockDocumentBuilder{} - nsr := NamespacedResource{Resource: "test"} - - resChan := make(chan *IndexEvent) - - processor := newIndexQueueProcessor(mockIndex, nsr, 10, mockBuilder, resChan) - - // Test data - key := resourcepb.ResourceKey{Resource: "test", Name: "obj1", Namespace: "default"} - evt := &WrittenEvent{ - Key: &key, - ResourceVersion: time.Now().UnixMicro(), - Type: resourcepb.WatchEvent_ADDED, - Value: []byte(`{"test": "data"}`), - } - - // Setup expectations - mockBuilder.On("BuildDocument", mock.Anything, &key, evt.ResourceVersion, evt.Value).Return(&IndexableDocument{Key: &key}, nil) - mockIndex.On("BulkIndex", mock.MatchedBy(func(req *BulkIndexRequest) bool { - return len(req.Items) == 1 && req.Items[0].Action == ActionIndex - })).Return(nil) - - // Start processor and wait for the document to be indexed - processor.Add(evt) - - resp := <-resChan - assert.NotNil(t, resp) - assert.Nil(t, resp.Err) - assert.Equal(t, &key, resp.IndexableDocument.Key) - - mockBuilder.AssertExpectations(t) - mockIndex.AssertExpectations(t) -} - -func TestIndexQueueProcessor_BatchProcessing(t *testing.T) { - mockIndex := &MockResourceIndex{} - mockBuilder := &MockDocumentBuilder{} - nsr := NamespacedResource{Namespace: "default", Resource: "test"} - - resChan := make(chan *IndexEvent) - - processor := newIndexQueueProcessor(mockIndex, nsr, 2, mockBuilder, resChan) - - // Test data for two events - events := []*WrittenEvent{ - { - Key: &resourcepb.ResourceKey{Resource: "test", Name: "obj1", Namespace: "default"}, - ResourceVersion: time.Now().UnixMicro(), - Type: resourcepb.WatchEvent_ADDED, - Value: []byte(`{"test": "data1"}`), - }, - { - Key: &resourcepb.ResourceKey{Resource: "test", Name: "obj2", Namespace: "default"}, - ResourceVersion: time.Now().UnixMicro(), - Type: resourcepb.WatchEvent_DELETED, - }, - } - - // Setup expectations - mockBuilder.On("BuildDocument", mock.Anything, events[0].Key, events[0].ResourceVersion, events[0].Value). - Return(&IndexableDocument{Key: events[0].Key}, nil) - mockIndex.On("BulkIndex", mock.MatchedBy(func(req *BulkIndexRequest) bool { - return len(req.Items) == 2 && - req.Items[0].Action == ActionIndex && - req.Items[1].Action == ActionDelete - })).Return(nil) - - // Start processor and add events - processor.Add(events[0]) - processor.Add(events[1]) - - r0 := <-resChan - assert.Nil(t, r0.Err) - assert.Equal(t, events[0].Key, r0.IndexableDocument.Key) - - r1 := <-resChan - assert.Nil(t, r1.Err) - assert.Nil(t, r1.IndexableDocument) // deleted event - - mockBuilder.AssertExpectations(t) - mockIndex.AssertExpectations(t) -} - -func TestIndexQueueProcessor_BuildDocumentError(t *testing.T) { - mockIndex := &MockResourceIndex{} - mockBuilder := &MockDocumentBuilder{} - nsr := NamespacedResource{Resource: "test"} - - resChan := make(chan *IndexEvent) - - processor := newIndexQueueProcessor(mockIndex, nsr, 10, mockBuilder, resChan) - - evt := &WrittenEvent{ - Key: &resourcepb.ResourceKey{Resource: "test", Name: "obj1", Namespace: "default"}, - ResourceVersion: time.Now().UnixMicro(), - Type: resourcepb.WatchEvent_ADDED, - Value: []byte(`invalid json`), - } - - // Setup expectations for error case - mockBuilder.On("BuildDocument", mock.Anything, evt.Key, evt.ResourceVersion, evt.Value). - Return(nil, assert.AnError) - - // The bulk index should not be called since document building failed - mockIndex.On("BulkIndex", mock.Anything).Return(nil).Maybe() - - processor.Add(evt) - - resp := <-resChan - assert.NotNil(t, resp) - assert.Error(t, resp.Err) - assert.Nil(t, resp.IndexableDocument) - - mockBuilder.AssertExpectations(t) - mockIndex.AssertExpectations(t) -} - -func TestIndexQueueProcessor_BulkIndexError(t *testing.T) { - mockIndex := &MockResourceIndex{} - mockBuilder := &MockDocumentBuilder{} - nsr := NamespacedResource{Resource: "test"} - - resChan := make(chan *IndexEvent) - - processor := newIndexQueueProcessor(mockIndex, nsr, 10, mockBuilder, resChan) - - evt := &WrittenEvent{ - Key: &resourcepb.ResourceKey{Resource: "test", Name: "obj1", Namespace: "default"}, - ResourceVersion: time.Now().UnixMicro(), - Type: resourcepb.WatchEvent_ADDED, - Value: []byte(`{"test": "data"}`), - } - - // Setup expectations - mockBuilder.On("BuildDocument", mock.Anything, evt.Key, evt.ResourceVersion, evt.Value). - Return(&IndexableDocument{Key: evt.Key}, nil) - mockIndex.On("BulkIndex", mock.Anything).Return(assert.AnError) - - processor.Add(evt) - - resp := <-resChan - assert.NotNil(t, resp) - assert.Error(t, resp.Err) - - mockBuilder.AssertExpectations(t) - mockIndex.AssertExpectations(t) -} diff --git a/pkg/storage/unified/resource/search_test.go b/pkg/storage/unified/resource/search_test.go index 0b42f90e4e8..b4e7108383c 100644 --- a/pkg/storage/unified/resource/search_test.go +++ b/pkg/storage/unified/resource/search_test.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "iter" - "log/slog" "sync" "testing" "time" @@ -128,17 +127,15 @@ type mockSearchBackend struct { } type buildIndexCall struct { - key NamespacedResource - size int64 - resourceVersion int64 - fields SearchableDocumentFields + key NamespacedResource + size int64 + fields SearchableDocumentFields } type buildEmptyIndexCall struct { - key NamespacedResource - size int64 // should be 0 for empty indexes - resourceVersion int64 - fields SearchableDocumentFields + key NamespacedResource + size int64 // should be 0 for empty indexes + fields SearchableDocumentFields } func (m *mockSearchBackend) GetIndex(ctx context.Context, key NamespacedResource) (ResourceIndex, error) { @@ -147,7 +144,7 @@ func (m *mockSearchBackend) GetIndex(ctx context.Context, key NamespacedResource return m.cache[key], nil } -func (m *mockSearchBackend) BuildIndex(ctx context.Context, key NamespacedResource, size int64, resourceVersion int64, fields SearchableDocumentFields, reason string, builder BuildFn, updater UpdateFn, rebuild bool, searchAfterWrite bool) (ResourceIndex, error) { +func (m *mockSearchBackend) BuildIndex(ctx context.Context, key NamespacedResource, size int64, fields SearchableDocumentFields, reason string, builder BuildFn, updater UpdateFn, rebuild bool) (ResourceIndex, error) { index := &MockResourceIndex{} index.On("BulkIndex", mock.Anything).Return(nil).Maybe() index.On("DocCount", mock.Anything, mock.Anything).Return(int64(0), nil).Maybe() @@ -171,18 +168,16 @@ func (m *mockSearchBackend) BuildIndex(ctx context.Context, key NamespacedResour if size == 0 { // This is an empty index (buildEmptyIndex was called) m.buildEmptyIndexCalls = append(m.buildEmptyIndexCalls, buildEmptyIndexCall{ - key: key, - size: size, - resourceVersion: resourceVersion, - fields: fields, + key: key, + size: size, + fields: fields, }) } else { // This is a normal index (build was called) m.buildIndexCalls = append(m.buildIndexCalls, buildIndexCall{ - key: key, - size: size, - resourceVersion: resourceVersion, - fields: fields, + key: key, + size: size, + fields: fields, }) } @@ -193,122 +188,6 @@ func (m *mockSearchBackend) TotalDocs() int64 { return 0 } -func TestBuildIndexes_MaxCountThreshold(t *testing.T) { - tests := []struct { - name string - initMaxSize int - resourceStats []ResourceStats - expectedNormalBuilds []string // expected NamespacedResource strings that should be built normally - expectedEmptyBuilds []string // expected NamespacedResource strings that should be built as empty - }{ - { - name: "max count disabled (0) - all resources built normally", - initMaxSize: 0, - resourceStats: []ResourceStats{ - {NamespacedResource: NamespacedResource{Namespace: "ns1", Group: "group1", Resource: "resource1"}, Count: 50}, - {NamespacedResource: NamespacedResource{Namespace: "ns1", Group: "group1", Resource: "resource2"}, Count: 150}, - {NamespacedResource: NamespacedResource{Namespace: "ns1", Group: "group2", Resource: "resource1"}, Count: 250}, - }, - expectedNormalBuilds: []string{ - "ns1/group1/resource1", - "ns1/group1/resource2", - "ns1/group2/resource1", - }, - expectedEmptyBuilds: []string{}, - }, - { - name: "max count 100 - resources above threshold get empty indexes", - initMaxSize: 100, - resourceStats: []ResourceStats{ - {NamespacedResource: NamespacedResource{Namespace: "ns1", Group: "group1", Resource: "resource1"}, Count: 50}, // normal build - {NamespacedResource: NamespacedResource{Namespace: "ns1", Group: "group1", Resource: "resource2"}, Count: 150}, // empty build - {NamespacedResource: NamespacedResource{Namespace: "ns1", Group: "group2", Resource: "resource1"}, Count: 250}, // empty build - {NamespacedResource: NamespacedResource{Namespace: "ns1", Group: "group2", Resource: "resource2"}, Count: 80}, // normal build - }, - expectedNormalBuilds: []string{ - "ns1/group1/resource1", - "ns1/group2/resource2", - }, - expectedEmptyBuilds: []string{ - "ns1/group1/resource2", - "ns1/group2/resource1", - }, - }, - { - name: "max count 300 - no resources exceed threshold", - initMaxSize: 300, - resourceStats: []ResourceStats{ - {NamespacedResource: NamespacedResource{Namespace: "ns1", Group: "group1", Resource: "resource1"}, Count: 50}, // normal build - {NamespacedResource: NamespacedResource{Namespace: "ns1", Group: "group1", Resource: "resource2"}, Count: 150}, // normal build - {NamespacedResource: NamespacedResource{Namespace: "ns1", Group: "group2", Resource: "resource1"}, Count: 250}, // normal build - }, - expectedNormalBuilds: []string{ - "ns1/group1/resource1", - "ns1/group1/resource2", - "ns1/group2/resource1", - }, - expectedEmptyBuilds: []string{}, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - // Setup mock implementations - storage := &mockStorageBackend{ - resourceStats: tt.resourceStats, - } - search := &mockSearchBackend{ - buildIndexCalls: []buildIndexCall{}, - buildEmptyIndexCalls: []buildEmptyIndexCall{}, - } - supplier := &TestDocumentBuilderSupplier{ - GroupsResources: map[string]string{ - "group1": "resource1", - "group2": "resource2", - }, - } - - // Create search support with the specified initMaxSize - opts := SearchOptions{ - Backend: search, - Resources: supplier, - WorkerThreads: 1, - InitMinCount: 1, // set min count to default for this test - InitMaxCount: tt.initMaxSize, - } - - support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil, false) - require.NoError(t, err) - require.NotNil(t, support) - - // Call buildIndexes - ctx := context.Background() - indexesBuilt, err := support.buildIndexes(ctx, false) - require.NoError(t, err) - - // Verify the correct number of indexes were built (normal + empty) - expectedTotal := len(tt.expectedNormalBuilds) + len(tt.expectedEmptyBuilds) - require.Equal(t, expectedTotal, indexesBuilt) - - // Verify the correct resources were built normally - actualNormalBuilds := make([]string, len(search.buildIndexCalls)) - for i, call := range search.buildIndexCalls { - actualNormalBuilds[i] = call.key.String() - } - require.ElementsMatch(t, tt.expectedNormalBuilds, actualNormalBuilds) - - // Verify the correct resources were built as empty indexes - actualEmptyBuilds := make([]string, len(search.buildEmptyIndexCalls)) - for i, call := range search.buildEmptyIndexCalls { - actualEmptyBuilds[i] = call.key.String() - // Verify that empty indexes are built with size 0 - require.Equal(t, int64(0), call.size, "Empty index should be built with size 0") - } - require.ElementsMatch(t, tt.expectedEmptyBuilds, actualEmptyBuilds) - }) - } -} - func TestSearchGetOrCreateIndex(t *testing.T) { // Setup mock implementations storage := &mockStorageBackend{ @@ -332,10 +211,9 @@ func TestSearchGetOrCreateIndex(t *testing.T) { Resources: supplier, WorkerThreads: 1, InitMinCount: 1, // set min count to default for this test - InitMaxCount: 0, } - support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil, false) + support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil) require.NoError(t, err) require.NotNil(t, support) @@ -361,12 +239,6 @@ func TestSearchGetOrCreateIndex(t *testing.T) { require.NotEmpty(t, search.buildIndexCalls) require.Less(t, len(search.buildIndexCalls), concurrency, "Should not have built index more than a few times (ideally once)") require.Equal(t, int64(50), search.buildIndexCalls[0].size) - require.Equal(t, int64(11111111), search.buildIndexCalls[0].resourceVersion) - - // Verify that UpdateIndex was not called at all, since searchAfterWrite is not enabled. - idx, err := support.getOrCreateIndex(context.Background(), NamespacedResource{Namespace: "ns", Group: "group", Resource: "resource"}, "test") - require.NoError(t, err) - checkMockIndexUpdateCalls(t, idx, nil) } func TestSearchGetOrCreateIndexWithIndexUpdate(t *testing.T) { @@ -399,11 +271,10 @@ func TestSearchGetOrCreateIndexWithIndexUpdate(t *testing.T) { Resources: supplier, WorkerThreads: 1, InitMinCount: 1, // set min count to default for this test - InitMaxCount: 0, } // Enable searchAfterWrite - support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil, true) + support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil) require.NoError(t, err) require.NotNil(t, support) @@ -452,10 +323,9 @@ func TestSearchGetOrCreateIndexWithCancellation(t *testing.T) { Resources: supplier, WorkerThreads: 1, InitMinCount: 1, // set min count to default for this test - InitMaxCount: 0, } - support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil, false) + support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil) require.NoError(t, err) require.NotNil(t, support) @@ -484,49 +354,6 @@ func TestSearchGetOrCreateIndexWithCancellation(t *testing.T) { require.NoError(t, err) } -func TestSearchWillUpdateIndexOnQueueProcessor(t *testing.T) { - // Regression test: Indexes were being closed when being rebuilt, but not updated on the queue processor. This was causing new events to - // be added to a closed index, resulting in an error and missing docs in the index. - - // Create mock components - mockIndex1 := &MockResourceIndex{} - mockIndex2 := &MockResourceIndex{} // Different index to test replacement - mockBuilder := &MockDocumentBuilder{} - - // Create searchSupport instance - s := &searchSupport{ - log: slog.Default(), - indexQueueProcessors: make(map[string]*indexQueueProcessor), - indexQueueProcessorsMutex: sync.Mutex{}, - indexEventsChan: make(chan *IndexEvent, 10), - } - - nsr := NamespacedResource{ - Namespace: "test-namespace", - Group: "test-group", - Resource: "test-resource", - } - - // Pre-populate the processor to avoid the builders.get() call - key := fmt.Sprintf("%s/%s/%s", nsr.Namespace, nsr.Group, nsr.Resource) - processor1 := newIndexQueueProcessor(mockIndex1, nsr, 10, mockBuilder, s.indexEventsChan) - s.indexQueueProcessors[key] = processor1 - - // Verify initial state - require.Same(t, mockIndex1, processor1.index) - - // Call getOrCreateIndexQueueProcessor with a different index - // This should return the existing processor but update its index - processor2, err := s.getOrCreateIndexQueueProcessor(mockIndex2, nsr) - require.NoError(t, err) - require.NotNil(t, processor2) - - // Same processor instance, but index was replaced - require.Same(t, processor1, processor2, "Should return the same processor instance") - require.Same(t, mockIndex2, processor2.index, "Index should be replaced with mockIndex2") - require.Same(t, mockIndex2, processor1.index, "Original processor should have updated index") -} - type slowSearchBackendWithCache struct { mockSearchBackend wg sync.WaitGroup @@ -538,7 +365,7 @@ func (m *slowSearchBackendWithCache) GetIndex(ctx context.Context, key Namespace return m.cache[key], nil } -func (m *slowSearchBackendWithCache) BuildIndex(ctx context.Context, key NamespacedResource, size int64, resourceVersion int64, fields SearchableDocumentFields, reason string, builder BuildFn, updater UpdateFn, rebuild bool, searchAfterWrite bool) (ResourceIndex, error) { +func (m *slowSearchBackendWithCache) BuildIndex(ctx context.Context, key NamespacedResource, size int64, fields SearchableDocumentFields, reason string, builder BuildFn, updater UpdateFn, rebuild bool) (ResourceIndex, error) { m.wg.Add(1) defer m.wg.Done() @@ -548,7 +375,7 @@ func (m *slowSearchBackendWithCache) BuildIndex(ctx context.Context, key Namespa if ctx.Err() != nil { return nil, ctx.Err() } - idx, err := m.mockSearchBackend.BuildIndex(ctx, key, size, resourceVersion, fields, reason, builder, updater, rebuild, searchAfterWrite) + idx, err := m.mockSearchBackend.BuildIndex(ctx, key, size, fields, reason, builder, updater, rebuild) if err != nil { return nil, err } diff --git a/pkg/storage/unified/resource/server.go b/pkg/storage/unified/resource/server.go index 18412d4a853..62f17050540 100644 --- a/pkg/storage/unified/resource/server.go +++ b/pkg/storage/unified/resource/server.go @@ -22,6 +22,7 @@ import ( claims "github.com/grafana/authlib/types" "github.com/grafana/dskit/backoff" "github.com/grafana/dskit/ring" + "github.com/grafana/grafana/pkg/apimachinery/utils" secrets "github.com/grafana/grafana/pkg/registry/apis/secret/contracts" "github.com/grafana/grafana/pkg/storage/unified/resourcepb" @@ -182,13 +183,6 @@ type SearchOptions struct { // Skip building index on startup for small indexes InitMinCount int - // Build empty index on startup for large indexes so that - // we don't re-attempt to build the index later. - InitMaxCount int - - // Channel to watch for index events (for testing) - IndexEventsChan chan *IndexEvent - // Interval for periodic index rebuilds (0 disables periodic rebuilds) RebuildInterval time.Duration @@ -243,9 +237,6 @@ type ResourceServerOptions struct { Ring *ring.Ring RingLifecycler *ring.BasicLifecycler - - // Enable strong consistency for searches. When enabled, index is always updated with latest changes before search. - SearchAfterWrite bool } func NewResourceServer(opts ResourceServerOptions) (*server, error) { @@ -343,7 +334,7 @@ func NewResourceServer(opts ResourceServerOptions) (*server, error) { if opts.Search.Resources != nil { var err error - s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.Tracer, opts.IndexMetrics, opts.Ring, opts.RingLifecycler, opts.SearchAfterWrite) + s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.Tracer, opts.IndexMetrics, opts.Ring, opts.RingLifecycler) if err != nil { return nil, err } diff --git a/pkg/storage/unified/search/bleve.go b/pkg/storage/unified/search/bleve.go index 44fdffe27a8..88ddca29ee5 100644 --- a/pkg/storage/unified/search/bleve.go +++ b/pkg/storage/unified/search/bleve.go @@ -213,8 +213,7 @@ func newBleveIndex(path string, mapper mapping.IndexMapping) (bleve.Index, error // 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). -// An index in the file system is considered to be valid if the requested resourceVersion is smaller than or equal to -// the resourceVersion used to build the index and the number of indexed objects matches the expected size. +// 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. // The return value of "builder" should be the RV returned from List. This will be stored as the index RV // //nolint:gocyclo @@ -222,13 +221,11 @@ func (b *bleveBackend) BuildIndex( ctx context.Context, key resource.NamespacedResource, size int64, - resourceVersion int64, fields resource.SearchableDocumentFields, indexBuildReason string, builder resource.BuildFn, updater resource.UpdateFn, rebuild bool, - searchAfterWrite bool, ) (resource.ResourceIndex, error) { _, span := b.tracer.Start(ctx, tracingPrexfixBleve+"BuildIndex") defer span.End() @@ -238,7 +235,6 @@ func (b *bleveBackend) BuildIndex( attribute.String("group", key.Group), attribute.String("resource", key.Resource), attribute.Int64("size", size), - attribute.Int64("rv", resourceVersion), attribute.String("reason", indexBuildReason), ) @@ -254,7 +250,7 @@ func (b *bleveBackend) BuildIndex( return nil, err } - logWithDetails := b.log.With("namespace", key.Namespace, "group", key.Group, "resource", key.Resource, "size", size, "rv", resourceVersion, "reason", indexBuildReason) + logWithDetails := b.log.With("namespace", key.Namespace, "group", key.Group, "resource", key.Resource, "size", size, "reason", indexBuildReason) // Close the newly created/opened index by default. closeIndex := true @@ -290,8 +286,8 @@ func (b *bleveBackend) BuildIndex( // 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. - if cachedIndex == nil && resourceVersion > 0 && !rebuild { - index, fileIndexName, indexRV = b.findPreviousFileBasedIndex(resourceDir, resourceVersion, size, searchAfterWrite) + if cachedIndex == nil && !rebuild { + index, fileIndexName, indexRV = b.findPreviousFileBasedIndex(resourceDir) } if index != nil { @@ -509,7 +505,7 @@ func formatIndexName(now time.Time) string { return now.Format("20060102-150405") } -func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string, resourceVersion int64, size int64, searchAfterWrite bool) (bleve.Index, string, int64) { +func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) (bleve.Index, string, int64) { entries, err := os.ReadDir(resourceDir) if err != nil { return nil, "", 0 @@ -528,21 +524,6 @@ func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string, resourceVe continue } - if !searchAfterWrite { - cnt, err := idx.DocCount() - if err != nil { - b.log.Debug("error getting count from index", "indexDir", indexDir, "err", err) - _ = idx.Close() - continue - } - - if uint64(size) != cnt { - b.log.Debug("index count mismatch. ignoring index", "indexDir", indexDir, "size", size, "cnt", cnt) - _ = idx.Close() - continue - } - } - indexRV, err := getRV(idx) if err != nil { b.log.Error("error getting rv from index", "indexDir", indexDir, "err", err) @@ -552,13 +533,6 @@ func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string, resourceVe continue } - // if searchAfterWrite is enabled, we don't need to re-build the index, as it will be updated at request time - if !searchAfterWrite && indexRV < resourceVersion { - b.log.Debug("indexRV is less than requested resourceVersion. ignoring index", "indexDir", indexDir, "rv", indexRV, "resourceVersion", resourceVersion) - _ = idx.Close() - continue - } - return idx, indexName, indexRV } diff --git a/pkg/storage/unified/search/bleve_search_test.go b/pkg/storage/unified/search/bleve_search_test.go index 423355b28c5..250001dc824 100644 --- a/pkg/storage/unified/search/bleve_search_test.go +++ b/pkg/storage/unified/search/bleve_search_test.go @@ -237,7 +237,6 @@ func newTestDashboardsIndex(t testing.TB, threshold int64, size int64, batchSize t.Cleanup(backend.CloseAllIndexes) - rv := int64(10) ctx := identity.WithRequester(context.Background(), &user.SignedInUser{Namespace: "ns"}) info, err := search.DashboardBuilder(func(ctx context.Context, namespace string, blob resource.BlobSupport) (resource.DocumentBuilder, error) { @@ -254,7 +253,7 @@ func newTestDashboardsIndex(t testing.TB, threshold int64, size int64, batchSize Namespace: key.Namespace, Group: key.Group, Resource: key.Resource, - }, size, rv, info.Fields, "test", writer, nil, false, false) + }, size, info.Fields, "test", writer, nil, false) require.NoError(t, err) return index diff --git a/pkg/storage/unified/search/bleve_test.go b/pkg/storage/unified/search/bleve_test.go index 9af5a0c1fd5..b3c140cbdd6 100644 --- a/pkg/storage/unified/search/bleve_test.go +++ b/pkg/storage/unified/search/bleve_test.go @@ -89,7 +89,7 @@ func TestBleveBackend(t *testing.T) { Namespace: key.Namespace, Group: key.Group, Resource: key.Resource, - }, 2, rv, info.Fields, "test", func(index resource.ResourceIndex) (int64, error) { + }, 2, info.Fields, "test", func(index resource.ResourceIndex) (int64, error) { err := index.BulkIndex(&resource.BulkIndexRequest{ Items: []*resource.BulkIndexItem{ { @@ -192,7 +192,7 @@ func TestBleveBackend(t *testing.T) { return 0, err } return rv, nil - }, nil, false, false) + }, nil, false) require.NoError(t, err) require.NotNil(t, index) dashboardsIndex = index @@ -370,7 +370,7 @@ func TestBleveBackend(t *testing.T) { Namespace: key.Namespace, Group: key.Group, Resource: key.Resource, - }, 2, rv, fields, "test", func(index resource.ResourceIndex) (int64, error) { + }, 2, fields, "test", func(index resource.ResourceIndex) (int64, error) { err := index.BulkIndex(&resource.BulkIndexRequest{ Items: []*resource.BulkIndexItem{ { @@ -421,7 +421,7 @@ func TestBleveBackend(t *testing.T) { return 0, err } return rv, nil - }, nil, false, false) + }, nil, false) require.NoError(t, err) require.NotNil(t, index) foldersIndex = index @@ -786,7 +786,7 @@ func TestBuildIndexExpiration(t *testing.T) { t.Run("memory based indexes should expire", func(t *testing.T) { backend, reg := setupBleveBackend(t, 5, time.Nanosecond, "") - builtIndex, err := backend.BuildIndex(context.Background(), ns, 1 /* below FileThreshold */, 100, nil, "test", indexTestDocs(ns, 1, 100), nil, false, false) + builtIndex, err := backend.BuildIndex(context.Background(), ns, 1 /* below FileThreshold */, nil, "test", indexTestDocs(ns, 1, 100), nil, false) require.NoError(t, err) // Wait for index expiration, which is 1ns @@ -807,7 +807,7 @@ func TestBuildIndexExpiration(t *testing.T) { backend, reg := setupBleveBackend(t, 5, time.Nanosecond, "") // size=100 is above FileThreshold, this will be file-based index - builtIndex, err := backend.BuildIndex(context.Background(), ns, 100, 100, nil, "test", indexTestDocs(ns, 1, 100), nil, false, false) + builtIndex, err := backend.BuildIndex(context.Background(), ns, 100, nil, "test", indexTestDocs(ns, 1, 100), nil, false) require.NoError(t, err) // Wait for index expiration, which is 1ns @@ -839,9 +839,9 @@ func TestCloseAllIndexes(t *testing.T) { tmpDir := t.TempDir() backend1, reg := setupBleveBackend(t, 5, time.Nanosecond, tmpDir) - _, err := backend1.BuildIndex(context.Background(), ns, 10 /* file based */, 100, nil, "test", indexTestDocs(ns, 10, 100), nil, false, false) + _, 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 */, 100, nil, "test", indexTestDocs(ns, 10, 100), nil, false, false) + _, err = backend1.BuildIndex(context.Background(), ns2, 1 /* memory based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false) require.NoError(t, err) // Verify two open indexes. @@ -865,91 +865,76 @@ func TestBuildIndex(t *testing.T) { RVBiggerThan RV = "more" RVSame RV = "same" ) - for _, searchAfterWrite := range []bool{false, true} { - for _, rebuild := range []bool{false, true} { - for _, sameSize := range []bool{false, true} { - for _, documentRV := range []RV{RVLessThan, RVSame, RVBiggerThan} { - shouldRebuild := false - if rebuild { - shouldRebuild = true - } else if !searchAfterWrite && !sameSize { - shouldRebuild = true - } else if !searchAfterWrite && documentRV == RVBiggerThan { - shouldRebuild = true - } + 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 " - } - - if searchAfterWrite { - testName += "and searchAfterWrite is true" - } else { - testName += "and searchAfterWrite 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, searchAfterWrite) - backend1.CloseAllIndexes() - - if !sameSize { - size = 11 - } - switch documentRV { - case RVBiggerThan: - rv = 101 - case RVLessThan: - rv = 99 - case RVSame: - } - backend2, idx := createBleveBackendAndIndex(t, tmpDir, ns, size, rv, 1000, rebuild, searchAfterWrite) - - cnt, err := idx.DocCount(context.Background(), "") - require.NoError(t, err) - if shouldRebuild { - require.Equal(t, int64(1000), cnt, "Index has been not rebuilt") - } else { - require.Equal(t, int64(10), cnt, "Index has not been reused") - } - backend2.CloseAllIndexes() - }) + 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 + } + switch documentRV { + case RVBiggerThan: + rv = 101 + case RVLessThan: + rv = 99 + case RVSame: + } + backend2, idx := createBleveBackendAndIndex(t, tmpDir, ns, size, rv, 1000, rebuild) + + cnt, err := idx.DocCount(context.Background(), "") + require.NoError(t, err) + if shouldRebuild { + require.Equal(t, int64(1000), cnt, "Index has been not rebuilt") + } else { + require.Equal(t, int64(10), cnt, "Index has not been reused") + } + backend2.CloseAllIndexes() + }) } } } } -func createBleveBackendAndIndex(t *testing.T, tmpDir string, ns resource.NamespacedResource, size, rv int64, docCount int, rebuild, searchAfterWrite bool) (*bleveBackend, resource.ResourceIndex) { +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 */, rv, nil, "test", indexTestDocs(ns, docCount, rv), nil, rebuild, searchAfterWrite) + 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 } @@ -977,7 +962,7 @@ func TestRebuildingIndexClosesPreviousCachedIndex(t *testing.T) { if testCase.firstInMemory { firstSize = 1 } - firstIndex, err := backend.BuildIndex(context.Background(), ns, int64(firstSize), 100, nil, "test", indexTestDocs(ns, firstSize, 100), nil, false, false) + firstIndex, err := backend.BuildIndex(context.Background(), ns, int64(firstSize), nil, "test", indexTestDocs(ns, firstSize, 100), nil, false) require.NoError(t, err) if testCase.firstInMemory { @@ -993,7 +978,7 @@ func TestRebuildingIndexClosesPreviousCachedIndex(t *testing.T) { secondSize = 1 openInMemoryIndexes = 1 } - secondIndex, err := backend.BuildIndex(context.Background(), ns, int64(secondSize), 100, nil, "test", indexTestDocs(ns, secondSize, 100), nil, false, false) + secondIndex, err := backend.BuildIndex(context.Background(), ns, int64(secondSize), nil, "test", indexTestDocs(ns, secondSize, 100), nil, false) require.NoError(t, err) if testCase.secondInMemory { @@ -1142,13 +1127,13 @@ func testBleveIndexWithFailures(t *testing.T, fileBased bool) { // size=100 is above FileThreshold (5), make it a file-based index. size = 100 } - _, err := backend.BuildIndex(context.Background(), ns, size, 100, nil, "test", func(index resource.ResourceIndex) (int64, error) { + _, err := backend.BuildIndex(context.Background(), ns, size, nil, "test", func(index resource.ResourceIndex) (int64, error) { return 0, fmt.Errorf("fail") - }, nil, false, false) + }, nil, false) require.Error(t, err) // Even though previous build of the index failed, new building of the index should work. - _, err = backend.BuildIndex(context.Background(), ns, size, 100, nil, "test", indexTestDocs(ns, int(size), 100), nil, false, false) + _, err = backend.BuildIndex(context.Background(), ns, size, nil, "test", indexTestDocs(ns, int(size), 100), nil, false) require.NoError(t, err) } @@ -1160,7 +1145,7 @@ func TestIndexUpdate(t *testing.T) { } be, _ := setupBleveBackend(t, 5, 1*time.Minute, "") - idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, 100, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false, false) + idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false) require.NoError(t, err) resp := searchTitle(t, idx, "gen", 10, ns) @@ -1214,7 +1199,7 @@ func TestConcurrentIndexUpdateAndBuildIndex(t *testing.T) { return sinceRV + int64(5), 5, err } - idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, 100, nil, "test", indexTestDocs(ns, 10, 100), updaterFn, false, false) + idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updaterFn, false) require.NoError(t, err) ctx, cancel := context.WithCancel(context.Background()) @@ -1222,7 +1207,7 @@ func TestConcurrentIndexUpdateAndBuildIndex(t *testing.T) { _, err = idx.UpdateIndex(ctx, "test") require.NoError(t, err) - _, err = be.BuildIndex(t.Context(), ns, 10 /* file based */, 100, nil, "test", indexTestDocs(ns, 10, 100), updaterFn, false, false) + _, err = be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updaterFn, false) require.NoError(t, err) _, err = idx.UpdateIndex(ctx, "test") @@ -1238,7 +1223,7 @@ func TestConcurrentIndexUpdateSearchAndRebuild(t *testing.T) { be, _ := setupBleveBackend(t, 5, 1*time.Minute, "") - _, err := be.BuildIndex(t.Context(), ns, 10, 0, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false, false) + _, err := be.BuildIndex(t.Context(), ns, 10, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false) require.NoError(t, err) wg := sync.WaitGroup{} @@ -1301,7 +1286,7 @@ func TestConcurrentIndexUpdateSearchAndRebuild(t *testing.T) { go func() { defer wg.Done() for ctx.Err() == nil { - _, err := be.BuildIndex(t.Context(), ns, 10, 0, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false, false) + _, err := be.BuildIndex(t.Context(), ns, 10, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false) require.NoError(t, err) rebuilds.Inc() } @@ -1324,7 +1309,7 @@ func TestConcurrentIndexUpdateAndSearch(t *testing.T) { be, _ := setupBleveBackend(t, 5, 1*time.Minute, "") - idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, 100, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false, false) + idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false) require.NoError(t, err) wg := sync.WaitGroup{} @@ -1388,7 +1373,7 @@ func TestIndexUpdateWithErrors(t *testing.T) { time.Sleep(100 * time.Millisecond) return 0, 0, updateErr } - idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, 100, nil, "test", indexTestDocs(ns, 10, 100), updaterFn, false, false) + idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updaterFn, false) require.NoError(t, err) t.Run("update fail", func(t *testing.T) { diff --git a/pkg/storage/unified/search/options.go b/pkg/storage/unified/search/options.go index 0093da44264..d261a445a12 100644 --- a/pkg/storage/unified/search/options.go +++ b/pkg/storage/unified/search/options.go @@ -39,7 +39,6 @@ func NewSearchOptions(features featuremgmt.FeatureToggles, cfg *setting.Cfg, tra Resources: docs, WorkerThreads: cfg.IndexWorkers, InitMinCount: cfg.IndexMinCount, - InitMaxCount: cfg.IndexMaxCount, RebuildInterval: cfg.IndexRebuildInterval, }, nil } diff --git a/pkg/storage/unified/sql/server.go b/pkg/storage/unified/sql/server.go index 4037e4253c6..b2debf0acdc 100644 --- a/pkg/storage/unified/sql/server.go +++ b/pkg/storage/unified/sql/server.go @@ -12,6 +12,7 @@ import ( "github.com/grafana/authlib/types" "github.com/grafana/dskit/ring" "github.com/grafana/dskit/services" + infraDB "github.com/grafana/grafana/pkg/infra/db" secrets "github.com/grafana/grafana/pkg/registry/apis/secret/contracts" inlinesecurevalue "github.com/grafana/grafana/pkg/registry/apis/secret/inline" @@ -119,7 +120,6 @@ func NewResourceServer( serverOptions.QOSQueue = opts.QOSQueue serverOptions.Ring = opts.Ring serverOptions.RingLifecycler = opts.RingLifecycler - serverOptions.SearchAfterWrite = opts.Features.IsEnabledGlobally(featuremgmt.FlagUnifiedStorageSearchAfterWriteExperimentalAPI) return resource.NewResourceServer(serverOptions) } diff --git a/pkg/storage/unified/sql/test/benchmark_test.go b/pkg/storage/unified/sql/test/benchmark_test.go index 540b17c0bbf..128a344d891 100644 --- a/pkg/storage/unified/sql/test/benchmark_test.go +++ b/pkg/storage/unified/sql/test/benchmark_test.go @@ -2,18 +2,14 @@ package test import ( "context" - "os" "testing" "time" "github.com/stretchr/testify/require" "github.com/grafana/grafana/pkg/infra/db" - "github.com/grafana/grafana/pkg/infra/tracing" - "github.com/grafana/grafana/pkg/services/featuremgmt" "github.com/grafana/grafana/pkg/setting" "github.com/grafana/grafana/pkg/storage/unified/resource" - "github.com/grafana/grafana/pkg/storage/unified/search" "github.com/grafana/grafana/pkg/storage/unified/sql" "github.com/grafana/grafana/pkg/storage/unified/sql/db/dbimpl" test "github.com/grafana/grafana/pkg/storage/unified/testing" @@ -46,48 +42,3 @@ func TestIntegrationBenchmarkSQLStorageBackend(t *testing.T) { } test.BenchmarkStorageBackend(t, newTestBackend(t), opts) } - -func TestIntegrationBenchmarkResourceServer(t *testing.T) { - t.Skip("skipping slow test, causing CI to fail due to timeout") - - testutil.SkipIntegrationTestInShortMode(t) - - ctx := context.Background() - opts := &test.BenchmarkOptions{ - NumResources: 1000, - Concurrency: 10, - NumNamespaces: 1, - NumGroups: 1, - NumResourceTypes: 1, - } - tempDir := t.TempDir() - t.Cleanup(func() { - _ = os.RemoveAll(tempDir) - }) - // Create a new bleve backend - search, err := search.NewBleveBackend(search.BleveOptions{ - Root: tempDir, - }, tracing.NewNoopTracerService(), featuremgmt.WithFeatures(), nil) - require.NoError(t, err) - require.NotNil(t, search) - - t.Cleanup(search.CloseAllIndexes) - - // Create a new resource backend - dbstore := db.InitTestDB(t) - eDB, err := dbimpl.ProvideResourceDB(dbstore, setting.NewCfg(), nil) - require.NoError(t, err) - require.NotNil(t, eDB) - - storage, err := sql.NewBackend(sql.BackendOptions{ - DBProvider: eDB, - IsHA: false, - }) - require.NoError(t, err) - require.NotNil(t, storage) - - err = storage.Init(ctx) - require.NoError(t, err) - - test.BenchmarkIndexServer(t, ctx, storage, search, opts) -} diff --git a/pkg/storage/unified/testing/benchmark.go b/pkg/storage/unified/testing/benchmark.go index b43c8f0ad8b..b8bf86c1e9d 100644 --- a/pkg/storage/unified/testing/benchmark.go +++ b/pkg/storage/unified/testing/benchmark.go @@ -215,9 +215,9 @@ func runSearchBackendBenchmarkWriteThroughput(ctx context.Context, backend resou // Build initial index size := int64(10000) // force the index to be on disk - index, err := backend.BuildIndex(ctx, nr, size, 0, nil, "benchmark", func(index resource.ResourceIndex) (int64, error) { + index, err := backend.BuildIndex(ctx, nr, size, nil, "benchmark", func(index resource.ResourceIndex) (int64, error) { return 0, nil - }, nil, false, false) + }, nil, false) if err != nil { return nil, fmt.Errorf("failed to initialize backend: %w", err) } @@ -334,91 +334,3 @@ func BenchmarkSearchBackend(tb testing.TB, backend resource.SearchBackend, opts tb.Logf("P90 Latency: %v", result.P90Latency) tb.Logf("P99 Latency: %v", result.P99Latency) } - -func BenchmarkIndexServer(tb testing.TB, ctx context.Context, backend resource.StorageBackend, searchBackend resource.SearchBackend, opts *BenchmarkOptions) { - events := make(chan *resource.IndexEvent, opts.NumResources) - groupsResources := make(map[string]string) - for g := 0; g < opts.NumGroups; g++ { - for r := 0; r < opts.NumResourceTypes; r++ { - groupsResources[fmt.Sprintf("group-%d", g)] = fmt.Sprintf("resource-%d", r) - } - } - server, err := resource.NewResourceServer(resource.ResourceServerOptions{ - Backend: backend, - Search: resource.SearchOptions{ - Backend: searchBackend, - IndexEventsChan: events, - Resources: &resource.TestDocumentBuilderSupplier{GroupsResources: groupsResources}, - }, - }) - require.NoError(tb, err) - require.NotNil(tb, server) - - // Initialize the backend - err = initializeBackend(ctx, backend, opts) - require.NoError(tb, err) - - // Discard the latencies from the initial index build. - for i := 0; i < (opts.NumGroups * opts.NumResourceTypes * opts.NumNamespaces); i++ { - <-events - } - - // Run the storage backend benchmark write throughput to create events - startTime := time.Now() - var result *BenchmarkResult - // Channel to signal when the benchmark goroutine completes - benchmarkDone := make(chan struct{}) - - go func() { - defer close(benchmarkDone) - result, err = runStorageBackendBenchmark(ctx, backend, opts) - require.NoError(tb, err) - }() - - // Wait for all events to be processed - latencies := make([]float64, 0, opts.NumResources) - for i := 0; i < opts.NumResources; i++ { - evt := <-events - latencies = append(latencies, evt.Latency.Seconds()) - } - totalDuration := time.Since(startTime) - - <-benchmarkDone - // Calculate index latency percentiles - sort.Float64s(latencies) - var p50, p90, p99 float64 - if len(latencies) > 0 { - p50 = latencies[len(latencies)*50/100] - p90 = latencies[len(latencies)*90/100] - p99 = latencies[len(latencies)*99/100] - } - - // Report metrics if running a benchmark - if b, ok := tb.(*testing.B); ok { - b.ReportMetric(result.Throughput, "writes/sec") - b.ReportMetric(float64(result.P50Latency.Milliseconds()), "p50-latency-ms") - b.ReportMetric(float64(result.P90Latency.Milliseconds()), "p90-latency-ms") - b.ReportMetric(float64(result.P99Latency.Milliseconds()), "p99-latency-ms") - b.ReportMetric(p50, "p50-index-latency-s") - b.ReportMetric(p90, "p90-index-latency-s") - b.ReportMetric(p99, "p99-index-latency-s") - } - - // Log results for better visibility - tb.Logf("Benchmark Configuration: Workers=%d, Resources=%d, Namespaces=%d, Groups=%d, Resource Types=%d", - opts.Concurrency, opts.NumResources, opts.NumNamespaces, opts.NumGroups, opts.NumResourceTypes) - tb.Logf("") - tb.Logf("Storage Benchmark Results:") - tb.Logf("Total Duration: %v", result.TotalDuration) - tb.Logf("Storage Write Count: %d", result.WriteCount) - tb.Logf("Storage Write Throughput: %.2f writes/sec", result.Throughput) - tb.Logf("P50 Write Latency: %v", result.P50Latency) - tb.Logf("P90 Write Latency: %v", result.P90Latency) - tb.Logf("P99 Write Latency: %v", result.P99Latency) - tb.Logf("") - tb.Logf("Index Latency Results:") - tb.Logf("Indexing Throughput: %.2f events/sec", float64(len(latencies))/totalDuration.Seconds()) - tb.Logf("P50 Index Latency: %.3fs", p50) - tb.Logf("P90 Index Latency: %.3fs", p90) - tb.Logf("P99 Index Latency: %.3fs", p99) -} diff --git a/pkg/storage/unified/testing/search_and_storage.go b/pkg/storage/unified/testing/search_and_storage.go index 7a15223c0fa..5cb4d13488c 100644 --- a/pkg/storage/unified/testing/search_and_storage.go +++ b/pkg/storage/unified/testing/search_and_storage.go @@ -94,7 +94,6 @@ func RunTestSearchAndStorage(t *testing.T, ctx context.Context, backend resource require.Greater(t, rv, int64(0)) } }) - ch := make(chan *resource.IndexEvent) t.Run("Create a resource server with both backends", func(t *testing.T) { // Create a resource server with both backends @@ -108,7 +107,6 @@ func RunTestSearchAndStorage(t *testing.T, ctx context.Context, backend resource "test.grafana.app": "testresources", }, }, - IndexEventsChan: ch, }, }) require.NoError(t, err) @@ -192,9 +190,6 @@ func RunTestSearchAndStorage(t *testing.T, ctx context.Context, backend resource require.NoError(t, err) require.NotNil(t, createResp) require.Nil(t, createResp.Error) - - ev := <-ch - require.NotNil(t, ev) } }) diff --git a/pkg/storage/unified/testing/search_backend.go b/pkg/storage/unified/testing/search_backend.go index 704d608316b..301a3a202e4 100644 --- a/pkg/storage/unified/testing/search_backend.go +++ b/pkg/storage/unified/testing/search_backend.go @@ -64,7 +64,7 @@ func runTestSearchBackendBuildIndex(t *testing.T, backend resource.SearchBackend require.Nil(t, index) // Build the index - index, err = backend.BuildIndex(ctx, ns, 0, 0, nil, "test", func(index resource.ResourceIndex) (int64, error) { + index, err = backend.BuildIndex(ctx, ns, 0, nil, "test", func(index resource.ResourceIndex) (int64, error) { // Write a test document err := index.BulkIndex(&resource.BulkIndexRequest{ Items: []*resource.BulkIndexItem{ @@ -86,7 +86,7 @@ func runTestSearchBackendBuildIndex(t *testing.T, backend resource.SearchBackend return 0, err } return 1, nil - }, nil, false, false) + }, nil, false) require.NoError(t, err) require.NotNil(t, index) @@ -111,7 +111,7 @@ func runTestResourceIndex(t *testing.T, backend resource.SearchBackend, nsPrefix } // Build initial index with some test documents - index, err := backend.BuildIndex(ctx, ns, 3, 0, nil, "test", func(index resource.ResourceIndex) (int64, error) { + index, err := backend.BuildIndex(ctx, ns, 3, nil, "test", func(index resource.ResourceIndex) (int64, error) { err := index.BulkIndex(&resource.BulkIndexRequest{ Items: []*resource.BulkIndexItem{ { @@ -152,7 +152,7 @@ func runTestResourceIndex(t *testing.T, backend resource.SearchBackend, nsPrefix }) require.NoError(t, err) return int64(2), nil - }, nil, false, false) + }, nil, false) require.NoError(t, err) require.NotNil(t, index) @@ -235,7 +235,7 @@ func runTestResourceIndex(t *testing.T, backend resource.SearchBackend, nsPrefix t.Run("Search by LibraryPanel reference", func(t *testing.T) { // Build index with dashboards that have LibraryPanel references - index, err := backend.BuildIndex(ctx, ns, 3, 0, nil, "test", func(index resource.ResourceIndex) (int64, error) { + index, err := backend.BuildIndex(ctx, ns, 3, nil, "test", func(index resource.ResourceIndex) (int64, error) { err := index.BulkIndex(&resource.BulkIndexRequest{ Items: []*resource.BulkIndexItem{ { @@ -294,7 +294,7 @@ func runTestResourceIndex(t *testing.T, backend resource.SearchBackend, nsPrefix }) require.NoError(t, err) return int64(3), nil - }, nil, false, false) + }, nil, false) require.NoError(t, err) require.NotNil(t, index)