public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(unified): err on timeout to open index (#115953)

Browse Source 
* fix(unified): default index path to ephemeral storage mount path

Signed-off-by: Rafael Paulovic <rafael.paulovic@grafana.com>

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

This reverts commit dc4c106e91.

* fix(unified): set index_path for tests

* chore(unified): re-add bolt open timeout and test for error handling

* chore(unified): return err on timeout

* chore(unified): revert changes to default, use DataPath if index_path not set

* chore(unified): add defaults.ini entry for unified_storage

This is needed to override using env. vars

* chore: revert unrelated diff

* chore: address code review comments

- reduce bolt timeout to 1s
- remove errIndexLocked err type
- add more information about index_path in defaults.ini

---------

Signed-off-by: Rafael Paulovic <rafael.paulovic@grafana.com>
This commit is contained in:
Rafael Bortolon Paulovic
2026-01-09 13:48:54 +01:00
committed by GitHub
parent 71a65e1f80
commit 1c5caeb987
4 changed files with 47 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
conf/defaults.ini
+7
View File
@@ -2281,3 +2281,10 @@ allow_image_rendering = true
# will check if there has been any changes to the repository not propagated by a webhook.
# The minimum value is 10 seconds.
min_sync_interval = 10s
#################################### Unified Storage ####################################
[unified_storage]
# index_path is the path where unified storage can store its index files for search.
# If empty, defaults to "<data_dir>/unified-search/bleve" (see [paths] section).
# Please note that sharing the same index_path between multiple running Grafana instances is not supported.
index_path =
pkg/setting/setting.go
+4 -2
View File
@@ -588,8 +588,10 @@ type Cfg struct {
// Unified Storage
UnifiedStorage map[string]UnifiedStorageConfig
// DisableDataMigrations will disable resources data migration to unified storage at startup
DisableDataMigrations bool
MaxPageSizeBytes int
DisableDataMigrations bool
MaxPageSizeBytes int
// IndexPath the directory where index files are stored.
// Note: Bleve locks index files, so mounts cannot be shared between multiple instances.
IndexPath string
IndexWorkers int
IndexRebuildWorkers int
pkg/storage/unified/search/bleve.go
+26 -45
View File
@@ -45,7 +45,7 @@ import (
const (
indexStorageMemory = "memory"
indexStorageFile = "file"
boltTimeout = "500ms"
boltTimeout = "1s"
)
// Keys used to store internal data in index.
@@ -417,25 +417,18 @@ 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 {
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
var findErr error
index, fileIndexName, indexRV, findErr = b.findPreviousFileBasedIndex(resourceDir)
if findErr != nil {
return nil, findErr
}
}
if newIndexType == indexStorageFile && index != nil {
if 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 if newIndexType == indexStorageFile {
} else {
// 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.
@@ -462,9 +455,7 @@ func (b *bleveBackend) BuildIndex(
logWithDetails.Info("Building index using filesystem", "directory", indexDir)
defer closeIndexOnExit(index, indexDir) // Close index, and delete new index directory.
}
}
if newIndexType == indexStorageMemory {
} else {
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)
@@ -567,7 +558,7 @@ func cleanFileSegment(input string) string {
return input
}
// cleanOldIndexes deletes all subdirectories inside resourceDir, skipping directory with "skipName".
// cleanOldIndexes deletes all subdirectories inside dir, skipping directory with "skipName".
// "skipName" can be empty.
func (b *bleveBackend) cleanOldIndexes(resourceDir string, skipName string) {
entries, err := os.ReadDir(resourceDir)
@@ -578,19 +569,19 @@ func (b *bleveBackend) cleanOldIndexes(resourceDir string, skipName string) {
b.log.Warn("error cleaning folders from", "directory", resourceDir, "error", err)
return
}
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)
for _, entry := range entries {
if entry.IsDir() && entry.Name() != skipName {
entryDir := filepath.Join(resourceDir, entry.Name())
if !isPathWithinRoot(entryDir, b.opts.Root) {
b.log.Warn("Skipping cleanup of directory", "directory", entryDir)
continue
}
err = os.RemoveAll(indexDir)
err = os.RemoveAll(entryDir)
if err != nil {
b.log.Error("Unable to remove old index folder", "directory", indexDir, "error", err)
b.log.Error("Unable to remove old index folder", "directory", entryDir, "error", err)
} else {
b.log.Info("Removed old index folder", "directory", indexDir)
b.log.Info("Removed old index folder", "directory", entryDir)
}
}
}
@@ -637,17 +628,10 @@ func formatIndexName(now time.Time) string {
return now.Format("20060102-150405")
}
type fileIndex struct {
Index bleve.Index
Name string
RV int64
IsOpen bool
}
func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) *fileIndex {
func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) (bleve.Index, string, int64, error) {
entries, err := os.ReadDir(resourceDir)
if err != nil {
return nil
return nil, "", 0, nil
}
for _, ent := range entries {
@@ -657,14 +641,15 @@ func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) *fileIndex
indexName := ent.Name()
indexDir := filepath.Join(resourceDir, indexName)
idx, err := bleve.OpenUsing(indexDir, map[string]interface{}{"bolt_timeout": boltTimeout})
if err != nil {
// On timeout, the file probably is locked by another process.
// This indicates a setup issue that should be fixed rather than worked around by creating a new index file.
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.Error("index is locked by another process", "indexDir", indexDir, "err", err)
return nil, "", 0, fmt.Errorf("index is locked by another process: indexDir=%s, err=%w", indexDir, err)
}
b.log.Debug("error opening index", "indexDir", indexDir, "err", err)
b.log.Error("error opening index", "indexDir", indexDir, "err", err)
continue
}
@@ -675,14 +660,10 @@ func (b *bleveBackend) findPreviousFileBasedIndex(resourceDir string) *fileIndex
continue
}
return &fileIndex{
Index: idx,
Name: indexName,
RV: indexRV,
}
return idx, indexName, indexRV, nil
}
return nil
return nil, "", 0, nil
}
// Stop closes all indexes and stops background tasks.
pkg/storage/unified/search/bleve_test.go
+10 -43
View File
@@ -18,6 +18,7 @@ import (
"github.com/prometheus/client_golang/prometheus/testutil"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
bolterrors "go.etcd.io/bbolt/errors"
"go.uber.org/atomic"
"go.uber.org/goleak"
@@ -1584,7 +1585,7 @@ func docCount(t *testing.T, idx resource.ResourceIndex) int {
return int(cnt)
}
func TestBleveBackendFallsBackToMemory(t *testing.T) {
func TestBuildIndexReturnsErrorWhenIndexLocked(t *testing.T) {
ns := resource.NamespacedResource{
Namespace: "test",
Group: "group",
@@ -1605,53 +1606,19 @@ func TestBleveBackendFallsBackToMemory(t *testing.T) {
require.Equal(t, indexStorageFile, bleveIdx1.indexStorage)
checkOpenIndexes(t, reg1, 0, 1)
// Now create a second backend using the same directory
// This simulates another instance trying to open the same index
backend2, reg2 := setupBleveBackend(t, withRootDir(tmpDir))
// BuildIndex should detect the file is locked and fallback to memory
index2, err := backend2.BuildIndex(context.Background(), ns, 100 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false)
require.NoError(t, err)
require.NotNil(t, index2)
// Verify second index fell back to in-memory despite size being above file threshold
bleveIdx2, ok := index2.(*bleveIndex)
require.True(t, ok)
require.Equal(t, indexStorageMemory, bleveIdx2.indexStorage)
// Verify metrics show 1 memory index and 0 file indexes for backend2
checkOpenIndexes(t, reg2, 1, 0)
// Verify the in-memory index works correctly
require.Equal(t, 10, docCount(t, index2))
// Clean up: close first backend to release the file lock
backend1.Stop()
}
func TestBleveSkipCleanOldIndexesOnMemoryFallback(t *testing.T) {
ns := resource.NamespacedResource{
Namespace: "test",
Group: "group",
Resource: "resource",
}
tmpDir := t.TempDir()
backend1, _ := setupBleveBackend(t, withRootDir(tmpDir))
_, err := backend1.BuildIndex(context.Background(), ns, 100 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false)
require.NoError(t, err)
// Now create a second backend using the same directory
// This simulates another instance trying to open the same index
backend2, _ := setupBleveBackend(t, withRootDir(tmpDir))
// BuildIndex should detect the file is locked and fallback to memory
_, err = backend2.BuildIndex(context.Background(), ns, 100 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false)
// BuildIndex should detect the file is locked and return an error after timeout
now := time.Now()
timeout, err := time.ParseDuration(boltTimeout)
require.NoError(t, err)
// Verify that the index directory still exists (i.e., cleanOldIndexes was skipped)
verifyDirEntriesCount(t, backend2.getResourceDir(ns), 1)
index2, err := backend2.BuildIndex(context.Background(), ns, 100 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false)
require.Error(t, err)
require.ErrorIs(t, err, bolterrors.ErrTimeout)
require.Nil(t, index2)
require.GreaterOrEqual(t, time.Since(now).Milliseconds(), timeout.Milliseconds()-500, "BuildIndex should have waited for approximately boltTimeout duration")
// Clean up: close first backend to release the file lock
backend1.Stop()