Search: use SQL search as a fallback during bluge's initial indexing (#54095)

* Search: use SQL search as a fallback when bluge indexing is ongoing

* Search: lint

* Search: feedback fixes - return an empty frame with a special name

* Search: revert readiness check query type

* Search: remove println

* remove sleep, get coffee

pkg/services/searchV2/index.go

@@ -84,6 +84,9 @@ type searchIndex struct {
 	mu                      sync.RWMutex
 	loader                  dashboardLoader
 	perOrgIndex             map[int64]*orgIndex
+	initializedOrgs         map[int64]bool
+	initialIndexingComplete bool
+	initializationMutex     sync.RWMutex
 	eventStore              eventStore
 	logger                  log.Logger
 	buildSignals            chan buildSignal
@@ -97,6 +100,7 @@ func newSearchIndex(dashLoader dashboardLoader, evStore eventStore, extender Doc
 		loader:          dashLoader,
 		eventStore:      evStore,
 		perOrgIndex:     map[int64]*orgIndex{},
+		initializedOrgs: map[int64]bool{},
 		logger:          log.New("searchIndex"),
 		buildSignals:    make(chan buildSignal),
 		extender:        extender,
@@ -105,6 +109,50 @@ func newSearchIndex(dashLoader dashboardLoader, evStore eventStore, extender Doc
 	}
 }
 
+func (i *searchIndex) isInitialized(_ context.Context, orgId int64) IsSearchReadyResponse {
+	i.initializationMutex.RLock()
+	orgInitialized := i.initializedOrgs[orgId]
+	initialInitComplete := i.initialIndexingComplete
+	i.initializationMutex.RUnlock()
+
+	if orgInitialized && initialInitComplete {
+		return IsSearchReadyResponse{IsReady: true}
+	}
+
+	if !initialInitComplete {
+		return IsSearchReadyResponse{IsReady: false, Reason: "initial-indexing-ongoing"}
+	}
+
+	i.triggerBuildingOrgIndex(orgId)
+	return IsSearchReadyResponse{IsReady: false, Reason: "org-indexing-ongoing"}
+}
+
+func (i *searchIndex) triggerBuildingOrgIndex(orgId int64) {
+	go func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+		defer cancel()
+
+		doneIndexing := make(chan error, 1)
+		signal := buildSignal{orgID: orgId, done: doneIndexing}
+		select {
+		case i.buildSignals <- signal:
+		case <-ctx.Done():
+			i.logger.Warn("Failed to send a build signal to initialize org index", "orgId", orgId)
+			return
+		}
+		select {
+		case err := <-doneIndexing:
+			if err != nil {
+				i.logger.Error("Failed to build org index", "orgId", orgId, "error", err)
+			} else {
+				i.logger.Debug("Successfully built org index", "orgId", orgId)
+			}
+		case <-ctx.Done():
+			i.logger.Warn("Building org index timeout", "orgId", orgId)
+		}
+	}()
+}
+
 func (i *searchIndex) sync(ctx context.Context) error {
 	doneCh := make(chan struct{}, 1)
 	select {
@@ -149,6 +197,10 @@ func (i *searchIndex) run(ctx context.Context, orgIDs []int64, reIndexSignalCh c
 	// Channel to handle signals about asynchronous full re-indexing completion.
 	reIndexDoneCh := make(chan int64, 1)
 
+	i.initializationMutex.Lock()
+	i.initialIndexingComplete = true
+	i.initializationMutex.Unlock()
+
 	for {
 		select {
 		case doneCh := <-i.syncCh:
@@ -421,6 +473,10 @@ func (i *searchIndex) buildOrgIndex(ctx context.Context, orgID int64) (int, erro
 	i.perOrgIndex[orgID] = index
 	i.mu.Unlock()
 
+	i.initializationMutex.Lock()
+	i.initializedOrgs[orgID] = true
+	i.initializationMutex.Unlock()
+
 	if orgID == 1 {
 		go func() {
 			if reader, cancel, err := index.readerForIndex(indexTypeDashboard); err == nil {

pkg/services/searchV2/search_service_mock.go

@@ -45,6 +45,20 @@ func (_m *MockSearchService) IsDisabled() bool {
 	return r0
 }
 
+// IsReady provides a mock function with given fields: ctx, orgId
+func (_m *MockSearchService) IsReady(ctx context.Context, orgId int64) IsSearchReadyResponse {
+	ret := _m.Called(ctx, orgId)
+
+	var r0 IsSearchReadyResponse
+	if rf, ok := ret.Get(0).(func(context.Context, int64) IsSearchReadyResponse); ok {
+		r0 = rf(ctx, orgId)
+	} else {
+		r0 = ret.Get(0).(IsSearchReadyResponse)
+	}
+
+	return r0
+}
+
 // RegisterDashboardIndexExtender provides a mock function with given fields: ext
 func (_m *MockSearchService) RegisterDashboardIndexExtender(ext DashboardIndexExtender) {
 	_m.Called(ext)

pkg/services/searchV2/service.go

@@ -64,6 +64,10 @@ type StandardSearchService struct {
 	reIndexCh      chan struct{}
 }
 
+func (s *StandardSearchService) IsReady(ctx context.Context, orgId int64) IsSearchReadyResponse {
+	return s.dashboardIndex.isInitialized(ctx, orgId)
+}
+
 func ProvideService(cfg *setting.Cfg, sql *sqlstore.SQLStore, entityEventStore store.EntityEventsService, ac accesscontrol.Service) SearchService {
 	extender := &NoopExtender{}
 	s := &StandardSearchService{

pkg/services/searchV2/stub.go

@@ -10,6 +10,10 @@ import (
 type stubSearchService struct {
 }
 
+func (s *stubSearchService) IsReady(ctx context.Context, orgId int64) IsSearchReadyResponse {
+	return IsSearchReadyResponse{}
+}
+
 func (s *stubSearchService) IsDisabled() bool {
 	return true
 }

pkg/services/searchV2/types.go

@@ -31,11 +31,17 @@ type DashboardQuery struct {
 	From               int          `json:"from,omitempty"`       // for paging
 }
 
+type IsSearchReadyResponse struct {
+	IsReady bool
+	Reason  string // initial-indexing-ongoing, org-indexing-ongoing
+}
+
 //go:generate mockery --name SearchService --structname MockSearchService --inpackage --filename search_service_mock.go
 type SearchService interface {
 	registry.CanBeDisabled
 	registry.BackgroundService
 	DoDashboardQuery(ctx context.Context, user *backend.User, orgId int64, query DashboardQuery) *backend.DataResponse
+	IsReady(ctx context.Context, orgId int64) IsSearchReadyResponse
 	RegisterDashboardIndexExtender(ext DashboardIndexExtender)
 	TriggerReIndex()
 }

pkg/tsdb/grafanads/grafana.go

@@ -9,13 +9,15 @@ import (
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
-
 	"github.com/grafana/grafana/pkg/components/simplejson"
+	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/services/datasources"
 	"github.com/grafana/grafana/pkg/services/searchV2"
 	"github.com/grafana/grafana/pkg/services/store"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/grafana/grafana/pkg/tsdb/testdatasource"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
 // DatasourceName is the string constant used as the datasource name in requests
@@ -36,6 +38,17 @@ const DatasourceUID = "grafana"
 var (
 	_                                       backend.QueryDataHandler   = (*Service)(nil)
 	_                                       backend.CheckHealthHandler = (*Service)(nil)
+	namespace                                                          = "grafana"
+	subsystem                                                          = "grafanads"
+	dashboardSearchNotServedRequestsCounter                            = promauto.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "dashboard_search_requests_not_served_total",
+			Help:      "A counter for dashboard search requests that could not be served due to an ongoing search engine indexing",
+		},
+		[]string{"reason"},
+	)
 )
 
 func ProvideService(cfg *setting.Cfg, search searchV2.SearchService, store store.StorageService) *Service {
@@ -46,6 +59,7 @@ func newService(cfg *setting.Cfg, search searchV2.SearchService, store store.Sto
 	s := &Service{
 		search: search,
 		store:  store,
+		log:    log.New("grafanads"),
 	}
 
 	return s
@@ -55,6 +69,7 @@ func newService(cfg *setting.Cfg, search searchV2.SearchService, store store.Sto
 type Service struct {
 	search searchV2.SearchService
 	store  store.StorageService
+	log    log.Logger
 }
 
 func DataSourceModel(orgId int64) *datasources.DataSource {
@@ -157,6 +172,21 @@ func (s *Service) doRandomWalk(query backend.DataQuery) backend.DataResponse {
 }
 
 func (s *Service) doSearchQuery(ctx context.Context, req *backend.QueryDataRequest, query backend.DataQuery) backend.DataResponse {
+	searchReadinessCheckResp := s.search.IsReady(ctx, req.PluginContext.OrgID)
+	if !searchReadinessCheckResp.IsReady {
+		dashboardSearchNotServedRequestsCounter.With(prometheus.Labels{
+			"reason": searchReadinessCheckResp.Reason,
+		}).Inc()
+
+		return backend.DataResponse{
+			Frames: data.Frames{
+				&data.Frame{
+					Name: "Loading",
+				},
+			},
+		}
+	}
+
 	m := requestModel{}
 	err := json.Unmarshal(query.JSON, &m)
 	if err != nil {

public/app/features/search/service/bluge.ts

@@ -10,12 +10,18 @@ import { replaceCurrentFolderQuery } from './utils';
 
 import { DashboardQueryResult, GrafanaSearcher, QueryResponse, SearchQuery, SearchResultMeta } from '.';
 
+// The backend returns an empty frame with a special name to indicate that the indexing engine is being rebuilt,
+// and that it can not serve any search requests. We are temporarily using the old SQL Search API as a fallback when that happens.
+const loadingFrameName = 'Loading';
+
 export class BlugeSearcher implements GrafanaSearcher {
+  constructor(private fallbackSearcher: GrafanaSearcher) {}
+
   async search(query: SearchQuery): Promise<QueryResponse> {
     if (query.facet?.length) {
       throw new Error('facets not supported!');
     }
-    return doSearchQuery(query);
+    return this.doSearchQuery(query);
   }
 
   async starred(query: SearchQuery): Promise<QueryResponse> {
@@ -28,7 +34,7 @@ export class BlugeSearcher implements GrafanaSearcher {
       uid: starsUIDS,
       query: query.query ?? '*',
     };
-    return doSearchQuery(starredQuery);
+    return this.doSearchQuery(starredQuery);
   }
 
   async tags(query: SearchQuery): Promise<TermCount[]> {
@@ -52,6 +58,11 @@ export class BlugeSearcher implements GrafanaSearcher {
         } as any)
       )
     ).data as DataFrame[];
+
+    if (data?.[0]?.name === loadingFrameName) {
+      return this.fallbackSearcher.tags(query);
+    }
+
     for (const frame of data) {
       if (frame.fields[0].name === 'tag') {
         return getTermCountsFrom(frame);
@@ -80,12 +91,8 @@ export class BlugeSearcher implements GrafanaSearcher {
 
     return Promise.resolve(opts);
   }
-}
 
-const firstPageSize = 50;
+  async doSearchQuery(query: SearchQuery): Promise<QueryResponse> {
-const nextPageSizes = 100;
-
-async function doSearchQuery(query: SearchQuery): Promise<QueryResponse> {
     query = await replaceCurrentFolderQuery(query);
     const ds = await getGrafanaDatasource();
     const target = {
@@ -104,6 +111,11 @@ async function doSearchQuery(query: SearchQuery): Promise<QueryResponse> {
     );
 
     const first = (rsp.data?.[0] as DataFrame) ?? { fields: [], length: 0 };
+
+    if (first.name === loadingFrameName) {
+      return this.fallbackSearcher.search(query);
+    }
+
     for (const field of first.fields) {
       field.display = getDisplayProcessor({ field, theme: config.theme2 });
     }
@@ -204,8 +216,12 @@ async function doSearchQuery(query: SearchQuery): Promise<QueryResponse> {
         return index < view.dataFrame.length;
       },
     };
+  }
 }
 
+const firstPageSize = 50;
+const nextPageSizes = 100;
+
 function getTermCountsFrom(frame: DataFrame): TermCount[] {
   const keys = frame.fields[0].values;
   const vals = frame.fields[1].values;

public/app/features/search/service/searcher.ts

@@ -7,9 +7,10 @@ import { GrafanaSearcher } from './types';
 let searcher: GrafanaSearcher | undefined = undefined;
 
 export function getGrafanaSearcher(): GrafanaSearcher {
+  const sqlSearcher = new SQLSearcher();
   if (!searcher) {
     const useBluge = config.featureToggles.panelTitleSearch;
-    searcher = useBluge ? new BlugeSearcher() : new SQLSearcher();
+    searcher = useBluge ? new BlugeSearcher(sqlSearcher) : sqlSearcher;
   }
   return searcher!;
 }