[MM-31094] Replication Lag (#16888)
* MM-31094: Adds tooling to develop and test using a MySQL instance with replication lag. Adds some lazy lookups to fallback to master if results are not found. * MM-31094: Removes mysql-read-replica from default docker services. * MM-31094: Switches (store..SessionStore).Get and (store.TeamStore).GetMember to using context.Context. * MM-31094: Updates (store.UsersStore).Get to use context. * MM-31094: Updates (store.PostStore).Get to use context. * MM-31094: Removes feature flag and config setting. * MM-31094: Rolls back some master reads. * MM-31094: Rolls a non-cache read. * MM-31094: Removes feature flag from the store. * MM-31094: Removes unused constant and struct field. * MM-31094: Removes some old feature flag references. * MM-31094: Fixes some tests. * MM-31094: App layers fix. * MM-31094: Fixes mocks. * MM-31094: Don't reparse flag. * MM-31094: No reparse. * MM-31094: Removed unused FeatureFlags field. * MM-31094: Removes unnecessary feature flags variable declarations. * MM-31094: Fixes copy-paste error. * MM-31094: Fixes logical error. * MM-30194: Removes test method from store. * Revert "MM-30194: Removes test method from store." This reverts commit d5a6e8529bd5f4d993824c828e239d009b05e567. * MM-31094: Conforming to make's strange syntax. * MM-31094: Configures helper for read replica with option. * MM-31094: Adds some missing ctx's. * MM-31094: WIP * MM-31094: Updates test names. * MM-31094: WIP * MM-31094: Removes unnecessary master reads. * MM-31094: ID case changes out of scope. * MM-31094: Removes unused context. * MM-31094: Switches to a helper. Removes some var naming changes. Fixes a merge error. * MM-31094: Removes SQLITE db driver ref. * MM-31094: Layer generate fix. * MM-31094: Removes unnecessary changes. * MM-31094: Moves test method. * MM-31094: Re-add previous fix. * MM-31094: Removes make command for dev. * MM-31094: Fix for login. Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
Этот коммит содержится в:
коммит произвёл
GitHub
родитель
f31a9ed1a8
Коммит
58dce5930e
@@ -16,6 +16,7 @@ import (
|
||||
"github.com/mattermost/mattermost-server/v5/shared/i18n"
|
||||
"github.com/mattermost/mattermost-server/v5/shared/mlog"
|
||||
"github.com/mattermost/mattermost-server/v5/store"
|
||||
"github.com/mattermost/mattermost-server/v5/store/sqlstore"
|
||||
"github.com/mattermost/mattermost-server/v5/utils"
|
||||
)
|
||||
|
||||
@@ -1379,7 +1380,7 @@ func (a *App) addUserToChannel(user *model.User, channel *model.Channel) (*model
|
||||
}
|
||||
|
||||
func (a *App) AddUserToChannel(user *model.User, channel *model.Channel) (*model.ChannelMember, *model.AppError) {
|
||||
teamMember, nErr := a.Srv().Store.Team().GetMember(channel.TeamId, user.Id)
|
||||
teamMember, nErr := a.Srv().Store.Team().GetMember(sqlstore.WithMaster(context.Background()), channel.TeamId, user.Id)
|
||||
if nErr != nil {
|
||||
var nfErr *store.ErrNotFound
|
||||
switch {
|
||||
|
||||
@@ -6,6 +6,7 @@ package app
|
||||
import (
|
||||
"archive/zip"
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
@@ -406,7 +407,7 @@ func (a *App) MigrateFilenamesToFileInfos(post *model.Post) []*model.FileInfo {
|
||||
fileMigrationLock.Lock()
|
||||
defer fileMigrationLock.Unlock()
|
||||
|
||||
result, nErr := a.Srv().Store.Post().Get(post.Id, false, false, false)
|
||||
result, nErr := a.Srv().Store.Post().Get(context.Background(), post.Id, false, false, false)
|
||||
if nErr != nil {
|
||||
mlog.Error("Unable to get post when migrating post to use FileInfos", mlog.Err(nErr), mlog.String("post_id", post.Id))
|
||||
return []*model.FileInfo{}
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"testing"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v5/shared/mlog"
|
||||
@@ -11,11 +12,18 @@ import (
|
||||
)
|
||||
|
||||
var mainHelper *testlib.MainHelper
|
||||
var replicaFlag bool
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
if f := flag.Lookup("mysql-replica"); f == nil {
|
||||
flag.BoolVar(&replicaFlag, "mysql-replica", false, "")
|
||||
flag.Parse()
|
||||
}
|
||||
|
||||
var options = testlib.HelperOptions{
|
||||
EnableStore: true,
|
||||
EnableResources: true,
|
||||
WithReadReplica: replicaFlag,
|
||||
}
|
||||
|
||||
mlog.DisableZap()
|
||||
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
"github.com/mattermost/mattermost-server/v5/shared/i18n"
|
||||
"github.com/mattermost/mattermost-server/v5/shared/mlog"
|
||||
"github.com/mattermost/mattermost-server/v5/store"
|
||||
"github.com/mattermost/mattermost-server/v5/store/sqlstore"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -186,7 +187,7 @@ func (a *App) CreatePost(post *model.Post, channel *model.Channel, triggerWebhoo
|
||||
if post.RootId != "" {
|
||||
pchan = make(chan store.StoreResult, 1)
|
||||
go func() {
|
||||
r, pErr := a.Srv().Store.Post().Get(post.RootId, false, false, false)
|
||||
r, pErr := a.Srv().Store.Post().Get(sqlstore.WithMaster(context.Background()), post.RootId, false, false, false)
|
||||
pchan <- store.StoreResult{Data: r, NErr: pErr}
|
||||
close(pchan)
|
||||
}()
|
||||
@@ -537,7 +538,7 @@ func (a *App) DeleteEphemeralPost(userID, postID string) {
|
||||
func (a *App) UpdatePost(post *model.Post, safeUpdate bool) (*model.Post, *model.AppError) {
|
||||
post.SanitizeProps()
|
||||
|
||||
postLists, nErr := a.Srv().Store.Post().Get(post.Id, false, false, false)
|
||||
postLists, nErr := a.Srv().Store.Post().Get(context.Background(), post.Id, false, false, false)
|
||||
if nErr != nil {
|
||||
var nfErr *store.ErrNotFound
|
||||
var invErr *store.ErrInvalidInput
|
||||
@@ -742,7 +743,7 @@ func (a *App) GetSinglePost(postID string) (*model.Post, *model.AppError) {
|
||||
}
|
||||
|
||||
func (a *App) GetPostThread(postID string, skipFetchThreads, collapsedThreads, collapsedThreadsExtended bool) (*model.PostList, *model.AppError) {
|
||||
posts, err := a.Srv().Store.Post().Get(postID, skipFetchThreads, collapsedThreads, collapsedThreadsExtended)
|
||||
posts, err := a.Srv().Store.Post().Get(context.Background(), postID, skipFetchThreads, collapsedThreads, collapsedThreadsExtended)
|
||||
if err != nil {
|
||||
var nfErr *store.ErrNotFound
|
||||
var invErr *store.ErrInvalidInput
|
||||
@@ -787,7 +788,7 @@ func (a *App) GetFlaggedPostsForChannel(userID, channelID string, offset int, li
|
||||
}
|
||||
|
||||
func (a *App) GetPermalinkPost(postID string, userID string) (*model.PostList, *model.AppError) {
|
||||
list, nErr := a.Srv().Store.Post().Get(postID, false, false, false)
|
||||
list, nErr := a.Srv().Store.Post().Get(context.Background(), postID, false, false, false)
|
||||
if nErr != nil {
|
||||
var nfErr *store.ErrNotFound
|
||||
var invErr *store.ErrInvalidInput
|
||||
|
||||
@@ -1970,3 +1970,43 @@ func TestCollapsedThreadFetch(t *testing.T) {
|
||||
require.NotEmpty(t, l.Posts[postRoot.Id].Participants[0].Email)
|
||||
})
|
||||
}
|
||||
|
||||
func TestReplyToPostWithLag(t *testing.T) {
|
||||
if !replicaFlag {
|
||||
t.Skipf("requires test flag -mysql-replica")
|
||||
}
|
||||
|
||||
th := Setup(t).InitBasic()
|
||||
defer th.TearDown()
|
||||
|
||||
if *th.App.Srv().Config().SqlSettings.DriverName != model.DATABASE_DRIVER_MYSQL {
|
||||
t.Skipf("requires %q database driver", model.DATABASE_DRIVER_MYSQL)
|
||||
}
|
||||
|
||||
mainHelper.SQLStore.UpdateLicense(model.NewTestLicense("somelicense"))
|
||||
|
||||
t.Run("replication lag time great than reply time", func(t *testing.T) {
|
||||
err := mainHelper.SetReplicationLagForTesting(5)
|
||||
require.Nil(t, err)
|
||||
defer mainHelper.SetReplicationLagForTesting(0)
|
||||
mainHelper.ToggleReplicasOn()
|
||||
defer mainHelper.ToggleReplicasOff()
|
||||
|
||||
root, err := th.App.CreatePost(&model.Post{
|
||||
UserId: th.BasicUser.Id,
|
||||
ChannelId: th.BasicChannel.Id,
|
||||
Message: "root post",
|
||||
}, th.BasicChannel, false, true)
|
||||
require.Nil(t, err)
|
||||
|
||||
reply, err := th.App.CreatePost(&model.Post{
|
||||
UserId: th.BasicUser2.Id,
|
||||
ChannelId: th.BasicChannel.Id,
|
||||
RootId: root.Id,
|
||||
ParentId: root.Id,
|
||||
Message: fmt.Sprintf("@%s", th.BasicUser2.Username),
|
||||
}, th.BasicChannel, false, true)
|
||||
require.Nil(t, err)
|
||||
require.NotNil(t, reply)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -59,7 +59,7 @@ func TestReadReplicaDisabledBasedOnLicense(t *testing.T) {
|
||||
} else {
|
||||
dsn = os.Getenv("TEST_DATABASE_MYSQL_DSN")
|
||||
}
|
||||
cfg.SqlSettings = *storetest.MakeSqlSettings(driverName)
|
||||
cfg.SqlSettings = *storetest.MakeSqlSettings(driverName, false)
|
||||
if dsn != "" {
|
||||
cfg.SqlSettings.DataSource = &dsn
|
||||
}
|
||||
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
"github.com/mattermost/mattermost-server/v5/model"
|
||||
"github.com/mattermost/mattermost-server/v5/shared/mlog"
|
||||
"github.com/mattermost/mattermost-server/v5/store"
|
||||
"github.com/mattermost/mattermost-server/v5/store/sqlstore"
|
||||
)
|
||||
|
||||
func (a *App) CreateSession(session *model.Session) (*model.Session, *model.AppError) {
|
||||
@@ -84,7 +85,7 @@ func (a *App) GetSession(token string) (*model.Session, *model.AppError) {
|
||||
|
||||
if session.Id == "" {
|
||||
var nErr error
|
||||
if session, nErr = a.Srv().Store.Session().Get(token); nErr == nil {
|
||||
if session, nErr = a.Srv().Store.Session().Get(sqlstore.WithMaster(context.Background()), token); nErr == nil {
|
||||
if session != nil {
|
||||
if session.Token != token {
|
||||
return nil, model.NewAppError("GetSession", "api.context.invalid_token.error", map[string]interface{}{"Token": token, "Error": ""}, "session token is different from the one in DB", http.StatusUnauthorized)
|
||||
@@ -287,7 +288,7 @@ func (a *App) RevokeSessionsForDeviceId(userID string, deviceID string, currentS
|
||||
}
|
||||
|
||||
func (a *App) GetSessionById(sessionID string) (*model.Session, *model.AppError) {
|
||||
session, err := a.Srv().Store.Session().Get(sessionID)
|
||||
session, err := a.Srv().Store.Session().Get(context.Background(), sessionID)
|
||||
if err != nil {
|
||||
return nil, model.NewAppError("GetSessionById", "app.session.get.app_error", nil, err.Error(), http.StatusBadRequest)
|
||||
}
|
||||
@@ -296,7 +297,7 @@ func (a *App) GetSessionById(sessionID string) (*model.Session, *model.AppError)
|
||||
}
|
||||
|
||||
func (a *App) RevokeSessionById(sessionID string) *model.AppError {
|
||||
session, err := a.Srv().Store.Session().Get(sessionID)
|
||||
session, err := a.Srv().Store.Session().Get(context.Background(), sessionID)
|
||||
if err != nil {
|
||||
return model.NewAppError("RevokeSessionById", "app.session.get.app_error", nil, err.Error(), http.StatusBadRequest)
|
||||
}
|
||||
@@ -534,7 +535,7 @@ func (a *App) createSessionForUserAccessToken(tokenString string) (*model.Sessio
|
||||
|
||||
func (a *App) RevokeUserAccessToken(token *model.UserAccessToken) *model.AppError {
|
||||
var session *model.Session
|
||||
session, _ = a.Srv().Store.Session().Get(token.Token)
|
||||
session, _ = a.Srv().Store.Session().Get(context.Background(), token.Token)
|
||||
|
||||
if err := a.Srv().Store.UserAccessToken().Delete(token.Id); err != nil {
|
||||
return model.NewAppError("RevokeUserAccessToken", "app.user_access_token.delete.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
@@ -549,7 +550,7 @@ func (a *App) RevokeUserAccessToken(token *model.UserAccessToken) *model.AppErro
|
||||
|
||||
func (a *App) DisableUserAccessToken(token *model.UserAccessToken) *model.AppError {
|
||||
var session *model.Session
|
||||
session, _ = a.Srv().Store.Session().Get(token.Token)
|
||||
session, _ = a.Srv().Store.Session().Get(context.Background(), token.Token)
|
||||
|
||||
if err := a.Srv().Store.UserAccessToken().UpdateTokenDisable(token.Id); err != nil {
|
||||
return model.NewAppError("DisableUserAccessToken", "app.user_access_token.update_token_disable.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
@@ -564,7 +565,7 @@ func (a *App) DisableUserAccessToken(token *model.UserAccessToken) *model.AppErr
|
||||
|
||||
func (a *App) EnableUserAccessToken(token *model.UserAccessToken) *model.AppError {
|
||||
var session *model.Session
|
||||
session, _ = a.Srv().Store.Session().Get(token.Token)
|
||||
session, _ = a.Srv().Store.Session().Get(context.Background(), token.Token)
|
||||
|
||||
err := a.Srv().Store.UserAccessToken().UpdateTokenEnable(token.Id)
|
||||
if err != nil {
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"testing"
|
||||
@@ -347,7 +348,7 @@ func TestApp_ExtendExpiryIfNeeded(t *testing.T) {
|
||||
require.Equal(t, session.ExpiresAt, cachedSession.ExpiresAt)
|
||||
|
||||
// check database was updated.
|
||||
storedSession, nErr := th.App.Srv().Store.Session().Get(session.Token)
|
||||
storedSession, nErr := th.App.Srv().Store.Session().Get(context.Background(), session.Token)
|
||||
require.NoError(t, nErr)
|
||||
require.Equal(t, session.ExpiresAt, storedSession.ExpiresAt)
|
||||
})
|
||||
|
||||
@@ -23,6 +23,7 @@ import (
|
||||
"github.com/mattermost/mattermost-server/v5/shared/i18n"
|
||||
"github.com/mattermost/mattermost-server/v5/shared/mlog"
|
||||
"github.com/mattermost/mattermost-server/v5/store"
|
||||
"github.com/mattermost/mattermost-server/v5/store/sqlstore"
|
||||
)
|
||||
|
||||
func (a *App) CreateTeam(team *model.Team) (*model.Team, *model.AppError) {
|
||||
@@ -355,7 +356,7 @@ func (a *App) GetSchemeRolesForTeam(teamID string) (string, string, string, *mod
|
||||
}
|
||||
|
||||
func (a *App) UpdateTeamMemberRoles(teamID string, userID string, newRoles string) (*model.TeamMember, *model.AppError) {
|
||||
member, nErr := a.Srv().Store.Team().GetMember(teamID, userID)
|
||||
member, nErr := a.Srv().Store.Team().GetMember(context.Background(), teamID, userID)
|
||||
if nErr != nil {
|
||||
var nfErr *store.ErrNotFound
|
||||
switch {
|
||||
@@ -701,7 +702,7 @@ func (a *App) joinUserToTeam(team *model.Team, user *model.User) (*model.TeamMem
|
||||
tm.SchemeAdmin = true
|
||||
}
|
||||
|
||||
rtm, err := a.Srv().Store.Team().GetMember(team.Id, user.Id)
|
||||
rtm, err := a.Srv().Store.Team().GetMember(context.Background(), team.Id, user.Id)
|
||||
if err != nil {
|
||||
// Membership appears to be missing. Lets try to add.
|
||||
tmr, nErr := a.Srv().Store.Team().SaveMember(tm, *a.Config().TeamSettings.MaxUsersPerTeam)
|
||||
@@ -998,7 +999,7 @@ func (a *App) GetTeamsForUser(userID string) ([]*model.Team, *model.AppError) {
|
||||
}
|
||||
|
||||
func (a *App) GetTeamMember(teamID, userID string) (*model.TeamMember, *model.AppError) {
|
||||
teamMember, err := a.Srv().Store.Team().GetMember(teamID, userID)
|
||||
teamMember, err := a.Srv().Store.Team().GetMember(sqlstore.WithMaster(context.Background()), teamID, userID)
|
||||
if err != nil {
|
||||
var nfErr *store.ErrNotFound
|
||||
switch {
|
||||
|
||||
@@ -173,7 +173,7 @@ func TestHubSessionRevokeRace(t *testing.T) {
|
||||
mockSessionStore := mocks.SessionStore{}
|
||||
mockSessionStore.On("UpdateLastActivityAt", "id1", mock.Anything).Return(nil)
|
||||
mockSessionStore.On("Save", mock.AnythingOfType("*model.Session")).Return(sess1, nil)
|
||||
mockSessionStore.On("Get", "id1").Return(sess1, nil)
|
||||
mockSessionStore.On("Get", mock.Anything, "id1").Return(sess1, nil)
|
||||
mockSessionStore.On("Remove", "id1").Return(nil)
|
||||
|
||||
mockStatusStore := mocks.StatusStore{}
|
||||
|
||||
Ссылка в новой задаче
Block a user