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

@@ -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")
+}

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")
+}

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")
+}

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)

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<string,Facet> 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;
+}
+

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))

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) {

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 {

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"}

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()

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())
+}

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)
+}

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,
 		},

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",

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")
+}