Unified Storage: Update index when indexqueueprocessor exists (#109351)

* Update readme

* When queue processor already exists, update the index on it in case it has been closed.

* Adds regression test.
This commit is contained in:
owensmallwood
2025-08-07 11:38:24 -06:00
committed by GitHub
parent f888f61097
commit e667dfed29
3 changed files with 57 additions and 4 deletions
+10 -4
View File
@@ -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
+2
View File
@@ -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
}
@@ -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