Adding bulk-indexing and improving a bit the name indexing for bleve and elasticsearch (#16704)

* Adding bulk-indexing and improving a bit the name indexing for bleve and elasticsearch * Update services/searchengine/bleveengine/bleve.go Co-authored-by: Mario de Frutos Dieguez <mario@defrutos.org> * Update store/sqlstore/file_info_store.go Co-authored-by: Mario de Frutos Dieguez <mario@defrutos.org> * Update store/sqlstore/file_info_store.go Co-authored-by: Mario de Frutos Dieguez <mario@defrutos.org> * Adding tests requested in the PR review * fixing tests * Adding a feature flag to avoid indexing files before the feature is released * Fixing i18n Co-authored-by: Mario de Frutos Dieguez <mario@defrutos.org> Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
2021-02-22 14:28:52 +01:00
--- a/store/opentracinglayer/opentracinglayer.go
+++ b/store/opentracinglayer/opentracinglayer.go
@@ -2946,6 +2946,24 @@ func (s *OpenTracingLayerFileInfoStore) ClearCaches() {

 }

+func (s *OpenTracingLayerFileInfoStore) CountAll() (int64, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "FileInfoStore.CountAll")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.FileInfoStore.CountAll()
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerFileInfoStore) DeleteForPost(postID string) (string, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "FileInfoStore.DeleteForPost")
@@ -3018,6 +3036,24 @@ func (s *OpenTracingLayerFileInfoStore) GetByPath(path string) (*model.FileInfo,
 	return result, err
 }

+func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "FileInfoStore.GetFilesBatchForIndexing")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, endTime, limit)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerFileInfoStore) GetForPost(postID string, readFromMaster bool, includeDeleted bool, allowFromCache bool) ([]*model.FileInfo, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "FileInfoStore.GetForPost")
--- a/store/retrylayer/retrylayer.go
+++ b/store/retrylayer/retrylayer.go
@@ -3150,6 +3150,26 @@ func (s *RetryLayerFileInfoStore) ClearCaches() {

 }

+func (s *RetryLayerFileInfoStore) CountAll() (int64, error) {
+
+	tries := 0
+	for {
+		result, err := s.FileInfoStore.CountAll()
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+	}
+
+}
+
 func (s *RetryLayerFileInfoStore) DeleteForPost(postID string) (string, error) {

 	tries := 0
@@ -3230,6 +3250,26 @@ func (s *RetryLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, error

 }

+func (s *RetryLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) {
+
+	tries := 0
+	for {
+		result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, endTime, limit)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+	}
+
+}
+
 func (s *RetryLayerFileInfoStore) GetForPost(postID string, readFromMaster bool, includeDeleted bool, allowFromCache bool) ([]*model.FileInfo, error) {

 	tries := 0
--- a/store/sqlstore/file_info_store.go
+++ b/store/sqlstore/file_info_store.go
@@ -593,3 +593,40 @@ func (fs SqlFileInfoStore) Search(paramsList []*model.SearchParams, userId, team
 	list.MakeNonNil()
 	return list, nil
 }
+
+func (fs SqlFileInfoStore) CountAll() (int64, error) {
+	query := fs.getQueryBuilder().
+		Select("COUNT(*)").
+		From("FileInfo").
+		Where("DeleteAt = 0")
+
+	queryString, args, err := query.ToSql()
+	if err != nil {
+		return int64(0), errors.Wrap(err, "count_tosql")
+	}
+
+	count, err := fs.GetReplica().SelectInt(queryString, args...)
+	if err != nil {
+		return int64(0), errors.Wrap(err, "failed to count Files")
+	}
+	return count, nil
+}
+
+func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime, endTime int64, limit int) ([]*model.FileForIndexing, error) {
+	var files []*model.FileForIndexing
+	sql, args, _ := fs.getQueryBuilder().
+		Select("fi.*, p.ChannelId").
+		From("FileInfo as fi").
+		LeftJoin("Posts AS p ON fi.PostId = p.Id").
+		Where(sq.GtOrEq{"fi.CreateAt": startTime}).
+		Where(sq.Lt{"fi.CreateAt": endTime}).
+		OrderBy("fi.CreateAt").
+		Limit(uint64(limit)).
+		ToSql()
+	_, err := fs.GetSearchReplica().Select(&files, sql, args...)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to find Files")
+	}
+
+	return files, nil
+}
--- a/store/store.go
+++ b/store/store.go
@@ -580,6 +580,8 @@ type FileInfoStore interface {
 	PermanentDeleteByUser(userId string) (int64, error)
 	SetContent(fileID, content string) error
 	Search(paramsList []*model.SearchParams, userId, teamID string, page, perPage int) (*model.FileInfoList, error)
+	CountAll() (int64, error)
+	GetFilesBatchForIndexing(startTime, endTime int64, limit int) ([]*model.FileForIndexing, error)
 	ClearCaches()
 }

--- a/store/storetest/file_info_store.go
+++ b/store/storetest/file_info_store.go
@@ -27,6 +27,8 @@ func TestFileInfoStore(t *testing.T, ss store.Store) {
 	t.Run("FileInfoPermanentDelete", func(t *testing.T) { testFileInfoPermanentDelete(t, ss) })
 	t.Run("FileInfoPermanentDeleteBatch", func(t *testing.T) { testFileInfoPermanentDeleteBatch(t, ss) })
 	t.Run("FileInfoPermanentDeleteByUser", func(t *testing.T) { testFileInfoPermanentDeleteByUser(t, ss) })
+	t.Run("GetFilesBatchForIndexing", func(t *testing.T) { testFileInfoStoreGetFilesBatchForIndexing(t, ss) })
+	t.Run("CountAll", func(t *testing.T) { testFileInfoStoreCountAll(t, ss) })
 }

 func testFileInfoSaveGet(t *testing.T, ss store.Store) {
@@ -602,3 +604,144 @@ func testFileInfoPermanentDeleteByUser(t *testing.T, ss store.Store) {
 	_, err = ss.FileInfo().PermanentDeleteByUser(userId)
 	require.NoError(t, err)
 }
+
+func testFileInfoStoreGetFilesBatchForIndexing(t *testing.T, ss store.Store) {
+	c1 := &model.Channel{}
+	c1.TeamId = model.NewId()
+	c1.DisplayName = "Channel1"
+	c1.Name = "zz" + model.NewId() + "b"
+	c1.Type = model.CHANNEL_OPEN
+	c1, _ = ss.Channel().Save(c1, -1)
+
+	c2 := &model.Channel{}
+	c2.TeamId = model.NewId()
+	c2.DisplayName = "Channel2"
+	c2.Name = "zz" + model.NewId() + "b"
+	c2.Type = model.CHANNEL_OPEN
+	c2, _ = ss.Channel().Save(c2, -1)
+
+	o1 := &model.Post{}
+	o1.ChannelId = c1.Id
+	o1.UserId = model.NewId()
+	o1.Message = "zz" + model.NewId() + "AAAAAAAAAAA"
+	o1, err := ss.Post().Save(o1)
+	require.Nil(t, err)
+	f1, err := ss.FileInfo().Save(&model.FileInfo{
+		PostId:    o1.Id,
+		CreatorId: model.NewId(),
+		Path:      "file1.txt",
+	})
+	require.Nil(t, err)
+	defer func() {
+		ss.FileInfo().PermanentDelete(f1.Id)
+	}()
+	time.Sleep(1 * time.Millisecond)
+
+	o2 := &model.Post{}
+	o2.ChannelId = c2.Id
+	o2.UserId = model.NewId()
+	o2.Message = "zz" + model.NewId() + "CCCCCCCCC"
+	o2, err = ss.Post().Save(o2)
+	require.Nil(t, err)
+
+	f2, err := ss.FileInfo().Save(&model.FileInfo{
+		PostId:    o2.Id,
+		CreatorId: model.NewId(),
+		Path:      "file2.txt",
+	})
+	require.Nil(t, err)
+	defer func() {
+		ss.FileInfo().PermanentDelete(f2.Id)
+	}()
+	time.Sleep(1 * time.Millisecond)
+
+	o3 := &model.Post{}
+	o3.ChannelId = c1.Id
+	o3.UserId = model.NewId()
+	o3.ParentId = o1.Id
+	o3.RootId = o1.Id
+	o3.Message = "zz" + model.NewId() + "QQQQQQQQQQ"
+	o3, err = ss.Post().Save(o3)
+	require.Nil(t, err)
+
+	f3, err := ss.FileInfo().Save(&model.FileInfo{
+		PostId:    o3.Id,
+		CreatorId: model.NewId(),
+		Path:      "file3.txt",
+	})
+	require.Nil(t, err)
+	defer func() {
+		ss.FileInfo().PermanentDelete(f3.Id)
+	}()
+
+	t.Run("get all files", func(t *testing.T) {
+		r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt, model.GetMillis()+100000, 100)
+		require.Nil(t, err)
+		require.Len(t, r, 3, "Expected 3 posts in results. Got %v", len(r))
+		for _, f := range r {
+			if f.Id == f1.Id {
+				require.Equal(t, f.ChannelId, o1.ChannelId, "Unexpected channel ID")
+				require.Equal(t, f.Path, "file1.txt", "Unexpected filename")
+			} else if f.Id == f2.Id {
+				require.Equal(t, f.ChannelId, o2.ChannelId, "Unexpected channel ID")
+				require.Equal(t, f.Path, "file2.txt", "Unexpected filename")
+			} else if f.Id == f3.Id {
+				require.Equal(t, f.ChannelId, o3.ChannelId, "Unexpected channel ID")
+				require.Equal(t, f.Path, "file3.txt", "Unexpected filename")
+			} else {
+				require.Fail(t, "unexpected file returned")
+			}
+		}
+	})
+
+	t.Run("get files after certain date", func(t *testing.T) {
+		r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt+1, model.GetMillis()+100000, 100)
+		require.Nil(t, err)
+		require.Len(t, r, 2, "Expected 2 posts in results. Got %v", len(r))
+		for _, f := range r {
+			if f.Id == f2.Id {
+				require.Equal(t, f.ChannelId, o2.ChannelId, "Unexpected channel ID")
+				require.Equal(t, f.Path, "file2.txt", "Unexpected filename")
+			} else if f.Id == f3.Id {
+				require.Equal(t, f.ChannelId, o3.ChannelId, "Unexpected channel ID")
+				require.Equal(t, f.Path, "file3.txt", "Unexpected filename")
+			} else {
+				require.Fail(t, "unexpected file returned")
+			}
+		}
+	})
+}
+
+func testFileInfoStoreCountAll(t *testing.T, ss store.Store) {
+	_, err := ss.FileInfo().PermanentDeleteBatch(model.GetMillis(), 100000)
+	require.Nil(t, err)
+	f1, err := ss.FileInfo().Save(&model.FileInfo{
+		PostId:    model.NewId(),
+		CreatorId: model.NewId(),
+		Path:      "file1.txt",
+	})
+	require.Nil(t, err)
+
+	_, err = ss.FileInfo().Save(&model.FileInfo{
+		PostId:    model.NewId(),
+		CreatorId: model.NewId(),
+		Path:      "file2.txt",
+	})
+	require.Nil(t, err)
+	_, err = ss.FileInfo().Save(&model.FileInfo{
+		PostId:    model.NewId(),
+		CreatorId: model.NewId(),
+		Path:      "file3.txt",
+	})
+	require.Nil(t, err)
+
+	count, err := ss.FileInfo().CountAll()
+	require.Nil(t, err)
+	require.Equal(t, int64(3), count)
+
+	_, err = ss.FileInfo().DeleteForPost(f1.PostId)
+	require.Nil(t, err)
+	count, err = ss.FileInfo().CountAll()
+	require.Nil(t, err)
+	require.Equal(t, int64(2), count)
+}
--- a/store/storetest/mocks/FileInfoStore.go
+++ b/store/storetest/mocks/FileInfoStore.go
@@ -33,6 +33,27 @@ func (_m *FileInfoStore) ClearCaches() {
 	_m.Called()
 }

+// CountAll provides a mock function with given fields:
+func (_m *FileInfoStore) CountAll() (int64, error) {
+	ret := _m.Called()
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func() int64); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func() error); ok {
+		r1 = rf()
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // DeleteForPost provides a mock function with given fields: postID
 func (_m *FileInfoStore) DeleteForPost(postID string) (string, error) {
 	ret := _m.Called(postID)
@@ -123,6 +144,29 @@ func (_m *FileInfoStore) GetByPath(path string) (*model.FileInfo, error) {
 	return r0, r1
 }

+// GetFilesBatchForIndexing provides a mock function with given fields: startTime, endTime, limit
+func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) {
+	ret := _m.Called(startTime, endTime, limit)
+
+	var r0 []*model.FileForIndexing
+	if rf, ok := ret.Get(0).(func(int64, int64, int) []*model.FileForIndexing); ok {
+		r0 = rf(startTime, endTime, limit)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*model.FileForIndexing)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int64, int64, int) error); ok {
+		r1 = rf(startTime, endTime, limit)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetForPost provides a mock function with given fields: postID, readFromMaster, includeDeleted, allowFromCache
 func (_m *FileInfoStore) GetForPost(postID string, readFromMaster bool, includeDeleted bool, allowFromCache bool) ([]*model.FileInfo, error) {
 	ret := _m.Called(postID, readFromMaster, includeDeleted, allowFromCache)
--- a/store/timerlayer/timerlayer.go
+++ b/store/timerlayer/timerlayer.go
@@ -2702,6 +2702,22 @@ func (s *TimerLayerFileInfoStore) ClearCaches() {
 	}
 }

+func (s *TimerLayerFileInfoStore) CountAll() (int64, error) {
+	start := timemodule.Now()
+
+	result, err := s.FileInfoStore.CountAll()
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("FileInfoStore.CountAll", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerFileInfoStore) DeleteForPost(postID string) (string, error) {
 	start := timemodule.Now()

@@ -2766,6 +2782,22 @@ func (s *TimerLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, error
 	return result, err
 }

+func (s *TimerLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) {
+	start := timemodule.Now()
+
+	result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, endTime, limit)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("FileInfoStore.GetFilesBatchForIndexing", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerFileInfoStore) GetForPost(postID string, readFromMaster bool, includeDeleted bool, allowFromCache bool) ([]*model.FileInfo, error) {
 	start := timemodule.Now()