Issue indexer queue redis support

custom/conf/app.ini.sample

@@ -264,6 +264,8 @@ ISSUE_INDEXER_QUEUE_TYPE = levelqueue
 ; When ISSUE_INDEXER_QUEUE_TYPE is levelqueue, this will be the queue will be saved path,
 ; default is indexers/issues.queue
 ISSUE_INDEXER_QUEUE_DIR = indexers/issues.queue
+; When `ISSUE_INDEXER_QUEUE_TYPE` is `redis`, this will store the redis connection string.
+ISSUE_INDEXER_QUEUE_CONN_STR = "addrs=127.0.0.1:6379 db=0"
 ; Batch queue number, default is 20
 ISSUE_INDEXER_QUEUE_BATCH_NUMBER = 20
 

docs/content/doc/advanced/config-cheat-sheet.en-us.md

@@ -158,9 +158,10 @@ Values containing `#` or `;` must be quoted using `` ` `` or `"""`.
 
 - `ISSUE_INDEXER_TYPE`: **bleve**: Issue indexer type, currently support: bleve or db, if it's db, below issue indexer item will be invalid.
 - `ISSUE_INDEXER_PATH`: **indexers/issues.bleve**: Index file used for issue search.
-- `ISSUE_INDEXER_QUEUE_TYPE`: **levelqueue**: Issue indexer queue, currently support: channel or levelqueue
-- `ISSUE_INDEXER_QUEUE_DIR`: **indexers/issues.queue**: When ISSUE_INDEXER_QUEUE_TYPE is levelqueue, this will be the queue will be saved path
-- `ISSUE_INDEXER_QUEUE_BATCH_NUMBER`: **20**: Batch queue number
+- `ISSUE_INDEXER_QUEUE_TYPE`: **levelqueue**: Issue indexer queue, currently supports:`channel`, `levelqueue`, `redis`.
+- `ISSUE_INDEXER_QUEUE_DIR`: **indexers/issues.queue**: When `ISSUE_INDEXER_QUEUE_TYPE` is `levelqueue`, this will be the queue will be saved path.
+- `ISSUE_INDEXER_QUEUE_CONN_STR`: **addrs=127.0.0.1:6379 db=0**: When `ISSUE_INDEXER_QUEUE_TYPE` is `redis`, this will store the redis connection string.
+- `ISSUE_INDEXER_QUEUE_BATCH_NUMBER`: **20**: Batch queue number.
 
 - `REPO_INDEXER_ENABLED`: **false**: Enables code search (uses a lot of disk space, about 6 times more than the repository size).
 - `REPO_INDEXER_PATH`: **indexers/repos.bleve**: Index file used for code search.

docs/content/doc/advanced/config-cheat-sheet.zh-cn.md

@@ -82,13 +82,13 @@ menu:
 - `PATH`: Tidb 或者 SQLite3 数据文件存放路径。
 - `LOG_SQL`: **true**: 显示生成的SQL，默认为真。
 
-
 ## Indexer (`indexer`)
 
 - `ISSUE_INDEXER_TYPE`: **bleve**: 工单索引类型，当前支持 `bleve` 或 `db`，当为 `db` 时其它工单索引项可不用设置。
 - `ISSUE_INDEXER_PATH`: **indexers/issues.bleve**: 工单索引文件存放路径，当索引类型为 `bleve` 时有效。
-- `ISSUE_INDEXER_QUEUE_TYPE`: **levelqueue**: 工单索引队列类型，当前支持 `channel` 或 `levelqueue`。
+- `ISSUE_INDEXER_QUEUE_TYPE`: **levelqueue**: 工单索引队列类型，当前支持 `channel`， `levelqueue` 或 `redis`。
 - `ISSUE_INDEXER_QUEUE_DIR`: **indexers/issues.queue**: 当 `ISSUE_INDEXER_QUEUE_TYPE` 为 `levelqueue` 时，保存索引队列的磁盘路径。
+- `ISSUE_INDEXER_QUEUE_CONN_STR`: **addrs=127.0.0.1:6379 db=0**: 当 `ISSUE_INDEXER_QUEUE_TYPE` 为 `redis` 时，保存Redis队列的连接字符串。
 - `ISSUE_INDEXER_QUEUE_BATCH_NUMBER`: **20**: 队列处理中批量提交数量。
 
 - `REPO_INDEXER_ENABLED`: **false**: 是否启用代码搜索（启用后会占用比较大的磁盘空间）。

go.mod

@@ -54,6 +54,7 @@ require (
 	github.com/go-macaron/inject v0.0.0-20160627170012-d8a0b8677191
 	github.com/go-macaron/session v0.0.0-20190131233854-0a0a789bf193
 	github.com/go-macaron/toolbox v0.0.0-20180818072302-a77f45a7ce90
+	github.com/go-redis/redis v6.15.2+incompatible
 	github.com/go-sql-driver/mysql v1.4.0
 	github.com/go-xorm/builder v0.3.3
 	github.com/go-xorm/core v0.6.0

go.sum

@@ -116,6 +116,8 @@ github.com/go-macaron/session v0.0.0-20190131233854-0a0a789bf193 h1:z/nqwd+ql/r6
 github.com/go-macaron/session v0.0.0-20190131233854-0a0a789bf193/go.mod h1:ScEJm9Gk+ez5JJTml5WlBIqavAfuE5nF8e4Gvyz/X+A=
 github.com/go-macaron/toolbox v0.0.0-20180818072302-a77f45a7ce90 h1:3wYKrRg9IjUMfaf3H0Hh7M5Li9ge79Y7aw2yujHa2jQ=
 github.com/go-macaron/toolbox v0.0.0-20180818072302-a77f45a7ce90/go.mod h1:Ut/NmkIMGVYlEdJBzEZgWVWG5ZpYS9BLmUgXfAgi+qM=
+github.com/go-redis/redis v6.15.2+incompatible h1:9SpNVG76gr6InJGxoZ6IuuxaCOQwDAhzyXg+Bs+0Sb4=
+github.com/go-redis/redis v6.15.2+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
 github.com/go-sql-driver/mysql v0.0.0-20181218123637-c45f530f8e7f h1:fbIzwEaXt5b2bl9mm+PIufKTSGKk6ZuwSSTQ7iZj7Lo=
 github.com/go-sql-driver/mysql v0.0.0-20181218123637-c45f530f8e7f/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-xorm/builder v0.3.2/go.mod h1:v8mE3MFBgtL+RGFNfUnAMUqqfk/Y4W5KuwCFQIEpQLk=

modules/indexer/issues/indexer.go

@@ -46,9 +46,9 @@ type Indexer interface {
 }
 
 var (
-	// issueIndexerUpdateQueue queue of issue ids to be updated
-	issueIndexerUpdateQueue Queue
-	issueIndexer            Indexer
+	// issueIndexerQueue queue of issue ids to be updated
+	issueIndexerQueue Queue
+	issueIndexer      Indexer
 )
 
 // InitIssueIndexer initialize issue indexer, syncReindex is true then reindex until
@@ -72,27 +72,36 @@ func InitIssueIndexer(syncReindex bool) error {
 	}
 
 	if dummyQueue {
-		issueIndexerUpdateQueue = &DummyQueue{}
+		issueIndexerQueue = &DummyQueue{}
 		return nil
 	}
 
 	var err error
-	switch setting.Indexer.IssueIndexerQueueType {
+	switch setting.Indexer.IssueQueueType {
 	case setting.LevelQueueType:
-		issueIndexerUpdateQueue, err = NewLevelQueue(
+		issueIndexerQueue, err = NewLevelQueue(
 			issueIndexer,
-			setting.Indexer.IssueIndexerQueueDir,
-			setting.Indexer.IssueIndexerQueueBatchNumber)
+			setting.Indexer.IssueQueueDir,
+			setting.Indexer.IssueQueueBatchNumber)
 		if err != nil {
 			return err
 		}
 	case setting.ChannelQueueType:
-		issueIndexerUpdateQueue = NewChannelQueue(issueIndexer, setting.Indexer.IssueIndexerQueueBatchNumber)
+		issueIndexerQueue = NewChannelQueue(issueIndexer, setting.Indexer.IssueQueueBatchNumber)
+	case setting.RedisQueueType:
+		addrs, pass, idx, err := parseConnStr(setting.Indexer.IssueQueueConnStr)
+		if err != nil {
+			return err
+		}
+		issueIndexerQueue, err = NewRedisQueue(addrs, pass, idx, issueIndexer, setting.Indexer.IssueQueueBatchNumber)
+		if err != nil {
+			return err
+		}
 	default:
-		return fmt.Errorf("Unsupported indexer queue type: %v", setting.Indexer.IssueIndexerQueueType)
+		return fmt.Errorf("Unsupported indexer queue type: %v", setting.Indexer.IssueQueueType)
 	}
 
-	go issueIndexerUpdateQueue.Run()
+	go issueIndexerQueue.Run()
 
 	if populate {
 		if syncReindex {
@@ -152,7 +161,7 @@ func UpdateIssueIndexer(issue *models.Issue) {
 			comments = append(comments, comment.Content)
 		}
 	}
-	issueIndexerUpdateQueue.Push(&IndexerData{
+	issueIndexerQueue.Push(&IndexerData{
 		ID:       issue.ID,
 		RepoID:   issue.RepoID,
 		Title:    issue.Title,
@@ -174,7 +183,7 @@ func DeleteRepoIssueIndexer(repo *models.Repository) {
 		return
 	}
 
-	issueIndexerUpdateQueue.Push(&IndexerData{
+	issueIndexerQueue.Push(&IndexerData{
 		IDs:      ids,
 		IsDelete: true,
 	})

modules/indexer/issues/indexer_test.go

@@ -29,7 +29,7 @@ func TestMain(m *testing.M) {
 func TestBleveSearchIssues(t *testing.T) {
 	assert.NoError(t, models.PrepareTestDatabase())
 
-	os.RemoveAll(setting.Indexer.IssueIndexerQueueDir)
+	os.RemoveAll(setting.Indexer.IssueQueueDir)
 	os.RemoveAll(setting.Indexer.IssuePath)
 	setting.Indexer.IssueType = "bleve"
 	if err := InitIssueIndexer(true); err != nil {

modules/indexer/issues/queue_redis.go

@@ -0,0 +1,145 @@
+// Copyright 2019 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package issues
+
+import (
+	"encoding/json"
+	"errors"
+	"strconv"
+	"strings"
+	"time"
+
+	"code.gitea.io/gitea/modules/log"
+	"github.com/go-redis/redis"
+)
+
+var (
+	_ Queue = &RedisQueue{}
+)
+
+type redisClient interface {
+	RPush(key string, args ...interface{}) *redis.IntCmd
+	LPop(key string) *redis.StringCmd
+	Ping() *redis.StatusCmd
+}
+
+// RedisQueue redis queue
+type RedisQueue struct {
+	client      redisClient
+	queueName   string
+	indexer     Indexer
+	batchNumber int
+}
+
+func parseConnStr(connStr string) (addrs, password string, dbIdx int, err error) {
+	fields := strings.Fields(connStr)
+	for _, f := range fields {
+		items := strings.SplitN(f, "=", 2)
+		if len(items) < 2 {
+			continue
+		}
+		switch strings.ToLower(items[0]) {
+		case "addrs":
+			addrs = items[1]
+		case "password":
+			password = items[1]
+		case "db":
+			dbIdx, err = strconv.Atoi(items[1])
+			if err != nil {
+				return
+			}
+		}
+	}
+	return
+}
+
+// NewRedisQueue creates single redis or cluster redis queue
+func NewRedisQueue(addrs string, password string, dbIdx int, indexer Indexer, batchNumber int) (*RedisQueue, error) {
+	dbs := strings.Split(addrs, ",")
+	var queue = RedisQueue{
+		queueName:   "issue_indexer_queue",
+		indexer:     indexer,
+		batchNumber: batchNumber,
+	}
+	if len(dbs) == 0 {
+		return nil, errors.New("no redis host found")
+	} else if len(dbs) == 1 {
+		queue.client = redis.NewClient(&redis.Options{
+			Addr:     strings.TrimSpace(dbs[0]), // use default Addr
+			Password: password,                  // no password set
+			DB:       dbIdx,                     // use default DB
+		})
+	} else {
+		queue.client = redis.NewClusterClient(&redis.ClusterOptions{
+			Addrs: dbs,
+		})
+	}
+	if err := queue.client.Ping().Err(); err != nil {
+		return nil, err
+	}
+	return &queue, nil
+}
+
+// Run runs the redis queue
+func (r *RedisQueue) Run() error {
+	var i int
+	var datas = make([]*IndexerData, 0, r.batchNumber)
+	for {
+		bs, err := r.client.LPop(r.queueName).Bytes()
+		if err != nil && err != redis.Nil {
+			log.Error("LPop faile: %v", err)
+			time.Sleep(time.Millisecond * 100)
+			continue
+		}
+
+		i++
+		if len(datas) > r.batchNumber || (len(datas) > 0 && i > 3) {
+			r.indexer.Index(datas)
+			datas = make([]*IndexerData, 0, r.batchNumber)
+			i = 0
+		}
+
+		if len(bs) <= 0 {
+			time.Sleep(time.Millisecond * 100)
+			continue
+		}
+
+		var data IndexerData
+		err = json.Unmarshal(bs, &data)
+		if err != nil {
+			log.Error("Unmarshal: %v", err)
+			time.Sleep(time.Millisecond * 100)
+			continue
+		}
+
+		log.Trace("RedisQueue: task found: %#v", data)
+
+		if data.IsDelete {
+			if data.ID > 0 {
+				if err = r.indexer.Delete(data.ID); err != nil {
+					log.Error("indexer.Delete: %v", err)
+				}
+			} else if len(data.IDs) > 0 {
+				if err = r.indexer.Delete(data.IDs...); err != nil {
+					log.Error("indexer.Delete: %v", err)
+				}
+			}
+			time.Sleep(time.Millisecond * 100)
+			continue
+		}
+
+		datas = append(datas, &data)
+		time.Sleep(time.Millisecond * 100)
+	}
+}
+
+// Push implements Queue
+func (r *RedisQueue) Push(data *IndexerData) error {
+	bs, err := json.Marshal(data)
+	if err != nil {
+		return err
+	}
+	return r.client.RPush(r.queueName, bs).Err()
+}

modules/setting/indexer.go

@@ -13,26 +13,29 @@ import (
 const (
 	LevelQueueType   = "levelqueue"
 	ChannelQueueType = "channel"
+	RedisQueueType   = "redis"
 )
 
 var (
 	// Indexer settings
 	Indexer = struct {
-		IssueType                    string
-		IssuePath                    string
-		RepoIndexerEnabled           bool
-		RepoPath                     string
-		UpdateQueueLength            int
-		MaxIndexerFileSize           int64
-		IssueIndexerQueueType        string
-		IssueIndexerQueueDir         string
-		IssueIndexerQueueBatchNumber int
+		IssueType             string
+		IssuePath             string
+		RepoIndexerEnabled    bool
+		RepoPath              string
+		UpdateQueueLength     int
+		MaxIndexerFileSize    int64
+		IssueQueueType        string
+		IssueQueueDir         string
+		IssueQueueConnStr     string
+		IssueQueueBatchNumber int
 	}{
-		IssueType:                    "bleve",
-		IssuePath:                    "indexers/issues.bleve",
-		IssueIndexerQueueType:        LevelQueueType,
-		IssueIndexerQueueDir:         "indexers/issues.queue",
-		IssueIndexerQueueBatchNumber: 20,
+		IssueType:             "bleve",
+		IssuePath:             "indexers/issues.bleve",
+		IssueQueueType:        LevelQueueType,
+		IssueQueueDir:         "indexers/issues.queue",
+		IssueQueueConnStr:     "",
+		IssueQueueBatchNumber: 20,
 	}
 )
 
@@ -50,7 +53,8 @@ func newIndexerService() {
 	}
 	Indexer.UpdateQueueLength = sec.Key("UPDATE_BUFFER_LEN").MustInt(20)
 	Indexer.MaxIndexerFileSize = sec.Key("MAX_FILE_SIZE").MustInt64(1024 * 1024)
-	Indexer.IssueIndexerQueueType = sec.Key("ISSUE_INDEXER_QUEUE_TYPE").MustString(LevelQueueType)
-	Indexer.IssueIndexerQueueDir = sec.Key("ISSUE_INDEXER_QUEUE_DIR").MustString(path.Join(AppDataPath, "indexers/issues.queue"))
-	Indexer.IssueIndexerQueueBatchNumber = sec.Key("ISSUE_INDEXER_QUEUE_BATCH_NUMBER").MustInt(20)
+	Indexer.IssueQueueType = sec.Key("ISSUE_INDEXER_QUEUE_TYPE").MustString(LevelQueueType)
+	Indexer.IssueQueueDir = sec.Key("ISSUE_INDEXER_QUEUE_DIR").MustString(path.Join(AppDataPath, "indexers/issues.queue"))
+	Indexer.IssueQueueConnStr = sec.Key("ISSUE_INDEXER_QUEUE_CONN_STR").MustString(path.Join(AppDataPath, ""))
+	Indexer.IssueQueueBatchNumber = sec.Key("ISSUE_INDEXER_QUEUE_BATCH_NUMBER").MustInt(20)
 }