From 1ca2e2b6c27141682d6e9bc08d534d5529f93923 Mon Sep 17 00:00:00 2001
From: Alexander Emelin <frvzmb@gmail.com>
Date: Wed, 22 Jun 2022 12:21:43 +0300
Subject: [PATCH] Search: Sync state on read for HA consistency (#50152)

---
 pkg/services/searchV2/index.go   | 62 ++++++++++++++++++++++++++------
 pkg/services/searchV2/service.go |  6 ++++
 2 files changed, 58 insertions(+), 10 deletions(-)

diff --git a/pkg/services/searchV2/index.go b/pkg/services/searchV2/index.go
index 0fc19d3bd44..2610e649ee0 100644
--- a/pkg/services/searchV2/index.go
+++ b/pkg/services/searchV2/index.go
@@ -60,6 +60,7 @@ type dashboardIndex struct {
 	buildSignals   chan int64
 	extender       DocumentExtender
 	folderIdLookup folderUIDLookup
+	syncCh         chan chan struct{}
 }
 
 func newDashboardIndex(dashLoader dashboardLoader, evStore eventStore, extender DocumentExtender, folderIDs folderUIDLookup) *dashboardIndex {
@@ -72,15 +73,33 @@ func newDashboardIndex(dashLoader dashboardLoader, evStore eventStore, extender
 		buildSignals:   make(chan int64),
 		extender:       extender,
 		folderIdLookup: folderIDs,
+		syncCh:         make(chan chan struct{}),
+	}
+}
+
+func (i *dashboardIndex) sync(ctx context.Context) error {
+	doneCh := make(chan struct{}, 1)
+	select {
+	case i.syncCh <- doneCh:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+	select {
+	case <-doneCh:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
 	}
 }
 
 func (i *dashboardIndex) run(ctx context.Context) error {
-	fullReIndexTicker := time.NewTicker(5 * time.Minute)
-	defer fullReIndexTicker.Stop()
+	reIndexInterval := 5 * time.Minute
+	fullReIndexTimer := time.NewTimer(reIndexInterval)
+	defer fullReIndexTimer.Stop()
 
-	partialUpdateTicker := time.NewTicker(5 * time.Second)
-	defer partialUpdateTicker.Stop()
+	partialUpdateInterval := 5 * time.Second
+	partialUpdateTimer := time.NewTimer(partialUpdateInterval)
+	defer partialUpdateTimer.Stop()
 
 	var lastEventID int64
 	lastEvent, err := i.eventStore.GetLastEvent(ctx)
@@ -96,12 +115,17 @@ func (i *dashboardIndex) run(ctx context.Context) error {
 		return fmt.Errorf("can't build initial dashboard search index: %w", err)
 	}
 
-	i.eventStore.OnEvent(i.applyEventOnIndex)
+	// Channel to handle signals about asynchronous full re-indexing completion.
+	reIndexDoneCh := make(chan int64, 1)
 
 	for {
 		select {
-		case <-partialUpdateTicker.C:
+		case doneCh := <-i.syncCh:
 			lastEventID = i.applyIndexUpdates(ctx, lastEventID)
+			close(doneCh)
+		case <-partialUpdateTimer.C:
+			lastEventID = i.applyIndexUpdates(ctx, lastEventID)
+			partialUpdateTimer.Reset(partialUpdateInterval)
 		case orgID := <-i.buildSignals:
 			i.mu.RLock()
 			_, ok := i.perOrgWriter[orgID]
@@ -112,10 +136,28 @@ func (i *dashboardIndex) run(ctx context.Context) error {
 			}
 			i.mu.RUnlock()
 			_, _ = i.buildOrgIndex(ctx, orgID)
-		case <-fullReIndexTicker.C:
-			started := time.Now()
-			i.reIndexFromScratch(ctx)
-			i.logger.Info("Full re-indexing finished", "fullReIndexElapsed", time.Since(started))
+		case <-fullReIndexTimer.C:
+			lastIndexedEventID := lastEventID
+			go func() {
+				// Do full re-index asynchronously to avoid blocking index synchronization
+				// on read for a long time.
+				started := time.Now()
+				i.logger.Info("Start re-indexing")
+				i.reIndexFromScratch(ctx)
+				i.logger.Info("Full re-indexing finished", "fullReIndexElapsed", time.Since(started))
+				reIndexDoneCh <- lastIndexedEventID
+			}()
+		case lastIndexedEventID := <-reIndexDoneCh:
+			// Asynchronous re-indexing is finished. Set lastEventID to the value which
+			// was actual at the re-indexing start – so that we could re-apply all the
+			// events happened during async index build process and make sure it's consistent.
+			if lastEventID != lastIndexedEventID {
+				i.logger.Info("Re-apply event ID to last indexed", "currentEventID", lastEventID, "lastIndexedEventID", lastIndexedEventID)
+				lastEventID = lastIndexedEventID
+				// Apply events immediately.
+				partialUpdateTimer.Reset(0)
+			}
+			fullReIndexTimer.Reset(reIndexInterval)
 		case <-ctx.Done():
 			return ctx.Err()
 		}
diff --git a/pkg/services/searchV2/service.go b/pkg/services/searchV2/service.go
index dd79b172a59..e329b2bfa8b 100644
--- a/pkg/services/searchV2/service.go
+++ b/pkg/services/searchV2/service.go
@@ -143,6 +143,12 @@ func (s *StandardSearchService) DoDashboardQuery(ctx context.Context, user *back
 		return rsp
 	}
 
+	err = s.dashboardIndex.sync(ctx)
+	if err != nil {
+		rsp.Error = err
+		return rsp
+	}
+
 	reader, ok := s.dashboardIndex.getOrgReader(orgID)
 	if !ok {
 		go func() {