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

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

Browse Source 
* 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

(cherry picked from commit 74158ed66b)
This commit is contained in:
Artur Wierzbicki
2022-08-26 08:38:15 -07:00
committed by GitHub
parent fe7c483b9b
commit 108c2575f8
8 changed files with 275 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/searchV2/index.go
+73 -17
View File
@@ -82,30 +82,78 @@ func (i *orgIndex) readerForIndex(idxType indexType) (*bluge.Reader, func(), err
}
type searchIndex struct {
mu sync.RWMutex
loader dashboardLoader
perOrgIndex map[int64]*orgIndex
eventStore eventStore
logger log.Logger
buildSignals chan buildSignal
extender DocumentExtender
folderIdLookup folderUIDLookup
syncCh chan chan 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
extender DocumentExtender
folderIdLookup folderUIDLookup
syncCh chan chan struct{}
}
func newSearchIndex(dashLoader dashboardLoader, evStore eventStore, extender DocumentExtender, folderIDs folderUIDLookup) *searchIndex {
return &searchIndex{
loader: dashLoader,
eventStore: evStore,
perOrgIndex: map[int64]*orgIndex{},
logger: log.New("searchIndex"),
buildSignals: make(chan buildSignal),
extender: extender,
folderIdLookup: folderIDs,
syncCh: make(chan chan struct{}),
loader: dashLoader,
eventStore: evStore,
perOrgIndex: map[int64]*orgIndex{},
initializedOrgs: map[int64]bool{},
logger: log.New("searchIndex"),
buildSignals: make(chan buildSignal),
extender: extender,
folderIdLookup: folderIDs,
syncCh: make(chan chan struct{}),
}
}
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 {
@@ -150,6 +198,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:
@@ -422,6 +474,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
+14
View File
@@ -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
+4
View File
@@ -62,6 +62,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.AccessControl) SearchService {
extender := &NoopExtender{}
s := &StandardSearchService{
pkg/services/searchV2/stub.go
+4
View File
@@ -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
+6
View File
@@ -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
+33 -3
View File
@@ -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
@@ -34,8 +36,19 @@ const DatasourceUID = "grafana"
// This is important to do since otherwise we will only get a
// not implemented error response from plugin at runtime.
var (
_ backend.QueryDataHandler = (*Service)(nil)
_ backend.CheckHealthHandler = (*Service)(nil)
_ 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 {