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

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

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