Migrate profileByIdsCache cache to cache layer (#13133)

* Partial advances

* Small advances

* invalidating cache with doInvalidateCacheCluster

* Renamed cache

* Starting with tests (failing by now)

* Passing tests :)

* Passing tests

* Fix ineffectual assignment in tests

* Revert of backward incompatible change

* Removed unused var

model/cluster_message.go

@@ -23,6 +23,7 @@ const (
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_USER_TEAMS                   = "inv_user_teams"
 	CLUSTER_EVENT_CLEAR_SESSION_CACHE_FOR_USER                      = "clear_session_user"
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_ROLES                        = "inv_roles"
+	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_PROFILE_BY_IDS               = "inv_profile_ids"
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_SCHEMES                      = "inv_schemes"
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_WEBHOOKS                     = "inv_webhooks"
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_EMOJIS_BY_ID                 = "inv_emojis_by_id"

store/localcachelayer/layer.go

@@ -32,6 +32,9 @@ const (
 	LAST_POSTS_CACHE_SIZE = 20000
 	LAST_POSTS_CACHE_SEC  = 30 * 60
 
+	USER_PROFILE_BY_ID_CACHE_SIZE = 20000
+	USER_PROFILE_BY_ID_SEC        = 30 * 60
+
 	CLEAR_CACHE_MESSAGE_DATA = ""
 )
 
@@ -54,6 +57,8 @@ type LocalCacheStore struct {
 	webhookCache             *utils.Cache
 	post                     LocalCachePostStore
 	postLastPostsCache       *utils.Cache
+	user                     LocalCacheUserStore
+	userProfileByIdsCache    *utils.Cache
 }
 
 func NewLocalCacheLayer(baseStore store.Store, metrics einterfaces.MetricsInterface, cluster einterfaces.ClusterInterface) LocalCacheStore {
@@ -77,6 +82,8 @@ func NewLocalCacheLayer(baseStore store.Store, metrics einterfaces.MetricsInterf
 	localCacheStore.channel = LocalCacheChannelStore{ChannelStore: baseStore.Channel(), rootStore: &localCacheStore}
 	localCacheStore.postLastPostsCache = utils.NewLruWithParams(LAST_POSTS_CACHE_SIZE, "LastPost", LAST_POSTS_CACHE_SEC, model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_LAST_POSTS)
 	localCacheStore.post = LocalCachePostStore{PostStore: baseStore.Post(), rootStore: &localCacheStore}
+	localCacheStore.userProfileByIdsCache = utils.NewLruWithParams(USER_PROFILE_BY_ID_CACHE_SIZE, "UserProfileByIds", USER_PROFILE_BY_ID_SEC, model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_PROFILE_BY_IDS)
+	localCacheStore.user = LocalCacheUserStore{UserStore: baseStore.User(), rootStore: &localCacheStore}
 
 	if cluster != nil {
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_REACTIONS, localCacheStore.reaction.handleClusterInvalidateReaction)
@@ -87,6 +94,7 @@ func NewLocalCacheLayer(baseStore store.Store, metrics einterfaces.MetricsInterf
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_EMOJIS_ID_BY_NAME, localCacheStore.emoji.handleClusterInvalidateEmojiIdByName)
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_CHANNEL_MEMBER_COUNTS, localCacheStore.channel.handleClusterInvalidateChannelMemberCounts)
 		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_LAST_POSTS, localCacheStore.post.handleClusterInvalidateLastPosts)
+		cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_PROFILE_BY_IDS, localCacheStore.user.handleClusterInvalidateScheme)
 	}
 	return localCacheStore
 }
@@ -119,6 +127,10 @@ func (s LocalCacheStore) Post() store.PostStore {
 	return s.post
 }
 
+func (s LocalCacheStore) User() store.UserStore {
+	return s.user
+}
+
 func (s LocalCacheStore) DropAllTables() {
 	s.Invalidate()
 	s.Store.DropAllTables()
@@ -174,4 +186,5 @@ func (s *LocalCacheStore) Invalidate() {
 	s.doClearCacheCluster(s.emojiIdCacheByName)
 	s.doClearCacheCluster(s.channelMemberCountsCache)
 	s.doClearCacheCluster(s.postLastPostsCache)
+	s.doClearCacheCluster(s.userProfileByIdsCache)
 }

store/localcachelayer/main_test.go

@@ -6,6 +6,8 @@ package localcachelayer
 import (
 	"testing"
 
+	"github.com/mattermost/mattermost-server/store"
+
 	"github.com/mattermost/mattermost-server/model"
 	"github.com/mattermost/mattermost-server/store/storetest/mocks"
 	"github.com/mattermost/mattermost-server/testlib"
@@ -71,6 +73,12 @@ func getMockStore() *mocks.Store {
 	mockPostStore.On("InvalidateLastPostTimeCache", "12360")
 	mockStore.On("Post").Return(&mockPostStore)
 
+	fakeUser := []*model.User{{Id: "123"}}
+	mockUserStore := mocks.UserStore{}
+	mockUserStore.On("GetProfileByIds", []string{"123"}, &store.UserGetByIdsOpts{}, true).Return(fakeUser, nil)
+	mockUserStore.On("GetProfileByIds", []string{"123"}, &store.UserGetByIdsOpts{}, false).Return(fakeUser, nil)
+	mockStore.On("User").Return(&mockUserStore)
+
 	return &mockStore
 }
 

store/localcachelayer/user_layer.go

@@ -0,0 +1,87 @@
+// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
+// See License.txt for license information.
+
+package localcachelayer
+
+import (
+	"net/http"
+
+	"github.com/mattermost/mattermost-server/model"
+	"github.com/mattermost/mattermost-server/store"
+)
+
+type LocalCacheUserStore struct {
+	store.UserStore
+	rootStore *LocalCacheStore
+}
+
+func (s *LocalCacheUserStore) handleClusterInvalidateScheme(msg *model.ClusterMessage) {
+	if msg.Data == CLEAR_CACHE_MESSAGE_DATA {
+		s.rootStore.userProfileByIdsCache.Purge()
+	} else {
+		s.rootStore.userProfileByIdsCache.Remove(msg.Data)
+	}
+}
+
+func (s LocalCacheUserStore) ClearCaches() {
+	s.rootStore.userProfileByIdsCache.Purge()
+
+	if s.rootStore.metrics != nil {
+		s.rootStore.metrics.IncrementMemCacheInvalidationCounter("Profile By Ids - Purge")
+	}
+}
+
+func (s LocalCacheUserStore) InvalidatProfileCacheForUser(userId string) {
+	s.rootStore.doInvalidateCacheCluster(s.rootStore.userProfileByIdsCache, userId)
+
+	if s.rootStore.metrics != nil {
+		s.rootStore.metrics.IncrementMemCacheInvalidationCounter("Profile By Ids - Remove")
+	}
+}
+
+func (s LocalCacheUserStore) GetProfileByIds(userIds []string, options *store.UserGetByIdsOpts, allowFromCache bool) ([]*model.User, *model.AppError) {
+	if !allowFromCache {
+		return s.UserStore.GetProfileByIds(userIds, options, false)
+	}
+
+	if options == nil {
+		options = &store.UserGetByIdsOpts{}
+	}
+
+	users := []*model.User{}
+	remainingUserIds := make([]string, 0)
+
+	for _, userId := range userIds {
+		if cacheItem := s.rootStore.doStandardReadCache(s.rootStore.userProfileByIdsCache, userId); cacheItem != nil {
+			u := &model.User{}
+			*u = *cacheItem.(*model.User)
+
+			if options.Since == 0 || u.UpdateAt > options.Since {
+				users = append(users, u)
+			}
+		} else {
+			remainingUserIds = append(remainingUserIds, userId)
+		}
+	}
+
+	if s.rootStore.metrics != nil {
+		s.rootStore.metrics.AddMemCacheHitCounter("Profile By Ids", float64(len(users)))
+		s.rootStore.metrics.AddMemCacheMissCounter("Profile By Ids", float64(len(remainingUserIds)))
+	}
+
+	if len(remainingUserIds) > 0 {
+		remainingUsers, err := s.UserStore.GetProfileByIds(remainingUserIds, options, false)
+		if err != nil {
+			return nil, model.NewAppError("SqlUserStore.GetProfileByIds", "store.sql_user.get_profiles.app_error", nil, err.Error(), http.StatusInternalServerError)
+		}
+
+		users = append(users, remainingUsers...)
+
+		for _, user := range remainingUsers {
+			s.rootStore.doStandardAddToCache(s.rootStore.userProfileByIdsCache, user.Id, user)
+		}
+
+	}
+
+	return users, nil
+}

store/localcachelayer/user_layer_test.go

@@ -0,0 +1,65 @@
+// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
+// See License.txt for license information.
+
+package localcachelayer
+
+import (
+	"testing"
+
+	"github.com/mattermost/mattermost-server/model"
+	"github.com/mattermost/mattermost-server/store"
+	"github.com/mattermost/mattermost-server/store/storetest"
+	"github.com/mattermost/mattermost-server/store/storetest/mocks"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUserStore(t *testing.T) {
+	StoreTestWithSqlSupplier(t, storetest.TestUserStore)
+}
+
+func TestUserStoreCache(t *testing.T) {
+	fakeUserIds := []string{"123"}
+	fakeUser := []*model.User{{Id: "123"}}
+
+	t.Run("first call not cached, second cached and returning same data", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		gotUser, err := cachedStore.User().GetProfileByIds(fakeUserIds, &store.UserGetByIdsOpts{}, true)
+		require.Nil(t, err)
+		assert.Equal(t, fakeUser, gotUser)
+		mockStore.User().(*mocks.UserStore).AssertNumberOfCalls(t, "GetProfileByIds", 1)
+
+		_, _ = cachedStore.User().GetProfileByIds(fakeUserIds, &store.UserGetByIdsOpts{}, true)
+		mockStore.User().(*mocks.UserStore).AssertNumberOfCalls(t, "GetProfileByIds", 1)
+	})
+
+	t.Run("first call not cached, second force no cached", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		gotUser, err := cachedStore.User().GetProfileByIds(fakeUserIds, &store.UserGetByIdsOpts{}, true)
+		require.Nil(t, err)
+		assert.Equal(t, fakeUser, gotUser)
+		mockStore.User().(*mocks.UserStore).AssertNumberOfCalls(t, "GetProfileByIds", 1)
+
+		_, _ = cachedStore.User().GetProfileByIds(fakeUserIds, &store.UserGetByIdsOpts{}, false)
+		mockStore.User().(*mocks.UserStore).AssertNumberOfCalls(t, "GetProfileByIds", 2)
+	})
+
+	t.Run("first call not cached, invalidate, and then not cached again", func(t *testing.T) {
+		mockStore := getMockStore()
+		cachedStore := NewLocalCacheLayer(mockStore, nil, nil)
+
+		gotUser, err := cachedStore.User().GetProfileByIds(fakeUserIds, &store.UserGetByIdsOpts{}, true)
+		require.Nil(t, err)
+		assert.Equal(t, fakeUser, gotUser)
+		mockStore.User().(*mocks.UserStore).AssertNumberOfCalls(t, "GetProfileByIds", 1)
+
+		cachedStore.User().InvalidatProfileCacheForUser("123")
+
+		_, _ = cachedStore.User().GetProfileByIds(fakeUserIds, &store.UserGetByIdsOpts{}, true)
+		mockStore.User().(*mocks.UserStore).AssertNumberOfCalls(t, "GetProfileByIds", 2)
+	})
+}

store/sqlstore/user_store.go

@@ -23,8 +23,6 @@ import (
 const (
 	PROFILES_IN_CHANNEL_CACHE_SIZE  = model.CHANNEL_CACHE_SIZE
 	PROFILES_IN_CHANNEL_CACHE_SEC   = 900 // 15 mins
-	PROFILE_BY_IDS_CACHE_SIZE       = model.SESSION_CACHE_SIZE
-	PROFILE_BY_IDS_CACHE_SEC        = 900 // 15 mins
 	MAX_GROUP_CHANNELS_FOR_PROFILES = 50
 )
 
@@ -44,25 +42,16 @@ type SqlUserStore struct {
 }
 
 var profilesInChannelCache *utils.Cache = utils.NewLru(PROFILES_IN_CHANNEL_CACHE_SIZE)
-var profileByIdsCache *utils.Cache = utils.NewLru(PROFILE_BY_IDS_CACHE_SIZE)
 
 func (us SqlUserStore) ClearCaches() {
 	profilesInChannelCache.Purge()
-	profileByIdsCache.Purge()
 
 	if us.metrics != nil {
 		us.metrics.IncrementMemCacheInvalidationCounter("Profiles in Channel - Purge")
-		us.metrics.IncrementMemCacheInvalidationCounter("Profile By Ids - Purge")
 	}
 }
 
-func (us SqlUserStore) InvalidatProfileCacheForUser(userId string) {
+func (us SqlUserStore) InvalidatProfileCacheForUser(userId string) {}
-	profileByIdsCache.Remove(userId)
-
-	if us.metrics != nil {
-		us.metrics.IncrementMemCacheInvalidationCounter("Profile By Ids - Remove")
-	}
-}
 
 func NewSqlUserStore(sqlStore SqlStore, metrics einterfaces.MetricsInterface) store.UserStore {
 	us := &SqlUserStore{
@@ -829,46 +818,15 @@ func (us SqlUserStore) GetNewUsersForTeam(teamId string, offset, limit int, view
 	return users, nil
 }
 
-func (us SqlUserStore) GetProfileByIds(userIds []string, options *store.UserGetByIdsOpts, allowFromCache bool) ([]*model.User, *model.AppError) {
+func (us SqlUserStore) GetProfileByIds(userIds []string, options *store.UserGetByIdsOpts, _ bool) ([]*model.User, *model.AppError) {
 	if options == nil {
 		options = &store.UserGetByIdsOpts{}
 	}
 
 	users := []*model.User{}
-	remainingUserIds := make([]string, 0)
-
-	if allowFromCache {
-		for _, userId := range userIds {
-			if cacheItem, ok := profileByIdsCache.Get(userId); ok {
-				u := &model.User{}
-				*u = *cacheItem.(*model.User)
-
-				if options.Since == 0 || u.UpdateAt > options.Since {
-					users = append(users, u)
-				}
-			} else {
-				remainingUserIds = append(remainingUserIds, userId)
-			}
-		}
-		if us.metrics != nil {
-			us.metrics.AddMemCacheHitCounter("Profile By Ids", float64(len(users)))
-			us.metrics.AddMemCacheMissCounter("Profile By Ids", float64(len(remainingUserIds)))
-		}
-	} else {
-		remainingUserIds = userIds
-		if us.metrics != nil {
-			us.metrics.AddMemCacheMissCounter("Profile By Ids", float64(len(remainingUserIds)))
-		}
-	}
-
-	// If everything came from the cache then just return
-	if len(remainingUserIds) == 0 {
-		return users, nil
-	}
-
 	query := us.usersQuery.
 		Where(map[string]interface{}{
-			"u.Id": remainingUserIds,
+			"u.Id": userIds,
 		}).
 		OrderBy("u.Username ASC")
 
@@ -891,10 +849,6 @@ func (us SqlUserStore) GetProfileByIds(userIds []string, options *store.UserGetB
 
 	for _, u := range users {
 		u.Sanitize(map[string]bool{})
-
-		cpy := &model.User{}
-		*cpy = *u
-		profileByIdsCache.AddWithExpiresInSecs(cpy.Id, cpy, PROFILE_BY_IDS_CACHE_SEC)
 	}
 
 	return users, nil