Implement automatic disabling and resume of code indexer queue

Author: lafriks

docs/content/doc/developers/hacking-on-gitea.en-us.md

@@ -187,6 +187,27 @@ make lint-frontend
 
 Note: When working on frontend code, set `USE_SERVICE_WORKER` to `false` in `app.ini` to prevent undesirable caching of frontend assets.
 
+### Configuring local ElasticSearch instance
+
+Start local ElasticSearch instance using docker:
+
+```sh
+mkdir -p $(pwd)/data/elasticsearch
+sudo chown -R 1000:1000 $(pwd)/data/elasticsearch
+docker run --rm -p 127.0.0.1:9200:9200 -p 127.0.0.1:9300:9300 -e "discovery.type=single-node" -v "$(pwd)/data/elasticsearch:/usr/share/elasticsearch/data" docker.elastic.co/elasticsearch/elasticsearch:7.16.3
+```
+
+Configure `app.ini`:
+
+```ini
+[indexer]
+ISSUE_INDEXER_TYPE = elasticsearch
+ISSUE_INDEXER_CONN_STR = http://elastic:changeme@localhost:9200
+REPO_INDEXER_ENABLED = true
+REPO_INDEXER_TYPE = elasticsearch
+REPO_INDEXER_CONN_STR = http://elastic:changeme@localhost:9200
+```
+
 ### Building and adding SVGs
 
 SVG icons are built using the `make svg` target which compiles the icon sources defined in `build/generate-svg.js` into the output directory `public/img/svg`. Custom icons can be added in the `web_src/svg` directory.

modules/context/repo.go

@@ -21,6 +21,7 @@ import (
 	user_model "code.gitea.io/gitea/models/user"
 	"code.gitea.io/gitea/modules/cache"
 	"code.gitea.io/gitea/modules/git"
+	code_indexer "code.gitea.io/gitea/modules/indexer/code"
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/markup/markdown"
 	"code.gitea.io/gitea/modules/setting"
@@ -522,6 +523,9 @@ func RepoAssignment(ctx *Context) (cancel context.CancelFunc) {
 	ctx.Data["ExposeAnonSSH"] = setting.SSH.ExposeAnonymous
 	ctx.Data["DisableHTTP"] = setting.Repository.DisableHTTPGit
 	ctx.Data["RepoSearchEnabled"] = setting.Indexer.RepoIndexerEnabled
+	if setting.Indexer.RepoIndexerEnabled {
+		ctx.Data["CodeIndexerUnavailable"] = !code_indexer.IsAvailable()
+	}
 	ctx.Data["CloneLink"] = repo.CloneLink()
 	ctx.Data["UncycloCloneLink"] = repo.UncycloCloneLink()
 

modules/indexer/code/bleve.go

@@ -271,6 +271,15 @@ func (b *BleveIndexer) Close() {
 	log.Info("PID: %d Repository Indexer closed", os.Getpid())
 }
 
+// SetAvailabilityChangeCallback does nothing
+func (b *BleveIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
+}
+
+// Ping does nothing
+func (b *BleveIndexer) Ping() bool {
+	return true
+}
+
 // Index indexes the data
 func (b *BleveIndexer) Index(ctx context.Context, repo *repo_model.Repository, sha string, changes *repoChanges) error {
 	batch := gitea_bleve.NewFlushingBatch(b.indexer, maxBatchSize)

modules/indexer/code/elastic_search.go

@@ -7,8 +7,10 @@ package code
 import (
 	"bufio"
 	"context"
+	"errors"
 	"fmt"
 	"io"
+	"net"
 	"strconv"
 	"strings"
 	"time"
@@ -39,8 +41,11 @@ 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{}
 }
 
 type elasticLogger struct {
@@ -78,7 +83,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()
@@ -126,14 +147,14 @@ func (b *ElasticSearchIndexer) init() (bool, error) {
 	ctx := context.Background()
 	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 +164,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 +173,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 +186,26 @@ 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.availabilityCallback = callback
+}
+
+// Ping checks if elastic is available
+func (b *ElasticSearchIndexer) Ping() bool {
+	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 +221,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)
 		}
 	}
 
@@ -275,7 +306,7 @@ func (b *ElasticSearchIndexer) Index(ctx context.Context, repo *repo_model.Repos
 			Index(b.indexerAliasName).
 			Add(reqs...).
 			Do(context.Background())
-		return err
+		return b.checkError(err)
 	}
 	return nil
 }
@@ -285,7 +316,7 @@ func (b *ElasticSearchIndexer) Delete(repoID int64) error {
 	_, err := b.client.DeleteByQuery(b.indexerAliasName).
 		Query(elastic.NewTermsQuery("repo_id", repoID)).
 		Do(context.Background())
-	return err
+	return b.checkError(err)
 }
 
 // indexPos find words positions for start and the following end on content. It will
@@ -409,7 +440,7 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
 			From(start).Size(pageSize).
 			Do(context.Background())
 		if err != nil {
-			return 0, nil, nil, err
+			return 0, nil, nil, b.checkError(err)
 		}
 
 		return convertResult(searchResult, kw, pageSize)
@@ -423,7 +454,7 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
 		Size(0). // We only needs stats information
 		Do(context.Background())
 	if err != nil {
-		return 0, nil, nil, err
+		return 0, nil, nil, b.checkError(err)
 	}
 
 	query = query.Must(langQuery)
@@ -440,7 +471,7 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
 		From(start).Size(pageSize).
 		Do(context.Background())
 	if err != nil {
-		return 0, nil, nil, err
+		return 0, nil, nil, b.checkError(err)
 	}
 
 	total, hits, _, err := convertResult(searchResult, kw, pageSize)
@@ -449,4 +480,33 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
 }
 
 // Close implements indexer
-func (b *ElasticSearchIndexer) Close() {}
+func (b *ElasticSearchIndexer) Close() {
+	close(b.stopTimer)
+}
+
+func (b *ElasticSearchIndexer) checkError(err error) error {
+	var opErr *net.OpError
+	if b.available && (elastic.IsConnErr(err) || (errors.As(err, &opErr) && (opErr.Op == "dial" || opErr.Op == "read"))) {
+		b.available = false
+		if b.availabilityCallback != nil {
+			b.availabilityCallback(b.available)
+		}
+	}
+	return err
+}
+
+func (b *ElasticSearchIndexer) checkAvailability() {
+	if b.available {
+		return
+	}
+
+	// Request cluster state to check if elastic is available again
+	_, err := b.client.ClusterState().Do(context.Background())
+	if err != nil {
+		return
+	}
+	b.available = true
+	if b.availabilityCallback != nil {
+		b.availabilityCallback(b.available)
+	}
+}

modules/indexer/code/indexer.go

@@ -42,6 +42,8 @@ type SearchResultLanguages struct {
 
 // Indexer defines an interface to index and search code contents
 type Indexer interface {
+	Ping() bool
+	SetAvailabilityChangeCallback(callback func(bool))
 	Index(ctx context.Context, repo *repo_model.Repository, sha string, changes *repoChanges) error
 	Delete(repoID int64) error
 	Search(repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error)
@@ -140,6 +142,7 @@ func Init() {
 				return data
 			}
 
+			unhandled := make([]queue.Data, 0, len(data))
 			for _, datum := range data {
 				indexerData, ok := datum.(*IndexerData)
 				if !ok {
@@ -150,10 +153,14 @@ func Init() {
 
 				if err := index(ctx, indexer, indexerData.RepoID); err != nil {
 					log.Error("index: %v", err)
-					continue
+					if indexer.Ping() {
+						continue
+					}
+					// Add back to queue
+					unhandled = append(unhandled, datum)
 				}
 			}
-			return nil
+			return unhandled
 		}
 
 		indexerQueue = queue.CreateUniqueQueue("code_indexer", handler, &IndexerData{})
@@ -212,6 +219,18 @@ func Init() {
 
 		indexer.set(rIndexer)
 
+		if queue, ok := indexerQueue.(queue.Pausable); ok {
+			rIndexer.SetAvailabilityChangeCallback(func(available bool) {
+				if !available {
+					log.Info("Code index queue paused")
+					queue.Pause()
+				} else {
+					log.Info("Code index queue resumed")
+					queue.Resume()
+				}
+			})
+		}
+
 		// Start processing the queue
 		go graceful.GetManager().RunWithShutdownFns(indexerQueue.Run)
 
@@ -262,6 +281,17 @@ func UpdateRepoIndexer(repo *repo_model.Repository) {
 	}
 }
 
+// IsAvailable checks if issue indexer is available
+func IsAvailable() bool {
+	idx, err := indexer.get()
+	if err != nil {
+		log.Error("IsAvailable(): unable to get indexer: %v", err)
+		return false
+	}
+
+	return idx.Ping()
+}
+
 // populateRepoIndexer populate the repo indexer with pre-existing data. This
 // should only be run when the indexer is created for the first time.
 func populateRepoIndexer(ctx context.Context) {

modules/indexer/code/wrapped.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 
 	repo_model "code.gitea.io/gitea/models/repo"
+	"code.gitea.io/gitea/modules/log"
 )
 
 var indexer = newWrappedIndexer()
@@ -56,6 +57,26 @@ func (w *wrappedIndexer) get() (Indexer, error) {
 	return w.internal, nil
 }
 
+// SetAvailabilityChangeCallback sets callback that will be triggered when availability changes
+func (w *wrappedIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
+	indexer, err := w.get()
+	if err != nil {
+		log.Error("Failed to get indexer: %v", err)
+		return
+	}
+	indexer.SetAvailabilityChangeCallback(callback)
+}
+
+// Ping checks if elastic is available
+func (w *wrappedIndexer) Ping() bool {
+	indexer, err := w.get()
+	if err != nil {
+		log.Warn("Failed to get indexer: %v", err)
+		return false
+	}
+	return indexer.Ping()
+}
+
 func (w *wrappedIndexer) Index(ctx context.Context, repo *repo_model.Repository, sha string, changes *repoChanges) error {
 	indexer, err := w.get()
 	if err != nil {

options/locale/locale_en-US.ini

@@ -268,6 +268,7 @@ search = Search
 code = Code
 search.fuzzy = Fuzzy
 search.match = Match
+code_search_unavailable = Currently code search is not available. Please contact your site administrator.
 repo_no_results = No matching repositories found.
 user_no_results = No matching users found.
 org_no_results = No matching organizations found.
@@ -1708,6 +1709,8 @@ search.search_repo = Search repository
 search.fuzzy = Fuzzy
 search.match = Match
 search.results = Search results for "%s" in <a href="%s">%s</a>
+search.code_no_results = No source code matching your search term found.
+search.code_search_unavailable = Currently code search is not available. Please contact your site administrator.
 
 settings = Settings
 settings.desc = Settings is where you can manage the settings for the repository