[pull] master from mattermost:master

server/channels/store/context.go

@@ -0,0 +1,46 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package store
+
+import (
+	"context"
+
+	"github.com/mattermost/mattermost/server/public/shared/request"
+)
+
+// storeContextKey is the base type for all context keys for the store.
+type storeContextKey string
+
+// contextValue is a type to hold some pre-determined context values.
+type contextValue string
+
+// Different possible values of contextValue.
+const (
+	useMaster contextValue = "useMaster"
+)
+
+// WithMaster adds the context value that master DB should be selected for this request.
+//
+// Deprecated: This method is deprecated and there's ongoing change to use `request.CTX` across
+// instead of `context.Context`. Please use `RequestContextWithMaster` instead.
+func WithMaster(ctx context.Context) context.Context {
+	return context.WithValue(ctx, storeContextKey(useMaster), true)
+}
+
+// RequestContextWithMaster adds the context value that master DB should be selected for this request.
+func RequestContextWithMaster(c request.CTX) request.CTX {
+	ctx := WithMaster(c.Context())
+	c = c.WithContext(ctx)
+	return c
+}
+
+// HasMaster is a helper function to check whether master DB should be selected or not.
+func HasMaster(ctx context.Context) bool {
+	if v := ctx.Value(storeContextKey(useMaster)); v != nil {
+		if res, ok := v.(bool); ok && res {
+			return true
+		}
+	}
+	return false
+}

server/channels/store/context_test.go

@@ -0,0 +1,55 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package store
+
+import (
+	"context"
+	"testing"
+
+	"github.com/mattermost/mattermost/server/public/shared/request"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestContextMaster(t *testing.T) {
+	ctx := context.Background()
+
+	m := WithMaster(ctx)
+	assert.True(t, HasMaster(m))
+}
+
+func TestRequestContextWithMaster(t *testing.T) {
+	t.Run("set and get", func(t *testing.T) {
+		var rctx request.CTX = request.TestContext(t)
+
+		rctx = RequestContextWithMaster(rctx)
+		assert.True(t, HasMaster(rctx.Context()))
+	})
+
+	t.Run("values get copied from original context", func(t *testing.T) {
+		var rctx request.CTX = request.TestContext(t)
+		rctx = RequestContextWithMaster(rctx)
+		rctxCopy := rctx
+
+		assert.True(t, HasMaster(rctx.Context()))
+		assert.True(t, HasMaster(rctxCopy.Context()))
+	})
+
+	t.Run("directly assigning does not cause the copy to alter the original context", func(t *testing.T) {
+		var rctx request.CTX = request.TestContext(t)
+		rctxCopy := rctx
+		rctxCopy = RequestContextWithMaster(rctxCopy)
+
+		assert.False(t, HasMaster(rctx.Context()))
+		assert.True(t, HasMaster(rctxCopy.Context()))
+	})
+
+	t.Run("directly assigning does not cause the original context to alter the copy", func(t *testing.T) {
+		var rctx request.CTX = request.TestContext(t)
+		rctxCopy := rctx
+		rctx = RequestContextWithMaster(rctx)
+
+		assert.True(t, HasMaster(rctx.Context()))
+		assert.False(t, HasMaster(rctxCopy.Context()))
+	})
+}

server/channels/store/searchlayer/channel_layer.go

@@ -47,7 +47,7 @@ func (c *SearchChannelStore) indexChannel(rctx request.CTX, channel *model.Chann
 		}
 	}
 
-	teamMemberIDs, err = c.GetTeamMembersForChannel(channel.Id)
+	teamMemberIDs, err = c.GetTeamMembersForChannel(rctx, channel.Id)
 	if err != nil {
 		rctx.Logger().Warn("Encountered error while indexing channel", mlog.String("channel_id", channel.Id), mlog.Err(err))
 		return
@@ -66,6 +66,30 @@ func (c *SearchChannelStore) indexChannel(rctx request.CTX, channel *model.Chann
 	}
 }
 
+func (c *SearchChannelStore) bulkIndexChannels(rctx request.CTX, channels []*model.Channel, teamMemberIDs []string) {
+	// Util function to get userIDs, only for private channels
+	getUserIDsForPrivateChannel := func(channel *model.Channel) ([]string, error) {
+		if channel.Type != model.ChannelTypePrivate {
+			return []string{}, nil
+		}
+		return c.GetAllChannelMemberIdsByChannelId(channel.Id)
+	}
+
+	for _, engine := range c.rootStore.searchEngine.GetActiveEngines() {
+		if !engine.IsIndexingEnabled() {
+			continue
+		}
+
+		runIndexFn(rctx, engine, func(engineCopy searchengine.SearchEngineInterface) {
+			appErr := engineCopy.SyncBulkIndexChannels(rctx, channels, getUserIDsForPrivateChannel, teamMemberIDs)
+			if appErr != nil {
+				rctx.Logger().Error("Failed to synchronously bulk-index channels.", mlog.String("search_engine", engineCopy.GetName()), mlog.Err(appErr))
+				return
+			}
+		})
+	}
+}
+
 func (c *SearchChannelStore) Save(rctx request.CTX, channel *model.Channel, maxChannels int64, channelOptions ...model.ChannelOption) (*model.Channel, error) {
 	newChannel, err := c.ChannelStore.Save(rctx, channel, maxChannels, channelOptions...)
 	if err == nil {

server/channels/store/searchlayer/layer.go

@@ -9,6 +9,7 @@ import (
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/mattermost/mattermost/server/public/shared/mlog"
 	"github.com/mattermost/mattermost/server/public/shared/request"
+	"github.com/mattermost/mattermost/server/public/utils"
 	"github.com/mattermost/mattermost/server/v8/channels/store"
 	"github.com/mattermost/mattermost/server/v8/platform/services/searchengine"
 )
@@ -111,6 +112,35 @@ func (s *SearchStore) indexUser(rctx request.CTX, user *model.User) {
 	}
 }
 
+func (s *SearchStore) indexChannelsForTeam(rctx request.CTX, teamID string) {
+	const perPage = 100
+	var (
+		channels []*model.Channel
+	)
+
+	channels, err := utils.Pager(func(page int) ([]*model.Channel, error) {
+		return s.channel.GetPublicChannelsForTeam(teamID, page*perPage, perPage)
+	}, perPage)
+	if err != nil {
+		rctx.Logger().Warn("Encountered error while retrieving public channels for indexing", mlog.String("team_id", teamID), mlog.Err(err))
+		return
+	}
+
+	if len(channels) == 0 {
+		return
+	}
+
+	// Use master context to avoid replica lag issues when reading team members
+	masterRctx := store.RequestContextWithMaster(rctx)
+	teamMemberIDs, err := s.channel.GetTeamMembersForChannel(masterRctx, channels[0].Id)
+	if err != nil {
+		rctx.Logger().Warn("Encountered error while retrieving team members for channel", mlog.String("channel_id", channels[0].Id), mlog.Err(err))
+		return
+	}
+
+	s.channel.bulkIndexChannels(rctx, channels, teamMemberIDs)
+}
+
 // Runs an indexing function synchronously or asynchronously depending on the engine
 func runIndexFn(rctx request.CTX, engine searchengine.SearchEngineInterface, indexFn func(searchengine.SearchEngineInterface)) {
 	if engine.IsIndexingSync() {

server/channels/store/searchlayer/team_layer.go

@@ -17,7 +17,11 @@ type SearchTeamStore struct {
 func (s SearchTeamStore) SaveMember(rctx request.CTX, teamMember *model.TeamMember, maxUsersPerTeam int) (*model.TeamMember, error) {
 	member, err := s.TeamStore.SaveMember(rctx, teamMember, maxUsersPerTeam)
 	if err == nil {
-		s.rootStore.indexUserFromID(rctx, member.UserId)
+		// Nothing to do if search engine is not active
+		if s.rootStore.searchEngine.ActiveEngine() != "database" && s.rootStore.searchEngine.ActiveEngine() != "none" {
+			s.rootStore.indexUserFromID(rctx, member.UserId)
+			s.rootStore.indexChannelsForTeam(rctx, member.TeamId)
+		}
 	}
 	return member, err
 }
@@ -33,15 +37,31 @@ func (s SearchTeamStore) UpdateMember(rctx request.CTX, teamMember *model.TeamMe
 func (s SearchTeamStore) RemoveMember(rctx request.CTX, teamId string, userId string) error {
 	err := s.TeamStore.RemoveMember(rctx, teamId, userId)
 	if err == nil {
-		s.rootStore.indexUserFromID(rctx, userId)
+		// Nothing to do if search engine is not active
+		if s.rootStore.searchEngine.ActiveEngine() != "database" && s.rootStore.searchEngine.ActiveEngine() != "none" {
+			s.rootStore.indexUserFromID(rctx, userId)
+			s.rootStore.indexChannelsForTeam(rctx, teamId)
+		}
 	}
 	return err
 }
 
 func (s SearchTeamStore) RemoveAllMembersByUser(rctx request.CTX, userId string) error {
+	if s.rootStore.searchEngine.ActiveEngine() != "database" && s.rootStore.searchEngine.ActiveEngine() != "none" {
+		memberships, err := s.TeamStore.GetTeamsForUser(rctx, userId, "", true)
+		if err != nil {
+			return err
+		}
+		for _, membership := range memberships {
+			s.rootStore.indexChannelsForTeam(rctx, membership.TeamId)
+		}
+	}
+
 	err := s.TeamStore.RemoveAllMembersByUser(rctx, userId)
 	if err == nil {
-		s.rootStore.indexUserFromID(rctx, userId)
+		if s.rootStore.searchEngine.ActiveEngine() != "database" && s.rootStore.searchEngine.ActiveEngine() != "none" {
+			s.rootStore.indexUserFromID(rctx, userId)
+		}
 	}
 	return err
 }

server/channels/store/sqlstore/channel_store.go

@@ -3046,9 +3046,9 @@ func (s SqlChannelStore) GetMembersForUserWithCursorPagination(userId string, pe
 	return dbMembers.ToModel(), nil
 }
 
-func (s SqlChannelStore) GetTeamMembersForChannel(channelID string) ([]string, error) {
+func (s SqlChannelStore) GetTeamMembersForChannel(rctx request.CTX, channelID string) ([]string, error) {
 	teamMemberIDs := []string{}
-	if err := s.GetReplica().Select(&teamMemberIDs, `SELECT tm.UserId
+	if err := s.DBXFromContext(rctx.Context()).Select(&teamMemberIDs, `SELECT tm.UserId
 		FROM Channels c, Teams t, TeamMembers tm
 		WHERE
 			c.TeamId=t.Id

server/channels/store/sqlstore/context.go

@@ -7,42 +7,25 @@ import (
 	"context"
 
 	"github.com/mattermost/mattermost/server/public/shared/request"
-)
-
-// storeContextKey is the base type for all context keys for the store.
-type storeContextKey string
-
-// contextValue is a type to hold some pre-determined context values.
-type contextValue string
-
-// Different possible values of contextValue.
-const (
-	useMaster contextValue = "useMaster"
+	"github.com/mattermost/mattermost/server/v8/channels/store"
 )
 
 // WithMaster adds the context value that master DB should be selected for this request.
 //
 // Deprecated: This method is deprecated and there's ongoing change to use `request.CTX` across
 // instead of `context.Context`. Please use `RequestContextWithMaster` instead.
 func WithMaster(ctx context.Context) context.Context {
-	return context.WithValue(ctx, storeContextKey(useMaster), true)
+	return store.WithMaster(ctx)
 }
 
 // RequestContextWithMaster adds the context value that master DB should be selected for this request.
 func RequestContextWithMaster(c request.CTX) request.CTX {
-	ctx := WithMaster(c.Context())
-	c = c.WithContext(ctx)
-	return c
+	return store.RequestContextWithMaster(c)
 }
 
 // HasMaster is a helper function to check whether master DB should be selected or not.
 func HasMaster(ctx context.Context) bool {
-	if v := ctx.Value(storeContextKey(useMaster)); v != nil {
-		if res, ok := v.(bool); ok && res {
-			return true
-		}
-	}
-	return false
+	return store.HasMaster(ctx)
 }
 
 // DBXFromContext is a helper utility that returns the sqlx DB handle from a given context.

server/channels/store/store.go

@@ -266,7 +266,7 @@ type ChannelStore interface {
 	AnalyticsDeletedTypeCount(teamID string, channelType model.ChannelType) (int64, error)
 	AnalyticsCountAll(teamID string) (map[model.ChannelType]int64, error)
 	GetMembersForUser(teamID string, userID string) (model.ChannelMembers, error)
-	GetTeamMembersForChannel(channelID string) ([]string, error)
+	GetTeamMembersForChannel(rctx request.CTX, channelID string) ([]string, error)
 	GetMembersForUserWithPagination(userID string, page, perPage int) (model.ChannelMembersWithTeamData, error)
 	GetMembersForUserWithCursorPagination(userId string, perPage int, fromChanneID string) (model.ChannelMembersWithTeamData, error)
 	Autocomplete(rctx request.CTX, userID, term string, includeDeleted, isGuest bool) (model.ChannelListWithTeamData, error)

server/enterprise/elasticsearch/common/common.go

@@ -34,8 +34,17 @@ const (
 	// At the moment, this number is hardcoded. If needed, we can expose
 	// this to the config.
 	BulkFlushInterval = 5 * time.Second
+
+	// Size of the largest request to be done, in bytes
+	BulkFlushBytes = 10 * 1024 * 1024 // 10 MiB
 )
 
+type BulkSettings struct {
+	FlushBytes    int
+	FlushInterval time.Duration
+	FlushNumReqs  int
+}
+
 var (
 	urlRe          = regexp.MustCompile(URLRegexpRE)
 	markdownLinkRe = regexp.MustCompile(URLMarkdownLinkRE)

server/enterprise/elasticsearch/common/indexing_job.go

@@ -569,7 +569,8 @@ func BulkIndexChannels(config *model.Config,
 			}
 		}
 
-		teamMemberIDs, err := store.Channel().GetTeamMembersForChannel(channel.Id)
+		rctx := request.EmptyContext(logger)
+		teamMemberIDs, err := store.Channel().GetTeamMembersForChannel(rctx, channel.Id)
 		if err != nil {
 			return nil, model.NewAppError("IndexerWorker.BulkIndexChannels", "ent.elasticsearch.getAllTeamMembers.error", nil, "", http.StatusInternalServerError).Wrap(err)
 		}