[MM-20979] Add first implementation of the Bleve search engine (#14562)
* [MM-20979] Add first implementation of the Bleve search engine * Fix i18n * Migrate searchengine utils tests * Fix linter * Don't add allTermsQ if both termQueries and notTermQueries are empty * Fix test that should work if user is system admin * Modify naming according to review comments * Abstract getIndexDir function * Extracting bleve engine name as a constant * Merge both Indexer interfaces into one * Add worker stopped message * Allow worker to be started/stopped with config change * Use constants for index names * Modify test order * Fix linter * Trying to unlock the CI
Этот коммит содержится в:
Miguel de la Cruz
коммит произвёл
GitHub
GitHub
родитель
8d0343d2eb
Коммит
0154b8059b
281
services/searchengine/bleveengine/bleve.go
Обычный файл
281
services/searchengine/bleveengine/bleve.go
Обычный файл
@@ -0,0 +1,281 @@
|
||||
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
||||
// See LICENSE.txt for license information.
|
||||
|
||||
package bleveengine
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v5/jobs"
|
||||
"github.com/mattermost/mattermost-server/v5/mlog"
|
||||
"github.com/mattermost/mattermost-server/v5/model"
|
||||
|
||||
"github.com/blevesearch/bleve"
|
||||
"github.com/blevesearch/bleve/analysis/analyzer/keyword"
|
||||
"github.com/blevesearch/bleve/analysis/analyzer/standard"
|
||||
"github.com/blevesearch/bleve/mapping"
|
||||
)
|
||||
|
||||
const (
|
||||
ENGINE_NAME = "bleve"
|
||||
POST_INDEX = "posts"
|
||||
USER_INDEX = "users"
|
||||
CHANNEL_INDEX = "channels"
|
||||
)
|
||||
|
||||
type BleveEngine struct {
|
||||
PostIndex bleve.Index
|
||||
UserIndex bleve.Index
|
||||
ChannelIndex bleve.Index
|
||||
Mutex sync.RWMutex
|
||||
ready int32
|
||||
cfg *model.Config
|
||||
jobServer *jobs.JobServer
|
||||
indexSync bool
|
||||
}
|
||||
|
||||
var keywordMapping *mapping.FieldMapping
|
||||
var standardMapping *mapping.FieldMapping
|
||||
var dateMapping *mapping.FieldMapping
|
||||
|
||||
func init() {
|
||||
keywordMapping = bleve.NewTextFieldMapping()
|
||||
keywordMapping.Analyzer = keyword.Name
|
||||
|
||||
standardMapping = bleve.NewTextFieldMapping()
|
||||
standardMapping.Analyzer = standard.Name
|
||||
|
||||
dateMapping = bleve.NewNumericFieldMapping()
|
||||
}
|
||||
|
||||
func getChannelIndexMapping() *mapping.IndexMappingImpl {
|
||||
channelMapping := bleve.NewDocumentMapping()
|
||||
channelMapping.AddFieldMappingsAt("Id", keywordMapping)
|
||||
channelMapping.AddFieldMappingsAt("TeamId", keywordMapping)
|
||||
channelMapping.AddFieldMappingsAt("NameSuggest", keywordMapping)
|
||||
|
||||
indexMapping := bleve.NewIndexMapping()
|
||||
indexMapping.AddDocumentMapping("_default", channelMapping)
|
||||
|
||||
return indexMapping
|
||||
}
|
||||
|
||||
func getPostIndexMapping() *mapping.IndexMappingImpl {
|
||||
postMapping := bleve.NewDocumentMapping()
|
||||
postMapping.AddFieldMappingsAt("Id", keywordMapping)
|
||||
postMapping.AddFieldMappingsAt("TeamId", keywordMapping)
|
||||
postMapping.AddFieldMappingsAt("ChannelId", keywordMapping)
|
||||
postMapping.AddFieldMappingsAt("UserId", keywordMapping)
|
||||
postMapping.AddFieldMappingsAt("CreateAt", dateMapping)
|
||||
postMapping.AddFieldMappingsAt("Message", standardMapping)
|
||||
postMapping.AddFieldMappingsAt("Type", keywordMapping)
|
||||
postMapping.AddFieldMappingsAt("Hashtags", standardMapping)
|
||||
postMapping.AddFieldMappingsAt("Attachments", standardMapping)
|
||||
|
||||
indexMapping := bleve.NewIndexMapping()
|
||||
indexMapping.AddDocumentMapping("_default", postMapping)
|
||||
|
||||
return indexMapping
|
||||
}
|
||||
|
||||
func getUserIndexMapping() *mapping.IndexMappingImpl {
|
||||
userMapping := bleve.NewDocumentMapping()
|
||||
userMapping.AddFieldMappingsAt("Id", keywordMapping)
|
||||
userMapping.AddFieldMappingsAt("SuggestionsWithFullname", keywordMapping)
|
||||
userMapping.AddFieldMappingsAt("SuggestionsWithoutFullname", keywordMapping)
|
||||
userMapping.AddFieldMappingsAt("TeamsIds", keywordMapping)
|
||||
userMapping.AddFieldMappingsAt("ChannelsIds", keywordMapping)
|
||||
|
||||
indexMapping := bleve.NewIndexMapping()
|
||||
indexMapping.AddDocumentMapping("_default", userMapping)
|
||||
|
||||
return indexMapping
|
||||
}
|
||||
|
||||
func NewBleveEngine(cfg *model.Config, jobServer *jobs.JobServer) *BleveEngine {
|
||||
return &BleveEngine{
|
||||
cfg: cfg,
|
||||
jobServer: jobServer,
|
||||
}
|
||||
}
|
||||
|
||||
func (b *BleveEngine) getIndexDir(indexName string) string {
|
||||
return filepath.Join(*b.cfg.BleveSettings.IndexDir, indexName+".bleve")
|
||||
}
|
||||
|
||||
func (b *BleveEngine) createOrOpenIndex(indexName string, mapping *mapping.IndexMappingImpl) (bleve.Index, error) {
|
||||
indexPath := b.getIndexDir(indexName)
|
||||
if index, err := bleve.Open(indexPath); err == nil {
|
||||
return index, nil
|
||||
}
|
||||
|
||||
index, err := bleve.New(indexPath, mapping)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return index, nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) openIndexes() *model.AppError {
|
||||
if atomic.LoadInt32(&b.ready) != 0 {
|
||||
return model.NewAppError("Bleveengine.Start", "bleveengine.already_started.error", nil, "", http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
var err error
|
||||
b.PostIndex, err = b.createOrOpenIndex(POST_INDEX, getPostIndexMapping())
|
||||
if err != nil {
|
||||
return model.NewAppError("Bleveengine.Start", "bleveengine.create_post_index.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
b.UserIndex, err = b.createOrOpenIndex(USER_INDEX, getUserIndexMapping())
|
||||
if err != nil {
|
||||
return model.NewAppError("Bleveengine.Start", "bleveengine.create_user_index.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
b.ChannelIndex, err = b.createOrOpenIndex(CHANNEL_INDEX, getChannelIndexMapping())
|
||||
if err != nil {
|
||||
return model.NewAppError("Bleveengine.Start", "bleveengine.create_channel_index.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
atomic.StoreInt32(&b.ready, 1)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) Start() *model.AppError {
|
||||
if !*b.cfg.BleveSettings.EnableIndexing || *b.cfg.BleveSettings.IndexDir == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
b.Mutex.Lock()
|
||||
defer b.Mutex.Unlock()
|
||||
|
||||
mlog.Info("EXPERIMENTAL: Starting Bleve")
|
||||
|
||||
return b.openIndexes()
|
||||
}
|
||||
|
||||
func (b *BleveEngine) closeIndexes() *model.AppError {
|
||||
if b.IsActive() {
|
||||
if err := b.PostIndex.Close(); err != nil {
|
||||
return model.NewAppError("Bleveengine.Stop", "bleveengine.stop_post_index.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
if err := b.UserIndex.Close(); err != nil {
|
||||
return model.NewAppError("Bleveengine.Stop", "bleveengine.stop_user_index.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
if err := b.ChannelIndex.Close(); err != nil {
|
||||
return model.NewAppError("Bleveengine.Stop", "bleveengine.stop_channel_index.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
}
|
||||
|
||||
atomic.StoreInt32(&b.ready, 0)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) Stop() *model.AppError {
|
||||
b.Mutex.Lock()
|
||||
defer b.Mutex.Unlock()
|
||||
|
||||
mlog.Info("Stopping Bleve")
|
||||
|
||||
return b.closeIndexes()
|
||||
}
|
||||
|
||||
func (b *BleveEngine) IsActive() bool {
|
||||
return atomic.LoadInt32(&b.ready) == 1
|
||||
}
|
||||
|
||||
func (b *BleveEngine) IsIndexingSync() bool {
|
||||
return b.indexSync
|
||||
}
|
||||
|
||||
func (b *BleveEngine) RefreshIndexes() *model.AppError {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) GetVersion() int {
|
||||
return 0
|
||||
}
|
||||
|
||||
func (b *BleveEngine) GetName() string {
|
||||
return ENGINE_NAME
|
||||
}
|
||||
|
||||
func (b *BleveEngine) TestConfig(cfg *model.Config) *model.AppError {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) deleteIndexes() *model.AppError {
|
||||
if err := os.RemoveAll(b.getIndexDir(POST_INDEX)); err != nil {
|
||||
return model.NewAppError("Bleveengine.PurgeIndexes", "bleveengine.purge_post_index.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
if err := os.RemoveAll(b.getIndexDir(USER_INDEX)); err != nil {
|
||||
return model.NewAppError("Bleveengine.PurgeIndexes", "bleveengine.purge_user_index.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
if err := os.RemoveAll(b.getIndexDir(CHANNEL_INDEX)); err != nil {
|
||||
return model.NewAppError("Bleveengine.PurgeIndexes", "bleveengine.purge_channel_index.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) PurgeIndexes() *model.AppError {
|
||||
if *b.cfg.BleveSettings.IndexDir == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
b.Mutex.Lock()
|
||||
defer b.Mutex.Unlock()
|
||||
|
||||
mlog.Info("PurgeIndexes Bleve")
|
||||
if err := b.closeIndexes(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := b.deleteIndexes(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return b.openIndexes()
|
||||
}
|
||||
|
||||
func (b *BleveEngine) DataRetentionDeleteIndexes(cutoff time.Time) *model.AppError {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) IsAutocompletionEnabled() bool {
|
||||
return *b.cfg.BleveSettings.EnableAutocomplete
|
||||
}
|
||||
|
||||
func (b *BleveEngine) IsIndexingEnabled() bool {
|
||||
return *b.cfg.BleveSettings.EnableIndexing
|
||||
}
|
||||
|
||||
func (b *BleveEngine) IsSearchEnabled() bool {
|
||||
return *b.cfg.BleveSettings.EnableSearching
|
||||
}
|
||||
|
||||
func (b *BleveEngine) UpdateConfig(cfg *model.Config) {
|
||||
b.Mutex.Lock()
|
||||
defer b.Mutex.Unlock()
|
||||
|
||||
mlog.Info("UpdateConf Bleve")
|
||||
|
||||
if *cfg.BleveSettings.EnableIndexing != *b.cfg.BleveSettings.EnableIndexing || *cfg.BleveSettings.IndexDir != *b.cfg.BleveSettings.IndexDir {
|
||||
if err := b.closeIndexes(); err != nil {
|
||||
mlog.Error("Error closing Bleve indexes to update the config", mlog.Err(err))
|
||||
return
|
||||
}
|
||||
b.cfg = cfg
|
||||
if err := b.openIndexes(); err != nil {
|
||||
mlog.Error("Error opening Bleve indexes after updating the config", mlog.Err(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
b.cfg = cfg
|
||||
}
|
||||
98
services/searchengine/bleveengine/bleve_test.go
Обычный файл
98
services/searchengine/bleveengine/bleve_test.go
Обычный файл
@@ -0,0 +1,98 @@
|
||||
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
||||
// See LICENSE.txt for license information.
|
||||
|
||||
package bleveengine
|
||||
|
||||
import (
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/suite"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v5/model"
|
||||
"github.com/mattermost/mattermost-server/v5/services/searchengine"
|
||||
"github.com/mattermost/mattermost-server/v5/store/searchlayer"
|
||||
"github.com/mattermost/mattermost-server/v5/store/searchtest"
|
||||
"github.com/mattermost/mattermost-server/v5/store/sqlstore"
|
||||
"github.com/mattermost/mattermost-server/v5/store/storetest"
|
||||
"github.com/mattermost/mattermost-server/v5/testlib"
|
||||
)
|
||||
|
||||
type BleveEngineTestSuite struct {
|
||||
suite.Suite
|
||||
|
||||
SQLSettings *model.SqlSettings
|
||||
SQLSupplier *sqlstore.SqlSupplier
|
||||
SearchEngine *searchengine.Broker
|
||||
Store *searchlayer.SearchStore
|
||||
IndexDir string
|
||||
}
|
||||
|
||||
func TestBleveEngineTestSuite(t *testing.T) {
|
||||
suite.Run(t, new(BleveEngineTestSuite))
|
||||
}
|
||||
|
||||
func (s *BleveEngineTestSuite) setupIndexes() {
|
||||
indexDir, err := ioutil.TempDir("", "mmbleve")
|
||||
if err != nil {
|
||||
s.Require().FailNow("Cannot setup bleveengine tests: %s", err.Error())
|
||||
}
|
||||
s.IndexDir = indexDir
|
||||
}
|
||||
|
||||
func (s *BleveEngineTestSuite) setupStore() {
|
||||
driverName := os.Getenv("MM_SQLSETTINGS_DRIVERNAME")
|
||||
if driverName == "" {
|
||||
driverName = model.DATABASE_DRIVER_POSTGRES
|
||||
}
|
||||
s.SQLSettings = storetest.MakeSqlSettings(driverName)
|
||||
s.SQLSupplier = sqlstore.NewSqlSupplier(*s.SQLSettings, nil)
|
||||
|
||||
cfg := &model.Config{}
|
||||
cfg.SetDefaults()
|
||||
cfg.BleveSettings.EnableIndexing = model.NewBool(true)
|
||||
cfg.BleveSettings.EnableSearching = model.NewBool(true)
|
||||
cfg.BleveSettings.EnableAutocomplete = model.NewBool(true)
|
||||
cfg.BleveSettings.IndexDir = model.NewString(s.IndexDir)
|
||||
cfg.SqlSettings.DisableDatabaseSearch = model.NewBool(true)
|
||||
|
||||
s.SearchEngine = searchengine.NewBroker(cfg, nil)
|
||||
s.Store = searchlayer.NewSearchLayer(&testlib.TestStore{Store: s.SQLSupplier}, s.SearchEngine, cfg)
|
||||
|
||||
bleveEngine := NewBleveEngine(cfg, nil)
|
||||
bleveEngine.indexSync = true
|
||||
s.SearchEngine.RegisterBleveEngine(bleveEngine)
|
||||
if err := bleveEngine.Start(); err != nil {
|
||||
s.Require().FailNow("Cannot start bleveengine: %s", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
func (s *BleveEngineTestSuite) SetupSuite() {
|
||||
s.setupIndexes()
|
||||
s.setupStore()
|
||||
}
|
||||
|
||||
func (s *BleveEngineTestSuite) TearDownSuite() {
|
||||
os.RemoveAll(s.IndexDir)
|
||||
s.SQLSupplier.Close()
|
||||
storetest.CleanupSqlSettings(s.SQLSettings)
|
||||
}
|
||||
|
||||
func (s *BleveEngineTestSuite) TestBleveSearchStoreTests() {
|
||||
searchTestEngine := &searchtest.SearchTestEngine{
|
||||
Driver: searchtest.ENGINE_BLEVE,
|
||||
}
|
||||
|
||||
s.Run("TestSearchChannelStore", func() {
|
||||
searchtest.TestSearchChannelStore(s.T(), s.Store, searchTestEngine)
|
||||
})
|
||||
|
||||
s.Run("TestSearchUserStore", func() {
|
||||
searchtest.TestSearchUserStore(s.T(), s.Store, searchTestEngine)
|
||||
})
|
||||
|
||||
s.Run("TestSearchPostStore", func() {
|
||||
searchtest.TestSearchPostStore(s.T(), s.Store, searchTestEngine)
|
||||
})
|
||||
}
|
||||
116
services/searchengine/bleveengine/common.go
Обычный файл
116
services/searchengine/bleveengine/common.go
Обычный файл
@@ -0,0 +1,116 @@
|
||||
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
||||
// See LICENSE.txt for license information.
|
||||
|
||||
package bleveengine
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v5/model"
|
||||
"github.com/mattermost/mattermost-server/v5/services/searchengine"
|
||||
)
|
||||
|
||||
type BLVChannel struct {
|
||||
Id string
|
||||
TeamId []string
|
||||
NameSuggest []string
|
||||
}
|
||||
|
||||
type BLVUser struct {
|
||||
Id string
|
||||
SuggestionsWithFullname []string
|
||||
SuggestionsWithoutFullname []string
|
||||
TeamsIds []string
|
||||
ChannelsIds []string
|
||||
}
|
||||
|
||||
type BLVPost struct {
|
||||
Id string
|
||||
TeamId string
|
||||
ChannelId string
|
||||
UserId string
|
||||
CreateAt int64
|
||||
Message string
|
||||
Type string
|
||||
Hashtags []string
|
||||
Attachments string
|
||||
}
|
||||
|
||||
func BLVChannelFromChannel(channel *model.Channel) *BLVChannel {
|
||||
displayNameInputs := searchengine.GetSuggestionInputsSplitBy(channel.DisplayName, " ")
|
||||
nameInputs := searchengine.GetSuggestionInputsSplitByMultiple(channel.Name, []string{"-", "_"})
|
||||
|
||||
return &BLVChannel{
|
||||
Id: channel.Id,
|
||||
TeamId: []string{channel.TeamId},
|
||||
NameSuggest: append(displayNameInputs, nameInputs...),
|
||||
}
|
||||
}
|
||||
|
||||
func BLVUserFromUserAndTeams(user *model.User, teamsIds, channelsIds []string) *BLVUser {
|
||||
usernameSuggestions := searchengine.GetSuggestionInputsSplitByMultiple(user.Username, []string{".", "-", "_"})
|
||||
|
||||
fullnameStrings := []string{}
|
||||
if user.FirstName != "" {
|
||||
fullnameStrings = append(fullnameStrings, user.FirstName)
|
||||
}
|
||||
if user.LastName != "" {
|
||||
fullnameStrings = append(fullnameStrings, user.LastName)
|
||||
}
|
||||
|
||||
fullnameSuggestions := []string{}
|
||||
if len(fullnameStrings) > 0 {
|
||||
fullname := strings.Join(fullnameStrings, " ")
|
||||
fullnameSuggestions = searchengine.GetSuggestionInputsSplitBy(fullname, " ")
|
||||
}
|
||||
|
||||
nicknameSuggesitons := []string{}
|
||||
if user.Nickname != "" {
|
||||
nicknameSuggesitons = searchengine.GetSuggestionInputsSplitBy(user.Nickname, " ")
|
||||
}
|
||||
|
||||
usernameAndNicknameSuggestions := append(usernameSuggestions, nicknameSuggesitons...)
|
||||
|
||||
return &BLVUser{
|
||||
Id: user.Id,
|
||||
SuggestionsWithFullname: append(usernameAndNicknameSuggestions, fullnameSuggestions...),
|
||||
SuggestionsWithoutFullname: usernameAndNicknameSuggestions,
|
||||
TeamsIds: teamsIds,
|
||||
ChannelsIds: channelsIds,
|
||||
}
|
||||
}
|
||||
|
||||
func BLVUserFromUserForIndexing(userForIndexing *model.UserForIndexing) *BLVUser {
|
||||
user := &model.User{
|
||||
Id: userForIndexing.Id,
|
||||
Username: userForIndexing.Username,
|
||||
Nickname: userForIndexing.Nickname,
|
||||
FirstName: userForIndexing.FirstName,
|
||||
LastName: userForIndexing.LastName,
|
||||
CreateAt: userForIndexing.CreateAt,
|
||||
DeleteAt: userForIndexing.DeleteAt,
|
||||
}
|
||||
|
||||
return BLVUserFromUserAndTeams(user, userForIndexing.TeamsIds, userForIndexing.ChannelsIds)
|
||||
}
|
||||
|
||||
func BLVPostFromPost(post *model.Post, teamId string) *BLVPost {
|
||||
p := &model.PostForIndexing{
|
||||
TeamId: teamId,
|
||||
}
|
||||
post.ShallowCopy(&p.Post)
|
||||
return BLVPostFromPostForIndexing(p)
|
||||
}
|
||||
|
||||
func BLVPostFromPostForIndexing(post *model.PostForIndexing) *BLVPost {
|
||||
return &BLVPost{
|
||||
Id: post.Id,
|
||||
TeamId: post.TeamId,
|
||||
ChannelId: post.ChannelId,
|
||||
UserId: post.UserId,
|
||||
CreateAt: post.CreateAt,
|
||||
Message: post.Message,
|
||||
Type: post.Type,
|
||||
Hashtags: strings.Fields(post.Hashtags),
|
||||
}
|
||||
}
|
||||
510
services/searchengine/bleveengine/indexer/indexing_job.go
Обычный файл
510
services/searchengine/bleveengine/indexer/indexing_job.go
Обычный файл
@@ -0,0 +1,510 @@
|
||||
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
||||
// See LICENSE.txt for license information.
|
||||
|
||||
package ebleveengine
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v5/app"
|
||||
"github.com/mattermost/mattermost-server/v5/jobs"
|
||||
tjobs "github.com/mattermost/mattermost-server/v5/jobs/interfaces"
|
||||
"github.com/mattermost/mattermost-server/v5/mlog"
|
||||
"github.com/mattermost/mattermost-server/v5/model"
|
||||
"github.com/mattermost/mattermost-server/v5/services/searchengine/bleveengine"
|
||||
)
|
||||
|
||||
const (
|
||||
BATCH_SIZE = 1000
|
||||
TIME_BETWEEN_BATCHES = 100
|
||||
ESTIMATED_POST_COUNT = 10000000
|
||||
ESTIMATED_CHANNEL_COUNT = 100000
|
||||
ESTIMATED_USER_COUNT = 10000
|
||||
)
|
||||
|
||||
func init() {
|
||||
app.RegisterJobsBleveIndexerInterface(func(s *app.Server) tjobs.IndexerJobInterface {
|
||||
return &BleveIndexerInterfaceImpl{s}
|
||||
})
|
||||
}
|
||||
|
||||
type BleveIndexerInterfaceImpl struct {
|
||||
Server *app.Server
|
||||
}
|
||||
|
||||
type BleveIndexerWorker struct {
|
||||
name string
|
||||
stop chan bool
|
||||
stopped chan bool
|
||||
jobs chan model.Job
|
||||
jobServer *jobs.JobServer
|
||||
|
||||
engine *bleveengine.BleveEngine
|
||||
}
|
||||
|
||||
func (bi *BleveIndexerInterfaceImpl) MakeWorker() model.Worker {
|
||||
return &BleveIndexerWorker{
|
||||
name: "BleveIndexer",
|
||||
stop: make(chan bool, 1),
|
||||
stopped: make(chan bool, 1),
|
||||
jobs: make(chan model.Job),
|
||||
jobServer: bi.Server.Jobs,
|
||||
|
||||
engine: bi.Server.SearchEngine.BleveEngine.(*bleveengine.BleveEngine),
|
||||
}
|
||||
}
|
||||
|
||||
type IndexingProgress struct {
|
||||
Now time.Time
|
||||
StartAtTime int64
|
||||
EndAtTime int64
|
||||
LastEntityTime int64
|
||||
TotalPostsCount int64
|
||||
DonePostsCount int64
|
||||
DonePosts bool
|
||||
TotalChannelsCount int64
|
||||
DoneChannelsCount int64
|
||||
DoneChannels bool
|
||||
TotalUsersCount int64
|
||||
DoneUsersCount int64
|
||||
DoneUsers bool
|
||||
}
|
||||
|
||||
func (ip *IndexingProgress) CurrentProgress() int64 {
|
||||
return (ip.DonePostsCount + ip.DoneChannelsCount + ip.DoneUsersCount) * 100 / (ip.TotalPostsCount + ip.TotalChannelsCount + ip.TotalUsersCount)
|
||||
}
|
||||
|
||||
func (ip *IndexingProgress) IsDone() bool {
|
||||
return ip.DonePosts && ip.DoneChannels && ip.DoneUsers
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) JobChannel() chan<- model.Job {
|
||||
return worker.jobs
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) Run() {
|
||||
mlog.Debug("Worker Started", mlog.String("workername", worker.name))
|
||||
|
||||
defer func() {
|
||||
mlog.Debug("Worker: Finished", mlog.String("workername", worker.name))
|
||||
worker.stopped <- true
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-worker.stop:
|
||||
mlog.Debug("Worker: Received stop signal", mlog.String("workername", worker.name))
|
||||
return
|
||||
case job := <-worker.jobs:
|
||||
mlog.Debug("Worker: Received a new candidate job.", mlog.String("workername", worker.name))
|
||||
worker.DoJob(&job)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) Stop() {
|
||||
mlog.Debug("Worker Stopping", mlog.String("workername", worker.name))
|
||||
worker.stop <- true
|
||||
<-worker.stopped
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) DoJob(job *model.Job) {
|
||||
claimed, err := worker.jobServer.ClaimJob(job)
|
||||
if err != nil {
|
||||
mlog.Warn("Worker: Error ocurred while trying to claim job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
|
||||
return
|
||||
}
|
||||
if !claimed {
|
||||
return
|
||||
}
|
||||
|
||||
mlog.Info("Worker: Indexing job claimed by worker", mlog.String("workername", worker.name), mlog.String("job_id", job.Id))
|
||||
|
||||
if !worker.engine.IsActive() {
|
||||
appError := model.NewAppError("BleveIndexerWorker", "bleveengine.indexer.do_job.engine_inactive", nil, "", http.StatusInternalServerError)
|
||||
if err := worker.jobServer.SetJobError(job, appError); err != nil {
|
||||
mlog.Error("Worker: Failed to run job as ")
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
progress := IndexingProgress{
|
||||
Now: time.Now(),
|
||||
DonePosts: false,
|
||||
DoneChannels: false,
|
||||
DoneUsers: false,
|
||||
StartAtTime: 0,
|
||||
EndAtTime: model.GetMillis(),
|
||||
}
|
||||
|
||||
// Extract the start and end times, if they are set.
|
||||
if startString, ok := job.Data["start_time"]; ok {
|
||||
startInt, err := strconv.ParseInt(startString, 10, 64)
|
||||
if err != nil {
|
||||
mlog.Error("Worker: Failed to parse start_time for job", mlog.String("workername", worker.name), mlog.String("start_time", startString), mlog.String("job_id", job.Id), mlog.Err(err))
|
||||
appError := model.NewAppError("BleveIndexerWorker", "bleveengine.indexer.do_job.parse_start_time.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
if err := worker.jobServer.SetJobError(job, appError); err != nil {
|
||||
mlog.Error("Worker: Failed to set job error", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err), mlog.NamedErr("set_error", appError))
|
||||
}
|
||||
return
|
||||
}
|
||||
progress.StartAtTime = startInt
|
||||
progress.LastEntityTime = progress.StartAtTime
|
||||
} else {
|
||||
// Set start time to oldest post in the database.
|
||||
oldestPost, err := worker.jobServer.Store.Post().GetOldest()
|
||||
if err != nil {
|
||||
mlog.Error("Worker: Failed to fetch oldest post for job.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.String("start_time", startString), mlog.Err(err))
|
||||
appError := model.NewAppError("BleveIndexerWorker", "bleveengine.indexer.do_job.get_oldest_post.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
if err := worker.jobServer.SetJobError(job, appError); err != nil {
|
||||
mlog.Error("Worker: Failed to set job error", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err), mlog.NamedErr("set_error", appError))
|
||||
}
|
||||
return
|
||||
}
|
||||
progress.StartAtTime = oldestPost.CreateAt
|
||||
progress.LastEntityTime = progress.StartAtTime
|
||||
}
|
||||
|
||||
if endString, ok := job.Data["end_time"]; ok {
|
||||
endInt, err := strconv.ParseInt(endString, 10, 64)
|
||||
if err != nil {
|
||||
mlog.Error("Worker: Failed to parse end_time for job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.String("end_time", endString), mlog.Err(err))
|
||||
appError := model.NewAppError("BleveIndexerWorker", "bleveengine.indexer.do_job.parse_end_time.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
if err := worker.jobServer.SetJobError(job, appError); err != nil {
|
||||
mlog.Error("Worker: Failed to set job errorv", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err), mlog.NamedErr("set_error", appError))
|
||||
}
|
||||
return
|
||||
}
|
||||
progress.EndAtTime = endInt
|
||||
}
|
||||
|
||||
// Counting all posts may fail or timeout when the posts table is large. If this happens, log a warning, but carry
|
||||
// on with the indexing job anyway. The only issue is that the progress % reporting will be inaccurate.
|
||||
if count, err := worker.jobServer.Store.Post().AnalyticsPostCount("", false, false); err != nil {
|
||||
mlog.Warn("Worker: Failed to fetch total post count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
|
||||
progress.TotalPostsCount = ESTIMATED_POST_COUNT
|
||||
} else {
|
||||
progress.TotalPostsCount = count
|
||||
}
|
||||
|
||||
// Same possible fail as above can happen when counting channels
|
||||
if count, err := worker.jobServer.Store.Channel().AnalyticsTypeCount("", "O"); err != nil {
|
||||
mlog.Warn("Worker: Failed to fetch total channel count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
|
||||
progress.TotalChannelsCount = ESTIMATED_CHANNEL_COUNT
|
||||
} else {
|
||||
progress.TotalChannelsCount = count
|
||||
}
|
||||
|
||||
// Same possible fail as above can happen when counting users
|
||||
if count, err := worker.jobServer.Store.User().Count(model.UserCountOptions{}); err != nil {
|
||||
mlog.Warn("Worker: Failed to fetch total user count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
|
||||
progress.TotalUsersCount = ESTIMATED_USER_COUNT
|
||||
} else {
|
||||
progress.TotalUsersCount = count
|
||||
}
|
||||
|
||||
cancelCtx, cancelCancelWatcher := context.WithCancel(context.Background())
|
||||
cancelWatcherChan := make(chan interface{}, 1)
|
||||
go worker.jobServer.CancellationWatcher(cancelCtx, job.Id, cancelWatcherChan)
|
||||
|
||||
defer cancelCancelWatcher()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-cancelWatcherChan:
|
||||
mlog.Info("Worker: Indexing job has been canceled via CancellationWatcher", mlog.String("workername", worker.name), mlog.String("job_id", job.Id))
|
||||
if err := worker.jobServer.SetJobCanceled(job); err != nil {
|
||||
mlog.Error("Worker: Failed to mark job as cancelled", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
|
||||
}
|
||||
return
|
||||
|
||||
case <-worker.stop:
|
||||
mlog.Info("Worker: Indexing has been canceled via Worker Stop", mlog.String("workername", worker.name), mlog.String("job_id", job.Id))
|
||||
if err := worker.jobServer.SetJobCanceled(job); err != nil {
|
||||
mlog.Error("Worker: Failed to mark job as canceled", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
|
||||
}
|
||||
return
|
||||
|
||||
case <-time.After(TIME_BETWEEN_BATCHES * time.Millisecond):
|
||||
var err *model.AppError
|
||||
if progress, err = worker.IndexBatch(progress); err != nil {
|
||||
mlog.Error("Worker: Failed to index batch for job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
|
||||
if err2 := worker.jobServer.SetJobError(job, err); err2 != nil {
|
||||
mlog.Error("Worker: Failed to set job error", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err2), mlog.NamedErr("set_error", err))
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if err := worker.jobServer.SetJobProgress(job, progress.CurrentProgress()); err != nil {
|
||||
mlog.Error("Worker: Failed to set progress for job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
|
||||
if err2 := worker.jobServer.SetJobError(job, err); err2 != nil {
|
||||
mlog.Error("Worker: Failed to set error for job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err2), mlog.NamedErr("set_error", err))
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if progress.IsDone() {
|
||||
if err := worker.jobServer.SetJobSuccess(job); err != nil {
|
||||
mlog.Error("Worker: Failed to set success for job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
|
||||
if err2 := worker.jobServer.SetJobError(job, err); err2 != nil {
|
||||
mlog.Error("Worker: Failed to set error for job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err2), mlog.NamedErr("set_error", err))
|
||||
}
|
||||
}
|
||||
mlog.Info("Worker: Indexing job finished successfully", mlog.String("workername", worker.name), mlog.String("job_id", job.Id))
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) IndexBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) {
|
||||
if !progress.DonePosts {
|
||||
return worker.IndexPostsBatch(progress)
|
||||
}
|
||||
if !progress.DoneChannels {
|
||||
return worker.IndexChannelsBatch(progress)
|
||||
}
|
||||
if !progress.DoneUsers {
|
||||
return worker.IndexUsersBatch(progress)
|
||||
}
|
||||
return progress, model.NewAppError("BleveIndexerWorker", "bleveengine.indexer.index_batch.nothing_left_to_index.error", nil, "", http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) IndexPostsBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) {
|
||||
endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000)
|
||||
|
||||
var posts []*model.PostForIndexing
|
||||
|
||||
tries := 0
|
||||
for posts == nil {
|
||||
var err *model.AppError
|
||||
posts, err = worker.jobServer.Store.Post().GetPostsBatchForIndexing(progress.LastEntityTime, endTime, BATCH_SIZE)
|
||||
if err != nil {
|
||||
if tries >= 10 {
|
||||
return progress, err
|
||||
} else {
|
||||
mlog.Warn("Failed to get posts batch for indexing. Retrying.", mlog.Err(err))
|
||||
|
||||
// Wait a bit before trying again.
|
||||
time.Sleep(15 * time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
tries++
|
||||
}
|
||||
|
||||
newLastMessageTime, err := worker.BulkIndexPosts(posts, progress)
|
||||
if err != nil {
|
||||
return progress, err
|
||||
}
|
||||
|
||||
// Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this
|
||||
// case, set the "newLastMessageTime" to the endTime so we don't get stuck running the same query in a loop.
|
||||
if len(posts) < BATCH_SIZE {
|
||||
newLastMessageTime = endTime
|
||||
}
|
||||
|
||||
// When to Stop: we index either until we pass a batch of messages where the last
|
||||
// message is created at or after the specified end time when setting up the batch
|
||||
// index, or until two consecutive full batches have the same end time of their final
|
||||
// messages. This second case is safe as long as the assumption that the database
|
||||
// cannot contain more messages with the same CreateAt time than the batch size holds.
|
||||
if progress.EndAtTime <= newLastMessageTime {
|
||||
progress.DonePosts = true
|
||||
progress.LastEntityTime = progress.StartAtTime
|
||||
} else if progress.LastEntityTime == newLastMessageTime && len(posts) == BATCH_SIZE {
|
||||
mlog.Error("More posts with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastMessageTime), mlog.Int("Batch Size", BATCH_SIZE))
|
||||
progress.DonePosts = true
|
||||
progress.LastEntityTime = progress.StartAtTime
|
||||
} else {
|
||||
progress.LastEntityTime = newLastMessageTime
|
||||
}
|
||||
|
||||
progress.DonePostsCount += int64(len(posts))
|
||||
|
||||
return progress, nil
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) BulkIndexPosts(posts []*model.PostForIndexing, progress IndexingProgress) (int64, *model.AppError) {
|
||||
lastCreateAt := int64(0)
|
||||
batch := worker.engine.PostIndex.NewBatch()
|
||||
|
||||
for _, post := range posts {
|
||||
if post.DeleteAt == 0 {
|
||||
searchPost := bleveengine.BLVPostFromPostForIndexing(post)
|
||||
batch.Index(searchPost.Id, searchPost)
|
||||
} else {
|
||||
batch.Delete(post.Id)
|
||||
}
|
||||
|
||||
lastCreateAt = post.CreateAt
|
||||
}
|
||||
|
||||
worker.engine.Mutex.RLock()
|
||||
defer worker.engine.Mutex.RUnlock()
|
||||
|
||||
if err := worker.engine.PostIndex.Batch(batch); err != nil {
|
||||
return 0, model.NewAppError("BleveIndexerWorker.BulkIndexPosts", "bleveengine.indexer.do_job.bulk_index_posts.batch_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return lastCreateAt, nil
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) IndexChannelsBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) {
|
||||
endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000)
|
||||
|
||||
var channels []*model.Channel
|
||||
|
||||
tries := 0
|
||||
for channels == nil {
|
||||
var err *model.AppError
|
||||
channels, err = worker.jobServer.Store.Channel().GetChannelsBatchForIndexing(progress.LastEntityTime, endTime, BATCH_SIZE)
|
||||
if err != nil {
|
||||
if tries >= 10 {
|
||||
return progress, err
|
||||
}
|
||||
|
||||
mlog.Warn("Failed to get channels batch for indexing. Retrying.", mlog.Err(err))
|
||||
|
||||
// Wait a bit before trying again.
|
||||
time.Sleep(15 * time.Second)
|
||||
}
|
||||
tries++
|
||||
}
|
||||
|
||||
newLastChannelTime, err := worker.BulkIndexChannels(channels, progress)
|
||||
if err != nil {
|
||||
return progress, err
|
||||
}
|
||||
|
||||
// Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this
|
||||
// case, set the "newLastChannelTime" to the endTime so we don't get stuck running the same query in a loop.
|
||||
if len(channels) < BATCH_SIZE {
|
||||
newLastChannelTime = endTime
|
||||
}
|
||||
|
||||
// When to Stop: we index either until we pass a batch of channels where the last
|
||||
// channel is created at or after the specified end time when setting up the batch
|
||||
// index, or until two consecutive full batches have the same end time of their final
|
||||
// channels. This second case is safe as long as the assumption that the database
|
||||
// cannot contain more channels with the same CreateAt time than the batch size holds.
|
||||
if progress.EndAtTime <= newLastChannelTime {
|
||||
progress.DoneChannels = true
|
||||
progress.LastEntityTime = progress.StartAtTime
|
||||
} else if progress.LastEntityTime == newLastChannelTime && len(channels) == BATCH_SIZE {
|
||||
mlog.Error("More channels with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastChannelTime), mlog.Int("Batch Size", BATCH_SIZE))
|
||||
progress.DoneChannels = true
|
||||
progress.LastEntityTime = progress.StartAtTime
|
||||
} else {
|
||||
progress.LastEntityTime = newLastChannelTime
|
||||
}
|
||||
|
||||
progress.DoneChannelsCount += int64(len(channels))
|
||||
|
||||
return progress, nil
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) BulkIndexChannels(channels []*model.Channel, progress IndexingProgress) (int64, *model.AppError) {
|
||||
lastCreateAt := int64(0)
|
||||
batch := worker.engine.ChannelIndex.NewBatch()
|
||||
|
||||
for _, channel := range channels {
|
||||
if channel.DeleteAt == 0 {
|
||||
searchChannel := bleveengine.BLVChannelFromChannel(channel)
|
||||
batch.Index(searchChannel.Id, searchChannel)
|
||||
} else {
|
||||
batch.Delete(channel.Id)
|
||||
}
|
||||
|
||||
lastCreateAt = channel.CreateAt
|
||||
}
|
||||
|
||||
worker.engine.Mutex.RLock()
|
||||
defer worker.engine.Mutex.RUnlock()
|
||||
|
||||
if err := worker.engine.ChannelIndex.Batch(batch); err != nil {
|
||||
return 0, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return lastCreateAt, nil
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) IndexUsersBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) {
|
||||
endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000)
|
||||
|
||||
var users []*model.UserForIndexing
|
||||
|
||||
tries := 0
|
||||
for users == nil {
|
||||
if usersBatch, err := worker.jobServer.Store.User().GetUsersBatchForIndexing(progress.LastEntityTime, endTime, BATCH_SIZE); err != nil {
|
||||
if tries >= 10 {
|
||||
return progress, err
|
||||
} else {
|
||||
mlog.Warn("Failed to get users batch for indexing. Retrying.", mlog.Err(err))
|
||||
|
||||
// Wait a bit before trying again.
|
||||
time.Sleep(15 * time.Second)
|
||||
}
|
||||
} else {
|
||||
users = usersBatch
|
||||
}
|
||||
|
||||
tries++
|
||||
}
|
||||
|
||||
newLastUserTime, err := worker.BulkIndexUsers(users, progress)
|
||||
if err != nil {
|
||||
return progress, err
|
||||
}
|
||||
|
||||
// Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this
|
||||
// case, set the "newLastUserTime" to the endTime so we don't get stuck running the same query in a loop.
|
||||
if len(users) < BATCH_SIZE {
|
||||
newLastUserTime = endTime
|
||||
}
|
||||
|
||||
// When to Stop: we index either until we pass a batch of users where the last
|
||||
// user is created at or after the specified end time when setting up the batch
|
||||
// index, or until two consecutive full batches have the same end time of their final
|
||||
// users. This second case is safe as long as the assumption that the database
|
||||
// cannot contain more users with the same CreateAt time than the batch size holds.
|
||||
if progress.EndAtTime <= newLastUserTime {
|
||||
progress.DoneUsers = true
|
||||
progress.LastEntityTime = progress.StartAtTime
|
||||
} else if progress.LastEntityTime == newLastUserTime && len(users) == BATCH_SIZE {
|
||||
mlog.Error("More users with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastUserTime), mlog.Int("Batch Size", BATCH_SIZE))
|
||||
progress.DoneUsers = true
|
||||
progress.LastEntityTime = progress.StartAtTime
|
||||
} else {
|
||||
progress.LastEntityTime = newLastUserTime
|
||||
}
|
||||
|
||||
progress.DoneUsersCount += int64(len(users))
|
||||
|
||||
return progress, nil
|
||||
}
|
||||
|
||||
func (worker *BleveIndexerWorker) BulkIndexUsers(users []*model.UserForIndexing, progress IndexingProgress) (int64, *model.AppError) {
|
||||
lastCreateAt := int64(0)
|
||||
batch := worker.engine.UserIndex.NewBatch()
|
||||
|
||||
for _, user := range users {
|
||||
if user.DeleteAt == 0 {
|
||||
searchUser := bleveengine.BLVUserFromUserForIndexing(user)
|
||||
batch.Index(searchUser.Id, searchUser)
|
||||
} else {
|
||||
batch.Delete(user.Id)
|
||||
}
|
||||
|
||||
lastCreateAt = user.CreateAt
|
||||
}
|
||||
|
||||
worker.engine.Mutex.RLock()
|
||||
defer worker.engine.Mutex.RUnlock()
|
||||
|
||||
if err := worker.engine.UserIndex.Batch(batch); err != nil {
|
||||
return 0, model.NewAppError("BleveIndexerWorker.BulkIndexUsers", "bleveengine.indexer.do_job.bulk_index_users.batch_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return lastCreateAt, nil
|
||||
}
|
||||
419
services/searchengine/bleveengine/search.go
Обычный файл
419
services/searchengine/bleveengine/search.go
Обычный файл
@@ -0,0 +1,419 @@
|
||||
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
||||
// See LICENSE.txt for license information.
|
||||
|
||||
package bleveengine
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v5/model"
|
||||
|
||||
"github.com/blevesearch/bleve"
|
||||
"github.com/blevesearch/bleve/search/query"
|
||||
)
|
||||
|
||||
func (b *BleveEngine) IndexPost(post *model.Post, teamId string) *model.AppError {
|
||||
b.Mutex.RLock()
|
||||
defer b.Mutex.RUnlock()
|
||||
|
||||
blvPost := BLVPostFromPost(post, teamId)
|
||||
if err := b.PostIndex.Index(blvPost.Id, blvPost); err != nil {
|
||||
return model.NewAppError("Bleveengine.IndexPost", "bleveengine.index_post.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) SearchPosts(channels *model.ChannelList, searchParams []*model.SearchParams, page, perPage int) ([]string, model.PostSearchMatches, *model.AppError) {
|
||||
channelQueries := []query.Query{}
|
||||
for _, channel := range *channels {
|
||||
channelIdQ := bleve.NewTermQuery(channel.Id)
|
||||
channelIdQ.SetField("ChannelId")
|
||||
channelQueries = append(channelQueries, channelIdQ)
|
||||
}
|
||||
channelDisjunctionQ := bleve.NewDisjunctionQuery(channelQueries...)
|
||||
|
||||
var termQueries []query.Query
|
||||
var notTermQueries []query.Query
|
||||
var filters []query.Query
|
||||
var notFilters []query.Query
|
||||
|
||||
typeQ := bleve.NewTermQuery("")
|
||||
typeQ.SetField("Type")
|
||||
filters = append(filters, typeQ)
|
||||
|
||||
for i, params := range searchParams {
|
||||
// Date, channels and FromUsers filters come in all
|
||||
// searchParams iteration, and as they are global to the
|
||||
// query, we only need to process them once
|
||||
if i == 0 {
|
||||
if len(params.InChannels) > 0 {
|
||||
inChannels := []query.Query{}
|
||||
for _, channelId := range params.InChannels {
|
||||
channelQ := bleve.NewTermQuery(channelId)
|
||||
channelQ.SetField("ChannelId")
|
||||
inChannels = append(inChannels, channelQ)
|
||||
}
|
||||
filters = append(filters, bleve.NewDisjunctionQuery(inChannels...))
|
||||
}
|
||||
|
||||
if len(params.ExcludedChannels) > 0 {
|
||||
excludedChannels := []query.Query{}
|
||||
for _, channelId := range params.ExcludedChannels {
|
||||
channelQ := bleve.NewTermQuery(channelId)
|
||||
channelQ.SetField("ChannelId")
|
||||
excludedChannels = append(excludedChannels, channelQ)
|
||||
}
|
||||
notFilters = append(notFilters, bleve.NewDisjunctionQuery(excludedChannels...))
|
||||
}
|
||||
|
||||
if len(params.FromUsers) > 0 {
|
||||
fromUsers := []query.Query{}
|
||||
for _, userId := range params.FromUsers {
|
||||
userQ := bleve.NewTermQuery(userId)
|
||||
userQ.SetField("UserId")
|
||||
fromUsers = append(fromUsers, userQ)
|
||||
}
|
||||
filters = append(filters, bleve.NewDisjunctionQuery(fromUsers...))
|
||||
}
|
||||
|
||||
if len(params.ExcludedUsers) > 0 {
|
||||
excludedUsers := []query.Query{}
|
||||
for _, userId := range params.ExcludedUsers {
|
||||
userQ := bleve.NewTermQuery(userId)
|
||||
userQ.SetField("UserId")
|
||||
excludedUsers = append(excludedUsers, userQ)
|
||||
}
|
||||
notFilters = append(notFilters, bleve.NewDisjunctionQuery(excludedUsers...))
|
||||
}
|
||||
|
||||
if params.OnDate != "" {
|
||||
before, after := params.GetOnDateMillis()
|
||||
beforeFloat64 := float64(before)
|
||||
afterFloat64 := float64(after)
|
||||
onDateQ := bleve.NewNumericRangeQuery(&beforeFloat64, &afterFloat64)
|
||||
onDateQ.SetField("CreateAt")
|
||||
filters = append(filters, onDateQ)
|
||||
} else {
|
||||
if params.AfterDate != "" || params.BeforeDate != "" {
|
||||
var min, max *float64
|
||||
if params.AfterDate != "" {
|
||||
minf := float64(params.GetAfterDateMillis())
|
||||
min = &minf
|
||||
}
|
||||
|
||||
if params.BeforeDate != "" {
|
||||
maxf := float64(params.GetBeforeDateMillis())
|
||||
max = &maxf
|
||||
}
|
||||
|
||||
dateQ := bleve.NewNumericRangeQuery(min, max)
|
||||
dateQ.SetField("CreateAt")
|
||||
filters = append(filters, dateQ)
|
||||
}
|
||||
|
||||
if params.ExcludedAfterDate != "" {
|
||||
minf := float64(params.GetExcludedAfterDateMillis())
|
||||
dateQ := bleve.NewNumericRangeQuery(&minf, nil)
|
||||
dateQ.SetField("CreateAt")
|
||||
notFilters = append(notFilters, dateQ)
|
||||
}
|
||||
|
||||
if params.ExcludedBeforeDate != "" {
|
||||
maxf := float64(params.GetExcludedBeforeDateMillis())
|
||||
dateQ := bleve.NewNumericRangeQuery(nil, &maxf)
|
||||
dateQ.SetField("CreateAt")
|
||||
notFilters = append(notFilters, dateQ)
|
||||
}
|
||||
|
||||
if params.ExcludedDate != "" {
|
||||
before, after := params.GetExcludedDateMillis()
|
||||
beforef := float64(before)
|
||||
afterf := float64(after)
|
||||
onDateQ := bleve.NewNumericRangeQuery(&beforef, &afterf)
|
||||
onDateQ.SetField("CreateAt")
|
||||
notFilters = append(notFilters, onDateQ)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if params.IsHashtag {
|
||||
if params.Terms != "" {
|
||||
hashtagQ := bleve.NewMatchQuery(params.Terms)
|
||||
hashtagQ.SetField("Hashtags")
|
||||
termQueries = append(termQueries, hashtagQ)
|
||||
} else if params.ExcludedTerms != "" {
|
||||
hashtagQ := bleve.NewMatchQuery(params.ExcludedTerms)
|
||||
hashtagQ.SetField("Hashtags")
|
||||
notTermQueries = append(notTermQueries, hashtagQ)
|
||||
}
|
||||
} else {
|
||||
if len(params.Terms) > 0 {
|
||||
query := bleve.NewBooleanQuery()
|
||||
messageQ := bleve.NewMatchQuery(params.Terms)
|
||||
messageQ.SetField("Message")
|
||||
|
||||
if searchParams[0].OrTerms {
|
||||
query.AddShould(messageQ)
|
||||
} else {
|
||||
query.AddMust(messageQ)
|
||||
}
|
||||
termQueries = append(termQueries, messageQ)
|
||||
}
|
||||
|
||||
if len(params.ExcludedTerms) > 0 {
|
||||
messageQ := bleve.NewMatchQuery(params.ExcludedTerms)
|
||||
messageQ.SetField("Message")
|
||||
notTermQueries = append(notTermQueries, messageQ)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
allTermsQ := bleve.NewBooleanQuery()
|
||||
allTermsQ.AddMustNot(notTermQueries...)
|
||||
if searchParams[0].OrTerms {
|
||||
allTermsQ.AddShould(termQueries...)
|
||||
} else {
|
||||
allTermsQ.AddMust(termQueries...)
|
||||
}
|
||||
|
||||
query := bleve.NewBooleanQuery()
|
||||
query.AddMust(channelDisjunctionQ)
|
||||
|
||||
if len(termQueries) > 0 || len(notTermQueries) > 0 {
|
||||
query.AddMust(allTermsQ)
|
||||
}
|
||||
|
||||
if len(filters) > 0 {
|
||||
query.AddMust(bleve.NewConjunctionQuery(filters...))
|
||||
}
|
||||
if len(notFilters) > 0 {
|
||||
query.AddMustNot(notFilters...)
|
||||
}
|
||||
|
||||
search := bleve.NewSearchRequest(query)
|
||||
results, err := b.PostIndex.Search(search)
|
||||
if err != nil {
|
||||
return nil, nil, model.NewAppError("Bleveengine.SearchPosts", "bleveengine.search_posts.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
postIds := []string{}
|
||||
matches := model.PostSearchMatches{}
|
||||
|
||||
for _, r := range results.Hits {
|
||||
postIds = append(postIds, r.ID)
|
||||
}
|
||||
|
||||
return postIds, matches, nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) DeletePost(post *model.Post) *model.AppError {
|
||||
b.Mutex.RLock()
|
||||
defer b.Mutex.RUnlock()
|
||||
|
||||
if err := b.PostIndex.Delete(post.Id); err != nil {
|
||||
return model.NewAppError("Bleveengine.DeletePost", "bleveengine.delete_post.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) IndexChannel(channel *model.Channel) *model.AppError {
|
||||
b.Mutex.RLock()
|
||||
defer b.Mutex.RUnlock()
|
||||
|
||||
blvChannel := BLVChannelFromChannel(channel)
|
||||
if err := b.ChannelIndex.Index(blvChannel.Id, blvChannel); err != nil {
|
||||
return model.NewAppError("Bleveengine.IndexChannel", "bleveengine.index_channel.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) SearchChannels(teamId, term string) ([]string, *model.AppError) {
|
||||
teamIdQ := bleve.NewTermQuery(teamId)
|
||||
teamIdQ.SetField("TeamId")
|
||||
queries := []query.Query{teamIdQ}
|
||||
|
||||
if term != "" {
|
||||
nameSuggestQ := bleve.NewPrefixQuery(strings.ToLower(term))
|
||||
nameSuggestQ.SetField("NameSuggest")
|
||||
queries = append(queries, nameSuggestQ)
|
||||
}
|
||||
|
||||
query := bleve.NewSearchRequest(bleve.NewConjunctionQuery(queries...))
|
||||
results, err := b.ChannelIndex.Search(query)
|
||||
if err != nil {
|
||||
return nil, model.NewAppError("Bleveengine.SearchChannels", "bleveengine.search_channels.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
channelIds := []string{}
|
||||
for _, result := range results.Hits {
|
||||
channelIds = append(channelIds, result.ID)
|
||||
}
|
||||
|
||||
return channelIds, nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) DeleteChannel(channel *model.Channel) *model.AppError {
|
||||
b.Mutex.RLock()
|
||||
defer b.Mutex.RUnlock()
|
||||
|
||||
if err := b.ChannelIndex.Delete(channel.Id); err != nil {
|
||||
return model.NewAppError("Bleveengine.DeleteChannel", "bleveengine.delete_channel.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) IndexUser(user *model.User, teamsIds, channelsIds []string) *model.AppError {
|
||||
b.Mutex.RLock()
|
||||
defer b.Mutex.RUnlock()
|
||||
|
||||
blvUser := BLVUserFromUserAndTeams(user, teamsIds, channelsIds)
|
||||
if err := b.UserIndex.Index(blvUser.Id, blvUser); err != nil {
|
||||
return model.NewAppError("Bleveengine.IndexUser", "bleveengine.index_user.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) SearchUsersInChannel(teamId, channelId string, restrictedToChannels []string, term string, options *model.UserSearchOptions) ([]string, []string, *model.AppError) {
|
||||
if restrictedToChannels != nil && len(restrictedToChannels) == 0 {
|
||||
return []string{}, []string{}, nil
|
||||
}
|
||||
|
||||
// users in channel
|
||||
var queries []query.Query
|
||||
if term != "" {
|
||||
termQ := bleve.NewPrefixQuery(strings.ToLower(term))
|
||||
if options.AllowFullNames {
|
||||
termQ.SetField("SuggestionsWithFullname")
|
||||
} else {
|
||||
termQ.SetField("SuggestionsWithoutFullname")
|
||||
}
|
||||
queries = append(queries, termQ)
|
||||
}
|
||||
|
||||
channelIdQ := bleve.NewTermQuery(channelId)
|
||||
channelIdQ.SetField("ChannelsIds")
|
||||
queries = append(queries, channelIdQ)
|
||||
|
||||
query := bleve.NewConjunctionQuery(queries...)
|
||||
|
||||
uchan, err := b.UserIndex.Search(bleve.NewSearchRequest(query))
|
||||
if err != nil {
|
||||
return nil, nil, model.NewAppError("Bleveengine.SearchUsersInChannel", "bleveengine.search_users_in_channel.uchan.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
// users not in channel
|
||||
boolQ := bleve.NewBooleanQuery()
|
||||
|
||||
if term != "" {
|
||||
termQ := bleve.NewPrefixQuery(strings.ToLower(term))
|
||||
if options.AllowFullNames {
|
||||
termQ.SetField("SuggestionsWithFullname")
|
||||
} else {
|
||||
termQ.SetField("SuggestionsWithoutFullname")
|
||||
}
|
||||
boolQ.AddMust(termQ)
|
||||
}
|
||||
|
||||
teamIdQ := bleve.NewTermQuery(teamId)
|
||||
teamIdQ.SetField("TeamsIds")
|
||||
boolQ.AddMust(teamIdQ)
|
||||
|
||||
outsideChannelIdQ := bleve.NewTermQuery(channelId)
|
||||
outsideChannelIdQ.SetField("ChannelsIds")
|
||||
boolQ.AddMustNot(outsideChannelIdQ)
|
||||
|
||||
if len(restrictedToChannels) > 0 {
|
||||
restrictedChannelsQ := bleve.NewDisjunctionQuery()
|
||||
for _, channelId := range restrictedToChannels {
|
||||
restrictedChannelQ := bleve.NewTermQuery(channelId)
|
||||
restrictedChannelsQ.AddQuery(restrictedChannelQ)
|
||||
}
|
||||
boolQ.AddMust(restrictedChannelsQ)
|
||||
}
|
||||
|
||||
nuchan, err := b.UserIndex.Search(bleve.NewSearchRequest(boolQ))
|
||||
if err != nil {
|
||||
return nil, nil, model.NewAppError("Bleveengine.SearchUsersInChannel", "bleveengine.search_users_in_channel.nuchan.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
uchanIds := []string{}
|
||||
for _, result := range uchan.Hits {
|
||||
uchanIds = append(uchanIds, result.ID)
|
||||
}
|
||||
|
||||
nuchanIds := []string{}
|
||||
for _, result := range nuchan.Hits {
|
||||
nuchanIds = append(nuchanIds, result.ID)
|
||||
}
|
||||
|
||||
return uchanIds, nuchanIds, nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) SearchUsersInTeam(teamId string, restrictedToChannels []string, term string, options *model.UserSearchOptions) ([]string, *model.AppError) {
|
||||
if restrictedToChannels != nil && len(restrictedToChannels) == 0 {
|
||||
return []string{}, nil
|
||||
}
|
||||
|
||||
var rootQ query.Query
|
||||
if term == "" && teamId == "" && restrictedToChannels == nil {
|
||||
rootQ = bleve.NewMatchAllQuery()
|
||||
} else {
|
||||
boolQ := bleve.NewBooleanQuery()
|
||||
|
||||
if term != "" {
|
||||
termQ := bleve.NewPrefixQuery(strings.ToLower(term))
|
||||
if options.AllowFullNames {
|
||||
termQ.SetField("SuggestionsWithFullname")
|
||||
} else {
|
||||
termQ.SetField("SuggestionsWithoutFullname")
|
||||
}
|
||||
boolQ.AddMust(termQ)
|
||||
}
|
||||
|
||||
if len(restrictedToChannels) > 0 {
|
||||
// restricted channels are already filtered by team, so we
|
||||
// can search only those matches
|
||||
restrictedChannelsQ := []query.Query{}
|
||||
for _, channelId := range restrictedToChannels {
|
||||
channelIdQ := bleve.NewTermQuery(channelId)
|
||||
channelIdQ.SetField("ChannelsIds")
|
||||
restrictedChannelsQ = append(restrictedChannelsQ, channelIdQ)
|
||||
}
|
||||
boolQ.AddMust(bleve.NewDisjunctionQuery(restrictedChannelsQ...))
|
||||
} else {
|
||||
// this means that we only need to restrict by team
|
||||
if teamId != "" {
|
||||
teamIdQ := bleve.NewTermQuery(teamId)
|
||||
teamIdQ.SetField("TeamsIds")
|
||||
boolQ.AddMust(teamIdQ)
|
||||
}
|
||||
}
|
||||
|
||||
rootQ = boolQ
|
||||
}
|
||||
|
||||
search := bleve.NewSearchRequest(rootQ)
|
||||
|
||||
results, err := b.UserIndex.Search(search)
|
||||
if err != nil {
|
||||
return nil, model.NewAppError("Bleveengine.SearchUsersInTeam", "bleveengine.search_users_in_team.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
usersIds := []string{}
|
||||
for _, r := range results.Hits {
|
||||
usersIds = append(usersIds, r.ID)
|
||||
}
|
||||
|
||||
return usersIds, nil
|
||||
}
|
||||
|
||||
func (b *BleveEngine) DeleteUser(user *model.User) *model.AppError {
|
||||
b.Mutex.RLock()
|
||||
defer b.Mutex.RUnlock()
|
||||
|
||||
if err := b.UserIndex.Delete(user.Id); err != nil {
|
||||
return model.NewAppError("Bleveengine.DeleteUser", "bleveengine.delete_user.error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -19,10 +19,15 @@ func (seb *Broker) RegisterElasticsearchEngine(es SearchEngineInterface) {
|
||||
seb.ElasticsearchEngine = es
|
||||
}
|
||||
|
||||
func (seb *Broker) RegisterBleveEngine(be SearchEngineInterface) {
|
||||
seb.BleveEngine = be
|
||||
}
|
||||
|
||||
type Broker struct {
|
||||
cfg *model.Config
|
||||
jobServer *jobs.JobServer
|
||||
ElasticsearchEngine SearchEngineInterface
|
||||
BleveEngine SearchEngineInterface
|
||||
}
|
||||
|
||||
func (seb *Broker) UpdateConfig(cfg *model.Config) *model.AppError {
|
||||
@@ -31,6 +36,10 @@ func (seb *Broker) UpdateConfig(cfg *model.Config) *model.AppError {
|
||||
seb.ElasticsearchEngine.UpdateConfig(cfg)
|
||||
}
|
||||
|
||||
if seb.BleveEngine != nil {
|
||||
seb.BleveEngine.UpdateConfig(cfg)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -39,5 +48,8 @@ func (seb *Broker) GetActiveEngines() []SearchEngineInterface {
|
||||
if seb.ElasticsearchEngine != nil && seb.ElasticsearchEngine.IsActive() {
|
||||
engines = append(engines, seb.ElasticsearchEngine)
|
||||
}
|
||||
if seb.BleveEngine != nil && seb.BleveEngine.IsActive() {
|
||||
engines = append(engines, seb.BleveEngine)
|
||||
}
|
||||
return engines
|
||||
}
|
||||
|
||||
44
services/searchengine/utils.go
Обычный файл
44
services/searchengine/utils.go
Обычный файл
@@ -0,0 +1,44 @@
|
||||
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
||||
// See LICENSE.txt for license information.
|
||||
|
||||
package searchengine
|
||||
|
||||
import (
|
||||
"regexp"
|
||||
"strings"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v5/utils"
|
||||
)
|
||||
|
||||
var EmailRegex = regexp.MustCompile(`^[^\s"]+@[^\s"]+$`)
|
||||
|
||||
func GetSuggestionInputsSplitBy(term, splitStr string) []string {
|
||||
splitTerm := strings.Split(strings.ToLower(term), splitStr)
|
||||
var initialSuggestionList []string
|
||||
for i := range splitTerm {
|
||||
initialSuggestionList = append(initialSuggestionList, strings.Join(splitTerm[i:], splitStr))
|
||||
}
|
||||
|
||||
suggestionList := []string{}
|
||||
// If splitStr is not an empty space, we create a suggestion with it at the beginning
|
||||
if splitStr == " " {
|
||||
suggestionList = initialSuggestionList
|
||||
} else {
|
||||
for i, suggestion := range initialSuggestionList {
|
||||
if i == 0 {
|
||||
suggestionList = append(suggestionList, suggestion)
|
||||
} else {
|
||||
suggestionList = append(suggestionList, splitStr+suggestion, suggestion)
|
||||
}
|
||||
}
|
||||
}
|
||||
return suggestionList
|
||||
}
|
||||
|
||||
func GetSuggestionInputsSplitByMultiple(term string, splitStrs []string) []string {
|
||||
suggestionList := []string{}
|
||||
for _, splitStr := range splitStrs {
|
||||
suggestionList = append(suggestionList, GetSuggestionInputsSplitBy(term, splitStr)...)
|
||||
}
|
||||
return utils.RemoveDuplicatesFromStringArray(suggestionList)
|
||||
}
|
||||
57
services/searchengine/utils_test.go
Обычный файл
57
services/searchengine/utils_test.go
Обычный файл
@@ -0,0 +1,57 @@
|
||||
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
||||
// See LICENSE.txt for license information.
|
||||
|
||||
package searchengine
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestElasticsearchGetSuggestionsSplitBy(t *testing.T) {
|
||||
testCases := []struct {
|
||||
Name string
|
||||
Term string
|
||||
SplitStr string ``
|
||||
Expected []string
|
||||
}{
|
||||
{
|
||||
Name: "Single string",
|
||||
Term: "string",
|
||||
SplitStr: " ",
|
||||
Expected: []string{"string"},
|
||||
},
|
||||
{
|
||||
Name: "String with spaces",
|
||||
Term: "String with spaces",
|
||||
SplitStr: " ",
|
||||
Expected: []string{"string with spaces", "with spaces", "spaces"},
|
||||
},
|
||||
{
|
||||
Name: "Username split by a dot",
|
||||
Term: "name.surname",
|
||||
SplitStr: ".",
|
||||
Expected: []string{"name.surname", ".surname", "surname"},
|
||||
},
|
||||
{
|
||||
Name: "String split by several dashes",
|
||||
Term: "one-two-three",
|
||||
SplitStr: "-",
|
||||
Expected: []string{"one-two-three", "-two-three", "two-three", "-three", "three"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.Name, func(t *testing.T) {
|
||||
res := GetSuggestionInputsSplitBy(tc.Term, tc.SplitStr)
|
||||
assert.ElementsMatch(t, res, tc.Expected)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestElasticsearchGetSuggestionsSplitByMultiple(t *testing.T) {
|
||||
r1 := GetSuggestionInputsSplitByMultiple("String with user.name", []string{" ", "."})
|
||||
expectedR1 := []string{"string with user.name", "with user.name", "user.name", ".name", "name"}
|
||||
assert.ElementsMatch(t, r1, expectedR1)
|
||||
}
|
||||
Ссылка в новой задаче
Block a user