From d61abe95ad042dfb58a33247425e191677513f8b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20=C5=A0tibran=C3=BD?= Date: Thu, 9 Oct 2025 16:42:02 +0200 Subject: [PATCH] unified-storage: Rebuild indexes with recently-imported resources (#112202) * 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. --- pkg/storage/unified/resource/bulk.go | 18 -------- pkg/storage/unified/resource/search.go | 49 ++++++++++++++++++--- pkg/storage/unified/resource/search_test.go | 47 ++++++++++++++++---- 3 files changed, 80 insertions(+), 34 deletions(-) diff --git a/pkg/storage/unified/resource/bulk.go b/pkg/storage/unified/resource/bulk.go index 913477fa3e3..935204384df 100644 --- a/pkg/storage/unified/resource/bulk.go +++ b/pkg/storage/unified/resource/bulk.go @@ -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) } diff --git a/pkg/storage/unified/resource/search.go b/pkg/storage/unified/resource/search.go index 50541be922f..edaaa0abba6 100644 --- a/pkg/storage/unified/resource/search.go +++ b/pkg/storage/unified/resource/search.go @@ -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) { diff --git a/pkg/storage/unified/resource/search_test.go b/pkg/storage/unified/resource/search_test.go index 15c6d9770f5..a35a8be7052 100644 --- a/pkg/storage/unified/resource/search_test.go +++ b/pkg/storage/unified/resource/search_test.go @@ -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}, }) }