From c183a5d380d43bae3b0a23c83784753b7341b5b3 Mon Sep 17 00:00:00 2001 From: Siyuan Liu Date: Tue, 2 Jun 2020 05:17:00 -0400 Subject: [PATCH] MM-25115- migrate post, session and status cache to cache2 (#14667) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * use cache2.Provider cache * use cache2.Provider cache * clean up imports * reset test runner * add close() call * Fixing i18n * address review comments Co-authored-by: Jesús Espino Co-authored-by: mattermod --- app/post.go | 18 ++++++++++-------- app/server.go | 32 ++++++++++++++++++++++++++------ app/session.go | 29 ++++++++++++++++------------- app/session_test.go | 20 ++++++++++++-------- app/status.go | 29 +++++++++++++++-------------- i18n/en.json | 4 ++++ 6 files changed, 83 insertions(+), 49 deletions(-) diff --git a/app/post.go b/app/post.go index 7e6f33962a..847470f966 100644 --- a/app/post.go +++ b/app/post.go @@ -14,6 +14,7 @@ import ( "github.com/mattermost/mattermost-server/v5/mlog" "github.com/mattermost/mattermost-server/v5/model" "github.com/mattermost/mattermost-server/v5/plugin" + "github.com/mattermost/mattermost-server/v5/services/cache2" "github.com/mattermost/mattermost-server/v5/store" "github.com/mattermost/mattermost-server/v5/utils" ) @@ -111,15 +112,16 @@ func (a *App) deduplicateCreatePost(post *model.Post) (foundPost *model.Post, er // Query the cache atomically for the given pending post id, saving a record if // it hasn't previously been seen. - value, loaded := a.Srv().seenPendingPostIdsCache.GetOrAdd(post.PendingPostId, unknownPostId, PENDING_POST_IDS_CACHE_TTL) - - // If we were the first thread to save this pending post id into the cache, - // proceed with create post normally. - if !loaded { + var postId string + nErr := a.Srv().seenPendingPostIdsCache.Get(post.PendingPostId, &postId) + if nErr == cache2.ErrKeyNotFound { + a.Srv().seenPendingPostIdsCache.SetWithExpiry(post.PendingPostId, unknownPostId, PENDING_POST_IDS_CACHE_TTL) return nil, nil } - postId := value.(string) + if nErr != nil { + return nil, model.NewAppError("errorGetPostId", "api.post.error_get_post_id.pending", nil, "", http.StatusInternalServerError) + } // If another thread saved the cache record, but hasn't yet updated it with the actual post // id (because it's still saving), notify the client with an error. Ideally, we'd wait @@ -161,7 +163,7 @@ func (a *App) CreatePost(post *model.Post, channel *model.Channel, triggerWebhoo return } - a.Srv().seenPendingPostIdsCache.AddWithExpiresInSecs(post.PendingPostId, savedPost.Id, int64(PENDING_POST_IDS_CACHE_TTL.Seconds())) + a.Srv().seenPendingPostIdsCache.SetWithExpiry(post.PendingPostId, savedPost.Id, PENDING_POST_IDS_CACHE_TTL) }() post.SanitizeProps() @@ -288,7 +290,7 @@ func (a *App) CreatePost(post *model.Post, channel *model.Channel, triggerWebhoo // Update the mapping from pending post id to the actual post id, for any clients that // might be duplicating requests. - a.Srv().seenPendingPostIdsCache.AddWithExpiresInSecs(post.PendingPostId, rpost.Id, int64(PENDING_POST_IDS_CACHE_TTL.Seconds())) + a.Srv().seenPendingPostIdsCache.SetWithExpiry(post.PendingPostId, rpost.Id, PENDING_POST_IDS_CACHE_TTL) if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil { a.Srv().Go(func() { diff --git a/app/server.go b/app/server.go index d692136a80..a3a8c56566 100644 --- a/app/server.go +++ b/app/server.go @@ -37,6 +37,7 @@ import ( "github.com/mattermost/mattermost-server/v5/plugin" "github.com/mattermost/mattermost-server/v5/services/cache" "github.com/mattermost/mattermost-server/v5/services/cache/lru" + "github.com/mattermost/mattermost-server/v5/services/cache2" "github.com/mattermost/mattermost-server/v5/services/filesstore" "github.com/mattermost/mattermost-server/v5/services/httpservice" "github.com/mattermost/mattermost-server/v5/services/imageproxy" @@ -105,9 +106,9 @@ type Server struct { newStore func() store.Store htmlTemplateWatcher *utils.HTMLTemplateWatcher - sessionCache cache.Cache - seenPendingPostIdsCache cache.Cache - statusCache cache.Cache + sessionCache cache2.Cache + seenPendingPostIdsCache cache2.Cache + statusCache cache2.Cache configListenerId string licenseListenerId string logListenerId string @@ -158,6 +159,8 @@ type Server struct { CacheProvider cache.Provider + CacheProvider2 cache2.Provider + tracer *tracing.Tracer timestampLastDiagnosticSent time.Time } @@ -256,9 +259,20 @@ func NewServer(options ...Option) (*Server, error) { s.CacheProvider.Connect() - s.sessionCache = s.CacheProvider.NewCache(model.SESSION_CACHE_SIZE) - s.seenPendingPostIdsCache = s.CacheProvider.NewCache(PENDING_POST_IDS_CACHE_SIZE) - s.statusCache = s.CacheProvider.NewCache(model.STATUS_CACHE_SIZE) + s.CacheProvider2 = cache2.NewProvider() + if err := s.CacheProvider2.Connect(); err != nil { + return nil, errors.Wrapf(err, "Unable to connect to cache provider") + } + + s.sessionCache = s.CacheProvider2.NewCache(&cache2.CacheOptions{ + Size: model.SESSION_CACHE_SIZE, + }) + s.seenPendingPostIdsCache = s.CacheProvider2.NewCache(&cache2.CacheOptions{ + Size: PENDING_POST_IDS_CACHE_SIZE, + }) + s.statusCache = s.CacheProvider2.NewCache(&cache2.CacheOptions{ + Size: model.STATUS_CACHE_SIZE, + }) if err := s.RunOldAppInitialization(); err != nil { return nil, err @@ -486,6 +500,12 @@ func (s *Server) Shutdown() error { s.CacheProvider.Close() } + if s.CacheProvider2 != nil { + if err = s.CacheProvider2.Close(); err != nil { + mlog.Error("Unable to cleanly shutdown cache", mlog.Err(err)) + } + } + mlog.Info("Server stopped") return nil } diff --git a/app/session.go b/app/session.go index 46dc19c43d..a9c8b028f3 100644 --- a/app/session.go +++ b/app/session.go @@ -6,6 +6,7 @@ package app import ( "math" "net/http" + "time" "github.com/mattermost/mattermost-server/v5/audit" "github.com/mattermost/mattermost-server/v5/mlog" @@ -30,8 +31,7 @@ func (a *App) GetSession(token string) (*model.Session, *model.AppError) { var session *model.Session var err *model.AppError - if ts, ok := a.Srv().sessionCache.Get(token); ok { - session = ts.(*model.Session) + if err := a.Srv().sessionCache.Get(token, &session); err == nil { if metrics != nil { metrics.IncrementMemCacheHitCounterSession() } @@ -178,15 +178,15 @@ func (a *App) ClearSessionCacheForAllUsers() { } func (a *App) ClearSessionCacheForUserSkipClusterSend(userId string) { - keys := a.Srv().sessionCache.Keys() - - for _, key := range keys { - if ts, ok := a.Srv().sessionCache.Get(key); ok { - session := ts.(*model.Session) - if session.UserId == userId { - a.Srv().sessionCache.Remove(key) - if a.Metrics() != nil { - a.Metrics().IncrementMemCacheInvalidationCounterSession() + if keys, err := a.Srv().sessionCache.Keys(); err == nil { + var session *model.Session + for _, key := range keys { + if err := a.Srv().sessionCache.Get(key, &session); err == nil { + if session.UserId == userId { + a.Srv().sessionCache.Remove(key) + if a.Metrics() != nil { + a.Metrics().IncrementMemCacheInvalidationCounterSession() + } } } } @@ -201,11 +201,14 @@ func (a *App) ClearSessionCacheForAllUsersSkipClusterSend() { } func (a *App) AddSessionToCache(session *model.Session) { - a.Srv().sessionCache.AddWithExpiresInSecs(session.Token, session, int64(*a.Config().ServiceSettings.SessionCacheInMinutes*60)) + a.Srv().sessionCache.SetWithExpiry(session.Token, session, time.Duration(int64(*a.Config().ServiceSettings.SessionCacheInMinutes))*time.Minute) } func (a *App) SessionCacheLength() int { - return a.Srv().sessionCache.Len() + if l, err := a.Srv().sessionCache.Len(); err == nil { + return l + } + return 0 } func (a *App) RevokeSessionsForDeviceId(userId string, deviceId string, currentSessionId string) *model.AppError { diff --git a/app/session_test.go b/app/session_test.go index 42724118f1..138e7f032f 100644 --- a/app/session_test.go +++ b/app/session_test.go @@ -6,6 +6,7 @@ package app import ( "fmt" "testing" + "time" "github.com/mattermost/mattermost-server/v5/model" "github.com/stretchr/testify/assert" @@ -28,21 +29,24 @@ func TestCache(t *testing.T) { UserId: model.NewId(), } - th.App.Srv().sessionCache.AddWithExpiresInSecs(session.Token, session, 5*60) - th.App.Srv().sessionCache.AddWithExpiresInSecs(session2.Token, session2, 5*60) + th.App.Srv().sessionCache.SetWithExpiry(session.Token, session, 5*time.Minute) + th.App.Srv().sessionCache.SetWithExpiry(session2.Token, session2, 5*time.Minute) - keys := th.App.Srv().sessionCache.Keys() + keys, err := th.App.Srv().sessionCache.Keys() + require.Nil(t, err) require.NotEmpty(t, keys) th.App.ClearSessionCacheForUser(session.UserId) - rkeys := th.App.Srv().sessionCache.Keys() + rkeys, err := th.App.Srv().sessionCache.Keys() + require.Nil(t, err) require.Lenf(t, rkeys, len(keys)-1, "should have one less: %d - %d != 1", len(keys), len(rkeys)) require.NotEmpty(t, rkeys) th.App.ClearSessionCacheForAllUsers() - rkeys = th.App.Srv().sessionCache.Keys() + rkeys, err = th.App.Srv().sessionCache.Keys() + require.Nil(t, err) require.Empty(t, rkeys) } @@ -291,9 +295,9 @@ func TestApp_ExtendExpiryIfNeeded(t *testing.T) { require.False(t, session.IsExpired()) // check cache was updated - ts, ok := th.App.Srv().sessionCache.Get(session.Token) - require.True(t, ok) - cachedSession := ts.(*model.Session) + var cachedSession *model.Session + errGet := th.App.Srv().sessionCache.Get(session.Token, &cachedSession) + require.Nil(t, errGet) require.Equal(t, session.ExpiresAt, cachedSession.ExpiresAt) // check database was updated. diff --git a/app/status.go b/app/status.go index c768523a0f..e01732c459 100644 --- a/app/status.go +++ b/app/status.go @@ -9,7 +9,7 @@ import ( ) func (a *App) AddStatusCacheSkipClusterSend(status *model.Status) { - a.Srv().statusCache.Add(status.UserId, status) + a.Srv().statusCache.Set(status.UserId, status) } func (a *App) AddStatusCache(status *model.Status) { @@ -30,16 +30,15 @@ func (a *App) GetAllStatuses() map[string]*model.Status { return map[string]*model.Status{} } - userIds := a.Srv().statusCache.Keys() statusMap := map[string]*model.Status{} - - for _, userId := range userIds { - status := a.GetStatusFromCache(userId) - if status != nil { - statusMap[userId] = status + if userIds, err := a.Srv().statusCache.Keys(); err == nil { + for _, userId := range userIds { + status := a.GetStatusFromCache(userId) + if status != nil { + statusMap[userId] = status + } } } - return statusMap } @@ -53,8 +52,9 @@ func (a *App) GetStatusesByIds(userIds []string) (map[string]interface{}, *model missingUserIds := []string{} for _, userId := range userIds { - if result, ok := a.Srv().statusCache.Get(userId); ok { - statusMap[userId] = result.(*model.Status).Status + var status *model.Status + if err := a.Srv().statusCache.Get(userId, &status); err == nil { + statusMap[userId] = status.Status if metrics != nil { metrics.IncrementMemCacheHitCounter("Status") } @@ -100,8 +100,9 @@ func (a *App) GetUserStatusesByIds(userIds []string) ([]*model.Status, *model.Ap missingUserIds := []string{} for _, userId := range userIds { - if result, ok := a.Srv().statusCache.Get(userId); ok { - statusMap = append(statusMap, result.(*model.Status)) + var status *model.Status + if err := a.Srv().statusCache.Get(userId, &status); err == nil { + statusMap = append(statusMap, status) if metrics != nil { metrics.IncrementMemCacheHitCounter("Status") } @@ -321,8 +322,8 @@ func (a *App) SetStatusOutOfOffice(userId string) { } func (a *App) GetStatusFromCache(userId string) *model.Status { - if result, ok := a.Srv().statusCache.Get(userId); ok { - status := result.(*model.Status) + var status *model.Status + if err := a.Srv().statusCache.Get(userId, &status); err == nil { statusCopy := &model.Status{} *statusCopy = *status return statusCopy diff --git a/i18n/en.json b/i18n/en.json index 4f24b9bb17..a36e9c1c4c 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -1692,6 +1692,10 @@ "id": "api.post.do_action.action_integration.app_error", "translation": "Action integration error." }, + { + "id": "api.post.error_get_post_id.pending", + "translation": "Unable to get the pending post." + }, { "id": "api.post.get_message_for_notification.files_sent", "translation": {