From c2c443757dda6230b695526c27e78f47516ed0c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Calisto?= Date: Fri, 21 Nov 2025 16:42:15 +0000 Subject: [PATCH] Unified Storage: allow rebuilding indexes for resource with a new grpc endpoint (#113748) * Unified Storage: allow rebuilding indexes for resource from a new grpc endpoint * remove log line * fix trace def * lint * fix after rebase * addressing code review changes * update with one channel per rebuild request * other review suggestions * update with review suggestions * run mockery generate for MockResourceClient * update tests * update tests and lint * fix test --- pkg/registry/apis/dashboard/legacy/client.go | 5 + .../apis/dashboard/legacy/sql_dashboards.go | 4 + pkg/registry/apis/folders/validate_test.go | 5 + pkg/server/search_server_distributor_test.go | 23 ++ pkg/storage/unified/proto/search.proto | 22 ++ pkg/storage/unified/resource/client_mock.go | 76 ++++- pkg/storage/unified/resource/search.go | 95 ++++++- pkg/storage/unified/resource/search_client.go | 5 + .../unified/resource/search_client_test.go | 5 + .../resource/search_server_distributor.go | 98 +++++++ pkg/storage/unified/resource/search_test.go | 100 ++++++- pkg/storage/unified/resource/server.go | 8 + pkg/storage/unified/resourcepb/search.pb.go | 265 ++++++++++++++---- .../unified/resourcepb/search_grpc.pb.go | 42 ++- pkg/storage/unified/sql/backend.go | 8 +- 15 files changed, 681 insertions(+), 80 deletions(-) diff --git a/pkg/registry/apis/dashboard/legacy/client.go b/pkg/registry/apis/dashboard/legacy/client.go index 1eeebd7b689..913dd02d14d 100644 --- a/pkg/registry/apis/dashboard/legacy/client.go +++ b/pkg/registry/apis/dashboard/legacy/client.go @@ -90,3 +90,8 @@ func (d *directResourceClient) Watch(ctx context.Context, in *resourcepb.WatchRe func (d *directResourceClient) BulkProcess(ctx context.Context, opts ...grpc.CallOption) (resourcepb.BulkStore_BulkProcessClient, error) { return nil, fmt.Errorf("BulkProcess not supported with direct resource client") } + +// RebuildIndexes implements resource.ResourceClient. +func (b *directResourceClient) RebuildIndexes(ctx context.Context, req *resourcepb.RebuildIndexesRequest, opts ...grpc.CallOption) (*resourcepb.RebuildIndexesResponse, error) { + return nil, fmt.Errorf("not implemented") +} diff --git a/pkg/registry/apis/dashboard/legacy/sql_dashboards.go b/pkg/registry/apis/dashboard/legacy/sql_dashboards.go index 7c17ec8b6aa..d4997c39c7d 100644 --- a/pkg/registry/apis/dashboard/legacy/sql_dashboards.go +++ b/pkg/registry/apis/dashboard/legacy/sql_dashboards.go @@ -1046,3 +1046,7 @@ func parseLibraryPanelRow(p panel) (dashboardV0.LibraryPanel, error) { return item, nil } + +func (b *dashboardSqlAccess) RebuildIndexes(ctx context.Context, req *resourcepb.RebuildIndexesRequest) (*resourcepb.RebuildIndexesResponse, error) { + return nil, fmt.Errorf("not implemented") +} diff --git a/pkg/registry/apis/folders/validate_test.go b/pkg/registry/apis/folders/validate_test.go index 1f67c0da9d8..c4bf07bc71c 100644 --- a/pkg/registry/apis/folders/validate_test.go +++ b/pkg/registry/apis/folders/validate_test.go @@ -581,3 +581,8 @@ func (m *mockSearchClient) GetStats(ctx context.Context, in *resourcepb.Resource func (m *mockSearchClient) Search(ctx context.Context, in *resourcepb.ResourceSearchRequest, opts ...grpc.CallOption) (*resourcepb.ResourceSearchResponse, error) { return m.search, m.searchErr } + +// RebuildIndexes implements resourcepb.ResourceIndexClient. +func (m *mockSearchClient) RebuildIndexes(ctx context.Context, in *resourcepb.RebuildIndexesRequest, opts ...grpc.CallOption) (*resourcepb.RebuildIndexesResponse, error) { + return nil, fmt.Errorf("not implemented") +} diff --git a/pkg/server/search_server_distributor_test.go b/pkg/server/search_server_distributor_test.go index 92182586105..65f6346d0a8 100644 --- a/pkg/server/search_server_distributor_test.go +++ b/pkg/server/search_server_distributor_test.go @@ -8,6 +8,7 @@ import ( "net" "net/http" "strconv" + "strings" "sync" "testing" "time" @@ -175,6 +176,28 @@ func TestIntegrationDistributor(t *testing.T) { } }) + t.Run("RebuildIndexes", func(t *testing.T) { + instanceResponseCount := make(map[string]int) + + // simulate RebuildIndexes for a single namespace + testNamespace := testNamespaces[0] + + req := &resourcepb.RebuildIndexesRequest{ + Namespace: testNamespace, + Keys: []*resourcepb.ResourceKey{{ + Namespace: testNamespace, + Group: "folder.grafana.app", + Resource: "folders", + }}, + } + distributorRes := getDistributorResponse(t, req, distributorServer.resourceClient.RebuildIndexes, instanceResponseCount) + require.Nil(t, distributorRes.Error) + + // assert all instances got the response by looking at the merged details + count := strings.Count(distributorRes.Details, "{instance:") + require.Equal(t, len(testServers), count) + }) + var wg sync.WaitGroup for _, testServer := range testServers { wg.Add(1) diff --git a/pkg/storage/unified/proto/search.proto b/pkg/storage/unified/proto/search.proto index d206060daf4..5018c97c9db 100644 --- a/pkg/storage/unified/proto/search.proto +++ b/pkg/storage/unified/proto/search.proto @@ -13,6 +13,8 @@ service ResourceIndex { // Get the resource stats rpc GetStats(ResourceStatsRequest) returns (ResourceStatsResponse); + + rpc RebuildIndexes(RebuildIndexesRequest) returns (RebuildIndexesResponse); } // Get statistics across multiple resources @@ -138,3 +140,23 @@ message ResourceSearchResponse { // Facet results map facet = 7; } + +message RebuildIndexesRequest { + // Namespace (tenant) must be the same as all keys' namespace + string namespace = 1; + + // List of ResourceKeys (Namespace + Group + Resource) + repeated ResourceKey keys = 2; +} + +message RebuildIndexesResponse { + // Total count of rebuilt indexes + int64 rebuildCount = 1; + + // Result message + string details = 2; + + // Error details + ErrorResult error = 3; +} + diff --git a/pkg/storage/unified/resource/client_mock.go b/pkg/storage/unified/resource/client_mock.go index 23688b097ec..fcc7392880e 100644 --- a/pkg/storage/unified/resource/client_mock.go +++ b/pkg/storage/unified/resource/client_mock.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package resource @@ -838,6 +838,80 @@ func (_c *MockResourceClient_Read_Call) RunAndReturn(run func(context.Context, * return _c } +// RebuildIndexes provides a mock function with given fields: ctx, in, opts +func (_m *MockResourceClient) RebuildIndexes(ctx context.Context, in *resourcepb.RebuildIndexesRequest, opts ...grpc.CallOption) (*resourcepb.RebuildIndexesResponse, error) { + _va := make([]interface{}, len(opts)) + for _i := range opts { + _va[_i] = opts[_i] + } + var _ca []interface{} + _ca = append(_ca, ctx, in) + _ca = append(_ca, _va...) + ret := _m.Called(_ca...) + + if len(ret) == 0 { + panic("no return value specified for RebuildIndexes") + } + + var r0 *resourcepb.RebuildIndexesResponse + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, *resourcepb.RebuildIndexesRequest, ...grpc.CallOption) (*resourcepb.RebuildIndexesResponse, error)); ok { + return rf(ctx, in, opts...) + } + if rf, ok := ret.Get(0).(func(context.Context, *resourcepb.RebuildIndexesRequest, ...grpc.CallOption) *resourcepb.RebuildIndexesResponse); ok { + r0 = rf(ctx, in, opts...) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*resourcepb.RebuildIndexesResponse) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, *resourcepb.RebuildIndexesRequest, ...grpc.CallOption) error); ok { + r1 = rf(ctx, in, opts...) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// MockResourceClient_RebuildIndexes_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RebuildIndexes' +type MockResourceClient_RebuildIndexes_Call struct { + *mock.Call +} + +// RebuildIndexes is a helper method to define mock.On call +// - ctx context.Context +// - in *resourcepb.RebuildIndexesRequest +// - opts ...grpc.CallOption +func (_e *MockResourceClient_Expecter) RebuildIndexes(ctx interface{}, in interface{}, opts ...interface{}) *MockResourceClient_RebuildIndexes_Call { + return &MockResourceClient_RebuildIndexes_Call{Call: _e.mock.On("RebuildIndexes", + append([]interface{}{ctx, in}, opts...)...)} +} + +func (_c *MockResourceClient_RebuildIndexes_Call) Run(run func(ctx context.Context, in *resourcepb.RebuildIndexesRequest, opts ...grpc.CallOption)) *MockResourceClient_RebuildIndexes_Call { + _c.Call.Run(func(args mock.Arguments) { + variadicArgs := make([]grpc.CallOption, len(args)-2) + for i, a := range args[2:] { + if a != nil { + variadicArgs[i] = a.(grpc.CallOption) + } + } + run(args[0].(context.Context), args[1].(*resourcepb.RebuildIndexesRequest), variadicArgs...) + }) + return _c +} + +func (_c *MockResourceClient_RebuildIndexes_Call) Return(_a0 *resourcepb.RebuildIndexesResponse, _a1 error) *MockResourceClient_RebuildIndexes_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *MockResourceClient_RebuildIndexes_Call) RunAndReturn(run func(context.Context, *resourcepb.RebuildIndexesRequest, ...grpc.CallOption) (*resourcepb.RebuildIndexesResponse, error)) *MockResourceClient_RebuildIndexes_Call { + _c.Call.Return(run) + return _c +} + // Search provides a mock function with given fields: ctx, in, opts func (_m *MockResourceClient) Search(ctx context.Context, in *resourcepb.ResourceSearchRequest, opts ...grpc.CallOption) (*resourcepb.ResourceSearchResponse, error) { _va := make([]interface{}, len(opts)) diff --git a/pkg/storage/unified/resource/search.go b/pkg/storage/unified/resource/search.go index dce40f884cf..bca5fb98491 100644 --- a/pkg/storage/unified/resource/search.go +++ b/pkg/storage/unified/resource/search.go @@ -234,6 +234,9 @@ func combineRebuildRequests(a, b rebuildRequest) (c rebuildRequest, ok bool) { ret.lastImportTime = b.lastImportTime } + // Combine complete channels + ret.completeChannels = append(a.completeChannels, b.completeChannels...) + return ret, true } @@ -512,6 +515,52 @@ func (s *searchSupport) GetStats(ctx context.Context, req *resourcepb.ResourceSt return rsp, nil } +func (s *searchSupport) RebuildIndexes(ctx context.Context, req *resourcepb.RebuildIndexesRequest) (*resourcepb.RebuildIndexesResponse, error) { + ctx, span := tracer.Start(ctx, "resource.searchSupport.RebuildIndexes") + defer span.End() + + filterKeys := make([]NamespacedResource, len(req.Keys)) + for _, key := range req.Keys { + if req.Namespace != key.Namespace { + return &resourcepb.RebuildIndexesResponse{ + Error: NewBadRequestError("key namespace does not match request namespace"), + }, nil + } + filterKeys = append(filterKeys, NamespacedResource{ + Namespace: key.Namespace, + Group: key.Group, + Resource: key.Resource, + }) + } + + importTimes, err := s.getLastImportTimes(ctx) + if err != nil { + return &resourcepb.RebuildIndexesResponse{ + Error: AsErrorResult(err), + }, nil + } + + completeChs := s.findIndexesToRebuild(importTimes, filterKeys, time.Now()) + rebuildCount := len(completeChs) + for _, ch := range completeChs { + select { + case <-ch: + continue + case <-ctx.Done(): // request was done before all indexes rebuilt + return &resourcepb.RebuildIndexesResponse{ + RebuildCount: int64(rebuildCount), + Details: fmt.Sprintf("returning before all index rebuilds completed for %d indexes", rebuildCount), + }, nil + } + } + + // All rebuilds completed successfully + return &resourcepb.RebuildIndexesResponse{ + RebuildCount: int64(rebuildCount), + Details: fmt.Sprintf("completed %d index rebuilds", rebuildCount), + }, nil +} + func (s *searchSupport) buildIndexes(ctx context.Context) (int, error) { totalBatchesIndexed := 0 group := errgroup.Group{} @@ -602,16 +651,23 @@ func (s *searchSupport) runPeriodicScanForIndexesToRebuild(ctx context.Context) if err != nil { s.log.Error("failed to get import times", "error", err) } - s.findIndexesToRebuild(importTimes, time.Now()) + s.findIndexesToRebuild(importTimes, nil, time.Now()) } } } -func (s *searchSupport) findIndexesToRebuild(lastImportTimes map[NamespacedResource]time.Time, now time.Time) { +func (s *searchSupport) findIndexesToRebuild(lastImportTimes map[NamespacedResource]time.Time, filterKeys []NamespacedResource, now time.Time) []chan struct{} { // 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. - keys := s.search.GetOpenIndexes() + var keys []NamespacedResource + if filterKeys != nil { + keys = filterKeys + } else { + keys = s.search.GetOpenIndexes() + } + + var completeChs []chan struct{} for _, key := range keys { idx := s.search.GetIndex(key) if idx == nil { @@ -638,18 +694,17 @@ func (s *searchSupport) findIndexesToRebuild(lastImportTimes map[NamespacedResou } if shouldRebuildIndex(bi, s.minBuildVersion, minBuildTime, lastImportTime, nil) { - s.rebuildQueue.Add(rebuildRequest{ - NamespacedResource: key, - minBuildTime: minBuildTime, - minBuildVersion: s.minBuildVersion, - lastImportTime: lastImportTime, - }) + completeCh := make(chan struct{}) + completeChs = append(completeChs, completeCh) + rebuildReq := newRebuildRequest(key, minBuildTime, lastImportTime, s.minBuildVersion, completeCh) + s.rebuildQueue.Add(rebuildReq) if s.indexMetrics != nil { s.indexMetrics.RebuildQueueLength.Set(float64(s.rebuildQueue.Len())) } } } + return completeChs } func (s *searchSupport) getLastImportTimes(ctx context.Context) (map[NamespacedResource]time.Time, error) { @@ -690,6 +745,12 @@ func (s *searchSupport) rebuildIndex(ctx context.Context, req rebuildRequest) { l := s.log.New("namespace", req.Namespace, "group", req.Group, "resource", req.Resource) + defer func() { + for _, ch := range req.completeChannels { + close(ch) + } + }() + idx := s.search.GetIndex(req.NamespacedResource) if idx == nil { span.AddEvent("index not found") @@ -782,6 +843,22 @@ type rebuildRequest struct { 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. + + completeChannels []chan<- struct{} // signal rebuild index is complete +} + +func newRebuildRequest(key NamespacedResource, minBuildTime, lastImportTime time.Time, minBuildVersion *semver.Version, completeCh chan<- struct{}) rebuildRequest { + var completeChannels []chan<- struct{} // setup a list as requests can be combined + if completeCh != nil { + completeChannels = []chan<- struct{}{completeCh} + } + return rebuildRequest{ + NamespacedResource: key, + minBuildTime: minBuildTime, + minBuildVersion: minBuildVersion, + lastImportTime: lastImportTime, + completeChannels: completeChannels, + } } func (s *searchSupport) getOrCreateIndex(ctx context.Context, stats *SearchStats, key NamespacedResource, reason string) (ResourceIndex, error) { diff --git a/pkg/storage/unified/resource/search_client.go b/pkg/storage/unified/resource/search_client.go index c4320114ba5..cfe0b0e594f 100644 --- a/pkg/storage/unified/resource/search_client.go +++ b/pkg/storage/unified/resource/search_client.go @@ -211,6 +211,11 @@ func (s *searchWrapper) Search(ctx context.Context, in *resourcepb.ResourceSearc return client.Search(ctx, in, opts...) } +func (s *searchWrapper) RebuildIndexes(ctx context.Context, in *resourcepb.RebuildIndexesRequest, + opts ...grpc.CallOption) (*resourcepb.RebuildIndexesResponse, error) { + return s.unifiedClient.RebuildIndexes(ctx, in, opts...) +} + // compareSearchResults compares legacy and unified search results and logs/metrics the outcome func (s *searchWrapper) compareSearchResults(legacyResponse, unifiedResponse *resourcepb.ResourceSearchResponse, requestKey *resourcepb.ResourceKey) { if legacyResponse == nil || unifiedResponse == nil { diff --git a/pkg/storage/unified/resource/search_client_test.go b/pkg/storage/unified/resource/search_client_test.go index 4b0a73895ec..f7676ad981e 100644 --- a/pkg/storage/unified/resource/search_client_test.go +++ b/pkg/storage/unified/resource/search_client_test.go @@ -112,6 +112,11 @@ func (m *MockResourceIndexClient) GetStats(ctx context.Context, in *resourcepb.R return args.Get(0).(*resourcepb.ResourceStatsResponse), args.Error(1) } +func (m *MockResourceIndexClient) RebuildIndexes(ctx context.Context, in *resourcepb.RebuildIndexesRequest, opts ...grpc.CallOption) (*resourcepb.RebuildIndexesResponse, error) { + args := m.Called(ctx, in, opts) + return args.Get(0).(*resourcepb.RebuildIndexesResponse), args.Error(1) +} + func setupTestSearchClient(t *testing.T) (schema.GroupResource, *MockResourceIndexClient, *MockResourceIndexClient, featuremgmt.FeatureToggles) { t.Helper() gr := schema.GroupResource{Group: "test", Resource: "items"} diff --git a/pkg/storage/unified/resource/search_server_distributor.go b/pkg/storage/unified/resource/search_server_distributor.go index 1363c2452f3..cceb0845b11 100644 --- a/pkg/storage/unified/resource/search_server_distributor.go +++ b/pkg/storage/unified/resource/search_server_distributor.go @@ -2,8 +2,12 @@ package resource import ( "context" + "errors" + "fmt" "hash/fnv" "math/rand" + "sync" + "sync/atomic" "time" "github.com/grafana/dskit/ring" @@ -114,6 +118,100 @@ func (ds *distributorServer) GetStats(ctx context.Context, r *resourcepb.Resourc return client.GetStats(ctx, r) } +func (ds *distributorServer) RebuildIndexes(ctx context.Context, r *resourcepb.RebuildIndexesRequest) (*resourcepb.RebuildIndexesResponse, error) { + ctx, span := ds.tracing.Start(ctx, "distributor.RebuildIndexes") + defer span.End() + + // validate input + for _, key := range r.Keys { + if r.Namespace != key.Namespace { + return &resourcepb.RebuildIndexesResponse{ + Error: NewBadRequestError("key namespace does not match request namespace"), + }, nil + } + } + + // distribute the request to all search pods to minimize risk of stale index + // it will not rebuild on those which don't have the index open + rs, err := ds.ring.GetAllHealthy(searchRingRead) + if err != nil { + return nil, fmt.Errorf("failed to get all healthy instances from the ring") + } + + err = grpc.SetHeader(ctx, metadata.Pairs("proxied-instance-id", "all")) + if err != nil { + ds.log.Debug("error setting grpc header", "err", err) + } + + md, ok := metadata.FromIncomingContext(ctx) + if !ok { + md = make(metadata.MD) + } + rCtx := userutils.InjectOrgID(metadata.NewOutgoingContext(ctx, md), r.Namespace) + + var wg sync.WaitGroup + var totalRebuildCount atomic.Int64 + detailsCh := make(chan string, len(rs.Instances)) + errorCh := make(chan error, len(rs.Instances)) + + for _, inst := range rs.Instances { + wg.Add(1) + go func() { + defer wg.Done() + + client, err := ds.clientPool.GetClientForInstance(inst) + if err != nil { + errorCh <- fmt.Errorf("instance %s: failed to get client, %w", inst.Id, err) + return + } + + rsp, err := client.(*RingClient).Client.RebuildIndexes(rCtx, r) + if err != nil { + errorCh <- fmt.Errorf("instance %s: failed to distribute rebuild index request, %w", inst.Id, err) + return + } + + if rsp.Error != nil { + errorCh <- fmt.Errorf("instance %s: rebuild index request returned the error %s", inst.Id, rsp.Error.Message) + return + } + + if rsp.Details != "" { + detailsCh <- fmt.Sprintf("{instance: %s, details: %s}", inst.Id, rsp.Details) + } + + totalRebuildCount.Add(rsp.RebuildCount) + }() + } + + wg.Wait() + close(errorCh) + close(detailsCh) + + errs := make([]error, 0, len(errorCh)) + for err := range errorCh { + ds.log.Error("rebuild indexes call failed with %w", err) + errs = append(errs, err) + } + + var details string + for d := range detailsCh { + if len(details) > 0 { + details += ", " + } + details += d + } + + response := &resourcepb.RebuildIndexesResponse{ + RebuildCount: totalRebuildCount.Load(), + Details: details, + } + if len(errs) > 0 { + response.Error = AsErrorResult(errors.Join(errs...)) + } + return response, nil +} + func (ds *distributorServer) CountManagedObjects(ctx context.Context, r *resourcepb.CountManagedObjectsRequest) (*resourcepb.CountManagedObjectsResponse, error) { ctx, span := ds.tracing.Start(ctx, "distributor.CountManagedObjects") defer span.End() diff --git a/pkg/storage/unified/resource/search_test.go b/pkg/storage/unified/resource/search_test.go index 215a88cd2cd..154608a4eee 100644 --- a/pkg/storage/unified/resource/search_test.go +++ b/pkg/storage/unified/resource/search_test.go @@ -10,6 +10,8 @@ import ( "time" "github.com/Masterminds/semver" + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" "github.com/grafana/authlib/types" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -85,7 +87,8 @@ func (m *MockDocumentBuilder) BuildDocument(ctx context.Context, key *resourcepb // mockStorageBackend implements StorageBackend for testing type mockStorageBackend struct { - resourceStats []ResourceStats + resourceStats []ResourceStats + lastImportTimes []ResourceLastImportTime } func (m *mockStorageBackend) GetResourceStats(ctx context.Context, nsr NamespacedResource, minCount int) ([]ResourceStats, error) { @@ -127,7 +130,11 @@ func (m *mockStorageBackend) ListModifiedSince(ctx context.Context, key Namespac func (m *mockStorageBackend) GetResourceLastImportTimes(ctx context.Context) iter.Seq2[ResourceLastImportTime, error] { return func(yield func(ResourceLastImportTime, error) bool) { - yield(ResourceLastImportTime{}, errors.New("not implemented")) + for _, ti := range m.lastImportTimes { + if !yield(ti, nil) { + return + } + } } } @@ -605,14 +612,14 @@ func TestFindIndexesForRebuild(t *testing.T) { {Namespace: "resource-v6", Group: "group", Resource: dashboardv1.DASHBOARD_RESOURCE}: lastImportTime, } - support.findIndexesToRebuild(importTimes, now) + support.findIndexesToRebuild(importTimes, nil, 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(importTimes, now5m) + support.findIndexesToRebuild(importTimes, nil, now5m) require.Equal(t, 7, support.rebuildQueue.Len()) // Values that we expect to find in rebuild requests. @@ -621,7 +628,7 @@ func TestFindIndexesForRebuild(t *testing.T) { minBuildTimeDashboard := now5m.Add(-1 * time.Hour) vals := support.rebuildQueue.Elements() - require.ElementsMatch(t, vals, []rebuildRequest{ + expected := []rebuildRequest{ {NamespacedResource: NamespacedResource{Namespace: "resource-2h-v5", Group: "group", Resource: "folder"}, minBuildVersion: minBuildVersion, minBuildTime: minBuildTime}, {NamespacedResource: NamespacedResource{Namespace: "resource-10h-v5", Group: "group", Resource: "folder"}, minBuildVersion: minBuildVersion, minBuildTime: minBuildTime}, {NamespacedResource: NamespacedResource{Namespace: "resource-10h-v6", Group: "group", Resource: "folder"}, minBuildVersion: minBuildVersion, minBuildTime: minBuildTime}, @@ -631,7 +638,10 @@ func TestFindIndexesForRebuild(t *testing.T) { {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}, - }) + } + if diff := cmp.Diff(expected, vals, cmpopts.IgnoreFields(rebuildRequest{}, "completeChannels"), cmp.AllowUnexported(rebuildRequest{})); diff != "" { + t.Errorf("rebuildQueue mismatch (-want +got):\n%s", diff) + } } func TestRebuildIndexes(t *testing.T) { @@ -748,3 +758,81 @@ func checkRebuildIndex(t *testing.T, support *searchSupport, req rebuildRequest, require.Nil(t, idxAfter, "index should not exist after rebuildIndex") } } + +func TestRebuildIndexesForResource(t *testing.T) { + key := NamespacedResource{Namespace: "ns", Group: "group", Resource: "resource"} + + storage := &mockStorageBackend{ + resourceStats: []ResourceStats{ + {NamespacedResource: key, Count: 50, ResourceVersion: 11111111}, + }, + lastImportTimes: []ResourceLastImportTime{{ + NamespacedResource: key, + LastImportTime: time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC), + }}, + } + + search := &mockSearchBackend{} + supplier := &TestDocumentBuilderSupplier{ + GroupsResources: map[string]string{ + "group": "resource", + }, + } + + opts := SearchOptions{ + Backend: search, + Resources: supplier, + InitMinCount: 1, + } + + support, err := newSearchSupport(opts, storage, nil, nil, nil, nil) + require.NoError(t, err) + require.NotNil(t, support) + + err = support.init(t.Context()) + require.NoError(t, err) + + require.Equal(t, 0, support.rebuildQueue.Len()) + + // invalid request + rebuildReq := &resourcepb.RebuildIndexesRequest{ + Namespace: "some-other-namespace", + Keys: []*resourcepb.ResourceKey{{ + Namespace: key.Namespace, + Group: key.Group, + Resource: key.Resource, + }}} + rsp, err := support.RebuildIndexes(t.Context(), rebuildReq) + require.NoError(t, err) + require.Equal(t, "key namespace does not match request namespace", rsp.Error.Message) + + rebuildReq.Namespace = key.Namespace + + // cached index info + search.cache[key] = &MockResourceIndex{ + buildInfo: IndexBuildInfo{BuildVersion: semver.MustParse("5.0.0"), BuildTime: time.Date(2025, 1, 1, 12, 0, 0, 0, time.UTC)}, + } + + // old import time will not be rebuilt + storage.lastImportTimes = []ResourceLastImportTime{{ + NamespacedResource: key, + LastImportTime: time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC), + }} + rsp, err = support.RebuildIndexes(t.Context(), rebuildReq) + require.NoError(t, err) + require.Equal(t, int64(0), rsp.RebuildCount) + require.Equal(t, 0, support.rebuildQueue.Len()) + + // recent import time gets added to rebuild queue and processed + storage.lastImportTimes = []ResourceLastImportTime{{ + NamespacedResource: key, + LastImportTime: time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC), + }} + + rsp, err = support.RebuildIndexes(t.Context(), rebuildReq) + require.NoError(t, err) + require.Equal(t, int64(1), rsp.RebuildCount) + + // rebuild waited for rebuild queue to process + require.Equal(t, 0, support.rebuildQueue.Len()) +} diff --git a/pkg/storage/unified/resource/server.go b/pkg/storage/unified/resource/server.go index ca9f87896e1..4c0adddf3c8 100644 --- a/pkg/storage/unified/resource/server.go +++ b/pkg/storage/unified/resource/server.go @@ -1541,3 +1541,11 @@ func (s *server) runInQueue(ctx context.Context, tenantID string, runnable func( return queueCtx.Err() // Timed out or canceled while waiting for execution. } } + +func (s *server) RebuildIndexes(ctx context.Context, req *resourcepb.RebuildIndexesRequest) (*resourcepb.RebuildIndexesResponse, error) { + if s.search == nil { + return nil, fmt.Errorf("search index not configured") + } + + return s.search.RebuildIndexes(ctx, req) +} diff --git a/pkg/storage/unified/resourcepb/search.pb.go b/pkg/storage/unified/resourcepb/search.pb.go index c10b2f2fe53..91f5aadfa97 100644 --- a/pkg/storage/unified/resourcepb/search.pb.go +++ b/pkg/storage/unified/resourcepb/search.pb.go @@ -386,6 +386,123 @@ func (x *ResourceSearchResponse) GetFacet() map[string]*ResourceSearchResponse_F return nil } +type RebuildIndexesRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Namespace (tenant) + Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` + // List of ResourceKeys (Namespace + Group + Resource) + Keys []*ResourceKey `protobuf:"bytes,2,rep,name=keys,proto3" json:"keys,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RebuildIndexesRequest) Reset() { + *x = RebuildIndexesRequest{} + mi := &file_search_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RebuildIndexesRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RebuildIndexesRequest) ProtoMessage() {} + +func (x *RebuildIndexesRequest) ProtoReflect() protoreflect.Message { + mi := &file_search_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use RebuildIndexesRequest.ProtoReflect.Descriptor instead. +func (*RebuildIndexesRequest) Descriptor() ([]byte, []int) { + return file_search_proto_rawDescGZIP(), []int{4} +} + +func (x *RebuildIndexesRequest) GetNamespace() string { + if x != nil { + return x.Namespace + } + return "" +} + +func (x *RebuildIndexesRequest) GetKeys() []*ResourceKey { + if x != nil { + return x.Keys + } + return nil +} + +type RebuildIndexesResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Total count of rebuilt indexes + RebuildCount int64 `protobuf:"varint,1,opt,name=rebuildCount,proto3" json:"rebuildCount,omitempty"` + // Result message + Details string `protobuf:"bytes,2,opt,name=details,proto3" json:"details,omitempty"` + // Error details + Error *ErrorResult `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RebuildIndexesResponse) Reset() { + *x = RebuildIndexesResponse{} + mi := &file_search_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RebuildIndexesResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RebuildIndexesResponse) ProtoMessage() {} + +func (x *RebuildIndexesResponse) ProtoReflect() protoreflect.Message { + mi := &file_search_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use RebuildIndexesResponse.ProtoReflect.Descriptor instead. +func (*RebuildIndexesResponse) Descriptor() ([]byte, []int) { + return file_search_proto_rawDescGZIP(), []int{5} +} + +func (x *RebuildIndexesResponse) GetRebuildCount() int64 { + if x != nil { + return x.RebuildCount + } + return 0 +} + +func (x *RebuildIndexesResponse) GetDetails() string { + if x != nil { + return x.Details + } + return "" +} + +func (x *RebuildIndexesResponse) GetError() *ErrorResult { + if x != nil { + return x.Error + } + return nil +} + type ResourceStatsResponse_Stats struct { state protoimpl.MessageState `protogen:"open.v1"` // Resource group @@ -400,7 +517,7 @@ type ResourceStatsResponse_Stats struct { func (x *ResourceStatsResponse_Stats) Reset() { *x = ResourceStatsResponse_Stats{} - mi := &file_search_proto_msgTypes[4] + mi := &file_search_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -412,7 +529,7 @@ func (x *ResourceStatsResponse_Stats) String() string { func (*ResourceStatsResponse_Stats) ProtoMessage() {} func (x *ResourceStatsResponse_Stats) ProtoReflect() protoreflect.Message { - mi := &file_search_proto_msgTypes[4] + mi := &file_search_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -459,7 +576,7 @@ type ResourceSearchRequest_Sort struct { func (x *ResourceSearchRequest_Sort) Reset() { *x = ResourceSearchRequest_Sort{} - mi := &file_search_proto_msgTypes[5] + mi := &file_search_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -471,7 +588,7 @@ func (x *ResourceSearchRequest_Sort) String() string { func (*ResourceSearchRequest_Sort) ProtoMessage() {} func (x *ResourceSearchRequest_Sort) ProtoReflect() protoreflect.Message { - mi := &file_search_proto_msgTypes[5] + mi := &file_search_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -511,7 +628,7 @@ type ResourceSearchRequest_Facet struct { func (x *ResourceSearchRequest_Facet) Reset() { *x = ResourceSearchRequest_Facet{} - mi := &file_search_proto_msgTypes[6] + mi := &file_search_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -523,7 +640,7 @@ func (x *ResourceSearchRequest_Facet) String() string { func (*ResourceSearchRequest_Facet) ProtoMessage() {} func (x *ResourceSearchRequest_Facet) ProtoReflect() protoreflect.Message { - mi := &file_search_proto_msgTypes[6] + mi := &file_search_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -568,7 +685,7 @@ type ResourceSearchResponse_Facet struct { func (x *ResourceSearchResponse_Facet) Reset() { *x = ResourceSearchResponse_Facet{} - mi := &file_search_proto_msgTypes[8] + mi := &file_search_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -580,7 +697,7 @@ func (x *ResourceSearchResponse_Facet) String() string { func (*ResourceSearchResponse_Facet) ProtoMessage() {} func (x *ResourceSearchResponse_Facet) ProtoReflect() protoreflect.Message { - mi := &file_search_proto_msgTypes[8] + mi := &file_search_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -634,7 +751,7 @@ type ResourceSearchResponse_TermFacet struct { func (x *ResourceSearchResponse_TermFacet) Reset() { *x = ResourceSearchResponse_TermFacet{} - mi := &file_search_proto_msgTypes[9] + mi := &file_search_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -646,7 +763,7 @@ func (x *ResourceSearchResponse_TermFacet) String() string { func (*ResourceSearchResponse_TermFacet) ProtoMessage() {} func (x *ResourceSearchResponse_TermFacet) ProtoReflect() protoreflect.Message { - mi := &file_search_proto_msgTypes[9] + mi := &file_search_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -780,22 +897,42 @@ var file_search_proto_rawDesc = string([]byte{ 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x26, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x61, 0x72, 0x63, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x2e, 0x46, - 0x61, 0x63, 0x65, 0x74, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x32, - 0xa9, 0x01, 0x0a, 0x0d, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x6e, 0x64, 0x65, - 0x78, 0x12, 0x4b, 0x0a, 0x06, 0x53, 0x65, 0x61, 0x72, 0x63, 0x68, 0x12, 0x1f, 0x2e, 0x72, 0x65, - 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, - 0x65, 0x61, 0x72, 0x63, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x72, - 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, - 0x53, 0x65, 0x61, 0x72, 0x63, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x4b, - 0x0a, 0x08, 0x47, 0x65, 0x74, 0x53, 0x74, 0x61, 0x74, 0x73, 0x12, 0x1e, 0x2e, 0x72, 0x65, 0x73, - 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x74, - 0x61, 0x74, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1f, 0x2e, 0x72, 0x65, 0x73, - 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x74, - 0x61, 0x74, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x3b, 0x5a, 0x39, 0x67, - 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x67, 0x72, 0x61, 0x66, 0x61, 0x6e, - 0x61, 0x2f, 0x67, 0x72, 0x61, 0x66, 0x61, 0x6e, 0x61, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x73, 0x74, - 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2f, 0x75, 0x6e, 0x69, 0x66, 0x69, 0x65, 0x64, 0x2f, 0x72, 0x65, - 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x61, 0x63, 0x65, 0x74, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, + 0x60, 0x0a, 0x15, 0x52, 0x65, 0x62, 0x75, 0x69, 0x6c, 0x64, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x65, + 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, + 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, + 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x12, 0x29, 0x0a, 0x04, 0x6b, 0x65, 0x79, 0x73, 0x18, 0x02, + 0x20, 0x03, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, + 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x4b, 0x65, 0x79, 0x52, 0x04, 0x6b, 0x65, 0x79, + 0x73, 0x22, 0x83, 0x01, 0x0a, 0x16, 0x52, 0x65, 0x62, 0x75, 0x69, 0x6c, 0x64, 0x49, 0x6e, 0x64, + 0x65, 0x78, 0x65, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x22, 0x0a, 0x0c, + 0x72, 0x65, 0x62, 0x75, 0x69, 0x6c, 0x64, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x01, 0x20, 0x01, + 0x28, 0x03, 0x52, 0x0c, 0x72, 0x65, 0x62, 0x75, 0x69, 0x6c, 0x64, 0x43, 0x6f, 0x75, 0x6e, 0x74, + 0x12, 0x18, 0x0a, 0x07, 0x64, 0x65, 0x74, 0x61, 0x69, 0x6c, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x07, 0x64, 0x65, 0x74, 0x61, 0x69, 0x6c, 0x73, 0x12, 0x2b, 0x0a, 0x05, 0x65, 0x72, + 0x72, 0x6f, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x72, 0x65, 0x73, 0x6f, + 0x75, 0x72, 0x63, 0x65, 0x2e, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, + 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x32, 0xfe, 0x01, 0x0a, 0x0d, 0x52, 0x65, 0x73, 0x6f, + 0x75, 0x72, 0x63, 0x65, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x12, 0x4b, 0x0a, 0x06, 0x53, 0x65, 0x61, + 0x72, 0x63, 0x68, 0x12, 0x1f, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, + 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x61, 0x72, 0x63, 0x68, 0x52, 0x65, 0x71, + 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, + 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x61, 0x72, 0x63, 0x68, 0x52, 0x65, + 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x4b, 0x0a, 0x08, 0x47, 0x65, 0x74, 0x53, 0x74, 0x61, + 0x74, 0x73, 0x12, 0x1e, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, + 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x74, 0x61, 0x74, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, + 0x73, 0x74, 0x1a, 0x1f, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, + 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x74, 0x61, 0x74, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, + 0x6e, 0x73, 0x65, 0x12, 0x53, 0x0a, 0x0e, 0x52, 0x65, 0x62, 0x75, 0x69, 0x6c, 0x64, 0x49, 0x6e, + 0x64, 0x65, 0x78, 0x65, 0x73, 0x12, 0x1f, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, + 0x2e, 0x52, 0x65, 0x62, 0x75, 0x69, 0x6c, 0x64, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x65, 0x73, 0x52, + 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, + 0x65, 0x2e, 0x52, 0x65, 0x62, 0x75, 0x69, 0x6c, 0x64, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x65, 0x73, + 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x3b, 0x5a, 0x39, 0x67, 0x69, 0x74, 0x68, + 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x67, 0x72, 0x61, 0x66, 0x61, 0x6e, 0x61, 0x2f, 0x67, + 0x72, 0x61, 0x66, 0x61, 0x6e, 0x61, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61, + 0x67, 0x65, 0x2f, 0x75, 0x6e, 0x69, 0x66, 0x69, 0x65, 0x64, 0x2f, 0x72, 0x65, 0x73, 0x6f, 0x75, + 0x72, 0x63, 0x65, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, }) var ( @@ -810,47 +947,53 @@ func file_search_proto_rawDescGZIP() []byte { return file_search_proto_rawDescData } -var file_search_proto_msgTypes = make([]protoimpl.MessageInfo, 11) +var file_search_proto_msgTypes = make([]protoimpl.MessageInfo, 13) var file_search_proto_goTypes = []any{ (*ResourceStatsRequest)(nil), // 0: resource.ResourceStatsRequest (*ResourceStatsResponse)(nil), // 1: resource.ResourceStatsResponse (*ResourceSearchRequest)(nil), // 2: resource.ResourceSearchRequest (*ResourceSearchResponse)(nil), // 3: resource.ResourceSearchResponse - (*ResourceStatsResponse_Stats)(nil), // 4: resource.ResourceStatsResponse.Stats - (*ResourceSearchRequest_Sort)(nil), // 5: resource.ResourceSearchRequest.Sort - (*ResourceSearchRequest_Facet)(nil), // 6: resource.ResourceSearchRequest.Facet - nil, // 7: resource.ResourceSearchRequest.FacetEntry - (*ResourceSearchResponse_Facet)(nil), // 8: resource.ResourceSearchResponse.Facet - (*ResourceSearchResponse_TermFacet)(nil), // 9: resource.ResourceSearchResponse.TermFacet - nil, // 10: resource.ResourceSearchResponse.FacetEntry - (*ErrorResult)(nil), // 11: resource.ErrorResult - (*ListOptions)(nil), // 12: resource.ListOptions - (*ResourceKey)(nil), // 13: resource.ResourceKey - (*ResourceTable)(nil), // 14: resource.ResourceTable + (*RebuildIndexesRequest)(nil), // 4: resource.RebuildIndexesRequest + (*RebuildIndexesResponse)(nil), // 5: resource.RebuildIndexesResponse + (*ResourceStatsResponse_Stats)(nil), // 6: resource.ResourceStatsResponse.Stats + (*ResourceSearchRequest_Sort)(nil), // 7: resource.ResourceSearchRequest.Sort + (*ResourceSearchRequest_Facet)(nil), // 8: resource.ResourceSearchRequest.Facet + nil, // 9: resource.ResourceSearchRequest.FacetEntry + (*ResourceSearchResponse_Facet)(nil), // 10: resource.ResourceSearchResponse.Facet + (*ResourceSearchResponse_TermFacet)(nil), // 11: resource.ResourceSearchResponse.TermFacet + nil, // 12: resource.ResourceSearchResponse.FacetEntry + (*ErrorResult)(nil), // 13: resource.ErrorResult + (*ListOptions)(nil), // 14: resource.ListOptions + (*ResourceKey)(nil), // 15: resource.ResourceKey + (*ResourceTable)(nil), // 16: resource.ResourceTable } var file_search_proto_depIdxs = []int32{ - 11, // 0: resource.ResourceStatsResponse.error:type_name -> resource.ErrorResult - 4, // 1: resource.ResourceStatsResponse.stats:type_name -> resource.ResourceStatsResponse.Stats - 12, // 2: resource.ResourceSearchRequest.options:type_name -> resource.ListOptions - 13, // 3: resource.ResourceSearchRequest.federated:type_name -> resource.ResourceKey - 5, // 4: resource.ResourceSearchRequest.sortBy:type_name -> resource.ResourceSearchRequest.Sort - 7, // 5: resource.ResourceSearchRequest.facet:type_name -> resource.ResourceSearchRequest.FacetEntry - 11, // 6: resource.ResourceSearchResponse.error:type_name -> resource.ErrorResult - 13, // 7: resource.ResourceSearchResponse.key:type_name -> resource.ResourceKey - 14, // 8: resource.ResourceSearchResponse.results:type_name -> resource.ResourceTable - 10, // 9: resource.ResourceSearchResponse.facet:type_name -> resource.ResourceSearchResponse.FacetEntry - 6, // 10: resource.ResourceSearchRequest.FacetEntry.value:type_name -> resource.ResourceSearchRequest.Facet - 9, // 11: resource.ResourceSearchResponse.Facet.terms:type_name -> resource.ResourceSearchResponse.TermFacet - 8, // 12: resource.ResourceSearchResponse.FacetEntry.value:type_name -> resource.ResourceSearchResponse.Facet - 2, // 13: resource.ResourceIndex.Search:input_type -> resource.ResourceSearchRequest - 0, // 14: resource.ResourceIndex.GetStats:input_type -> resource.ResourceStatsRequest - 3, // 15: resource.ResourceIndex.Search:output_type -> resource.ResourceSearchResponse - 1, // 16: resource.ResourceIndex.GetStats:output_type -> resource.ResourceStatsResponse - 15, // [15:17] is the sub-list for method output_type - 13, // [13:15] is the sub-list for method input_type - 13, // [13:13] is the sub-list for extension type_name - 13, // [13:13] is the sub-list for extension extendee - 0, // [0:13] is the sub-list for field type_name + 13, // 0: resource.ResourceStatsResponse.error:type_name -> resource.ErrorResult + 6, // 1: resource.ResourceStatsResponse.stats:type_name -> resource.ResourceStatsResponse.Stats + 14, // 2: resource.ResourceSearchRequest.options:type_name -> resource.ListOptions + 15, // 3: resource.ResourceSearchRequest.federated:type_name -> resource.ResourceKey + 7, // 4: resource.ResourceSearchRequest.sortBy:type_name -> resource.ResourceSearchRequest.Sort + 9, // 5: resource.ResourceSearchRequest.facet:type_name -> resource.ResourceSearchRequest.FacetEntry + 13, // 6: resource.ResourceSearchResponse.error:type_name -> resource.ErrorResult + 15, // 7: resource.ResourceSearchResponse.key:type_name -> resource.ResourceKey + 16, // 8: resource.ResourceSearchResponse.results:type_name -> resource.ResourceTable + 12, // 9: resource.ResourceSearchResponse.facet:type_name -> resource.ResourceSearchResponse.FacetEntry + 15, // 10: resource.RebuildIndexesRequest.keys:type_name -> resource.ResourceKey + 13, // 11: resource.RebuildIndexesResponse.error:type_name -> resource.ErrorResult + 8, // 12: resource.ResourceSearchRequest.FacetEntry.value:type_name -> resource.ResourceSearchRequest.Facet + 11, // 13: resource.ResourceSearchResponse.Facet.terms:type_name -> resource.ResourceSearchResponse.TermFacet + 10, // 14: resource.ResourceSearchResponse.FacetEntry.value:type_name -> resource.ResourceSearchResponse.Facet + 2, // 15: resource.ResourceIndex.Search:input_type -> resource.ResourceSearchRequest + 0, // 16: resource.ResourceIndex.GetStats:input_type -> resource.ResourceStatsRequest + 4, // 17: resource.ResourceIndex.RebuildIndexes:input_type -> resource.RebuildIndexesRequest + 3, // 18: resource.ResourceIndex.Search:output_type -> resource.ResourceSearchResponse + 1, // 19: resource.ResourceIndex.GetStats:output_type -> resource.ResourceStatsResponse + 5, // 20: resource.ResourceIndex.RebuildIndexes:output_type -> resource.RebuildIndexesResponse + 18, // [18:21] is the sub-list for method output_type + 15, // [15:18] is the sub-list for method input_type + 15, // [15:15] is the sub-list for extension type_name + 15, // [15:15] is the sub-list for extension extendee + 0, // [0:15] is the sub-list for field type_name } func init() { file_search_proto_init() } @@ -865,7 +1008,7 @@ func file_search_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_search_proto_rawDesc), len(file_search_proto_rawDesc)), NumEnums: 0, - NumMessages: 11, + NumMessages: 13, NumExtensions: 0, NumServices: 1, }, diff --git a/pkg/storage/unified/resourcepb/search_grpc.pb.go b/pkg/storage/unified/resourcepb/search_grpc.pb.go index 2fa3f16c611..d69cbd14e38 100644 --- a/pkg/storage/unified/resourcepb/search_grpc.pb.go +++ b/pkg/storage/unified/resourcepb/search_grpc.pb.go @@ -19,8 +19,9 @@ import ( const _ = grpc.SupportPackageIsVersion8 const ( - ResourceIndex_Search_FullMethodName = "/resource.ResourceIndex/Search" - ResourceIndex_GetStats_FullMethodName = "/resource.ResourceIndex/GetStats" + ResourceIndex_Search_FullMethodName = "/resource.ResourceIndex/Search" + ResourceIndex_GetStats_FullMethodName = "/resource.ResourceIndex/GetStats" + ResourceIndex_RebuildIndexes_FullMethodName = "/resource.ResourceIndex/RebuildIndexes" ) // ResourceIndexClient is the client API for ResourceIndex service. @@ -33,6 +34,7 @@ type ResourceIndexClient interface { Search(ctx context.Context, in *ResourceSearchRequest, opts ...grpc.CallOption) (*ResourceSearchResponse, error) // Get the resource stats GetStats(ctx context.Context, in *ResourceStatsRequest, opts ...grpc.CallOption) (*ResourceStatsResponse, error) + RebuildIndexes(ctx context.Context, in *RebuildIndexesRequest, opts ...grpc.CallOption) (*RebuildIndexesResponse, error) } type resourceIndexClient struct { @@ -63,6 +65,16 @@ func (c *resourceIndexClient) GetStats(ctx context.Context, in *ResourceStatsReq return out, nil } +func (c *resourceIndexClient) RebuildIndexes(ctx context.Context, in *RebuildIndexesRequest, opts ...grpc.CallOption) (*RebuildIndexesResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(RebuildIndexesResponse) + err := c.cc.Invoke(ctx, ResourceIndex_RebuildIndexes_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // ResourceIndexServer is the server API for ResourceIndex service. // All implementations should embed UnimplementedResourceIndexServer // for forward compatibility @@ -73,6 +85,7 @@ type ResourceIndexServer interface { Search(context.Context, *ResourceSearchRequest) (*ResourceSearchResponse, error) // Get the resource stats GetStats(context.Context, *ResourceStatsRequest) (*ResourceStatsResponse, error) + RebuildIndexes(context.Context, *RebuildIndexesRequest) (*RebuildIndexesResponse, error) } // UnimplementedResourceIndexServer should be embedded to have forward compatible implementations. @@ -85,6 +98,9 @@ func (UnimplementedResourceIndexServer) Search(context.Context, *ResourceSearchR func (UnimplementedResourceIndexServer) GetStats(context.Context, *ResourceStatsRequest) (*ResourceStatsResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method GetStats not implemented") } +func (UnimplementedResourceIndexServer) RebuildIndexes(context.Context, *RebuildIndexesRequest) (*RebuildIndexesResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method RebuildIndexes not implemented") +} // UnsafeResourceIndexServer may be embedded to opt out of forward compatibility for this service. // Use of this interface is not recommended, as added methods to ResourceIndexServer will @@ -133,6 +149,24 @@ func _ResourceIndex_GetStats_Handler(srv interface{}, ctx context.Context, dec f return interceptor(ctx, in, info, handler) } +func _ResourceIndex_RebuildIndexes_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(RebuildIndexesRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ResourceIndexServer).RebuildIndexes(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ResourceIndex_RebuildIndexes_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ResourceIndexServer).RebuildIndexes(ctx, req.(*RebuildIndexesRequest)) + } + return interceptor(ctx, in, info, handler) +} + // ResourceIndex_ServiceDesc is the grpc.ServiceDesc for ResourceIndex service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -148,6 +182,10 @@ var ResourceIndex_ServiceDesc = grpc.ServiceDesc{ MethodName: "GetStats", Handler: _ResourceIndex_GetStats_Handler, }, + { + MethodName: "RebuildIndexes", + Handler: _ResourceIndex_RebuildIndexes_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "search.proto", diff --git a/pkg/storage/unified/sql/backend.go b/pkg/storage/unified/sql/backend.go index d3dae729fd4..8de43f318b0 100644 --- a/pkg/storage/unified/sql/backend.go +++ b/pkg/storage/unified/sql/backend.go @@ -994,7 +994,9 @@ func (b *backend) GetResourceLastImportTimes(ctx context.Context) iter.Seq2[reso b.lastImportTimeDeletionTime.Store(now) } - rows, err := dbutil.QueryRows(ctx, b.db, sqlResourceLastImportTimeQuery, &sqlResourceLastImportTimeQueryRequest{SQLTemplate: sqltemplate.New(b.dialect)}) + rows, err := dbutil.QueryRows(ctx, b.db, sqlResourceLastImportTimeQuery, &sqlResourceLastImportTimeQueryRequest{ + SQLTemplate: sqltemplate.New(b.dialect), + }) if err != nil { return func(yield func(resource.ResourceLastImportTime, error) bool) { yield(resource.ResourceLastImportTime{}, err) @@ -1037,3 +1039,7 @@ func (b *backend) GetResourceLastImportTimes(ctx context.Context) iter.Seq2[reso } } } + +func (b *backend) RebuildIndexes(ctx context.Context, req *resourcepb.RebuildIndexesRequest) (*resourcepb.RebuildIndexesResponse, error) { + return nil, fmt.Errorf("rebuild indexes not supported by unistore sql backend") +}