diff --git a/pkg/storage/unified/resource/search.go b/pkg/storage/unified/resource/search.go index 5fa6ad34fb6..614d0dafb50 100644 --- a/pkg/storage/unified/resource/search.go +++ b/pkg/storage/unified/resource/search.go @@ -100,16 +100,17 @@ 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 + 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 ring *ring.Ring ringLifecycler *ring.BasicLifecycler @@ -133,7 +134,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) (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, searchAfterWrite bool) (support *searchSupport, err error) { // No backend search support if opts.Backend == nil { return nil, nil @@ -155,6 +156,7 @@ func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.A initWorkers: opts.WorkerThreads, initMinSize: opts.InitMinCount, initMaxSize: opts.InitMaxCount, + searchAfterWrite: searchAfterWrite, indexMetrics: indexMetrics, clientIndexEventsChan: opts.IndexEventsChan, indexEventsChan: make(chan *IndexEvent), @@ -479,26 +481,29 @@ func (s *searchSupport) init(ctx context.Context) error { span.AddEvent("namespaces indexed", trace.WithAttributes(attribute.Int("namespaced_indexed", totalBatchesIndexed))) - // Now start listening for new events watchctx := context.Background() // new context? - 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) + // 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 - go s.monitorIndexEvents(ctx) + // 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. diff --git a/pkg/storage/unified/resource/search_test.go b/pkg/storage/unified/resource/search_test.go index 5f614b5d9e9..07a95639ed6 100644 --- a/pkg/storage/unified/resource/search_test.go +++ b/pkg/storage/unified/resource/search_test.go @@ -248,7 +248,7 @@ func TestBuildIndexes_MaxCountThreshold(t *testing.T) { InitMaxCount: tt.initMaxSize, } - support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil) + support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil, false) require.NoError(t, err) require.NotNil(t, support) @@ -306,7 +306,7 @@ func TestSearchGetOrCreateIndex(t *testing.T) { InitMaxCount: 0, } - support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil) + support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil, false) require.NoError(t, err) require.NotNil(t, support) @@ -360,7 +360,7 @@ func TestSearchGetOrCreateIndexWithCancellation(t *testing.T) { InitMaxCount: 0, } - support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil) + support, err := newSearchSupport(opts, storage, nil, nil, noop.NewTracerProvider().Tracer("test"), nil, nil, nil, false) require.NoError(t, err) require.NotNil(t, support) diff --git a/pkg/storage/unified/resource/server.go b/pkg/storage/unified/resource/server.go index c41b187af16..1b52245157d 100644 --- a/pkg/storage/unified/resource/server.go +++ b/pkg/storage/unified/resource/server.go @@ -232,6 +232,9 @@ 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) { @@ -315,7 +318,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) + s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.Tracer, opts.IndexMetrics, opts.Ring, opts.RingLifecycler, opts.SearchAfterWrite) if err != nil { return nil, err } diff --git a/pkg/storage/unified/sql/server.go b/pkg/storage/unified/sql/server.go index 1eee524df75..6d2639007db 100644 --- a/pkg/storage/unified/sql/server.go +++ b/pkg/storage/unified/sql/server.go @@ -103,6 +103,7 @@ 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) }