From c07c0f27d2c33ca8bf00d243844aad3270cf62d5 Mon Sep 17 00:00:00 2001 From: owensmallwood Date: Fri, 8 Aug 2025 12:53:01 -0600 Subject: [PATCH] Unified Storage: Makes writing and reading index thread-safe in the queue processor (#109420) makes writing and reading index thread-safe --- pkg/storage/unified/resource/search.go | 2 +- pkg/storage/unified/resource/search_queue.go | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/pkg/storage/unified/resource/search.go b/pkg/storage/unified/resource/search.go index f605acc7e80..5fa6ad34fb6 100644 --- a/pkg/storage/unified/resource/search.go +++ b/pkg/storage/unified/resource/search.go @@ -944,7 +944,7 @@ func (s *searchSupport) getOrCreateIndexQueueProcessor(index ResourceIndex, nsr 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.index = index + indexQueueProcessor.updateIndex(index) return indexQueueProcessor, nil } diff --git a/pkg/storage/unified/resource/search_queue.go b/pkg/storage/unified/resource/search_queue.go index 08f3e5eeac4..9c71d913222 100644 --- a/pkg/storage/unified/resource/search_queue.go +++ b/pkg/storage/unified/resource/search_queue.go @@ -33,6 +33,12 @@ type IndexEvent struct { Err error } +func (b *indexQueueProcessor) updateIndex(newIndex ResourceIndex) { + b.mu.Lock() + defer b.mu.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{ @@ -128,7 +134,10 @@ func (b *indexQueueProcessor) process(batch []*WrittenEvent) { req.Items = append(req.Items, item) } + b.mu.Lock() err := b.index.BulkIndex(req) + b.mu.Unlock() + if err != nil { for _, r := range resp { r.Err = err