From 49396bac85d78adf79f2c684d2c1d667b70af01e Mon Sep 17 00:00:00 2001 From: Lauris BH Date: Fri, 21 Jan 2022 15:43:52 +0200 Subject: [PATCH 1/7] Automatically pause queue if issue index service is unavailable --- modules/indexer/issues/bleve.go | 9 +++ modules/indexer/issues/db.go | 9 +++ modules/indexer/issues/elastic_search.go | 82 ++++++++++++++++++++---- modules/indexer/issues/indexer.go | 38 ++++++++++- 4 files changed, 125 insertions(+), 13 deletions(-) diff --git a/modules/indexer/issues/bleve.go b/modules/indexer/issues/bleve.go index d986a0e55ef9c..0c23181fd74c0 100644 --- a/modules/indexer/issues/bleve.go +++ b/modules/indexer/issues/bleve.go @@ -186,6 +186,15 @@ func (b *BleveIndexer) Init() (bool, error) { return false, err } +// SetAvailabilityChangeCallback does nothing +func (b *BleveIndexer) SetAvailabilityChangeCallback(callback func(bool)) { +} + +// Ping does nothing +func (b *BleveIndexer) Ping() bool { + return true +} + // Close will close the bleve indexer func (b *BleveIndexer) Close() { if b.indexer != nil { diff --git a/modules/indexer/issues/db.go b/modules/indexer/issues/db.go index f02cbddce8fb5..acc6443ca7031 100644 --- a/modules/indexer/issues/db.go +++ b/modules/indexer/issues/db.go @@ -14,6 +14,15 @@ func (db *DBIndexer) Init() (bool, error) { return false, nil } +// SetAvailabilityChangeCallback dummy function +func (db *DBIndexer) SetAvailabilityChangeCallback(callback func(bool)) { +} + +// Ping checks if database is available +func (db *DBIndexer) Ping() bool { + return models.Ping() != nil +} + // Index dummy function func (db *DBIndexer) Index(issue []*IndexerData) error { return nil diff --git a/modules/indexer/issues/elastic_search.go b/modules/indexer/issues/elastic_search.go index 187b69b74946b..9dae15a2b0500 100644 --- a/modules/indexer/issues/elastic_search.go +++ b/modules/indexer/issues/elastic_search.go @@ -20,8 +20,11 @@ var _ Indexer = &ElasticSearchIndexer{} // ElasticSearchIndexer implements Indexer interface type ElasticSearchIndexer struct { - client *elastic.Client - indexerName string + client *elastic.Client + indexerName string + available bool + availabilityCallback func(bool) + stopTimer chan struct{} } type elasticLogger struct { @@ -56,10 +59,27 @@ func NewElasticSearchIndexer(url, indexerName string) (*ElasticSearchIndexer, er return nil, err } - return &ElasticSearchIndexer{ + indexer := &ElasticSearchIndexer{ client: client, indexerName: indexerName, - }, nil + 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 + } + } + }() + + return indexer, nil } const ( @@ -96,7 +116,7 @@ func (b *ElasticSearchIndexer) Init() (bool, error) { ctx := context.Background() exists, err := b.client.IndexExists(b.indexerName).Do(ctx) if err != nil { - return false, err + return false, b.checkError(err) } if !exists { @@ -104,7 +124,7 @@ func (b *ElasticSearchIndexer) Init() (bool, error) { createIndex, err := b.client.CreateIndex(b.indexerName).BodyString(mapping).Do(ctx) if err != nil { - return false, err + return false, b.checkError(err) } if !createIndex.Acknowledged { return false, errors.New("init failed") @@ -115,6 +135,16 @@ func (b *ElasticSearchIndexer) Init() (bool, error) { return true, 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 +} + // Index will save the index data func (b *ElasticSearchIndexer) Index(issues []*IndexerData) error { if len(issues) == 0 { @@ -132,7 +162,7 @@ func (b *ElasticSearchIndexer) Index(issues []*IndexerData) error { "comments": issue.Comments, }). Do(context.Background()) - return err + return b.checkError(err) } reqs := make([]elastic.BulkableRequest, 0) @@ -155,7 +185,7 @@ func (b *ElasticSearchIndexer) Index(issues []*IndexerData) error { Index(b.indexerName). Add(reqs...). Do(context.Background()) - return err + return b.checkError(err) } // Delete deletes indexes by ids @@ -167,7 +197,7 @@ func (b *ElasticSearchIndexer) Delete(ids ...int64) error { Index(b.indexerName). Id(fmt.Sprintf("%d", ids[0])). Do(context.Background()) - return err + return b.checkError(err) } reqs := make([]elastic.BulkableRequest, 0) @@ -183,7 +213,7 @@ func (b *ElasticSearchIndexer) Delete(ids ...int64) error { Index(b.indexerName). Add(reqs...). Do(context.Background()) - return err + return b.checkError(err) } // Search searches for issues by given conditions. @@ -207,7 +237,7 @@ func (b *ElasticSearchIndexer) Search(keyword string, repoIDs []int64, limit, st From(start).Size(limit). Do(context.Background()) if err != nil { - return nil, err + return nil, b.checkError(err) } hits := make([]Match, 0, limit) @@ -225,4 +255,32 @@ func (b *ElasticSearchIndexer) Search(keyword string, repoIDs []int64, limit, st } // Close implements indexer -func (b *ElasticSearchIndexer) Close() {} +func (b *ElasticSearchIndexer) Close() { + close(b.stopTimer) +} + +func (b *ElasticSearchIndexer) checkError(err error) error { + if elastic.IsConnErr(err) && b.available { + 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) + } +} diff --git a/modules/indexer/issues/indexer.go b/modules/indexer/issues/indexer.go index 729981ec71d2a..dbd7b02755436 100644 --- a/modules/indexer/issues/indexer.go +++ b/modules/indexer/issues/indexer.go @@ -47,6 +47,8 @@ type SearchResult struct { // Indexer defines an interface to indexer issues contents type Indexer interface { Init() (bool, error) + Ping() bool + SetAvailabilityChangeCallback(callback func(bool)) Index(issue []*IndexerData) error Delete(ids ...int64) error Search(kw string, repoIDs []int64, limit, start int) (*SearchResult, error) @@ -111,6 +113,7 @@ func InitIssueIndexer(syncReindex bool) { } iData := make([]*IndexerData, 0, len(data)) + unhandled := make([]queue.Data, 0, len(data)) for _, datum := range data { indexerData, ok := datum.(*IndexerData) if !ok { @@ -119,13 +122,34 @@ func InitIssueIndexer(syncReindex bool) { } log.Trace("IndexerData Process: %d %v %t", indexerData.ID, indexerData.IDs, indexerData.IsDelete) if indexerData.IsDelete { - _ = indexer.Delete(indexerData.IDs...) + if err := indexer.Delete(indexerData.IDs...); err != nil { + log.Error("Error whilst deleting from index: %v Error: %v", indexerData.IDs, err) + if indexer.Ping() { + continue + } + // Add back to queue + unhandled = append(unhandled, datum) + } continue } iData = append(iData, indexerData) } + if len(unhandled) > 0 { + for _, indexerData := range iData { + unhandled = append(unhandled, indexerData) + } + return unhandled + } if err := indexer.Index(iData); err != nil { log.Error("Error whilst indexing: %v Error: %v", iData, err) + if indexer.Ping() { + return nil + } + // Add back to queue + for _, indexerData := range iData { + unhandled = append(unhandled, indexerData) + } + return unhandled } return nil } @@ -193,6 +217,18 @@ func InitIssueIndexer(syncReindex bool) { log.Fatal("Unknown issue indexer type: %s", setting.Indexer.IssueType) } + if queue, ok := issueIndexerQueue.(queue.Pausable); ok { + holder.get().SetAvailabilityChangeCallback(func(available bool) { + if !available { + log.Info("Issue index queue paused") + queue.Pause() + } else { + log.Info("Issue index queue resumed") + queue.Resume() + } + }) + } + // Start processing the queue go graceful.GetManager().RunWithShutdownFns(issueIndexerQueue.Run) From 3acdb133b9441e3d01d1c6f981360dd2f60d9e0b Mon Sep 17 00:00:00 2001 From: Lauris BH Date: Fri, 21 Jan 2022 17:56:42 +0200 Subject: [PATCH 2/7] Handle keyword search error when issue indexer service is not available --- models/db/engine.go | 1 + modules/indexer/issues/db.go | 21 ++++++++++++--------- modules/indexer/issues/elastic_search.go | 4 +++- modules/indexer/issues/indexer.go | 11 +++++++++++ options/locale/locale_en-US.ini | 1 + routers/web/repo/issue.go | 7 +++++-- templates/shared/issuelist.tmpl | 5 +++++ 7 files changed, 38 insertions(+), 12 deletions(-) diff --git a/models/db/engine.go b/models/db/engine.go index 0604d939d3797..9f38af3c67060 100755 --- a/models/db/engine.go +++ b/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 diff --git a/modules/indexer/issues/db.go b/modules/indexer/issues/db.go index acc6443ca7031..b1bf8795f5db5 100644 --- a/modules/indexer/issues/db.go +++ b/modules/indexer/issues/db.go @@ -4,41 +4,44 @@ package issues -import "code.gitea.io/gitea/models" +import ( + "code.gitea.io/gitea/models" + "code.gitea.io/gitea/models/db" +) // DBIndexer implements Indexer interface to use database's like search type DBIndexer struct{} // Init dummy function -func (db *DBIndexer) Init() (bool, error) { +func (i *DBIndexer) Init() (bool, error) { return false, nil } // SetAvailabilityChangeCallback dummy function -func (db *DBIndexer) SetAvailabilityChangeCallback(callback func(bool)) { +func (i *DBIndexer) SetAvailabilityChangeCallback(callback func(bool)) { } // Ping checks if database is available -func (db *DBIndexer) Ping() bool { - return models.Ping() != nil +func (i *DBIndexer) Ping() bool { + return db.GetEngine(db.DefaultContext).Ping() != nil } // Index dummy function -func (db *DBIndexer) Index(issue []*IndexerData) error { +func (i *DBIndexer) Index(issue []*IndexerData) error { return nil } // Delete dummy function -func (db *DBIndexer) Delete(ids ...int64) error { +func (i *DBIndexer) Delete(ids ...int64) error { return nil } // Close dummy function -func (db *DBIndexer) Close() { +func (i *DBIndexer) Close() { } // Search dummy function -func (db *DBIndexer) Search(kw string, repoIDs []int64, limit, start int) (*SearchResult, error) { +func (i *DBIndexer) Search(kw string, repoIDs []int64, limit, start int) (*SearchResult, error) { total, ids, err := models.SearchIssueIDsByKeyword(kw, repoIDs, limit, start) if err != nil { return nil, err diff --git a/modules/indexer/issues/elastic_search.go b/modules/indexer/issues/elastic_search.go index 9dae15a2b0500..6ad061802b80a 100644 --- a/modules/indexer/issues/elastic_search.go +++ b/modules/indexer/issues/elastic_search.go @@ -8,6 +8,7 @@ import ( "context" "errors" "fmt" + "net" "strconv" "time" @@ -260,7 +261,8 @@ func (b *ElasticSearchIndexer) Close() { } func (b *ElasticSearchIndexer) checkError(err error) error { - if elastic.IsConnErr(err) && b.available { + 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) diff --git a/modules/indexer/issues/indexer.go b/modules/indexer/issues/indexer.go index dbd7b02755436..e9b0b9e5a26d4 100644 --- a/modules/indexer/issues/indexer.go +++ b/modules/indexer/issues/indexer.go @@ -387,3 +387,14 @@ func SearchIssuesByKeyword(repoIDs []int64, keyword string) ([]int64, error) { } return issueIDs, nil } + +// IsAvailable checks if issue indexer is available +func IsAvailable() bool { + indexer := holder.get() + if indexer == nil { + log.Error("IsAvailable(): unable to get indexer!") + return false + } + + return indexer.Ping() +} diff --git a/options/locale/locale_en-US.ini b/options/locale/locale_en-US.ini index 1f778d329bcde..d5b1768f6920c 100644 --- a/options/locale/locale_en-US.ini +++ b/options/locale/locale_en-US.ini @@ -1262,6 +1262,7 @@ issues.filter_sort.moststars = Most stars issues.filter_sort.feweststars = Fewest stars issues.filter_sort.mostforks = Most forks issues.filter_sort.fewestforks = Fewest forks +issues.keyword_search_unavailable = Currently searhing by keyword is not available. Please contact your site administrator. issues.action_open = Open issues.action_close = Close issues.action_label = Label diff --git a/routers/web/repo/issue.go b/routers/web/repo/issue.go index 4f2716763acd3..df1b9876c079a 100644 --- a/routers/web/repo/issue.go +++ b/routers/web/repo/issue.go @@ -163,8 +163,11 @@ func issues(ctx *context.Context, milestoneID, projectID int64, isPullOption uti if len(keyword) > 0 { issueIDs, err = issue_indexer.SearchIssuesByKeyword([]int64{repo.ID}, keyword) if err != nil { - ctx.ServerError("issueIndexer.Search", err) - return + if issue_indexer.IsAvailable() { + ctx.ServerError("issueIndexer.Search", err) + return + } + ctx.Data["IssueIndexerUnavailable"] = true } if len(issueIDs) == 0 { forceEmpty = true diff --git a/templates/shared/issuelist.tmpl b/templates/shared/issuelist.tmpl index 3050107123cb6..687bbdee9476d 100644 --- a/templates/shared/issuelist.tmpl +++ b/templates/shared/issuelist.tmpl @@ -139,5 +139,10 @@ {{end}} + {{if .IssueIndexerUnavailable}} +
+

{{$.i18n.Tr "repo.issues.keyword_search_unavailable"}}

+
+ {{end}} {{template "base/paginate" .}} From 3b28eefc0fdd8ce11c348380c7e52bf3499ed239 Mon Sep 17 00:00:00 2001 From: Lauris BH Date: Mon, 24 Jan 2022 16:59:01 +0200 Subject: [PATCH 3/7] Implement automatic disabling and resume of code indexer queue --- .../doc/developers/hacking-on-gitea.en-us.md | 21 +++++ modules/context/repo.go | 4 + modules/indexer/code/bleve.go | 9 ++ modules/indexer/code/elastic_search.go | 92 +++++++++++++++---- modules/indexer/code/indexer.go | 34 ++++++- modules/indexer/code/wrapped.go | 21 +++++ options/locale/locale_en-US.ini | 3 + routers/web/explore/code.go | 18 +++- routers/web/repo/search.go | 9 +- templates/explore/code.tmpl | 14 ++- templates/repo/home.tmpl | 9 +- templates/repo/search.tmpl | 88 ++++++++++-------- 12 files changed, 250 insertions(+), 72 deletions(-) diff --git a/docs/content/doc/developers/hacking-on-gitea.en-us.md b/docs/content/doc/developers/hacking-on-gitea.en-us.md index 5481a2f1fef47..6e9a98e1264cb 100644 --- a/docs/content/doc/developers/hacking-on-gitea.en-us.md +++ b/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. diff --git a/modules/context/repo.go b/modules/context/repo.go index 97b417ffd165a..76fe1c5676a51 100644 --- a/modules/context/repo.go +++ b/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() diff --git a/modules/indexer/code/bleve.go b/modules/indexer/code/bleve.go index cfadcfebd8a2f..14419e77a0b3e 100644 --- a/modules/indexer/code/bleve.go +++ b/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) diff --git a/modules/indexer/code/elastic_search.go b/modules/indexer/code/elastic_search.go index 9bd2fa301e875..c1d4c29b8a52f 100644 --- a/modules/indexer/code/elastic_search.go +++ b/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) + } +} diff --git a/modules/indexer/code/indexer.go b/modules/indexer/code/indexer.go index 9ae3abff60bab..f35afcc479004 100644 --- a/modules/indexer/code/indexer.go +++ b/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) { diff --git a/modules/indexer/code/wrapped.go b/modules/indexer/code/wrapped.go index 56baadd6fc6f6..86407bbad7221 100644 --- a/modules/indexer/code/wrapped.go +++ b/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 { diff --git a/options/locale/locale_en-US.ini b/options/locale/locale_en-US.ini index d5b1768f6920c..4923d8b7b72bb 100644 --- a/options/locale/locale_en-US.ini +++ b/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 %s +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 diff --git a/routers/web/explore/code.go b/routers/web/explore/code.go index d2acefde92793..ed1915d337ee7 100644 --- a/routers/web/explore/code.go +++ b/routers/web/explore/code.go @@ -89,15 +89,25 @@ func Code(ctx *context.Context) { total, searchResults, searchResultLanguages, err = code_indexer.PerformSearch(repoIDs, language, keyword, page, setting.UI.RepoSearchPagingNum, isMatch) if err != nil { - ctx.ServerError("SearchResults", err) - return + if code_indexer.IsAvailable() { + ctx.ServerError("SearchResults", err) + return + } + ctx.Data["CodeIndexerUnavailable"] = true + } else { + ctx.Data["CodeIndexerUnavailable"] = !code_indexer.IsAvailable() } // if non-login user or isAdmin, no need to check UnitTypeCode } else if (ctx.User == nil && len(repoIDs) > 0) || isAdmin { total, searchResults, searchResultLanguages, err = code_indexer.PerformSearch(repoIDs, language, keyword, page, setting.UI.RepoSearchPagingNum, isMatch) if err != nil { - ctx.ServerError("SearchResults", err) - return + if code_indexer.IsAvailable() { + ctx.ServerError("SearchResults", err) + return + } + ctx.Data["CodeIndexerUnavailable"] = true + } else { + ctx.Data["CodeIndexerUnavailable"] = !code_indexer.IsAvailable() } loadRepoIDs := make([]int64, 0, len(searchResults)) diff --git a/routers/web/repo/search.go b/routers/web/repo/search.go index 67539c3d7eb8a..04bb5b844a4b4 100644 --- a/routers/web/repo/search.go +++ b/routers/web/repo/search.go @@ -33,8 +33,13 @@ func Search(ctx *context.Context) { total, searchResults, searchResultLanguages, err := code_indexer.PerformSearch([]int64{ctx.Repo.Repository.ID}, language, keyword, page, setting.UI.RepoSearchPagingNum, isMatch) if err != nil { - ctx.ServerError("SearchResults", err) - return + if code_indexer.IsAvailable() { + ctx.ServerError("SearchResults", err) + return + } + ctx.Data["CodeIndexerUnavailable"] = true + } else { + ctx.Data["CodeIndexerUnavailable"] = !code_indexer.IsAvailable() } ctx.Data["Keyword"] = keyword ctx.Data["Language"] = language diff --git a/templates/explore/code.tmpl b/templates/explore/code.tmpl index 573c096f83f0e..e1056ab3129d7 100644 --- a/templates/explore/code.tmpl +++ b/templates/explore/code.tmpl @@ -5,21 +5,25 @@
- -
- {{svg "octicon-triangle-down" 14 "dropdown icon"}} + +
+ {{svg "octicon-triangle-down" 14 "dropdown icon"}}
{{.i18n.Tr (printf "explore.search.%s" (or .queryType "fuzzy"))}}
- +
- {{if .SearchResults}} + {{if .CodeIndexerUnavailable }} +
+

{{$.i18n.Tr "explore.code_search_unavailable"}}

+
+ {{else if .SearchResults}}

{{.i18n.Tr "explore.code_search_results" (.Keyword|Escape) | Str2html }}

diff --git a/templates/repo/home.tmpl b/templates/repo/home.tmpl index 9924ae547dcf0..6d525c24da46f 100644 --- a/templates/repo/home.tmpl +++ b/templates/repo/home.tmpl @@ -13,9 +13,12 @@
-
- -
diff --git a/templates/repo/search.tmpl b/templates/repo/search.tmpl index f2c2269af7fe6..fe15b89f81831 100644 --- a/templates/repo/search.tmpl +++ b/templates/repo/search.tmpl @@ -5,60 +5,68 @@
- -
- {{svg "octicon-triangle-down" 14 "dropdown icon"}} + +
+ {{svg "octicon-triangle-down" 14 "dropdown icon"}}
{{.i18n.Tr (printf "repo.search.%s" (or .queryType "fuzzy"))}}
- +
- {{if .Keyword}} + {{if .CodeIndexerUnavailable }} +
+

{{$.i18n.Tr "repo.search.code_search_unavailable"}}

+
+ {{else if .Keyword}}

{{.i18n.Tr "repo.search.results" (.Keyword|Escape) (.RepoLink|Escape) (.RepoName|Escape) | Str2html }}

-
- {{range $term := .SearchResultLanguages}} - - - {{$term.Language}} -
{{$term.Count}}
- - {{end}} -
-
- {{range $result := .SearchResults}} -
-

- {{.Filename}} - {{$.i18n.Tr "repo.diff.view_file"}} -

-
-
- - - - - - - -
- {{range .LineNumbers}} - {{.}} - {{end}} - {{.FormattedLines | Safe}}
+ {{if .SearchResults}} +
+ {{range $term := .SearchResultLanguages}} + + + {{$term.Language}} +
{{$term.Count}}
+ + {{end}} +
+
+ {{range $result := .SearchResults}} +
+

+ {{.Filename}} + {{$.i18n.Tr "repo.diff.view_file"}} +

+
+
+ + + + + + + +
+ {{range .LineNumbers}} + {{.}} + {{end}} + {{.FormattedLines | Safe}}
+
+ {{template "shared/searchbottom" dict "root" $ "result" .}}
- {{template "shared/searchbottom" dict "root" $ "result" .}} -
- {{end}} -
- {{template "base/paginate" .}} + {{end}} +
+ {{template "base/paginate" .}} + {{else}} +
{{$.i18n.Tr "repo.search.code_no_results"}}
+ {{end}} {{end}}
From 09ed40d2890d797ec89db882d694ea31de412154 Mon Sep 17 00:00:00 2001 From: Lauris BH Date: Mon, 24 Jan 2022 18:03:34 +0200 Subject: [PATCH 4/7] Fix tests --- integrations/repo_search_test.go | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/integrations/repo_search_test.go b/integrations/repo_search_test.go index 0782988363bcc..a1cc5811b9f43 100644 --- a/integrations/repo_search_test.go +++ b/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() From f6f7beb5a5dc30afe1c0f8c11de490f69ffff882 Mon Sep 17 00:00:00 2001 From: Lauris BH Date: Tue, 25 Jan 2022 22:16:39 +0200 Subject: [PATCH 5/7] Fix locking and correct context usage --- models/issue.go | 6 +-- models/issue_test.go | 11 ++--- modules/indexer/code/bleve.go | 4 +- modules/indexer/code/elastic_search.go | 51 +++++++++++++++++------- modules/indexer/code/indexer.go | 2 +- modules/indexer/code/indexer_test.go | 3 +- modules/indexer/code/search.go | 5 ++- modules/indexer/code/wrapped.go | 4 +- modules/indexer/issues/bleve.go | 5 ++- modules/indexer/issues/bleve_test.go | 3 +- modules/indexer/issues/db.go | 6 ++- modules/indexer/issues/elastic_search.go | 51 +++++++++++++++++------- modules/indexer/issues/indexer.go | 6 +-- modules/indexer/issues/indexer_test.go | 17 ++++---- routers/api/v1/repo/issue.go | 4 +- routers/web/repo/issue.go | 2 +- routers/web/user/home.go | 6 +-- 17 files changed, 118 insertions(+), 68 deletions(-) diff --git a/models/issue.go b/models/issue.go index cb5791be9e0c4..3a61b085dc3b2 100644 --- a/models/issue.go +++ b/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 } diff --git a/models/issue_test.go b/models/issue_test.go index e2759ba38f0f1..9344d385a7c75 100644 --- a/models/issue_test.go +++ b/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) diff --git a/modules/indexer/code/bleve.go b/modules/indexer/code/bleve.go index 14419e77a0b3e..e2e15320952d9 100644 --- a/modules/indexer/code/bleve.go +++ b/modules/indexer/code/bleve.go @@ -328,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 @@ -381,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 } diff --git a/modules/indexer/code/elastic_search.go b/modules/indexer/code/elastic_search.go index c1d4c29b8a52f..e541b0224ed18 100644 --- a/modules/indexer/code/elastic_search.go +++ b/modules/indexer/code/elastic_search.go @@ -13,12 +13,14 @@ import ( "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" @@ -46,6 +48,7 @@ type ElasticSearchIndexer struct { available bool availabilityCallback func(bool) stopTimer chan struct{} + lock sync.RWMutex } type elasticLogger struct { @@ -144,7 +147,7 @@ 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, b.checkError(err) @@ -198,11 +201,15 @@ func (b *ElasticSearchIndexer) init() (bool, error) { // 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 } @@ -305,7 +312,7 @@ func (b *ElasticSearchIndexer) Index(ctx context.Context, repo *repo_model.Repos _, err := b.client.Bulk(). Index(b.indexerAliasName). Add(reqs...). - Do(context.Background()) + Do(ctx) return b.checkError(err) } return nil @@ -315,7 +322,7 @@ 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()) + Do(graceful.GetManager().HammerContext()) return b.checkError(err) } @@ -397,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 @@ -438,7 +445,7 @@ 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, b.checkError(err) } @@ -452,7 +459,7 @@ 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, b.checkError(err) } @@ -469,7 +476,7 @@ 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, b.checkError(err) } @@ -486,27 +493,41 @@ func (b *ElasticSearchIndexer) Close() { 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) - } + 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.available { + if b.Ping() { return } // Request cluster state to check if elastic is available again - _, err := b.client.ClusterState().Do(context.Background()) + _, 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 = true + + 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) } } diff --git a/modules/indexer/code/indexer.go b/modules/indexer/code/indexer.go index f35afcc479004..d897fcccd5489 100644 --- a/modules/indexer/code/indexer.go +++ b/modules/indexer/code/indexer.go @@ -46,7 +46,7 @@ type Indexer interface { 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) + Search(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) Close() } diff --git a/modules/indexer/code/indexer_test.go b/modules/indexer/code/indexer_test.go index 0f9915c84b7b4..d56c33653f32e 100644 --- a/modules/indexer/code/indexer_test.go +++ b/modules/indexer/code/indexer_test.go @@ -5,6 +5,7 @@ package code import ( + "context" "path/filepath" "testing" @@ -65,7 +66,7 @@ func testIndexer(name string, t *testing.T, indexer Indexer) { for _, kw := range keywords { t.Run(kw.Keyword, func(t *testing.T) { - total, res, langs, err := indexer.Search(kw.RepoIDs, "", kw.Keyword, 1, 10, false) + total, res, langs, err := indexer.Search(context.TODO(), kw.RepoIDs, "", kw.Keyword, 1, 10, false) assert.NoError(t, err) assert.EqualValues(t, len(kw.IDs), total) assert.Len(t, langs, kw.Langs) diff --git a/modules/indexer/code/search.go b/modules/indexer/code/search.go index bb8dcf16b3f49..bb7715bafcdc6 100644 --- a/modules/indexer/code/search.go +++ b/modules/indexer/code/search.go @@ -6,6 +6,7 @@ package code import ( "bytes" + "context" "strings" "code.gitea.io/gitea/modules/highlight" @@ -106,12 +107,12 @@ func searchResult(result *SearchResult, startIndex, endIndex int) (*Result, erro } // PerformSearch perform a search on a repository -func PerformSearch(repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int, []*Result, []*SearchResultLanguages, error) { +func PerformSearch(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int, []*Result, []*SearchResultLanguages, error) { if len(keyword) == 0 { return 0, nil, nil, nil } - total, results, resultLanguages, err := indexer.Search(repoIDs, language, keyword, page, pageSize, isMatch) + total, results, resultLanguages, err := indexer.Search(ctx, repoIDs, language, keyword, page, pageSize, isMatch) if err != nil { return 0, nil, nil, err } diff --git a/modules/indexer/code/wrapped.go b/modules/indexer/code/wrapped.go index 86407bbad7221..ba58236fbad13 100644 --- a/modules/indexer/code/wrapped.go +++ b/modules/indexer/code/wrapped.go @@ -93,12 +93,12 @@ func (w *wrappedIndexer) Delete(repoID int64) error { return indexer.Delete(repoID) } -func (w *wrappedIndexer) Search(repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) { +func (w *wrappedIndexer) Search(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) { indexer, err := w.get() if err != nil { return 0, nil, nil, err } - return indexer.Search(repoIDs, language, keyword, page, pageSize, isMatch) + return indexer.Search(ctx, repoIDs, language, keyword, page, pageSize, isMatch) } func (w *wrappedIndexer) Close() { diff --git a/modules/indexer/issues/bleve.go b/modules/indexer/issues/bleve.go index 0c23181fd74c0..c298b7de3e1b0 100644 --- a/modules/indexer/issues/bleve.go +++ b/modules/indexer/issues/bleve.go @@ -5,6 +5,7 @@ package issues import ( + "context" "fmt" "os" "strconv" @@ -238,7 +239,7 @@ func (b *BleveIndexer) Delete(ids ...int64) error { // Search searches for issues by given conditions. // Returns the matching issue IDs -func (b *BleveIndexer) Search(keyword string, repoIDs []int64, limit, start int) (*SearchResult, error) { +func (b *BleveIndexer) Search(ctx context.Context, keyword string, repoIDs []int64, limit, start int) (*SearchResult, error) { var repoQueriesP []*query.NumericRangeQuery for _, repoID := range repoIDs { repoQueriesP = append(repoQueriesP, numericEqualityQuery(repoID, "RepoID")) @@ -258,7 +259,7 @@ func (b *BleveIndexer) Search(keyword string, repoIDs []int64, limit, start int) search := bleve.NewSearchRequestOptions(indexerQuery, limit, start, false) search.SortBy([]string{"-_score"}) - result, err := b.indexer.Search(search) + result, err := b.indexer.SearchInContext(ctx, search) if err != nil { return nil, err } diff --git a/modules/indexer/issues/bleve_test.go b/modules/indexer/issues/bleve_test.go index df036fb573370..926c32e242e8e 100644 --- a/modules/indexer/issues/bleve_test.go +++ b/modules/indexer/issues/bleve_test.go @@ -5,6 +5,7 @@ package issues import ( + "context" "os" "testing" @@ -84,7 +85,7 @@ func TestBleveIndexAndSearch(t *testing.T) { } for _, kw := range keywords { - res, err := indexer.Search(kw.Keyword, []int64{2}, 10, 0) + res, err := indexer.Search(context.TODO(), kw.Keyword, []int64{2}, 10, 0) assert.NoError(t, err) ids := make([]int64, 0, len(res.Hits)) diff --git a/modules/indexer/issues/db.go b/modules/indexer/issues/db.go index b1bf8795f5db5..e2badf64f2e60 100644 --- a/modules/indexer/issues/db.go +++ b/modules/indexer/issues/db.go @@ -5,6 +5,8 @@ package issues import ( + "context" + "code.gitea.io/gitea/models" "code.gitea.io/gitea/models/db" ) @@ -41,8 +43,8 @@ func (i *DBIndexer) Close() { } // Search dummy function -func (i *DBIndexer) Search(kw string, repoIDs []int64, limit, start int) (*SearchResult, error) { - total, ids, err := models.SearchIssueIDsByKeyword(kw, repoIDs, limit, start) +func (i *DBIndexer) Search(ctx context.Context, kw string, repoIDs []int64, limit, start int) (*SearchResult, error) { + total, ids, err := models.SearchIssueIDsByKeyword(ctx, kw, repoIDs, limit, start) if err != nil { return nil, err } diff --git a/modules/indexer/issues/elastic_search.go b/modules/indexer/issues/elastic_search.go index 6ad061802b80a..5e423ab0c15bd 100644 --- a/modules/indexer/issues/elastic_search.go +++ b/modules/indexer/issues/elastic_search.go @@ -10,8 +10,10 @@ import ( "fmt" "net" "strconv" + "sync" "time" + "code.gitea.io/gitea/modules/graceful" "code.gitea.io/gitea/modules/log" "github.com/olivere/elastic/v7" @@ -26,6 +28,7 @@ type ElasticSearchIndexer struct { available bool availabilityCallback func(bool) stopTimer chan struct{} + lock sync.RWMutex } type elasticLogger struct { @@ -114,7 +117,7 @@ const ( // Init will initialize the indexer func (b *ElasticSearchIndexer) Init() (bool, error) { - ctx := context.Background() + ctx := graceful.GetManager().HammerContext() exists, err := b.client.IndexExists(b.indexerName).Do(ctx) if err != nil { return false, b.checkError(err) @@ -138,11 +141,15 @@ func (b *ElasticSearchIndexer) Init() (bool, error) { // 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 } @@ -162,7 +169,7 @@ func (b *ElasticSearchIndexer) Index(issues []*IndexerData) error { "content": issue.Content, "comments": issue.Comments, }). - Do(context.Background()) + Do(graceful.GetManager().HammerContext()) return b.checkError(err) } @@ -185,7 +192,7 @@ func (b *ElasticSearchIndexer) Index(issues []*IndexerData) error { _, err := b.client.Bulk(). Index(b.indexerName). Add(reqs...). - Do(context.Background()) + Do(graceful.GetManager().HammerContext()) return b.checkError(err) } @@ -197,7 +204,7 @@ func (b *ElasticSearchIndexer) Delete(ids ...int64) error { _, err := b.client.Delete(). Index(b.indexerName). Id(fmt.Sprintf("%d", ids[0])). - Do(context.Background()) + Do(graceful.GetManager().HammerContext()) return b.checkError(err) } @@ -213,13 +220,13 @@ func (b *ElasticSearchIndexer) Delete(ids ...int64) error { _, err := b.client.Bulk(). Index(b.indexerName). Add(reqs...). - Do(context.Background()) + Do(graceful.GetManager().HammerContext()) return b.checkError(err) } // Search searches for issues by given conditions. // Returns the matching issue IDs -func (b *ElasticSearchIndexer) Search(keyword string, repoIDs []int64, limit, start int) (*SearchResult, error) { +func (b *ElasticSearchIndexer) Search(ctx context.Context, keyword string, repoIDs []int64, limit, start int) (*SearchResult, error) { kwQuery := elastic.NewMultiMatchQuery(keyword, "title", "content", "comments") query := elastic.NewBoolQuery() query = query.Must(kwQuery) @@ -236,7 +243,7 @@ func (b *ElasticSearchIndexer) Search(keyword string, repoIDs []int64, limit, st Query(query). Sort("_score", false). From(start).Size(limit). - Do(context.Background()) + Do(ctx) if err != nil { return nil, b.checkError(err) } @@ -262,27 +269,41 @@ func (b *ElasticSearchIndexer) Close() { 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) - } + 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.available { + if b.Ping() { return } // Request cluster state to check if elastic is available again - _, err := b.client.ClusterState().Do(context.Background()) + _, 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 = true + + 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) } } diff --git a/modules/indexer/issues/indexer.go b/modules/indexer/issues/indexer.go index e9b0b9e5a26d4..3aaa27eed213b 100644 --- a/modules/indexer/issues/indexer.go +++ b/modules/indexer/issues/indexer.go @@ -51,7 +51,7 @@ type Indexer interface { SetAvailabilityChangeCallback(callback func(bool)) Index(issue []*IndexerData) error Delete(ids ...int64) error - Search(kw string, repoIDs []int64, limit, start int) (*SearchResult, error) + Search(ctx context.Context, kw string, repoIDs []int64, limit, start int) (*SearchResult, error) Close() } @@ -370,7 +370,7 @@ func DeleteRepoIssueIndexer(repo *repo_model.Repository) { // SearchIssuesByKeyword search issue ids by keywords and repo id // WARNNING: You have to ensure user have permission to visit repoIDs' issues -func SearchIssuesByKeyword(repoIDs []int64, keyword string) ([]int64, error) { +func SearchIssuesByKeyword(ctx context.Context, repoIDs []int64, keyword string) ([]int64, error) { var issueIDs []int64 indexer := holder.get() @@ -378,7 +378,7 @@ func SearchIssuesByKeyword(repoIDs []int64, keyword string) ([]int64, error) { log.Error("SearchIssuesByKeyword(): unable to get indexer!") return nil, fmt.Errorf("unable to get issue indexer") } - res, err := indexer.Search(keyword, repoIDs, 50, 0) + res, err := indexer.Search(ctx, keyword, repoIDs, 50, 0) if err != nil { return nil, err } diff --git a/modules/indexer/issues/indexer_test.go b/modules/indexer/issues/indexer_test.go index ee6ebcdd18b96..d516615b56eb3 100644 --- a/modules/indexer/issues/indexer_test.go +++ b/modules/indexer/issues/indexer_test.go @@ -5,6 +5,7 @@ package issues import ( + "context" "os" "path" "path/filepath" @@ -56,19 +57,19 @@ func TestBleveSearchIssues(t *testing.T) { time.Sleep(5 * time.Second) - ids, err := SearchIssuesByKeyword([]int64{1}, "issue2") + ids, err := SearchIssuesByKeyword(context.TODO(), []int64{1}, "issue2") assert.NoError(t, err) assert.EqualValues(t, []int64{2}, ids) - ids, err = SearchIssuesByKeyword([]int64{1}, "first") + ids, err = SearchIssuesByKeyword(context.TODO(), []int64{1}, "first") assert.NoError(t, err) assert.EqualValues(t, []int64{1}, ids) - ids, err = SearchIssuesByKeyword([]int64{1}, "for") + ids, err = SearchIssuesByKeyword(context.TODO(), []int64{1}, "for") assert.NoError(t, err) assert.ElementsMatch(t, []int64{1, 2, 3, 5, 11}, ids) - ids, err = SearchIssuesByKeyword([]int64{1}, "good") + ids, err = SearchIssuesByKeyword(context.TODO(), []int64{1}, "good") assert.NoError(t, err) assert.EqualValues(t, []int64{1}, ids) } @@ -79,19 +80,19 @@ func TestDBSearchIssues(t *testing.T) { setting.Indexer.IssueType = "db" InitIssueIndexer(true) - ids, err := SearchIssuesByKeyword([]int64{1}, "issue2") + ids, err := SearchIssuesByKeyword(context.TODO(), []int64{1}, "issue2") assert.NoError(t, err) assert.EqualValues(t, []int64{2}, ids) - ids, err = SearchIssuesByKeyword([]int64{1}, "first") + ids, err = SearchIssuesByKeyword(context.TODO(), []int64{1}, "first") assert.NoError(t, err) assert.EqualValues(t, []int64{1}, ids) - ids, err = SearchIssuesByKeyword([]int64{1}, "for") + ids, err = SearchIssuesByKeyword(context.TODO(), []int64{1}, "for") assert.NoError(t, err) assert.ElementsMatch(t, []int64{1, 2, 3, 5, 11}, ids) - ids, err = SearchIssuesByKeyword([]int64{1}, "good") + ids, err = SearchIssuesByKeyword(context.TODO(), []int64{1}, "good") assert.NoError(t, err) assert.EqualValues(t, []int64{1}, ids) } diff --git a/routers/api/v1/repo/issue.go b/routers/api/v1/repo/issue.go index 5ce0c109e6c1e..e2afa724989f3 100644 --- a/routers/api/v1/repo/issue.go +++ b/routers/api/v1/repo/issue.go @@ -188,7 +188,7 @@ func SearchIssues(ctx *context.APIContext) { } var issueIDs []int64 if len(keyword) > 0 && len(repoIDs) > 0 { - if issueIDs, err = issue_indexer.SearchIssuesByKeyword(repoIDs, keyword); err != nil { + if issueIDs, err = issue_indexer.SearchIssuesByKeyword(ctx, repoIDs, keyword); err != nil { ctx.Error(http.StatusInternalServerError, "SearchIssuesByKeyword", err) return } @@ -379,7 +379,7 @@ func ListIssues(ctx *context.APIContext) { var issueIDs []int64 var labelIDs []int64 if len(keyword) > 0 { - issueIDs, err = issue_indexer.SearchIssuesByKeyword([]int64{ctx.Repo.Repository.ID}, keyword) + issueIDs, err = issue_indexer.SearchIssuesByKeyword(ctx, []int64{ctx.Repo.Repository.ID}, keyword) if err != nil { ctx.Error(http.StatusInternalServerError, "SearchIssuesByKeyword", err) return diff --git a/routers/web/repo/issue.go b/routers/web/repo/issue.go index df1b9876c079a..e0db617ed0d32 100644 --- a/routers/web/repo/issue.go +++ b/routers/web/repo/issue.go @@ -161,7 +161,7 @@ func issues(ctx *context.Context, milestoneID, projectID int64, isPullOption uti var issueIDs []int64 if len(keyword) > 0 { - issueIDs, err = issue_indexer.SearchIssuesByKeyword([]int64{repo.ID}, keyword) + issueIDs, err = issue_indexer.SearchIssuesByKeyword(ctx, []int64{repo.ID}, keyword) if err != nil { if issue_indexer.IsAvailable() { ctx.ServerError("issueIndexer.Search", err) diff --git a/routers/web/user/home.go b/routers/web/user/home.go index 13fa9bd8c4f7c..33512d97c06e4 100644 --- a/routers/web/user/home.go +++ b/routers/web/user/home.go @@ -438,7 +438,7 @@ func buildIssueOverview(ctx *context.Context, unitType unit.Type) { // Execute keyword search for issues. // USING NON-FINAL STATE OF opts FOR A QUERY. - issueIDsFromSearch, err := issueIDsFromSearch(ctxUser, keyword, opts) + issueIDsFromSearch, err := issueIDsFromSearch(ctx, ctxUser, keyword, opts) if err != nil { ctx.ServerError("issueIDsFromSearch", err) return @@ -673,7 +673,7 @@ func getRepoIDs(reposQuery string) []int64 { return repoIDs } -func issueIDsFromSearch(ctxUser *user_model.User, keyword string, opts *models.IssuesOptions) ([]int64, error) { +func issueIDsFromSearch(ctx *context.Context, ctxUser *user_model.User, keyword string, opts *models.IssuesOptions) ([]int64, error) { if len(keyword) == 0 { return []int64{}, nil } @@ -682,7 +682,7 @@ func issueIDsFromSearch(ctxUser *user_model.User, keyword string, opts *models.I if err != nil { return nil, fmt.Errorf("GetRepoIDsForIssuesOptions: %v", err) } - issueIDsFromSearch, err := issue_indexer.SearchIssuesByKeyword(searchRepoIDs, keyword) + issueIDsFromSearch, err := issue_indexer.SearchIssuesByKeyword(ctx, searchRepoIDs, keyword) if err != nil { return nil, fmt.Errorf("SearchIssuesByKeyword: %v", err) } From 05f888137af2bcd6bdbc8c0d196d795049be749a Mon Sep 17 00:00:00 2001 From: Lauris BH Date: Tue, 25 Jan 2022 22:21:01 +0200 Subject: [PATCH 6/7] Fix missing context argument --- routers/web/explore/code.go | 4 ++-- routers/web/repo/search.go | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/routers/web/explore/code.go b/routers/web/explore/code.go index ed1915d337ee7..640a5a0e4f308 100644 --- a/routers/web/explore/code.go +++ b/routers/web/explore/code.go @@ -87,7 +87,7 @@ func Code(ctx *context.Context) { ctx.Data["RepoMaps"] = rightRepoMap - total, searchResults, searchResultLanguages, err = code_indexer.PerformSearch(repoIDs, language, keyword, page, setting.UI.RepoSearchPagingNum, isMatch) + total, searchResults, searchResultLanguages, err = code_indexer.PerformSearch(ctx, repoIDs, language, keyword, page, setting.UI.RepoSearchPagingNum, isMatch) if err != nil { if code_indexer.IsAvailable() { ctx.ServerError("SearchResults", err) @@ -99,7 +99,7 @@ func Code(ctx *context.Context) { } // if non-login user or isAdmin, no need to check UnitTypeCode } else if (ctx.User == nil && len(repoIDs) > 0) || isAdmin { - total, searchResults, searchResultLanguages, err = code_indexer.PerformSearch(repoIDs, language, keyword, page, setting.UI.RepoSearchPagingNum, isMatch) + total, searchResults, searchResultLanguages, err = code_indexer.PerformSearch(ctx, repoIDs, language, keyword, page, setting.UI.RepoSearchPagingNum, isMatch) if err != nil { if code_indexer.IsAvailable() { ctx.ServerError("SearchResults", err) diff --git a/routers/web/repo/search.go b/routers/web/repo/search.go index 04bb5b844a4b4..e33fe38dea1f3 100644 --- a/routers/web/repo/search.go +++ b/routers/web/repo/search.go @@ -30,7 +30,7 @@ func Search(ctx *context.Context) { queryType := ctx.FormTrim("t") isMatch := queryType == "match" - total, searchResults, searchResultLanguages, err := code_indexer.PerformSearch([]int64{ctx.Repo.Repository.ID}, + total, searchResults, searchResultLanguages, err := code_indexer.PerformSearch(ctx, []int64{ctx.Repo.Repository.ID}, language, keyword, page, setting.UI.RepoSearchPagingNum, isMatch) if err != nil { if code_indexer.IsAvailable() { From 5941075f97f001c049a5f23ef87975140b8771f4 Mon Sep 17 00:00:00 2001 From: Lauris BH Date: Tue, 25 Jan 2022 22:38:24 +0200 Subject: [PATCH 7/7] Fix stop timer --- modules/indexer/code/elastic_search.go | 6 +++++- modules/indexer/issues/elastic_search.go | 6 +++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/modules/indexer/code/elastic_search.go b/modules/indexer/code/elastic_search.go index e541b0224ed18..db37b4f66c5e1 100644 --- a/modules/indexer/code/elastic_search.go +++ b/modules/indexer/code/elastic_search.go @@ -488,7 +488,11 @@ func (b *ElasticSearchIndexer) Search(ctx context.Context, repoIDs []int64, lang // Close implements indexer func (b *ElasticSearchIndexer) Close() { - close(b.stopTimer) + select { + case <-b.stopTimer: + default: + close(b.stopTimer) + } } func (b *ElasticSearchIndexer) checkError(err error) error { diff --git a/modules/indexer/issues/elastic_search.go b/modules/indexer/issues/elastic_search.go index 5e423ab0c15bd..97e32b89753be 100644 --- a/modules/indexer/issues/elastic_search.go +++ b/modules/indexer/issues/elastic_search.go @@ -264,7 +264,11 @@ func (b *ElasticSearchIndexer) Search(ctx context.Context, keyword string, repoI // Close implements indexer func (b *ElasticSearchIndexer) Close() { - close(b.stopTimer) + select { + case <-b.stopTimer: + default: + close(b.stopTimer) + } } func (b *ElasticSearchIndexer) checkError(err error) error {