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

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.

integrations/repo_search_test.go

@@ -17,10 +17,7 @@ import (
 )
 
 func resultFilenames(t testing.TB, doc *HTMLDoc) []string {
-	resultsSelection := doc.doc.Find(".repository.search")
-	assert.EqualValues(t, 1, resultsSelection.Length(),
-		"Invalid template (repo search template has changed?)")
-	filenameSelections := resultsSelection.Find(".repo-search-result").Find(".header").Find("span.file")
+	filenameSelections := doc.doc.Find(".repository.search").Find(".repo-search-result").Find(".header").Find("span.file")
 	result := make([]string, filenameSelections.Length())
 	filenameSelections.Each(func(i int, selection *goquery.Selection) {
 		result[i] = selection.Text()

models/db/engine.go

@@ -65,6 +65,7 @@ type Engine interface {
 	Query(...interface{}) ([]map[string][]byte, error)
 	Cols(...string) *xorm.Session
 	Context(ctx context.Context) *xorm.Session
+	Ping() error
 }
 
 // TableInfo returns table's information via an object

models/issue.go

@@ -1859,7 +1859,7 @@ func GetRepoIssueStats(repoID, uid int64, filterMode int, isPull bool) (numOpen,
 }
 
 // SearchIssueIDsByKeyword search issues on database
-func SearchIssueIDsByKeyword(kw string, repoIDs []int64, limit, start int) (int64, []int64, error) {
+func SearchIssueIDsByKeyword(ctx context.Context, kw string, repoIDs []int64, limit, start int) (int64, []int64, error) {
 	repoCond := builder.In("repo_id", repoIDs)
 	subQuery := builder.Select("id").From("issue").Where(repoCond)
 	kw = strings.ToUpper(kw)
@@ -1884,7 +1884,7 @@ func SearchIssueIDsByKeyword(kw string, repoIDs []int64, limit, start int) (int6
 		ID          int64
 		UpdatedUnix int64
 	}, 0, limit)
-	err := db.GetEngine(db.DefaultContext).Distinct("id", "updated_unix").Table("issue").Where(cond).
+	err := db.GetEngine(ctx).Distinct("id", "updated_unix").Table("issue").Where(cond).
 		OrderBy("`updated_unix` DESC").Limit(limit, start).
 		Find(&res)
 	if err != nil {
@@ -1894,7 +1894,7 @@ func SearchIssueIDsByKeyword(kw string, repoIDs []int64, limit, start int) (int6
 		ids = append(ids, r.ID)
 	}
 
-	total, err := db.GetEngine(db.DefaultContext).Distinct("id").Table("issue").Where(cond).Count()
+	total, err := db.GetEngine(ctx).Distinct("id").Table("issue").Where(cond).Count()
 	if err != nil {
 		return 0, nil, err
 	}

models/issue_test.go

@@ -5,6 +5,7 @@
 package models
 
 import (
+	"context"
 	"fmt"
 	"sort"
 	"sync"
@@ -303,23 +304,23 @@ func TestIssue_loadTotalTimes(t *testing.T) {
 
 func TestIssue_SearchIssueIDsByKeyword(t *testing.T) {
 	assert.NoError(t, unittest.PrepareTestDatabase())
-	total, ids, err := SearchIssueIDsByKeyword("issue2", []int64{1}, 10, 0)
+	total, ids, err := SearchIssueIDsByKeyword(context.TODO(), "issue2", []int64{1}, 10, 0)
 	assert.NoError(t, err)
 	assert.EqualValues(t, 1, total)
 	assert.EqualValues(t, []int64{2}, ids)
 
-	total, ids, err = SearchIssueIDsByKeyword("first", []int64{1}, 10, 0)
+	total, ids, err = SearchIssueIDsByKeyword(context.TODO(), "first", []int64{1}, 10, 0)
 	assert.NoError(t, err)
 	assert.EqualValues(t, 1, total)
 	assert.EqualValues(t, []int64{1}, ids)
 
-	total, ids, err = SearchIssueIDsByKeyword("for", []int64{1}, 10, 0)
+	total, ids, err = SearchIssueIDsByKeyword(context.TODO(), "for", []int64{1}, 10, 0)
 	assert.NoError(t, err)
 	assert.EqualValues(t, 5, total)
 	assert.ElementsMatch(t, []int64{1, 2, 3, 5, 11}, ids)
 
 	// issue1's comment id 2
-	total, ids, err = SearchIssueIDsByKeyword("good", []int64{1}, 10, 0)
+	total, ids, err = SearchIssueIDsByKeyword(context.TODO(), "good", []int64{1}, 10, 0)
 	assert.NoError(t, err)
 	assert.EqualValues(t, 1, total)
 	assert.EqualValues(t, []int64{1}, ids)
@@ -464,7 +465,7 @@ func TestCorrectIssueStats(t *testing.T) {
 	wg.Wait()
 
 	// Now we will get all issueID's that match the "Bugs are nasty" query.
-	total, ids, err := SearchIssueIDsByKeyword("Bugs are nasty", []int64{1}, issueAmount, 0)
+	total, ids, err := SearchIssueIDsByKeyword(context.TODO(), "Bugs are nasty", []int64{1}, issueAmount, 0)
 
 	// Just to be sure.
 	assert.NoError(t, err)

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["WikiCloneLink"] = repo.WikiCloneLink()
 

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)
@@ -319,7 +328,7 @@ func (b *BleveIndexer) Delete(repoID int64) error {
 
 // Search searches for files in the specified repo.
 // Returns the matching file-paths
-func (b *BleveIndexer) Search(repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) {
+func (b *BleveIndexer) Search(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) {
 	var (
 		indexerQuery query.Query
 		keywordQuery query.Query
@@ -372,7 +381,7 @@ func (b *BleveIndexer) Search(repoIDs []int64, language, keyword string, page, p
 		searchRequest.AddFacet("languages", bleve.NewFacetRequest("Language", 10))
 	}
 
-	result, err := b.indexer.Search(searchRequest)
+	result, err := b.indexer.SearchInContext(ctx, searchRequest)
 	if err != nil {
 		return 0, nil, nil, err
 	}