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

unified-storage: Rebuild indexes with recently-imported resources (#112202)

Browse Source 
* Use timestamps reported via GetResourceLastImportTimes to trigger index rebuilds.

* Add test for old last import time.

* Don't reindex after bulk-import. It is now done indirectly via LastImportTime on all instances that own the index.
This commit is contained in:
Peter Štibraný
2025-10-09 16:42:02 +02:00
committed by GitHub
parent bfcd8b8f48
commit d61abe95ad
3 changed files with 80 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/resource/bulk.go
-18
View File
@@ -250,24 +250,6 @@ func (s *server) BulkProcess(stream resourcepb.BulkStore_BulkProcessServer) erro
rsp.Error = AsErrorResult(runner.err)
}
if rsp.Error == nil && s.search != nil {
// Rebuild any changed indexes
for _, summary := range rsp.Summary {
_, err := s.search.build(ctx, NamespacedResource{
Namespace: summary.Namespace,
Group: summary.Group,
Resource: summary.Resource,
}, summary.Count, "rebuildAfterBatchLoad", true)
if err != nil {
s.log.Warn("error building search index after batch load", "err", err)
rsp.Error = &resourcepb.ErrorResult{
Code: http.StatusInternalServerError,
Message: "err building search index: " + summary.Resource,
Reason: err.Error(),
}
}
}
}
return sendAndClose(rsp)
}
pkg/storage/unified/resource/search.go
+42 -7
View File
@@ -236,6 +236,11 @@ func combineRebuildRequests(a, b rebuildRequest) (c rebuildRequest, ok bool) {
ret.minBuildTime = b.minBuildTime
}
// Using higher "last import time" is stricter condition, and causes more indexes to be rebuilt.
if a.lastImportTime.IsZero() || (!b.lastImportTime.IsZero() && b.lastImportTime.After(a.lastImportTime)) {
ret.lastImportTime = b.lastImportTime
}
return ret, true
}
@@ -540,12 +545,16 @@ func (s *searchSupport) runPeriodicScanForIndexesToRebuild(ctx context.Context)
s.log.Info("stopping periodic index rebuild due to context cancellation")
return
case <-ticker.C:
s.findIndexesToRebuild(time.Now())
importTimes, err := s.getLastImportTimes(ctx)
if err != nil {
s.log.Error("failed to get import times", "error", err)
}
s.findIndexesToRebuild(importTimes, time.Now())
}
}
}
func (s *searchSupport) findIndexesToRebuild(now time.Time) {
func (s *searchSupport) findIndexesToRebuild(lastImportTimes map[NamespacedResource]time.Time, now time.Time) {
// Check all open indexes and see if any of them need to be rebuilt.
// This is done periodically to make sure that the indexes are up to date.
@@ -567,17 +576,20 @@ func (s *searchSupport) findIndexesToRebuild(now time.Time) {
minBuildTime = now.Add(-maxAge)
}
lastImportTime := lastImportTimes[key] // Will be time.Time{} if not found.
bi, err := idx.BuildInfo()
if err != nil {
s.log.Error("failed to get build info for index to rebuild", "key", key, "error", err)
continue
}
if shouldRebuildIndex(s.minBuildVersion, bi, minBuildTime, nil) {
if shouldRebuildIndex(bi, s.minBuildVersion, minBuildTime, lastImportTime, nil) {
s.rebuildQueue.Add(rebuildRequest{
NamespacedResource: key,
minBuildTime: minBuildTime,
minBuildVersion: s.minBuildVersion,
lastImportTime: lastImportTime,
})
if s.indexMetrics != nil {
@@ -587,6 +599,18 @@ func (s *searchSupport) findIndexesToRebuild(now time.Time) {
}
}
func (s *searchSupport) getLastImportTimes(ctx context.Context) (map[NamespacedResource]time.Time, error) {
result := map[NamespacedResource]time.Time{}
for importTime, err := range s.storage.GetResourceLastImportTimes(ctx) {
if err != nil {
// We return times that we have collected so far, if any.
return result, err
}
result[importTime.NamespacedResource] = importTime.LastImportTime
}
return result, nil
}
// runIndexRebuilder is a goroutine waiting for rebuild requests, and rebuilds indexes specified in those requests.
// Rebuild requests can be generated periodically (if configured), or after new documents have been imported into the storage with old RVs.
func (s *searchSupport) runIndexRebuilder(ctx context.Context) {
@@ -626,7 +650,7 @@ func (s *searchSupport) rebuildIndex(ctx context.Context, req rebuildRequest) {
l.Error("failed to get build info for index to rebuild", "error", err)
}
rebuild := shouldRebuildIndex(req.minBuildVersion, bi, req.minBuildTime, l)
rebuild := shouldRebuildIndex(bi, req.minBuildVersion, req.minBuildTime, req.lastImportTime, l)
if !rebuild {
span.AddEvent("index not rebuilt")
l.Info("index doesn't need to be rebuilt")
@@ -662,7 +686,7 @@ func (s *searchSupport) rebuildIndex(ctx context.Context, req rebuildRequest) {
}
}
func shouldRebuildIndex(minBuildVersion *semver.Version, buildInfo IndexBuildInfo, minBuildTime time.Time, rebuildLogger *slog.Logger) bool {
func shouldRebuildIndex(buildInfo IndexBuildInfo, minBuildVersion *semver.Version, minBuildTime time.Time, lastImportTime time.Time, rebuildLogger *slog.Logger) bool {
if !minBuildTime.IsZero() {
if buildInfo.BuildTime.IsZero() || buildInfo.BuildTime.Before(minBuildTime) {
if rebuildLogger != nil {
@@ -672,6 +696,16 @@ func shouldRebuildIndex(minBuildVersion *semver.Version, buildInfo IndexBuildInf
}
}
// This is technically the same as minBuildTime, but we want to log a different message to make the rebuild reason clear.
if !lastImportTime.IsZero() {
if buildInfo.BuildTime.IsZero() || buildInfo.BuildTime.Before(lastImportTime) {
if rebuildLogger != nil {
rebuildLogger.Info("index build time is before lastImportTime, rebuilding the index", "indexBuildTime", buildInfo.BuildTime, "lastImportTime", lastImportTime)
}
return true
}
}
if minBuildVersion != nil {
if buildInfo.BuildVersion == nil || buildInfo.BuildVersion.Compare(minBuildVersion) < 0 {
if rebuildLogger != nil {
@@ -687,8 +721,9 @@ func shouldRebuildIndex(minBuildVersion *semver.Version, buildInfo IndexBuildInf
type rebuildRequest struct {
NamespacedResource
minBuildTime time.Time // if not zero, only rebuild index if it has been built before this timestamp
minBuildVersion *semver.Version // if not nil, only rebuild index with build version older than this.
minBuildTime time.Time // if not zero, rebuild index if it has been built before this timestamp
lastImportTime time.Time // if not zero, rebuild index if it has been built before this timestamp.
minBuildVersion *semver.Version // if not nil, rebuild index with build version older than this.
}
func (s *searchSupport) getOrCreateIndex(ctx context.Context, key NamespacedResource, reason string) (ResourceIndex, error) {
pkg/storage/unified/resource/search_test.go
+38 -9
View File
@@ -9,8 +9,6 @@ import (
"testing"
"time"
"log/slog"
"github.com/Masterminds/semver"
"github.com/grafana/authlib/types"
"github.com/stretchr/testify/mock"
@@ -18,7 +16,6 @@ import (
"go.opentelemetry.io/otel/trace/noop"
dashboardv1 "github.com/grafana/grafana/apps/dashboard/pkg/apis/dashboard/v1beta1"
"github.com/grafana/grafana/pkg/infra/log/logtest"
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
)
@@ -442,6 +439,7 @@ func TestShouldRebuildIndex(t *testing.T) {
type testcase struct {
buildInfo IndexBuildInfo
minTime time.Time
lastImportTime time.Time
minBuildVersion *semver.Version
expected bool
@@ -459,6 +457,11 @@ func TestShouldRebuildIndex(t *testing.T) {
minTime: now,
expected: true,
},
"empty build info, with lastImportTime": {
buildInfo: IndexBuildInfo{},
lastImportTime: now,
expected: true,
},
"empty build info, with minVersion": {
buildInfo: IndexBuildInfo{},
minBuildVersion: semver.MustParse("10.15.20"),
@@ -474,6 +477,16 @@ func TestShouldRebuildIndex(t *testing.T) {
minTime: now,
expected: false,
},
"build time before last import time": {
buildInfo: IndexBuildInfo{BuildTime: now.Add(-2 * time.Hour)},
lastImportTime: now,
expected: true,
},
"build time after last import time": {
buildInfo: IndexBuildInfo{BuildTime: now.Add(2 * time.Hour)},
lastImportTime: now,
expected: false,
},
"build version before min version": {
buildInfo: IndexBuildInfo{BuildVersion: semver.MustParse("10.15.19")},
minBuildVersion: semver.MustParse("10.15.20"),
@@ -486,7 +499,7 @@ func TestShouldRebuildIndex(t *testing.T) {
},
} {
t.Run(name, func(t *testing.T) {
res := shouldRebuildIndex(tc.minBuildVersion, tc.buildInfo, tc.minTime, slog.New(&logtest.NopHandler{}))
res := shouldRebuildIndex(tc.buildInfo, tc.minBuildVersion, tc.minTime, tc.lastImportTime, nil)
require.Equal(t, tc.expected, res)
})
}
@@ -499,7 +512,7 @@ func TestFindIndexesForRebuild(t *testing.T) {
},
}
now := time.Now()
now := time.Now().UTC()
search := &mockSearchBackend{
openIndexes: []NamespacedResource{
@@ -511,6 +524,7 @@ func TestFindIndexesForRebuild(t *testing.T) {
{Namespace: "resource-v6", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE},
{Namespace: "resource-2h-v5", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE},
{Namespace: "resource-2h-v6", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE},
{Namespace: "resource-recently-imported", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE},
// We report this index as open, but it's really not. This can happen if index expires between the call
// to GetOpenIndexes and the call to GetIndex.
@@ -557,6 +571,11 @@ func TestFindIndexesForRebuild(t *testing.T) {
{Namespace: "resource-2h-v6", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE}: &MockResourceIndex{
buildInfo: IndexBuildInfo{BuildTime: now.Add(-2 * time.Hour), BuildVersion: semver.MustParse("6.0.0")},
},
// Built recently, to be rebuilt because of last import time
{Namespace: "resource-recently-imported", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE}: &MockResourceIndex{
buildInfo: IndexBuildInfo{BuildTime: now.Add(-30 * time.Minute), BuildVersion: semver.MustParse("6.0.0")},
},
},
}
@@ -579,15 +598,23 @@ func TestFindIndexesForRebuild(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, support)
support.findIndexesToRebuild(now)
require.Equal(t, 6, support.rebuildQueue.Len())
lastImportTime := now.Add(-10 * time.Minute)
importTimes := map[NamespacedResource]time.Time{
{Namespace: "resource-recently-imported", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE}: lastImportTime,
// This index was "just" built, and should not be rebuilt.
{Namespace: "resource-v6", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE}: lastImportTime,
}
support.findIndexesToRebuild(importTimes, now)
require.Equal(t, 7, support.rebuildQueue.Len())
now5m := now.Add(5 * time.Minute)
// Running findIndexesToRebuild again should not add any new indexes to the rebuild queue, and all existing
// ones should be "combined" with new ones (this will "bump" minBuildTime)
support.findIndexesToRebuild(now5m)
require.Equal(t, 6, support.rebuildQueue.Len())
support.findIndexesToRebuild(importTimes, now5m)
require.Equal(t, 7, support.rebuildQueue.Len())
// Values that we expect to find in rebuild requests.
minBuildVersion := semver.MustParse("5.5.5")
@@ -603,6 +630,8 @@ func TestFindIndexesForRebuild(t *testing.T) {
{NamespacedResource: NamespacedResource{Namespace: "resource-v5", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE}, minBuildVersion: minBuildVersion, minBuildTime: minBuildTimeDashboard},
{NamespacedResource: NamespacedResource{Namespace: "resource-2h-v5", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE}, minBuildVersion: minBuildVersion, minBuildTime: minBuildTimeDashboard},
{NamespacedResource: NamespacedResource{Namespace: "resource-2h-v6", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE}, minBuildVersion: minBuildVersion, minBuildTime: minBuildTimeDashboard},
{NamespacedResource: NamespacedResource{Namespace: "resource-recently-imported", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE}, minBuildVersion: minBuildVersion, minBuildTime: minBuildTimeDashboard, lastImportTime: lastImportTime},
})
}