MM-25115- migrate post, session and status cache to cache2 (#14667)

* 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 <jespinog@gmail.com> Co-authored-by: mattermod <mattermod@users.noreply.github.com>
2020-06-02 05:17:00 -04:00
--- 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() {
--- 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
 }
--- 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 {
--- 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.
--- 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
--- 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": {