diff --git a/pkg/storage/unified/README.md b/pkg/storage/unified/README.md index f60b6c937db..b33d2aa3660 100644 --- a/pkg/storage/unified/README.md +++ b/pkg/storage/unified/README.md @@ -296,6 +296,11 @@ docker in case you don't have one: docker run -d --name db -e "MYSQL_DATABASE=grafana" -e "MYSQL_USER=grafana" -e "MYSQL_PASSWORD=grafana" -e "MYSQL_ROOT_PASSWORD=root" -p 3306:3306 docker.io/bitnami/mysql:8.0.31 ``` +or use our mysql docker block: +```sh +make devenv sources=mysql +``` + ### 2. Create dedicated ini files for every service Example distributor ini file: @@ -304,7 +309,7 @@ Example distributor ini file: * Bind and join `memberlist` on `127.0.0.1:7946` (default memberlist port) ```ini -target = distributor +target = search-server-distributor [server] http_port = 3000 @@ -404,6 +409,7 @@ path = grafana1.db [grafana-apiserver] address = 127.0.0.1:10000 storage_type = unified-grpc +search_server_address = 127.0.0.1:10000 [server] protocol = http @@ -443,9 +449,9 @@ Repeat for the other services. ./bin/grafana server target --config conf/storage-api-2.ini ./bin/grafana server target --config conf/storage-api-3.ini -./bin/grafana server target --config conf/grafana1.ini -./bin/grafana server target --config conf/grafana2.ini -./bin/grafana server target --config conf/grafana3.ini +./bin/grafana server --config conf/grafana1.ini +./bin/grafana server --config conf/grafana2.ini +./bin/grafana server --config conf/grafana3.ini ``` etc diff --git a/pkg/storage/unified/resource/search.go b/pkg/storage/unified/resource/search.go index 4752fa35cec..f605acc7e80 100644 --- a/pkg/storage/unified/resource/search.go +++ b/pkg/storage/unified/resource/search.go @@ -943,6 +943,8 @@ 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 return indexQueueProcessor, nil } diff --git a/pkg/storage/unified/resource/search_test.go b/pkg/storage/unified/resource/search_test.go index 677e24b8b8b..5f614b5d9e9 100644 --- a/pkg/storage/unified/resource/search_test.go +++ b/pkg/storage/unified/resource/search_test.go @@ -2,6 +2,8 @@ package resource import ( "context" + "fmt" + "log/slog" "sync" "testing" "time" @@ -387,6 +389,49 @@ func TestSearchGetOrCreateIndexWithCancellation(t *testing.T) { require.NoError(t, err) } +func TestSearchWillUpdateIndexOnQueueProcessor(t *testing.T) { + // Regression test: Indexes were being closed when being rebuilt, but not updated on the queue processor. This was causing new events to + // be added to a closed index, resulting in an error and missing docs in the index. + + // Create mock components + mockIndex1 := &MockResourceIndex{} + mockIndex2 := &MockResourceIndex{} // Different index to test replacement + mockBuilder := &MockDocumentBuilder{} + + // Create searchSupport instance + s := &searchSupport{ + log: slog.Default(), + indexQueueProcessors: make(map[string]*indexQueueProcessor), + indexQueueProcessorsMutex: sync.Mutex{}, + indexEventsChan: make(chan *IndexEvent, 10), + } + + nsr := NamespacedResource{ + Namespace: "test-namespace", + Group: "test-group", + Resource: "test-resource", + } + + // Pre-populate the processor to avoid the builders.get() call + key := fmt.Sprintf("%s/%s/%s", nsr.Namespace, nsr.Group, nsr.Resource) + processor1 := newIndexQueueProcessor(mockIndex1, nsr, 10, mockBuilder, s.indexEventsChan) + s.indexQueueProcessors[key] = processor1 + + // Verify initial state + require.Same(t, mockIndex1, processor1.index) + + // Call getOrCreateIndexQueueProcessor with a different index + // This should return the existing processor but update its index + processor2, err := s.getOrCreateIndexQueueProcessor(mockIndex2, nsr) + require.NoError(t, err) + require.NotNil(t, processor2) + + // Same processor instance, but index was replaced + require.Same(t, processor1, processor2, "Should return the same processor instance") + require.Same(t, mockIndex2, processor2.index, "Index should be replaced with mockIndex2") + require.Same(t, mockIndex2, processor1.index, "Original processor should have updated index") +} + type slowSearchBackendWithCache struct { mockSearchBackend wg sync.WaitGroup