mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

[MM-55595] Use annotated logger in search layer (#25468)

Просмотр исходного кода
Этот коммит содержится в:
Ben Schumacher
2023-12-04 18:34:57 +01:00
коммит произвёл GitHub
родитель 5a4dba8809
Коммит b2ec1ff8ae
130 изменённых файлов: 2107 добавлений и 1930 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

14
server/channels/store/sqlstore/adapters.go
Просмотреть файл

@@ -6,11 +6,7 @@ package sqlstore
import (
"bytes"
"database/sql/driver"
"fmt"
"strconv"
"strings"
"github.com/mattermost/mattermost/server/public/shared/mlog"
)
type jsonArray []string
@@ -50,16 +46,6 @@ func (str jsonKeyPath) Value() (driver.Value, error) {
return "{" + string(str) + "}", nil
}
type TraceOnAdapter struct{}
func (t *TraceOnAdapter) Printf(format string, v ...any) {
originalString := fmt.Sprintf(format, v...)
newString := strings.ReplaceAll(originalString, "\n", " ")
newString = strings.ReplaceAll(newString, "\t", " ")
newString = strings.ReplaceAll(newString, "\"", "")
mlog.Debug(newString)
}
type JSONSerializable interface {
ToJSON() string
}

29
server/channels/store/sqlstore/channel_store.go
Просмотреть файл

@@ -18,6 +18,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/v8/channels/store"
"github.com/mattermost/mattermost/server/v8/einterfaces"
"github.com/mattermost/mattermost/server/v8/platform/services/cache"
@@ -613,7 +614,7 @@ func (s SqlChannelStore) Save(channel *model.Channel, maxChannelsPerTeam int64)
return newChannel, err
}
func (s SqlChannelStore) CreateDirectChannel(user *model.User, otherUser *model.User, channelOptions ...model.ChannelOption) (*model.Channel, error) {
func (s SqlChannelStore) CreateDirectChannel(rctx request.CTX, user *model.User, otherUser *model.User, channelOptions ...model.ChannelOption) (*model.Channel, error) {
channel := new(model.Channel)
for _, option := range channelOptions {
@@ -641,10 +642,10 @@ func (s SqlChannelStore) CreateDirectChannel(user *model.User, otherUser *model.
SchemeUser: !otherUser.IsGuest(),
}
return s.SaveDirectChannel(channel, cm1, cm2)
return s.SaveDirectChannel(rctx, channel, cm1, cm2)
}
func (s SqlChannelStore) SaveDirectChannel(directChannel *model.Channel, member1 *model.ChannelMember, member2 *model.ChannelMember) (_ *model.Channel, err error) {
func (s SqlChannelStore) SaveDirectChannel(rctx request.CTX, directChannel *model.Channel, member1 *model.ChannelMember, member2 *model.ChannelMember) (_ *model.Channel, err error) {
if directChannel.DeleteAt != 0 {
return nil, store.NewErrInvalidInput("Channel", "DeleteAt", directChannel.DeleteAt)
}
@@ -721,7 +722,7 @@ func (s SqlChannelStore) saveChannelT(transaction *sqlxTxWrapper, channel *model
}
// Update writes the updated channel to the database.
func (s SqlChannelStore) Update(channel *model.Channel) (_ *model.Channel, err error) {
func (s SqlChannelStore) Update(rctx request.CTX, channel *model.Channel) (_ *model.Channel, err error) {
transaction, err := s.GetMasterX().Beginx()
if err != nil {
return nil, errors.Wrap(err, "begin_transaction")
@@ -973,7 +974,7 @@ func (s SqlChannelStore) permanentDeleteByTeamtT(transaction *sqlxTxWrapper, tea
}
// PermanentDelete removes the given channel from the database.
func (s SqlChannelStore) PermanentDelete(channelId string) (err error) {
func (s SqlChannelStore) PermanentDelete(rctx request.CTX, channelId string) (err error) {
transaction, err := s.GetMasterX().Beginx()
if err != nil {
return errors.Wrap(err, "PermanentDelete: begin_transaction")
@@ -1009,7 +1010,7 @@ func (s SqlChannelStore) permanentDeleteT(transaction *sqlxTxWrapper, channelId
return nil
}
func (s SqlChannelStore) PermanentDeleteMembersByChannel(channelId string) error {
func (s SqlChannelStore) PermanentDeleteMembersByChannel(rctx request.CTX, channelId string) error {
_, err := s.GetMasterX().Exec("DELETE FROM ChannelMembers WHERE ChannelId = ?", channelId)
if err != nil {
return errors.Wrapf(err, "failed to delete Channel with channelId=%s", channelId)
@@ -1836,7 +1837,7 @@ func (s SqlChannelStore) UpdateMultipleMembers(members []*model.ChannelMember) (
return updatedMembers, nil
}
func (s SqlChannelStore) UpdateMember(member *model.ChannelMember) (*model.ChannelMember, error) {
func (s SqlChannelStore) UpdateMember(rctx request.CTX, member *model.ChannelMember) (*model.ChannelMember, error) {
updatedMembers, err := s.UpdateMultipleMembers([]*model.ChannelMember{member})
if err != nil {
return nil, err
@@ -2485,7 +2486,7 @@ func (s SqlChannelStore) GetGuestCount(channelId string, allowFromCache bool) (i
return count, nil
}
func (s SqlChannelStore) RemoveMembers(channelId string, userIds []string) error {
func (s SqlChannelStore) RemoveMembers(rctx request.CTX, channelId string, userIds []string) error {
builder := s.getQueryBuilder().
Delete("ChannelMembers").
Where(sq.Eq{"ChannelId": channelId}).
@@ -2516,11 +2517,11 @@ func (s SqlChannelStore) RemoveMembers(channelId string, userIds []string) error
return nil
}
func (s SqlChannelStore) RemoveMember(channelId string, userId string) error {
return s.RemoveMembers(channelId, []string{userId})
func (s SqlChannelStore) RemoveMember(rctx request.CTX, channelId string, userId string) error {
return s.RemoveMembers(rctx, channelId, []string{userId})
}
func (s SqlChannelStore) RemoveAllDeactivatedMembers(channelId string) error {
func (s SqlChannelStore) RemoveAllDeactivatedMembers(rctx request.CTX, channelId string) error {
query := `
DELETE
FROM
@@ -2545,7 +2546,7 @@ func (s SqlChannelStore) RemoveAllDeactivatedMembers(channelId string) error {
return nil
}
func (s SqlChannelStore) PermanentDeleteMembersByUser(userId string) error {
func (s SqlChannelStore) PermanentDeleteMembersByUser(rctx request.CTX, userId string) error {
if _, err := s.GetMasterX().Exec("DELETE FROM ChannelMembers WHERE UserId = ?", userId); err != nil {
return errors.Wrapf(err, "failed to permanent delete ChannelMembers with userId=%s", userId)
}
@@ -3035,7 +3036,7 @@ func (s SqlChannelStore) GetTeamMembersForChannel(channelID string) ([]string, e
return teamMemberIDs, nil
}
func (s SqlChannelStore) Autocomplete(userID, term string, includeDeleted, isGuest bool) (model.ChannelListWithTeamData, error) {
func (s SqlChannelStore) Autocomplete(rctx request.CTX, userID, term string, includeDeleted, isGuest bool) (model.ChannelListWithTeamData, error) {
query := s.getQueryBuilder().Select("c.*",
"t.DisplayName AS TeamDisplayName",
"t.Name AS TeamName",
@@ -3090,7 +3091,7 @@ func (s SqlChannelStore) Autocomplete(userID, term string, includeDeleted, isGue
return channels, nil
}
func (s SqlChannelStore) AutocompleteInTeam(teamID, userID, term string, includeDeleted, isGuest bool) (model.ChannelList, error) {
func (s SqlChannelStore) AutocompleteInTeam(rctx request.CTX, teamID, userID, term string, includeDeleted, isGuest bool) (model.ChannelList, error) {
query := s.getQueryBuilder().Select("*").
From("Channels c").
Where(sq.Eq{"c.TeamId": teamID}).

1
server/channels/store/sqlstore/command_webhook_store.go
Просмотреть файл

@@ -90,7 +90,6 @@ func (s SqlCommandWebhookStore) TryUse(id string, limit int) error {
}
func (s SqlCommandWebhookStore) Cleanup() {
mlog.Debug("Cleaning up command webhook store.")
exptime := model.GetMillis() - model.CommandWebhookLifetime
query := s.getQueryBuilder().

23
server/channels/store/sqlstore/file_info_store.go
Просмотреть файл

@@ -15,6 +15,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/v8/channels/store"
"github.com/mattermost/mattermost/server/v8/einterfaces"
)
@@ -110,7 +111,7 @@ func newSqlFileInfoStore(sqlStore *SqlStore, metrics einterfaces.MetricsInterfac
return s
}
func (fs SqlFileInfoStore) Save(info *model.FileInfo) (*model.FileInfo, error) {
func (fs SqlFileInfoStore) Save(rctx request.CTX, info *model.FileInfo) (*model.FileInfo, error) {
info.PreSave()
if err := info.IsValid(); err != nil {
return nil, err
@@ -159,7 +160,7 @@ func (fs SqlFileInfoStore) GetByIds(ids []string) ([]*model.FileInfo, error) {
return infos, nil
}
func (fs SqlFileInfoStore) Upsert(info *model.FileInfo) (*model.FileInfo, error) {
func (fs SqlFileInfoStore) Upsert(rctx request.CTX, info *model.FileInfo) (*model.FileInfo, error) {
info.PreSave()
if err := info.IsValid(); err != nil {
return nil, err
@@ -202,7 +203,7 @@ func (fs SqlFileInfoStore) Upsert(info *model.FileInfo) (*model.FileInfo, error)
return nil, errors.Wrap(err, "unable to retrieve rows affected")
}
if count == 0 {
return fs.Save(info)
return fs.Save(rctx, info)
}
return info, nil
}
@@ -388,7 +389,7 @@ func (fs SqlFileInfoStore) GetForUser(userId string) ([]*model.FileInfo, error)
return infos, nil
}
func (fs SqlFileInfoStore) AttachToPost(fileId, postId, channelId, creatorId string) error {
func (fs SqlFileInfoStore) AttachToPost(rctx request.CTX, fileId, postId, channelId, creatorId string) error {
query := fs.getQueryBuilder().
Update("FileInfo").
Set("PostId", postId).
@@ -422,7 +423,7 @@ func (fs SqlFileInfoStore) AttachToPost(fileId, postId, channelId, creatorId str
return nil
}
func (fs SqlFileInfoStore) SetContent(fileId, content string) error {
func (fs SqlFileInfoStore) SetContent(rctx request.CTX, fileId, content string) error {
query := fs.getQueryBuilder().
Update("FileInfo").
Set("Content", content).
@@ -441,7 +442,7 @@ func (fs SqlFileInfoStore) SetContent(fileId, content string) error {
return nil
}
func (fs SqlFileInfoStore) DeleteForPost(postId string) (string, error) {
func (fs SqlFileInfoStore) DeleteForPost(rctx request.CTX, postId string) (string, error) {
if _, err := fs.GetMasterX().Exec(
`UPDATE
FileInfo
@@ -454,14 +455,14 @@ func (fs SqlFileInfoStore) DeleteForPost(postId string) (string, error) {
return postId, nil
}
func (fs SqlFileInfoStore) PermanentDelete(fileId string) error {
func (fs SqlFileInfoStore) PermanentDelete(rctx request.CTX, fileId string) error {
if _, err := fs.GetMasterX().Exec(`DELETE FROM FileInfo WHERE Id = ?`, fileId); err != nil {
return errors.Wrapf(err, "failed to delete FileInfo with id=%s", fileId)
}
return nil
}
func (fs SqlFileInfoStore) PermanentDeleteBatch(endTime int64, limit int64) (int64, error) {
func (fs SqlFileInfoStore) PermanentDeleteBatch(rctx request.CTX, endTime int64, limit int64) (int64, error) {
var query string
if fs.DriverName() == "postgres" {
query = "DELETE from FileInfo WHERE Id = any (array (SELECT Id FROM FileInfo WHERE CreateAt < ? LIMIT ?))"
@@ -482,7 +483,7 @@ func (fs SqlFileInfoStore) PermanentDeleteBatch(endTime int64, limit int64) (int
return rowsAffected, nil
}
func (fs SqlFileInfoStore) PermanentDeleteByUser(userId string) (int64, error) {
func (fs SqlFileInfoStore) PermanentDeleteByUser(rctx request.CTX, userId string) (int64, error) {
query := "DELETE from FileInfo WHERE CreatorId = ?"
sqlResult, err := fs.GetMasterX().Exec(query, userId)
@@ -498,7 +499,7 @@ func (fs SqlFileInfoStore) PermanentDeleteByUser(userId string) (int64, error) {
return rowsAffected, nil
}
func (fs SqlFileInfoStore) Search(paramsList []*model.SearchParams, userId, teamId string, page, perPage int) (*model.FileInfoList, error) {
func (fs SqlFileInfoStore) Search(rctx request.CTX, paramsList []*model.SearchParams, userId, teamId string, page, perPage int) (*model.FileInfoList, error) {
// Since we don't support paging for DB search, we just return nothing for later pages
if page > 0 {
return model.NewFileInfoList(), nil
@@ -659,7 +660,7 @@ func (fs SqlFileInfoStore) Search(paramsList []*model.SearchParams, userId, team
items := []fileInfoWithChannelID{}
err = fs.GetSearchReplicaX().Select(&items, queryString, args...)
if err != nil {
mlog.Warn("Query error searching files.", mlog.String("error", trimInput(err.Error())))
rctx.Logger().Warn("Query error searching files.", mlog.String("error", trimInput(err.Error())))
// Don't return the error to the caller as it is of no use to the user. Instead return an empty set of search results.
} else {
for _, item := range items {

8
server/channels/store/sqlstore/init_test.go
Просмотреть файл

@@ -3,8 +3,12 @@
package sqlstore
func InitTest() {
initStores()
import (
"github.com/mattermost/mattermost/server/public/shared/mlog"
)
func InitTest(logger mlog.LoggerIFace) {
initStores(logger)
}
func TearDownTest() {

16
server/channels/store/sqlstore/integrity_test.go
Просмотреть файл

@@ -116,12 +116,12 @@ func createEmoji(ss store.Store, userId string) *model.Emoji {
return emoji
}
func createFileInfo(ss store.Store, postId, channelId, userId string) *model.FileInfo {
func createFileInfo(rctx request.CTX, ss store.Store, postId, channelId, userId string) *model.FileInfo {
m := model.FileInfo{}
m.PostId = postId
m.CreatorId = userId
m.Path = "some/path/to/file"
info, _ := ss.FileInfo().Save(&m)
info, _ := ss.FileInfo().Save(rctx, &m)
return info
}
@@ -493,7 +493,7 @@ func TestCheckChannelsChannelMembersIntegrity(t *testing.T) {
require.Equal(t, model.OrphanedRecord{
ParentId: &member.ChannelId,
}, data.Records[0])
ss.Channel().PermanentDeleteMembersByChannel(member.ChannelId)
ss.Channel().PermanentDeleteMembersByChannel(rctx, member.ChannelId)
})
})
}
@@ -625,7 +625,7 @@ func TestCheckPostsFileInfoIntegrity(t *testing.T) {
t.Run("should generate a report with one record", func(t *testing.T) {
postId := model.NewId()
info := createFileInfo(ss, postId, model.NewId(), model.NewId())
info := createFileInfo(rctx, ss, postId, model.NewId(), model.NewId())
result := checkPostsFileInfoIntegrity(store)
require.NoError(t, result.Err)
data := result.Data.(model.RelationalIntegrityCheckData)
@@ -821,7 +821,7 @@ func TestCheckTeamsChannelsIntegrity(t *testing.T) {
channel := createChannelWithTeamId(ss, model.NewId())
userA := createUser(ss)
userB := createUser(ss)
direct, err := ss.Channel().CreateDirectChannel(userA, userB)
direct, err := ss.Channel().CreateDirectChannel(rctx, userA, userB)
require.NoError(t, err)
require.NotNil(t, direct)
result := checkTeamsChannelsIntegrity(store)
@@ -842,7 +842,7 @@ func TestCheckTeamsChannelsIntegrity(t *testing.T) {
channel := createChannelWithTeamId(ss, model.NewId())
userA := createUser(ss)
userB := createUser(ss)
direct, err := ss.Channel().CreateDirectChannel(userA, userB)
direct, err := ss.Channel().CreateDirectChannel(rctx, userA, userB)
require.NoError(t, err)
require.NotNil(t, direct)
_, err = dbmap.Exec(`UPDATE Channels SET TeamId = 'test' WHERE Id = '` + direct.Id + `'`)
@@ -1120,7 +1120,7 @@ func TestCheckUsersChannelMembersIntegrity(t *testing.T) {
ParentId: &member.UserId,
}, data.Records[0])
dbmap.Exec(`DELETE FROM Channels WHERE Id=?`, channel.Id)
ss.Channel().PermanentDeleteMembersByUser(member.UserId)
ss.Channel().PermanentDeleteMembersByUser(rctx, member.UserId)
})
})
}
@@ -1228,7 +1228,7 @@ func TestCheckUsersFileInfoIntegrity(t *testing.T) {
t.Run("should generate a report with one record", func(t *testing.T) {
user := createUser(ss)
userId := user.Id
info := createFileInfo(ss, model.NewId(), model.NewId(), userId)
info := createFileInfo(rctx, ss, model.NewId(), model.NewId(), userId)
dbmap.Exec(`DELETE FROM Users WHERE Id=?`, user.Id)
result := checkUsersFileInfoIntegrity(store)
require.NoError(t, result.Err)

2
server/channels/store/sqlstore/main_test.go
Просмотреть файл

@@ -16,7 +16,7 @@ func TestMain(m *testing.M) {
mainHelper = testlib.NewMainHelperWithOptions(nil)
defer mainHelper.Close()
sqlstore.InitTest()
sqlstore.InitTest(mainHelper.Logger)
mainHelper.Main(m)
sqlstore.TearDownTest()

5
server/channels/store/sqlstore/migrate.go
Просмотреть файл

@@ -28,11 +28,12 @@ type Migrator struct {
store *SqlStore
}
func NewMigrator(settings model.SqlSettings, dryRun bool) (*Migrator, error) {
func NewMigrator(settings model.SqlSettings, logger mlog.LoggerIFace, dryRun bool) (*Migrator, error) {
ss := &SqlStore{
rrCounter: 0,
srCounter: 0,
settings: &settings,
logger: logger,
quitMonitor: make(chan struct{}),
wgMonitor: &sync.WaitGroup{},
}
@@ -127,7 +128,7 @@ func (ss *SqlStore) initMorph(dryRun bool) (*morph.Morph, error) {
if err != nil {
return nil, err
}
db, err2 := SetupConnection("master", dataSource, ss.settings, DBPingAttempts)
db, err2 := SetupConnection(ss.Logger(), "master", dataSource, ss.settings, DBPingAttempts)
if err2 != nil {
return nil, err2
}

5
server/channels/store/sqlstore/migrate_test.go
Просмотреть файл

@@ -7,11 +7,14 @@ import (
"testing"
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/public/shared/mlog"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestUpAndDownMigrations(t *testing.T) {
logger := mlog.CreateTestLogger(t)
testDrivers := []string{
model.DatabaseDriverPostgres,
model.DatabaseDriverMysql,
@@ -24,7 +27,7 @@ func TestUpAndDownMigrations(t *testing.T) {
t.Skip(err)
}
store, err := New(*settings, nil)
store, err := New(*settings, logger, nil)
require.NoError(t, err)
defer store.Close()

13
server/channels/store/sqlstore/post_store.go
Просмотреть файл

@@ -18,6 +18,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/v8/channels/store"
"github.com/mattermost/mattermost/server/v8/channels/store/searchlayer"
"github.com/mattermost/mattermost/server/v8/channels/utils"
@@ -329,7 +330,7 @@ func (s *SqlPostStore) populateReplyCount(posts []*model.Post) error {
return nil
}
func (s *SqlPostStore) Update(newPost *model.Post, oldPost *model.Post) (*model.Post, error) {
func (s *SqlPostStore) Update(rctx request.CTX, newPost *model.Post, oldPost *model.Post) (*model.Post, error) {
newPost.UpdateAt = model.GetMillis()
newPost.PreCommit()
@@ -451,7 +452,7 @@ func (s *SqlPostStore) OverwriteMultiple(posts []*model.Post) (_ []*model.Post,
return posts, -1, nil
}
func (s *SqlPostStore) Overwrite(post *model.Post) (*model.Post, error) {
func (s *SqlPostStore) Overwrite(rctx request.CTX, post *model.Post) (*model.Post, error) {
posts, _, err := s.OverwriteMultiple([]*model.Post{post})
if err != nil {
return nil, err
@@ -870,7 +871,7 @@ func (s *SqlPostStore) GetEtag(channelId string, allowFromCache, collapsedThread
// Soft deletes a post
// and cleans up the thread if it's a comment
func (s *SqlPostStore) Delete(postID string, time int64, deleteByID string) (err error) {
func (s *SqlPostStore) Delete(rctx request.CTX, postID string, time int64, deleteByID string) (err error) {
transaction, err := s.GetMasterX().Beginx()
if err != nil {
return errors.Wrap(err, "begin_transaction")
@@ -1010,7 +1011,7 @@ func (s *SqlPostStore) permanentDeleteAllCommentByUser(userId string) (err error
// cleans up threads (removes said user from participants and decreases reply count),
// permanent delete all root posts by user,
// and delete threads and thread memberships for those root posts
func (s *SqlPostStore) PermanentDeleteByUser(userId string) error {
func (s *SqlPostStore) PermanentDeleteByUser(rctx request.CTX, userId string) error {
// First attempt to delete all the comments for a user
if err := s.permanentDeleteAllCommentByUser(userId); err != nil {
return err
@@ -1048,7 +1049,7 @@ func (s *SqlPostStore) PermanentDeleteByUser(userId string) error {
// deletes all threads and thread memberships
// deletes all reactions
// no thread comment cleanup needed, since we are deleting threads and thread memberships
func (s *SqlPostStore) PermanentDeleteByChannel(channelId string) (err error) {
func (s *SqlPostStore) PermanentDeleteByChannel(rctx request.CTX, channelId string) (err error) {
transaction, err := s.GetMasterX().Beginx()
if err != nil {
return errors.Wrap(err, "begin_transaction")
@@ -2713,7 +2714,7 @@ func (s *SqlPostStore) GetDirectPostParentsForExportAfter(limit int, afterId str
}
//nolint:unparam
func (s *SqlPostStore) SearchPostsForUser(paramsList []*model.SearchParams, userId, teamId string, page, perPage int) (*model.PostSearchResults, error) {
func (s *SqlPostStore) SearchPostsForUser(rctx request.CTX, paramsList []*model.SearchParams, userId, teamId string, page, perPage int) (*model.PostSearchResults, error) {
// Since we don't support paging for DB search, we just return nothing for later pages
if page > 0 {
return model.MakePostSearchResults(model.NewPostList(), nil), nil

1
server/channels/store/sqlstore/preference_store.go
Просмотреть файл

@@ -22,7 +22,6 @@ func newSqlPreferenceStore(sqlStore *SqlStore) store.PreferenceStore {
}
func (s SqlPreferenceStore) deleteUnusedFeatures() {
mlog.Debug("Deleting any unused pre-release features")
sql, args, err := s.getQueryBuilder().
Delete("Preferences").
Where(sq.Eq{"Category": model.PreferenceCategoryAdvancedSettings}).

2
server/channels/store/sqlstore/sqlx_wrapper_test.go
Просмотреть файл

@@ -13,6 +13,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/public/shared/mlog"
)
func TestSqlX(t *testing.T) {
@@ -32,6 +33,7 @@ func TestSqlX(t *testing.T) {
rrCounter: 0,
srCounter: 0,
settings: settings,
logger: mlog.CreateConsoleTestLogger(t),
quitMonitor: make(chan struct{}),
wgMonitor: &sync.WaitGroup{},
}

34
server/channels/store/sqlstore/store.go
Просмотреть файл

@@ -131,6 +131,7 @@ type SqlStore struct {
context context.Context
license *model.License
licenseMutex sync.RWMutex
logger mlog.LoggerIFace
metrics einterfaces.MetricsInterface
isBinaryParam bool
@@ -140,12 +141,13 @@ type SqlStore struct {
wgMonitor *sync.WaitGroup
}
func New(settings model.SqlSettings, metrics einterfaces.MetricsInterface) (*SqlStore, error) {
func New(settings model.SqlSettings, logger mlog.LoggerIFace, metrics einterfaces.MetricsInterface) (*SqlStore, error) {
store := &SqlStore{
rrCounter: 0,
srCounter: 0,
settings: &settings,
metrics: metrics,
logger: logger,
quitMonitor: make(chan struct{}),
wgMonitor: &sync.WaitGroup{},
}
@@ -241,16 +243,22 @@ func New(settings model.SqlSettings, metrics einterfaces.MetricsInterface) (*Sql
// SetupConnection sets up the connection to the database and pings it to make sure it's alive.
// It also applies any database configuration settings that are required.
func SetupConnection(connType string, dataSource string, settings *model.SqlSettings, attempts int) (*dbsql.DB, error) {
func SetupConnection(logger mlog.LoggerIFace, connType string, dataSource string, settings *model.SqlSettings, attempts int) (*dbsql.DB, error) {
db, err := dbsql.Open(*settings.DriverName, dataSource)
if err != nil {
return nil, errors.Wrap(err, "failed to open SQL connection")
}
// At this point, we have passed sql.Open, so we deliberately ignore any errors.
sanitized, _ := SanitizeDataSource(*settings.DriverName, dataSource)
logger = logger.With(
mlog.String("database", connType),
mlog.String("dataSource", sanitized),
)
for i := 0; i < attempts; i++ {
// At this point, we have passed sql.Open, so we deliberately ignore any errors.
sanitized, _ := SanitizeDataSource(*settings.DriverName, dataSource)
mlog.Info("Pinging SQL", mlog.String("database", connType), mlog.String("dataSource", sanitized))
logger.Info("Pinging SQL")
ctx, cancel := context.WithTimeout(context.Background(), DBPingTimeoutSecs*time.Second)
defer cancel()
err = db.PingContext(ctx)
@@ -258,7 +266,7 @@ func SetupConnection(connType string, dataSource string, settings *model.SqlSett
if i == attempts-1 {
return nil, err
}
mlog.Error("Failed to ping DB", mlog.Err(err), mlog.Int("retrying in seconds", DBPingTimeoutSecs))
logger.Error("Failed to ping DB", mlog.Int("retrying in seconds", DBPingTimeoutSecs), mlog.Err(err))
time.Sleep(DBPingTimeoutSecs * time.Second)
continue
}
@@ -293,6 +301,10 @@ func (ss *SqlStore) Context() context.Context {
return ss.context
}
func (ss *SqlStore) Logger() mlog.LoggerIFace {
return ss.logger
}
func noOpMapper(s string) string { return s }
func (ss *SqlStore) initConnection() error {
@@ -308,7 +320,7 @@ func (ss *SqlStore) initConnection() error {
}
}
handle, err := SetupConnection("master", dataSource, ss.settings, DBPingAttempts)
handle, err := SetupConnection(ss.Logger(), "master", dataSource, ss.settings, DBPingAttempts)
if err != nil {
return err
}
@@ -326,7 +338,7 @@ func (ss *SqlStore) initConnection() error {
ss.ReplicaXs = make([]*atomic.Pointer[sqlxDBWrapper], len(ss.settings.DataSourceReplicas))
for i, replica := range ss.settings.DataSourceReplicas {
ss.ReplicaXs[i] = &atomic.Pointer[sqlxDBWrapper]{}
handle, err = SetupConnection(fmt.Sprintf("replica-%v", i), replica, ss.settings, DBPingAttempts)
handle, err = SetupConnection(ss.Logger(), fmt.Sprintf("replica-%v", i), replica, ss.settings, DBPingAttempts)
if err != nil {
// Initializing to be offline
ss.ReplicaXs[i].Store(&sqlxDBWrapper{isOnline: &atomic.Bool{}})
@@ -341,7 +353,7 @@ func (ss *SqlStore) initConnection() error {
ss.searchReplicaXs = make([]*atomic.Pointer[sqlxDBWrapper], len(ss.settings.DataSourceSearchReplicas))
for i, replica := range ss.settings.DataSourceSearchReplicas {
ss.searchReplicaXs[i] = &atomic.Pointer[sqlxDBWrapper]{}
handle, err = SetupConnection(fmt.Sprintf("search-replica-%v", i), replica, ss.settings, DBPingAttempts)
handle, err = SetupConnection(ss.Logger(), fmt.Sprintf("search-replica-%v", i), replica, ss.settings, DBPingAttempts)
if err != nil {
// Initializing to be offline
ss.searchReplicaXs[i].Store(&sqlxDBWrapper{isOnline: &atomic.Bool{}})
@@ -358,7 +370,7 @@ func (ss *SqlStore) initConnection() error {
if src.DataSource == nil {
continue
}
ss.replicaLagHandles[i], err = SetupConnection(fmt.Sprintf(replicaLagPrefix+"-%d", i), *src.DataSource, ss.settings, DBPingAttempts)
ss.replicaLagHandles[i], err = SetupConnection(ss.Logger(), fmt.Sprintf(replicaLagPrefix+"-%d", i), *src.DataSource, ss.settings, DBPingAttempts)
if err != nil {
mlog.Warn("Failed to setup replica lag handle. Skipping..", mlog.String("db", fmt.Sprintf(replicaLagPrefix+"-%d", i)), mlog.Err(err))
continue
@@ -513,7 +525,7 @@ func (ss *SqlStore) monitorReplicas() {
return
}
handle, err := SetupConnection(name, dsn, ss.settings, 1)
handle, err := SetupConnection(ss.Logger(), name, dsn, ss.settings, 1)
if err != nil {
mlog.Warn("Failed to setup connection. Skipping..", mlog.String("db", name), mlog.Err(err))
return

37
server/channels/store/sqlstore/store_test.go
Просмотреть файл

@@ -24,6 +24,7 @@ import (
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/public/plugin/plugintest/mock"
"github.com/mattermost/mattermost/server/public/shared/mlog"
"github.com/mattermost/mattermost/server/public/shared/request"
"github.com/mattermost/mattermost/server/v8/channels/db"
"github.com/mattermost/mattermost/server/v8/channels/store"
@@ -106,7 +107,7 @@ func StoreTestWithSqlStore(t *testing.T, f func(*testing.T, request.CTX, store.S
}
}
func initStores() {
func initStores(logger mlog.LoggerIFace) {
if testing.Short() {
return
}
@@ -138,7 +139,7 @@ func initStores() {
st := st
eg.Go(func() error {
var err error
st.SqlStore, err = New(*st.SqlSettings, nil)
st.SqlStore, err = New(*st.SqlSettings, logger, nil)
if err != nil {
return err
}
@@ -183,12 +184,14 @@ func tearDownStores() {
// before the fix in MM-28397.
// Keeping it here to help avoiding future regressions.
func TestStoreLicenseRace(t *testing.T) {
logger := mlog.CreateTestLogger(t)
settings, err := makeSqlSettings(model.DatabaseDriverPostgres)
if err != nil {
t.Skip(err)
}
store, err := New(*settings, nil)
store, err := New(*settings, logger, nil)
require.NoError(t, err)
defer func() {
store.Close()
@@ -218,6 +221,9 @@ func TestStoreLicenseRace(t *testing.T) {
func TestGetReplica(t *testing.T) {
t.Parallel()
logger := mlog.CreateTestLogger(t)
testCases := []struct {
Description string
DataSourceReplicaNum int
@@ -289,7 +295,7 @@ func TestGetReplica(t *testing.T) {
settings.DataSourceReplicas = dataSourceReplicas
settings.DataSourceSearchReplicas = dataSourceSearchReplicas
store, err := New(*settings, nil)
store, err := New(*settings, logger, nil)
require.NoError(t, err)
defer func() {
store.Close()
@@ -362,7 +368,7 @@ func TestGetReplica(t *testing.T) {
settings.DataSourceReplicas = dataSourceReplicas
settings.DataSourceSearchReplicas = dataSourceSearchReplicas
store, err := New(*settings, nil)
store, err := New(*settings, logger, nil)
require.NoError(t, err)
defer func() {
store.Close()
@@ -416,6 +422,8 @@ func TestGetReplica(t *testing.T) {
}
func TestGetDbVersion(t *testing.T) {
logger := mlog.CreateTestLogger(t)
testDrivers := []string{
model.DatabaseDriverPostgres,
model.DatabaseDriverMysql,
@@ -430,7 +438,7 @@ func TestGetDbVersion(t *testing.T) {
t.Skip(err)
}
store, err := New(*settings, nil)
store, err := New(*settings, logger, nil)
require.NoError(t, err)
version, err := store.GetDbVersion(false)
@@ -573,6 +581,9 @@ func TestIsBinaryParamEnabled(t *testing.T) {
func TestGetAllConns(t *testing.T) {
t.Parallel()
logger := mlog.CreateConsoleTestLogger(t)
testCases := []struct {
Description string
DataSourceReplicaNum int
@@ -654,7 +665,7 @@ func TestGetAllConns(t *testing.T) {
settings.DataSourceReplicas = dataSourceReplicas
settings.DataSourceSearchReplicas = dataSourceSearchReplicas
store, err := New(*settings, nil)
store, err := New(*settings, logger, nil)
require.NoError(t, err)
defer func() {
store.Close()
@@ -764,6 +775,7 @@ func TestReplicaLagQuery(t *testing.T) {
srCounter: 0,
settings: settings,
metrics: mockMetrics,
logger: mlog.CreateConsoleTestLogger(t),
quitMonitor: make(chan struct{}),
wgMonitor: &sync.WaitGroup{},
}
@@ -841,6 +853,7 @@ func TestMySQLReadTimeout(t *testing.T) {
store := &SqlStore{
settings: settings,
logger: mlog.CreateConsoleTestLogger(t),
quitMonitor: make(chan struct{}),
wgMonitor: &sync.WaitGroup{},
}
@@ -857,6 +870,7 @@ func TestGetDBSchemaVersion(t *testing.T) {
model.DatabaseDriverMysql,
}
logger := mlog.CreateConsoleTestLogger(t)
assets := db.Assets()
for _, d := range testDrivers {
@@ -867,7 +881,7 @@ func TestGetDBSchemaVersion(t *testing.T) {
if err != nil {
t.Skip(err)
}
store, err := New(*settings, nil)
store, err := New(*settings, logger, nil)
require.NoError(t, err)
assetsList, err := assets.ReadDir(filepath.Join("migrations", driver))
@@ -896,6 +910,8 @@ func TestGetLocalSchemaVersion(t *testing.T) {
model.DatabaseDriverMysql,
}
logger := mlog.CreateConsoleTestLogger(t)
for _, d := range testDrivers {
driver := d
t.Run(driver, func(t *testing.T) {
@@ -903,7 +919,7 @@ func TestGetLocalSchemaVersion(t *testing.T) {
if err != nil {
t.Skip(err)
}
store, err := New(*settings, nil)
store, err := New(*settings, logger, nil)
require.NoError(t, err)
ver, err := store.GetLocalSchemaVersion()
@@ -922,6 +938,7 @@ func TestGetAppliedMigrations(t *testing.T) {
model.DatabaseDriverMysql,
}
logger := mlog.CreateConsoleTestLogger(t)
assets := db.Assets()
for _, d := range testDrivers {
@@ -932,7 +949,7 @@ func TestGetAppliedMigrations(t *testing.T) {
if err != nil {
t.Skip(err)
}
store, err := New(*settings, nil)
store, err := New(*settings, logger, nil)
require.NoError(t, err)
assetsList, err := assets.ReadDir(filepath.Join("migrations", driver))

10
server/channels/store/sqlstore/team_store.go
Просмотреть файл

@@ -936,7 +936,7 @@ func (s SqlTeamStore) UpdateMultipleMembers(members []*model.TeamMember) ([]*mod
return updatedMembers, nil
}
func (s SqlTeamStore) UpdateMember(member *model.TeamMember) (*model.TeamMember, error) {
func (s SqlTeamStore) UpdateMember(rctx request.CTX, member *model.TeamMember) (*model.TeamMember, error) {
members, err := s.UpdateMultipleMembers([]*model.TeamMember{member})
if err != nil {
return nil, err
@@ -1184,7 +1184,7 @@ func (s SqlTeamStore) GetChannelUnreadsForTeam(teamId, userId string) ([]*model.
return channels, nil
}
func (s SqlTeamStore) RemoveMembers(teamId string, userIds []string) error {
func (s SqlTeamStore) RemoveMembers(rctx request.CTX, teamId string, userIds []string) error {
builder := s.getQueryBuilder().
Delete("TeamMembers").
Where(sq.Eq{"TeamId": teamId}).
@@ -1202,8 +1202,8 @@ func (s SqlTeamStore) RemoveMembers(teamId string, userIds []string) error {
}
// RemoveMember remove from the database the team members that match the userId and teamId passed as parameter.
func (s SqlTeamStore) RemoveMember(teamId string, userId string) error {
return s.RemoveMembers(teamId, []string{userId})
func (s SqlTeamStore) RemoveMember(rctx request.CTX, teamId string, userId string) error {
return s.RemoveMembers(rctx, teamId, []string{userId})
}
// RemoveAllMembersByTeam removes from the database the team members that belong to the teamId passed as parameter.
@@ -1223,7 +1223,7 @@ func (s SqlTeamStore) RemoveAllMembersByTeam(teamId string) error {
}
// RemoveAllMembersByUser removes from the database the team members that match the userId passed as parameter.
func (s SqlTeamStore) RemoveAllMembersByUser(userId string) error {
func (s SqlTeamStore) RemoveAllMembersByUser(rctx request.CTX, userId string) error {
query, args, err := s.getQueryBuilder().
Delete("TeamMembers").
Where(sq.Eq{"UserId": userId}).ToSql()

7
server/channels/store/sqlstore/user_store.go
Просмотреть файл

@@ -18,6 +18,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/v8/channels/store"
"github.com/mattermost/mattermost/server/v8/einterfaces"
)
@@ -162,7 +163,7 @@ func (us SqlUserStore) DeactivateGuests() ([]string, error) {
return userIds, nil
}
func (us SqlUserStore) Update(user *model.User, trustedUpdateData bool) (*model.UserUpdate, error) {
func (us SqlUserStore) Update(rctx request.CTX, user *model.User, trustedUpdateData bool) (*model.UserUpdate, error) {
user.PreUpdate()
if err := user.IsValid(); err != nil {
@@ -1511,7 +1512,7 @@ func (us SqlUserStore) GetAnyUnreadPostCountForChannel(userId string, channelId
return count, nil
}
func (us SqlUserStore) Search(teamId string, term string, options *model.UserSearchOptions) ([]*model.User, error) {
func (us SqlUserStore) Search(rctx request.CTX, teamId string, term string, options *model.UserSearchOptions) ([]*model.User, error) {
query := us.usersQuery.
OrderBy("Username ASC").
Limit(uint64(options.Limit))
@@ -2143,7 +2144,7 @@ func (us SqlUserStore) DemoteUserToGuest(userID string) (_ *model.User, err erro
return user, nil
}
func (us SqlUserStore) AutocompleteUsersInChannel(teamId, channelId, term string, options *model.UserSearchOptions) (*model.UserAutocompleteInChannel, error) {
func (us SqlUserStore) AutocompleteUsersInChannel(rctx request.CTX, teamId, channelId, term string, options *model.UserSearchOptions) (*model.UserAutocompleteInChannel, error) {
var usersInChannel, usersNotInChannel []*model.User
g := errgroup.Group{}
g.Go(func() (err error) {