fix: use memory index if index file already open (#115720)

* feat: add lock structure into bleve index files

* fix: another approach

* fix: new check

* fix: build in memory if index file already open

* fix: update workspace

* fix: add test

* refactor: update func signature

* fix: address comments

* fix: make const

pkg/storage/unified/search/bleve.go

@@ -25,6 +25,7 @@ import (
 	bleveSearch "github.com/blevesearch/bleve/v2/search/searcher"
 	index "github.com/blevesearch/bleve_index_api"
 	"github.com/prometheus/client_golang/prometheus"
+	bolterrors "go.etcd.io/bbolt/errors"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.uber.org/atomic"
@@ -44,6 +45,7 @@ import (
 const (
 	indexStorageMemory = "memory"
 	indexStorageFile   = "file"
+	boltTimeout        = "500ms"
 )
 
 // Keys used to store internal data in index.
@@ -415,14 +417,25 @@ func (b *bleveBackend) BuildIndex(
 		// 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 && !rebuild {
-			index, fileIndexName, indexRV = b.findPreviousFileBasedIndex(resourceDir)
+			result := b.findPreviousFileBasedIndex(resourceDir)
+			if result != nil && result.IsOpen {
+				// Index file exists but is opened by another process, fallback to memory.
+				// Keep the name so we can skip cleanup of that directory.
+				newIndexType = indexStorageMemory
+				fileIndexName = result.Name
+			} else if result != nil && result.Index != nil {
+				// Found and opened existing index successfully
+				index = result.Index
+				fileIndexName = result.Name
+				indexRV = result.RV
+			}
 		}
 
-		if index != nil {
+		if newIndexType == indexStorageFile && index != nil {
 			build = false
 			logWithDetails.Debug("Existing index found on filesystem", "indexRV", indexRV, "directory", filepath.Join(resourceDir, fileIndexName))
 			defer closeIndexOnExit(index, "") // Close index, but don't delete directory.
-		} else {
+		} else if newIndexType == indexStorageFile {
 			// Building index from scratch. Index name has a time component in it to be unique, but if
 			// we happen to create non-unique name, we bump the time and try again.
 
@@ -449,7 +462,9 @@ func (b *bleveBackend) BuildIndex(
 			logWithDetails.Info("Building index using filesystem", "directory", indexDir)
 			defer closeIndexOnExit(index, indexDir) // Close index, and delete new index directory.
 		}
-	} else {
+	}
+
+	if newIndexType == indexStorageMemory {
 		index, err = newBleveIndex("", mapper, time.Now(), b.opts.BuildVersion)
 		if err != nil {
 			return nil, fmt.Errorf("error creating new in-memory bleve index: %w", err)
@@ -552,30 +567,30 @@ func cleanFileSegment(input string) string {
 	return input
 }
 
-// cleanOldIndexes deletes all subdirectories inside dir, skipping directory with "skipName".
+// cleanOldIndexes deletes all subdirectories inside resourceDir, skipping directory with "skipName".
 // "skipName" can be empty.
-func (b *bleveBackend) cleanOldIndexes(dir string, skipName string) {
-	files, err := os.ReadDir(dir)
+func (b *bleveBackend) cleanOldIndexes(resourceDir string, skipName string) {
+	entries, err := os.ReadDir(resourceDir)
 	if err != nil {
 		if os.IsNotExist(err) {
 			return
 		}
-		b.log.Warn("error cleaning folders from", "directory", dir, "error", err)
+		b.log.Warn("error cleaning folders from", "directory", resourceDir, "error", err)
 		return
 	}
-	for _, file := range files {
-		if file.IsDir() && file.Name() != skipName {
-			fpath := filepath.Join(dir, file.Name())
-			if !isPathWithinRoot(fpath, b.opts.Root) {
-				b.log.Warn("Skipping cleanup of directory", "directory", fpath)
+	for _, ent := range entries {
+		if ent.IsDir() && ent.Name() != skipName {
+			indexDir := filepath.Join(resourceDir, ent.Name())
+			if !isPathWithinRoot(indexDir, b.opts.Root) {
+				b.log.Warn("Skipping cleanup of directory", "directory", indexDir)
 				continue
 			}
 
-			err = os.RemoveAll(fpath)
+			err = os.RemoveAll(indexDir)
 			if err != nil {
-				b.log.Error("Unable to remove old index folder", "directory", fpath, "error", err)
+				b.log.Error("Unable to remove old index folder", "directory", indexDir, "error", err)
 			} else {
-				b.log.Info("Removed old index folder", "directory", fpath)
+				b.log.Info("Removed old index folder", "directory", indexDir)
 			}
 		}
 	}
@@ -622,10 +637,17 @@ func formatIndexName(now time.Time) string {
 	return now.Format("20060102-150405")
 }
 
-func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) (bleve.Index, string, int64) {
+type fileIndex struct {
+	Index  bleve.Index
+	Name   string
+	RV     int64
+	IsOpen bool
+}
+
+func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) *fileIndex {
 	entries, err := os.ReadDir(resourceDir)
 	if err != nil {
-		return nil, "", 0
+		return nil
 	}
 
 	for _, ent := range entries {
@@ -635,8 +657,13 @@ func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) (bleve.Ind
 
 		indexName := ent.Name()
 		indexDir := filepath.Join(resourceDir, indexName)
-		idx, err := bleve.Open(indexDir)
+
+		idx, err := bleve.OpenUsing(indexDir, map[string]interface{}{"bolt_timeout": boltTimeout})
 		if err != nil {
+			if errors.Is(err, bolterrors.ErrTimeout) {
+				b.log.Debug("Index is opened by another process (timeout), skipping", "indexDir", indexDir)
+				return &fileIndex{Name: indexName, IsOpen: true}
+			}
 			b.log.Debug("error opening index", "indexDir", indexDir, "err", err)
 			continue
 		}
@@ -648,10 +675,14 @@ func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) (bleve.Ind
 			continue
 		}
 
-		return idx, indexName, indexRV
+		return &fileIndex{
+			Index: idx,
+			Name:  indexName,
+			RV:    indexRV,
+		}
 	}
 
-	return nil, "", 0
+	return nil
 }
 
 // Stop closes all indexes and stops background tasks.