unified-storage: disable watcher when search-after-write feature flag is enabled (#109634)

* disable indexer watcher when search-after-write feature flag is enabled

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.

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)
 

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
 		}

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)
 }