Automatically pause queue if index service is unavailable (#15066)

* Handle keyword search error when issue indexer service is not available

* Implement automatic disabling and resume of code indexer queue

1 files changed, 108 insertions, 23 deletions

diff --git a/modules/indexer/code/elastic_search.go b/modules/indexer/code/elastic_search.go
index 9bd2fa301e..db37b4f66c 100644
--- a/modules/indexer/code/elastic_search.go
+++ b/modules/indexer/code/elastic_search.go
@@ -7,16 +7,20 @@ package code
 import (
 	"bufio"
 	"context"
+	"errors"
 	"fmt"
 	"io"
+	"net"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	repo_model "code.gitea.io/gitea/models/repo"
 	"code.gitea.io/gitea/modules/analyze"
 	"code.gitea.io/gitea/modules/charset"
 	"code.gitea.io/gitea/modules/git"
+	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/json"
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/setting"
@@ -39,8 +43,12 @@ var _ Indexer = &ElasticSearchIndexer{}
 
 // ElasticSearchIndexer implements Indexer interface
 type ElasticSearchIndexer struct {
-	client           *elastic.Client
-	indexerAliasName string
+	client               *elastic.Client
+	indexerAliasName     string
+	available            bool
+	availabilityCallback func(bool)
+	stopTimer            chan struct{}
+	lock                 sync.RWMutex
 }
 
 type elasticLogger struct {
@@ -78,7 +86,23 @@ func NewElasticSearchIndexer(url, indexerName string) (*ElasticSearchIndexer, bo
 	indexer := &ElasticSearchIndexer{
 		client:           client,
 		indexerAliasName: indexerName,
+		available:        true,
+		stopTimer:        make(chan struct{}),
 	}
+
+	ticker := time.NewTicker(10 * time.Second)
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				indexer.checkAvailability()
+			case <-indexer.stopTimer:
+				ticker.Stop()
+				return
+			}
+		}
+	}()
+
 	exists, err := indexer.init()
 	if err != nil {
 		indexer.Close()
@@ -123,17 +147,17 @@ func (b *ElasticSearchIndexer) realIndexerName() string {
 
 // Init will initialize the indexer
 func (b *ElasticSearchIndexer) init() (bool, error) {
-	ctx := context.Background()
+	ctx := graceful.GetManager().HammerContext()
 	exists, err := b.client.IndexExists(b.realIndexerName()).Do(ctx)
 	if err != nil {
-		return false, err
+		return false, b.checkError(err)
 	}
 	if !exists {
 		mapping := defaultMapping
 
 		createIndex, err := b.client.CreateIndex(b.realIndexerName()).BodyString(mapping).Do(ctx)
 		if err != nil {
-			return false, err
+			return false, b.checkError(err)
 		}
 		if !createIndex.Acknowledged {
 			return false, fmt.Errorf("create index %s with %s failed", b.realIndexerName(), mapping)
@@ -143,7 +167,7 @@ func (b *ElasticSearchIndexer) init() (bool, error) {
 	// check version
 	r, err := b.client.Aliases().Do(ctx)
 	if err != nil {
-		return false, err
+		return false, b.checkError(err)
 	}
 
 	realIndexerNames := r.IndicesByAlias(b.indexerAliasName)
@@ -152,10 +176,10 @@ func (b *ElasticSearchIndexer) init() (bool, error) {
 			Add(b.realIndexerName(), b.indexerAliasName).
 			Do(ctx)
 		if err != nil {
-			return false, err
+			return false, b.checkError(err)
 		}
 		if !res.Acknowledged {
-			return false, fmt.Errorf("")
+			return false, fmt.Errorf("create alias %s to index %s failed", b.indexerAliasName, b.realIndexerName())
 		}
 	} else if len(realIndexerNames) >= 1 && realIndexerNames[0] < b.realIndexerName() {
 		log.Warn("Found older gitea indexer named %s, but we will create a new one %s and keep the old NOT DELETED. You can delete the old version after the upgrade succeed.",
@@ -165,16 +189,30 @@ func (b *ElasticSearchIndexer) init() (bool, error) {
 			Add(b.realIndexerName(), b.indexerAliasName).
 			Do(ctx)
 		if err != nil {
-			return false, err
+			return false, b.checkError(err)
 		}
 		if !res.Acknowledged {
-			return false, fmt.Errorf("")
+			return false, fmt.Errorf("change alias %s to index %s failed", b.indexerAliasName, b.realIndexerName())
 		}
 	}
 
 	return exists, nil
 }
 
+// SetAvailabilityChangeCallback sets callback that will be triggered when availability changes
+func (b *ElasticSearchIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+	b.availabilityCallback = callback
+}
+
+// Ping checks if elastic is available
+func (b *ElasticSearchIndexer) Ping() bool {
+	b.lock.RLock()
+	defer b.lock.RUnlock()
+	return b.available
+}
+
 func (b *ElasticSearchIndexer) addUpdate(ctx context.Context, batchWriter git.WriteCloserError, batchReader *bufio.Reader, sha string, update fileUpdate, repo *repo_model.Repository) ([]elastic.BulkableRequest, error) {
 	// Ignore vendored files in code search
 	if setting.Indexer.ExcludeVendored && analyze.IsVendor(update.Filename) {
@@ -190,7 +228,7 @@ func (b *ElasticSearchIndexer) addUpdate(ctx context.Context, batchWriter git.Wr
 			return nil, err
 		}
 		if size, err = strconv.ParseInt(strings.TrimSpace(stdout), 10, 64); err != nil {
-			return nil, fmt.Errorf("Misformatted git cat-file output: %v", err)
+			return nil, fmt.Errorf("misformatted git cat-file output: %v", err)
 		}
 	}
 
@@ -274,8 +312,8 @@ func (b *ElasticSearchIndexer) Index(ctx context.Context, repo *repo_model.Repos
 		_, err := b.client.Bulk().
 			Index(b.indexerAliasName).
 			Add(reqs...).
-			Do(context.Background())
-		return err
+			Do(ctx)
+		return b.checkError(err)
 	}
 	return nil
 }
@@ -284,8 +322,8 @@ func (b *ElasticSearchIndexer) Index(ctx context.Context, repo *repo_model.Repos
 func (b *ElasticSearchIndexer) Delete(repoID int64) error {
 	_, err := b.client.DeleteByQuery(b.indexerAliasName).
 		Query(elastic.NewTermsQuery("repo_id", repoID)).
-		Do(context.Background())
-	return err
+		Do(graceful.GetManager().HammerContext())
+	return b.checkError(err)
 }
 
 // indexPos find words positions for start and the following end on content. It will
@@ -366,7 +404,7 @@ func extractAggs(searchResult *elastic.SearchResult) []*SearchResultLanguages {
 }
 
 // Search searches for codes and language stats by given conditions.
-func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) {
+func (b *ElasticSearchIndexer) Search(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) {
 	searchType := esMultiMatchTypeBestFields
 	if isMatch {
 		searchType = esMultiMatchTypePhrasePrefix
@@ -407,9 +445,9 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
 			).
 			Sort("repo_id", true).
 			From(start).Size(pageSize).
-			Do(context.Background())
+			Do(ctx)
 		if err != nil {
-			return 0, nil, nil, err
+			return 0, nil, nil, b.checkError(err)
 		}
 
 		return convertResult(searchResult, kw, pageSize)
@@ -421,9 +459,9 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
 		Aggregation("language", aggregation).
 		Query(query).
 		Size(0). // We only needs stats information
-		Do(context.Background())
+		Do(ctx)
 	if err != nil {
-		return 0, nil, nil, err
+		return 0, nil, nil, b.checkError(err)
 	}
 
 	query = query.Must(langQuery)
@@ -438,9 +476,9 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
 		).
 		Sort("repo_id", true).
 		From(start).Size(pageSize).
-		Do(context.Background())
+		Do(ctx)
 	if err != nil {
-		return 0, nil, nil, err
+		return 0, nil, nil, b.checkError(err)
 	}
 
 	total, hits, _, err := convertResult(searchResult, kw, pageSize)
@@ -449,4 +487,51 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
 }
 
 // Close implements indexer
-func (b *ElasticSearchIndexer) Close() {}
+func (b *ElasticSearchIndexer) Close() {
+	select {
+	case <-b.stopTimer:
+	default:
+		close(b.stopTimer)
+	}
+}
+
+func (b *ElasticSearchIndexer) checkError(err error) error {
+	var opErr *net.OpError
+	if !(elastic.IsConnErr(err) || (errors.As(err, &opErr) && (opErr.Op == "dial" || opErr.Op == "read"))) {
+		return err
+	}
+
+	b.setAvailability(false)
+
+	return err
+}
+
+func (b *ElasticSearchIndexer) checkAvailability() {
+	if b.Ping() {
+		return
+	}
+
+	// Request cluster state to check if elastic is available again
+	_, err := b.client.ClusterState().Do(graceful.GetManager().ShutdownContext())
+	if err != nil {
+		b.setAvailability(false)
+		return
+	}
+
+	b.setAvailability(true)
+}
+
+func (b *ElasticSearchIndexer) setAvailability(available bool) {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+
+	if b.available == available {
+		return
+	}
+
+	b.available = available
+	if b.availabilityCallback != nil {
+		// Call the callback from within the lock to ensure that the ordering remains correct
+		b.availabilityCallback(b.available)
+	}
+}