mirror of
https://codeberg.org/forgejo/forgejo.git
synced 2024-11-14 18:24:14 +01:00
6f9c278559
# ⚠️ Breaking Many deprecated queue config options are removed (actually, they should have been removed in 1.18/1.19). If you see the fatal message when starting Gitea: "Please update your app.ini to remove deprecated config options", please follow the error messages to remove these options from your app.ini. Example: ``` 2023/05/06 19:39:22 [E] Removed queue option: `[indexer].ISSUE_INDEXER_QUEUE_TYPE`. Use new options in `[queue.issue_indexer]` 2023/05/06 19:39:22 [E] Removed queue option: `[indexer].UPDATE_BUFFER_LEN`. Use new options in `[queue.issue_indexer]` 2023/05/06 19:39:22 [F] Please update your app.ini to remove deprecated config options ``` Many options in `[queue]` are are dropped, including: `WRAP_IF_NECESSARY`, `MAX_ATTEMPTS`, `TIMEOUT`, `WORKERS`, `BLOCK_TIMEOUT`, `BOOST_TIMEOUT`, `BOOST_WORKERS`, they can be removed from app.ini. # The problem The old queue package has some legacy problems: * complexity: I doubt few people could tell how it works. * maintainability: Too many channels and mutex/cond are mixed together, too many different structs/interfaces depends each other. * stability: due to the complexity & maintainability, sometimes there are strange bugs and difficult to debug, and some code doesn't have test (indeed some code is difficult to test because a lot of things are mixed together). * general applicability: although it is called "queue", its behavior is not a well-known queue. * scalability: it doesn't seem easy to make it work with a cluster without breaking its behaviors. It came from some very old code to "avoid breaking", however, its technical debt is too heavy now. It's a good time to introduce a better "queue" package. # The new queue package It keeps using old config and concept as much as possible. * It only contains two major kinds of concepts: * The "base queue": channel, levelqueue, redis * They have the same abstraction, the same interface, and they are tested by the same testing code. * The "WokerPoolQueue", it uses the "base queue" to provide "worker pool" function, calls the "handler" to process the data in the base queue. * The new code doesn't do "PushBack" * Think about a queue with many workers, the "PushBack" can't guarantee the order for re-queued unhandled items, so in new code it just does "normal push" * The new code doesn't do "pause/resume" * The "pause/resume" was designed to handle some handler's failure: eg: document indexer (elasticsearch) is down * If a queue is paused for long time, either the producers blocks or the new items are dropped. * The new code doesn't do such "pause/resume" trick, it's not a common queue's behavior and it doesn't help much. * If there are unhandled items, the "push" function just blocks for a few seconds and then re-queue them and retry. * The new code doesn't do "worker booster" * Gitea's queue's handlers are light functions, the cost is only the go-routine, so it doesn't make sense to "boost" them. * The new code only use "max worker number" to limit the concurrent workers. * The new "Push" never blocks forever * Instead of creating more and more blocking goroutines, return an error is more friendly to the server and to the end user. There are more details in code comments: eg: the "Flush" problem, the strange "code.index" hanging problem, the "immediate" queue problem. Almost ready for review. TODO: * [x] add some necessary comments during review * [x] add some more tests if necessary * [x] update documents and config options * [x] test max worker / active worker * [x] re-run the CI tasks to see whether any test is flaky * [x] improve the `handleOldLengthConfiguration` to provide more friendly messages * [x] fine tune default config values (eg: length?) ## Code coverage: ![image](https://user-images.githubusercontent.com/2114189/236620635-55576955-f95d-4810-b12f-879026a3afdf.png)
276 lines
7.6 KiB
Go
276 lines
7.6 KiB
Go
// Copyright 2018 The Gitea Authors. All rights reserved.
|
|
// SPDX-License-Identifier: MIT
|
|
|
|
package issues
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"strconv"
|
|
|
|
gitea_bleve "code.gitea.io/gitea/modules/indexer/bleve"
|
|
"code.gitea.io/gitea/modules/log"
|
|
"code.gitea.io/gitea/modules/util"
|
|
|
|
"github.com/blevesearch/bleve/v2"
|
|
"github.com/blevesearch/bleve/v2/analysis/analyzer/custom"
|
|
"github.com/blevesearch/bleve/v2/analysis/token/camelcase"
|
|
"github.com/blevesearch/bleve/v2/analysis/token/lowercase"
|
|
"github.com/blevesearch/bleve/v2/analysis/token/unicodenorm"
|
|
"github.com/blevesearch/bleve/v2/analysis/tokenizer/unicode"
|
|
"github.com/blevesearch/bleve/v2/index/upsidedown"
|
|
"github.com/blevesearch/bleve/v2/mapping"
|
|
"github.com/blevesearch/bleve/v2/search/query"
|
|
"github.com/ethantkoenig/rupture"
|
|
)
|
|
|
|
const (
|
|
issueIndexerAnalyzer = "issueIndexer"
|
|
issueIndexerDocType = "issueIndexerDocType"
|
|
issueIndexerLatestVersion = 2
|
|
)
|
|
|
|
// indexerID a bleve-compatible unique identifier for an integer id
|
|
func indexerID(id int64) string {
|
|
return strconv.FormatInt(id, 36)
|
|
}
|
|
|
|
// idOfIndexerID the integer id associated with an indexer id
|
|
func idOfIndexerID(indexerID string) (int64, error) {
|
|
id, err := strconv.ParseInt(indexerID, 36, 64)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("Unexpected indexer ID %s: %w", indexerID, err)
|
|
}
|
|
return id, nil
|
|
}
|
|
|
|
// numericEqualityQuery a numeric equality query for the given value and field
|
|
func numericEqualityQuery(value int64, field string) *query.NumericRangeQuery {
|
|
f := float64(value)
|
|
tru := true
|
|
q := bleve.NewNumericRangeInclusiveQuery(&f, &f, &tru, &tru)
|
|
q.SetField(field)
|
|
return q
|
|
}
|
|
|
|
func newMatchPhraseQuery(matchPhrase, field, analyzer string) *query.MatchPhraseQuery {
|
|
q := bleve.NewMatchPhraseQuery(matchPhrase)
|
|
q.FieldVal = field
|
|
q.Analyzer = analyzer
|
|
return q
|
|
}
|
|
|
|
const unicodeNormalizeName = "unicodeNormalize"
|
|
|
|
func addUnicodeNormalizeTokenFilter(m *mapping.IndexMappingImpl) error {
|
|
return m.AddCustomTokenFilter(unicodeNormalizeName, map[string]interface{}{
|
|
"type": unicodenorm.Name,
|
|
"form": unicodenorm.NFC,
|
|
})
|
|
}
|
|
|
|
const maxBatchSize = 16
|
|
|
|
// openIndexer open the index at the specified path, checking for metadata
|
|
// updates and bleve version updates. If index needs to be created (or
|
|
// re-created), returns (nil, nil)
|
|
func openIndexer(path string, latestVersion int) (bleve.Index, error) {
|
|
_, err := os.Stat(path)
|
|
if err != nil && os.IsNotExist(err) {
|
|
return nil, nil
|
|
} else if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
metadata, err := rupture.ReadIndexMetadata(path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if metadata.Version < latestVersion {
|
|
// the indexer is using a previous version, so we should delete it and
|
|
// re-populate
|
|
return nil, util.RemoveAll(path)
|
|
}
|
|
|
|
index, err := bleve.Open(path)
|
|
if err != nil && err == upsidedown.IncompatibleVersion {
|
|
// the indexer was built with a previous version of bleve, so we should
|
|
// delete it and re-populate
|
|
return nil, util.RemoveAll(path)
|
|
} else if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return index, nil
|
|
}
|
|
|
|
// BleveIndexerData an update to the issue indexer
|
|
type BleveIndexerData IndexerData
|
|
|
|
// Type returns the document type, for bleve's mapping.Classifier interface.
|
|
func (i *BleveIndexerData) Type() string {
|
|
return issueIndexerDocType
|
|
}
|
|
|
|
// createIssueIndexer create an issue indexer if one does not already exist
|
|
func createIssueIndexer(path string, latestVersion int) (bleve.Index, error) {
|
|
mapping := bleve.NewIndexMapping()
|
|
docMapping := bleve.NewDocumentMapping()
|
|
|
|
numericFieldMapping := bleve.NewNumericFieldMapping()
|
|
numericFieldMapping.IncludeInAll = false
|
|
docMapping.AddFieldMappingsAt("RepoID", numericFieldMapping)
|
|
|
|
textFieldMapping := bleve.NewTextFieldMapping()
|
|
textFieldMapping.Store = false
|
|
textFieldMapping.IncludeInAll = false
|
|
docMapping.AddFieldMappingsAt("Title", textFieldMapping)
|
|
docMapping.AddFieldMappingsAt("Content", textFieldMapping)
|
|
docMapping.AddFieldMappingsAt("Comments", textFieldMapping)
|
|
|
|
if err := addUnicodeNormalizeTokenFilter(mapping); err != nil {
|
|
return nil, err
|
|
} else if err = mapping.AddCustomAnalyzer(issueIndexerAnalyzer, map[string]interface{}{
|
|
"type": custom.Name,
|
|
"char_filters": []string{},
|
|
"tokenizer": unicode.Name,
|
|
"token_filters": []string{unicodeNormalizeName, camelcase.Name, lowercase.Name},
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
mapping.DefaultAnalyzer = issueIndexerAnalyzer
|
|
mapping.AddDocumentMapping(issueIndexerDocType, docMapping)
|
|
mapping.AddDocumentMapping("_all", bleve.NewDocumentDisabledMapping())
|
|
|
|
index, err := bleve.New(path, mapping)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err = rupture.WriteIndexMetadata(path, &rupture.IndexMetadata{
|
|
Version: latestVersion,
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
return index, nil
|
|
}
|
|
|
|
var _ Indexer = &BleveIndexer{}
|
|
|
|
// BleveIndexer implements Indexer interface
|
|
type BleveIndexer struct {
|
|
indexDir string
|
|
indexer bleve.Index
|
|
}
|
|
|
|
// NewBleveIndexer creates a new bleve local indexer
|
|
func NewBleveIndexer(indexDir string) *BleveIndexer {
|
|
return &BleveIndexer{
|
|
indexDir: indexDir,
|
|
}
|
|
}
|
|
|
|
// Init will initialize the indexer
|
|
func (b *BleveIndexer) Init() (bool, error) {
|
|
var err error
|
|
b.indexer, err = openIndexer(b.indexDir, issueIndexerLatestVersion)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if b.indexer != nil {
|
|
return true, nil
|
|
}
|
|
|
|
b.indexer, err = createIssueIndexer(b.indexDir, issueIndexerLatestVersion)
|
|
return false, err
|
|
}
|
|
|
|
// Ping does nothing
|
|
func (b *BleveIndexer) Ping() bool {
|
|
return true
|
|
}
|
|
|
|
// Close will close the bleve indexer
|
|
func (b *BleveIndexer) Close() {
|
|
if b.indexer != nil {
|
|
if err := b.indexer.Close(); err != nil {
|
|
log.Error("Error whilst closing indexer: %v", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Index will save the index data
|
|
func (b *BleveIndexer) Index(issues []*IndexerData) error {
|
|
batch := gitea_bleve.NewFlushingBatch(b.indexer, maxBatchSize)
|
|
for _, issue := range issues {
|
|
if err := batch.Index(indexerID(issue.ID), struct {
|
|
RepoID int64
|
|
Title string
|
|
Content string
|
|
Comments []string
|
|
}{
|
|
RepoID: issue.RepoID,
|
|
Title: issue.Title,
|
|
Content: issue.Content,
|
|
Comments: issue.Comments,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return batch.Flush()
|
|
}
|
|
|
|
// Delete deletes indexes by ids
|
|
func (b *BleveIndexer) Delete(ids ...int64) error {
|
|
batch := gitea_bleve.NewFlushingBatch(b.indexer, maxBatchSize)
|
|
for _, id := range ids {
|
|
if err := batch.Delete(indexerID(id)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return batch.Flush()
|
|
}
|
|
|
|
// Search searches for issues by given conditions.
|
|
// Returns the matching issue IDs
|
|
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"))
|
|
}
|
|
repoQueries := make([]query.Query, len(repoQueriesP))
|
|
for i, v := range repoQueriesP {
|
|
repoQueries[i] = query.Query(v)
|
|
}
|
|
|
|
indexerQuery := bleve.NewConjunctionQuery(
|
|
bleve.NewDisjunctionQuery(repoQueries...),
|
|
bleve.NewDisjunctionQuery(
|
|
newMatchPhraseQuery(keyword, "Title", issueIndexerAnalyzer),
|
|
newMatchPhraseQuery(keyword, "Content", issueIndexerAnalyzer),
|
|
newMatchPhraseQuery(keyword, "Comments", issueIndexerAnalyzer),
|
|
))
|
|
search := bleve.NewSearchRequestOptions(indexerQuery, limit, start, false)
|
|
search.SortBy([]string{"-_score"})
|
|
|
|
result, err := b.indexer.SearchInContext(ctx, search)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
ret := SearchResult{
|
|
Hits: make([]Match, 0, len(result.Hits)),
|
|
}
|
|
for _, hit := range result.Hits {
|
|
id, err := idOfIndexerID(hit.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ret.Hits = append(ret.Hits, Match{
|
|
ID: id,
|
|
})
|
|
}
|
|
return &ret, nil
|
|
}
|