MM-59932: Migrate remaining caches to Redis (#27880)

- We introduce 2 new APIs:
1. Scan: this allows incremental iteration
without blocking the Redis server and is the
recommended way to iterate over keys. With this,
we have entirely removed the need for Keys.
2. RemoveMulti: this allows deletion of multiple
keys in a single operation which optimizes
network round trips.

- While here, we make a small improvement to
GetStatusFromCache, where we remove the shallow
copy which wasn't necessary because we always
serialize the data from the cache.
- We do not use Redis for session cache because of
frequent requests to iterate the entire cache which leads
to a lot of `SCAN` calls.
- Avoid broadcasting status update messages for Redis case.
- Setting cache expiry for status cache
- Removing .Set method altogether to prevent
any chances of setting an item with no expiry.

https://mattermost.atlassian.net/browse/MM-59932

```release-note
NONE
```

server/channels/app/admin.go

@@ -229,7 +229,7 @@ func (a *App) GetLatestVersion(rctx request.CTX, latestVersionUrl string) (*mode
 		return nil, model.NewAppError("GetLatestVersion", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(validErr)
 	}
 
-	err = latestVersionCache.Set("latest_version_cache", releaseInfoResponse)
+	err = latestVersionCache.SetWithExpiry("latest_version_cache", releaseInfoResponse, 24*time.Hour)
 	if err != nil {
 		return nil, model.NewAppError("GetLatestVersion", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err)
 	}

server/channels/app/platform/cluster_handlers.go

@@ -60,7 +60,7 @@ func (ps *PlatformService) ClusterUpdateStatusHandler(msg *model.ClusterMessage)
 		ps.logger.Warn("Failed to decode status from JSON")
 	}
 
-	ps.statusCache.Set(status.UserId, status)
+	ps.statusCache.SetWithDefaultExpiry(status.UserId, status)
 }
 
 func (ps *PlatformService) ClusterInvalidateAllCachesHandler(msg *model.ClusterMessage) {
@@ -128,7 +128,7 @@ func (ps *PlatformService) InvalidateAllCachesSkipSend() {
 func (ps *PlatformService) InvalidateAllCaches() *model.AppError {
 	ps.InvalidateAllCachesSkipSend()
 
-	if ps.clusterIFace != nil {
+	if ps.clusterIFace != nil && *ps.Config().CacheSettings.CacheType == model.CacheTypeLRU {
 		msg := &model.ClusterMessage{
 			Event:            model.ClusterEventInvalidateAllCaches,
 			SendType:         model.ClusterSendReliable,

server/channels/app/platform/config_test.go

@@ -126,7 +126,7 @@ func TestIsFirstUserAccount(t *testing.T) {
 	}
 
 	// create a session, this should not affect IsFirstUserAccount
-	th.Service.sessionCache.Set("mock_session", 1)
+	th.Service.sessionCache.SetWithDefaultExpiry("mock_session", 1)
 
 	for _, te := range tests {
 		t.Run(te.name, func(t *testing.T) {

server/channels/app/platform/service.go

@@ -11,6 +11,7 @@ import (
 	"runtime"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/mattermost/mattermost/server/public/plugin"
@@ -293,16 +294,21 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
 	}
 
 	// Needed before loading license
-	ps.statusCache, err = cache.NewProvider().NewCache(&cache.CacheOptions{
+	ps.statusCache, err = ps.cacheProvider.NewCache(&cache.CacheOptions{
 		Name:           "Status",
 		Size:           model.StatusCacheSize,
 		Striped:        true,
 		StripedBuckets: maxInt(runtime.NumCPU()-1, 1),
+		DefaultExpiry:  30 * time.Minute,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("unable to create status cache: %w", err)
 	}
 
+	// Note: we hardcode the session cache to LRU because the session invalidation
+	// path always iterates through the entire cache, leading to a lot of SCAN calls
+	// in case of Redis. We could potentially have a reverse mapping of userIDs to
+	// session IDs, but leaving this one for now.
 	ps.sessionCache, err = cache.NewProvider().NewCache(&cache.CacheOptions{
 		Name:           "Session",
 		Size:           model.SessionCacheSize,

server/channels/app/platform/session.go

@@ -11,6 +11,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/platform/services/cache"
 )
 
 func (ps *PlatformService) ReturnSessionToPool(session *model.Session) {
@@ -50,18 +51,52 @@ func (ps *PlatformService) AddSessionToCache(session *model.Session) {
 }
 
 func (ps *PlatformService) ClearUserSessionCacheLocal(userID string) {
-	if keys, err := ps.sessionCache.Keys(); err == nil {
-		var session *model.Session
-		for _, key := range keys {
-			if err := ps.sessionCache.Get(key, &session); err == nil {
-				if session.UserId == userID {
-					ps.sessionCache.Remove(key)
-					if m := ps.metricsIFace; m != nil {
-						m.IncrementMemCacheInvalidationCounterSession()
-					}
+	var toDelete []string
+	// First, we iterate over the entire session cache.
+	err := ps.sessionCache.Scan(func(keys []string) error {
+		if len(keys) == 0 {
+			return nil
+		}
+
+		toPass := make([]any, 0, len(keys))
+		for i := 0; i < len(keys); i++ {
+			var session *model.Session
+			toPass = append(toPass, &session)
+		}
+
+		errs := ps.sessionCache.GetMulti(keys, toPass)
+		for i, err := range errs {
+			if err != nil {
+				if err != cache.ErrKeyNotFound {
+					return err
+				}
+				continue
+			}
+			gotSession := *(toPass[i].(**model.Session))
+			if gotSession == nil {
+				ps.logger.Warn("Found nil session in ClearUserSessionCacheLocal. This is not expected")
+				continue
+			}
+			// If we find the userID matches the passed userID,
+			// we mark it up for deletion.
+			if gotSession.UserId == userID {
+				toDelete = append(toDelete, keys[i])
+				if m := ps.metricsIFace; m != nil {
+					m.IncrementMemCacheInvalidationCounterSession()
 				}
 			}
 		}
+		return nil
+	})
+	if err != nil {
+		ps.logger.Warn("Error while scanning in ClearUserSessionCacheLocal", mlog.Err(err))
+		return
+	}
+	// Now, we delete everything.
+	err = ps.sessionCache.RemoveMulti(toDelete)
+	if err != nil {
+		ps.logger.Warn("Error while removing keys in ClearUserSessionCacheLocal", mlog.Err(err))
+		return
 	}
 }
 

server/channels/app/platform/session_test.go

@@ -37,22 +37,35 @@ func TestCache(t *testing.T) {
 	th.Service.sessionCache.SetWithExpiry(session.Token, session, 5*time.Minute)
 	th.Service.sessionCache.SetWithExpiry(session2.Token, session2, 5*time.Minute)
 
-	keys, err := th.Service.sessionCache.Keys()
+	var keys []string
+	err := th.Service.sessionCache.Scan(func(in []string) error {
+		keys = append(keys, in...)
+		return nil
+	})
 	require.NoError(t, err)
 	require.NotEmpty(t, keys)
 
 	th.Service.ClearUserSessionCache(session.UserId)
 
-	rkeys, err := th.Service.sessionCache.Keys()
+	var rkeys []string
+	err = th.Service.sessionCache.Scan(func(in []string) error {
+		rkeys = append(rkeys, in...)
+		return nil
+	})
 	require.NoError(t, err)
 	require.Lenf(t, rkeys, len(keys)-1, "should have one less: %d - %d != 1", len(keys), len(rkeys))
 	require.NotEmpty(t, rkeys)
+	clear(rkeys)
+	rkeys = []string{}
 
 	th.Service.ClearAllUsersSessionCache()
 
-	rkeys, err = th.Service.sessionCache.Keys()
+	err = th.Service.sessionCache.Scan(func(in []string) error {
+		rkeys = append(rkeys, in...)
+		return nil
+	})
 	require.NoError(t, err)
-	require.Empty(t, rkeys)
+	require.Len(t, rkeys, 0)
 }
 
 func TestSetSessionExpireInHours(t *testing.T) {

server/channels/app/platform/status.go

@@ -11,16 +11,17 @@ import (
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/mattermost/mattermost/server/public/shared/mlog"
 	"github.com/mattermost/mattermost/server/v8/channels/store"
+	"github.com/mattermost/mattermost/server/v8/platform/services/cache"
 )
 
 func (ps *PlatformService) AddStatusCacheSkipClusterSend(status *model.Status) {
-	ps.statusCache.Set(status.UserId, status)
+	ps.statusCache.SetWithDefaultExpiry(status.UserId, status)
 }
 
 func (ps *PlatformService) AddStatusCache(status *model.Status) {
 	ps.AddStatusCacheSkipClusterSend(status)
 
-	if ps.Cluster() != nil {
+	if ps.Cluster() != nil && *ps.Config().CacheSettings.CacheType == model.CacheTypeLRU {
 		statusJSON, err := json.Marshal(status)
 		if err != nil {
 			ps.logger.Warn("Failed to encode status to JSON", mlog.Err(err))
@@ -40,13 +41,36 @@ func (ps *PlatformService) GetAllStatuses() map[string]*model.Status {
 	}
 
 	statusMap := map[string]*model.Status{}
-	if userIDs, err := ps.statusCache.Keys(); err == nil {
-		for _, userID := range userIDs {
-			status := ps.GetStatusFromCache(userID)
-			if status != nil {
-				statusMap[userID] = status
-			}
+	err := ps.statusCache.Scan(func(keys []string) error {
+		if len(keys) == 0 {
+			return nil
 		}
+
+		toPass := make([]any, 0, len(keys))
+		for i := 0; i < len(keys); i++ {
+			var status *model.Status
+			toPass = append(toPass, &status)
+		}
+		errs := ps.statusCache.GetMulti(keys, toPass)
+		for i, err := range errs {
+			if err != nil {
+				if err != cache.ErrKeyNotFound {
+					return err
+				}
+				continue
+			}
+			gotStatus := *(toPass[i].(**model.Status))
+			if gotStatus != nil {
+				statusMap[keys[i]] = gotStatus
+				continue
+			}
+			ps.logger.Warn("Found nil status in GetAllStatuses. This is not expected")
+		}
+		return nil
+	})
+	if err != nil {
+		ps.logger.Warn("Error while getting all status in GetAllStatuses", mlog.Err(err))
+		return nil
 	}
 	return statusMap
 }
@@ -58,24 +82,40 @@ func (ps *PlatformService) GetStatusesByIds(userIDs []string) (map[string]any, *
 
 	statusMap := map[string]any{}
 	metrics := ps.Metrics()
-
 	missingUserIds := []string{}
-	for _, userID := range userIDs {
+
+	toPass := make([]any, 0, len(userIDs))
+	for i := 0; i < len(userIDs); i++ {
 		var status *model.Status
-		if err := ps.statusCache.Get(userID, &status); err == nil {
-			statusMap[userID] = status.Status
-			if metrics != nil {
-				metrics.IncrementMemCacheHitCounter(ps.statusCache.Name())
+		toPass = append(toPass, &status)
+	}
+	// First, we do a GetMulti to get all the status objects.
+	errs := ps.statusCache.GetMulti(userIDs, toPass)
+	for i, err := range errs {
+		if err != nil {
+			if err != cache.ErrKeyNotFound {
+				ps.logger.Warn("Error in GetStatusesByIds: ", mlog.Err(err))
 			}
-		} else {
-			missingUserIds = append(missingUserIds, userID)
+			missingUserIds = append(missingUserIds, userIDs[i])
 			if metrics != nil {
 				metrics.IncrementMemCacheMissCounter(ps.statusCache.Name())
 			}
+		} else {
+			// If we get a hit, we need to cast it back to the right type.
+			gotStatus := *(toPass[i].(**model.Status))
+			if gotStatus == nil {
+				ps.logger.Warn("Found nil in GetStatusesByIds. This is not expected")
+				continue
+			}
+			statusMap[userIDs[i]] = gotStatus.Status
+			if metrics != nil {
+				metrics.IncrementMemCacheHitCounter(ps.statusCache.Name())
+			}
 		}
 	}
 
 	if len(missingUserIds) > 0 {
+		// For cache misses, we fill them back from the DB.
 		statuses, err := ps.Store.Status().GetByIds(missingUserIds)
 		if err != nil {
 			return nil, model.NewAppError("GetStatusesByIds", "app.status.get.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
@@ -107,22 +147,38 @@ func (ps *PlatformService) GetUserStatusesByIds(userIDs []string) ([]*model.Stat
 	metrics := ps.Metrics()
 
 	missingUserIds := []string{}
-	for _, userID := range userIDs {
+	toPass := make([]any, 0, len(userIDs))
+	for i := 0; i < len(userIDs); i++ {
 		var status *model.Status
-		if err := ps.statusCache.Get(userID, &status); err == nil {
-			statusMap = append(statusMap, status)
-			if metrics != nil {
-				metrics.IncrementMemCacheHitCounter(ps.statusCache.Name())
+		toPass = append(toPass, &status)
+	}
+	// First, we do a GetMulti to get all the status objects.
+	errs := ps.statusCache.GetMulti(userIDs, toPass)
+	for i, err := range errs {
+		if err != nil {
+			if err != cache.ErrKeyNotFound {
+				ps.logger.Warn("Error in GetUserStatusesByIds: ", mlog.Err(err))
 			}
-		} else {
-			missingUserIds = append(missingUserIds, userID)
+			missingUserIds = append(missingUserIds, userIDs[i])
 			if metrics != nil {
 				metrics.IncrementMemCacheMissCounter(ps.statusCache.Name())
 			}
+		} else {
+			// If we get a hit, we need to cast it back to the right type.
+			gotStatus := *(toPass[i].(**model.Status))
+			if gotStatus == nil {
+				ps.logger.Warn("Found nil in GetUserStatusesByIds. This is not expected")
+				continue
+			}
+			statusMap = append(statusMap, gotStatus)
+			if metrics != nil {
+				metrics.IncrementMemCacheHitCounter(ps.statusCache.Name())
+			}
 		}
 	}
 
 	if len(missingUserIds) > 0 {
+		// For cache misses, we fill them back from the DB.
 		statuses, err := ps.Store.Status().GetByIds(missingUserIds)
 		if err != nil {
 			return nil, model.NewAppError("GetUserStatusesByIds", "app.status.get.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
@@ -179,9 +235,7 @@ func (ps *PlatformService) SaveAndBroadcastStatus(status *model.Status) {
 func (ps *PlatformService) GetStatusFromCache(userID string) *model.Status {
 	var status *model.Status
 	if err := ps.statusCache.Get(userID, &status); err == nil {
-		statusCopy := &model.Status{}
-		*statusCopy = *status
-		return statusCopy
+		return status
 	}
 
 	return nil

server/channels/store/localcachelayer/layer.go

@@ -340,7 +340,7 @@ func NewLocalCacheLayer(baseStore store.Store, metrics einterfaces.MetricsInterf
 	}); err != nil {
 		return
 	}
-	if localCacheStore.profilesInChannelCache, err = cache.NewProvider().NewCache(&cache.CacheOptions{
+	if localCacheStore.profilesInChannelCache, err = cacheProvider.NewCache(&cache.CacheOptions{
 		Size:                   ProfilesInChannelCacheSize,
 		Name:                   "ProfilesInChannel",
 		DefaultExpiry:          ProfilesInChannelCacheSec * time.Second,
@@ -450,7 +450,10 @@ func (s LocalCacheStore) DropAllTables() {
 }
 
 func (s *LocalCacheStore) doInvalidateCacheCluster(cache cache.Cache, key string, props map[string]string) {
-	cache.Remove(key)
+	err := cache.Remove(key)
+	if err != nil {
+		s.logger.Warn("Error while removing cache entry", mlog.Err(err), mlog.String("cache_name", cache.Name()))
+	}
 	if s.cluster != nil && s.cacheType == model.CacheTypeLRU {
 		msg := &model.ClusterMessage{
 			Event:    cache.GetInvalidateClusterEvent(),
@@ -464,20 +467,46 @@ func (s *LocalCacheStore) doInvalidateCacheCluster(cache cache.Cache, key string
 	}
 }
 
-func (s *LocalCacheStore) doStandardAddToCache(cache cache.Cache, key string, value any) {
-	cache.SetWithDefaultExpiry(key, value)
+func (s *LocalCacheStore) doMultiInvalidateCacheCluster(cache cache.Cache, keys []string, props map[string]string) {
+	err := cache.RemoveMulti(keys)
+	if err != nil {
+		s.logger.Warn("Error while removing cache entry", mlog.Err(err), mlog.String("cache_name", cache.Name()))
+	}
+	if s.cluster != nil && s.cacheType == model.CacheTypeLRU {
+		for _, key := range keys {
+			msg := &model.ClusterMessage{
+				Event:    cache.GetInvalidateClusterEvent(),
+				SendType: model.ClusterSendBestEffort,
+				Data:     []byte(key),
+			}
+			if props != nil {
+				msg.Props = props
+			}
+			s.cluster.SendClusterMessage(msg)
+		}
+	}
 }
 
-func (s *LocalCacheStore) doStandardReadCache(cache cache.Cache, key string, value any) error {
-	err := cache.Get(key, value)
+func (s *LocalCacheStore) doStandardAddToCache(cache cache.Cache, key string, value any) {
+	err := cache.SetWithDefaultExpiry(key, value)
+	if err != nil {
+		s.logger.Warn("Error while setting cache entry", mlog.Err(err), mlog.String("cache_name", cache.Name()))
+	}
+}
+
+func (s *LocalCacheStore) doStandardReadCache(c cache.Cache, key string, value any) error {
+	err := c.Get(key, value)
 	if err == nil {
 		if s.metrics != nil {
-			s.metrics.IncrementMemCacheHitCounter(cache.Name())
+			s.metrics.IncrementMemCacheHitCounter(c.Name())
 		}
 		return nil
 	}
+	if err != cache.ErrKeyNotFound {
+		s.logger.Warn("Error while reading from cache", mlog.Err(err), mlog.String("cache_name", c.Name()))
+	}
 	if s.metrics != nil {
-		s.metrics.IncrementMemCacheMissCounter(cache.Name())
+		s.metrics.IncrementMemCacheMissCounter(c.Name())
 	}
 	return err
 }

server/channels/store/localcachelayer/role_layer.go

@@ -78,9 +78,9 @@ func (s LocalCacheRoleStore) GetByNames(names []string) ([]*model.Role, error) {
 			gotRole := *(toPass[i].(**model.Role))
 			if gotRole != nil {
 				foundRoles = append(foundRoles, gotRole)
-			} else {
-				s.rootStore.logger.Warn("Found nil role in GetByNames. This is not expected")
+				continue
 			}
+			s.rootStore.logger.Warn("Found nil role in GetByNames. This is not expected")
 		}
 	}
 

server/channels/store/localcachelayer/user_layer.go

@@ -76,22 +76,42 @@ func (s *LocalCacheUserStore) InvalidateProfileCacheForUser(userId string) {
 }
 
 func (s *LocalCacheUserStore) InvalidateProfilesInChannelCacheByUser(userId string) {
-	// TODO: use scan here
-	keys, err := s.rootStore.profilesInChannelCache.Keys()
-	if err == nil {
-		for _, key := range keys {
-			// TODO: use MGET here on batches of keys
+	var toDelete []string
+	err := s.rootStore.profilesInChannelCache.Scan(func(keys []string) error {
+		if len(keys) == 0 {
+			return nil
+		}
+
+		toPass := make([]any, 0, len(keys))
+		for i := 0; i < len(keys); i++ {
+			// Note: keep https://github.com/mattermost/mattermost/pull/27830 in mind.
 			var userMap map[string]*model.User
-			if err = s.rootStore.profilesInChannelCache.Get(key, &userMap); err == nil {
-				if _, userInCache := userMap[userId]; userInCache {
-					s.rootStore.doInvalidateCacheCluster(s.rootStore.profilesInChannelCache, key, nil)
-					if s.rootStore.metrics != nil {
-						s.rootStore.metrics.IncrementMemCacheInvalidationCounter(s.rootStore.profilesInChannelCache.Name())
-					}
+			toPass = append(toPass, &userMap)
+		}
+		errs := s.rootStore.doMultiReadCache(s.rootStore.profilesInChannelCache, keys, toPass)
+		for i, err := range errs {
+			if err != nil {
+				if err != cache.ErrKeyNotFound {
+					return err
 				}
+				continue
+			}
+			gotMap := *(toPass[i].(*map[string]*model.User))
+			if gotMap == nil {
+				s.rootStore.logger.Warn("Found nil userMap in InvalidateProfilesInChannelCacheByUser. This is not expected")
+				continue
+			}
+			if _, ok := gotMap[userId]; ok {
+				toDelete = append(toDelete, keys[i])
 			}
 		}
+		return nil
+	})
+	if err != nil {
+		s.rootStore.logger.Warn("Error while scanning in InvalidateProfilesInChannelCacheByUser", mlog.Err(err))
+		return
 	}
+	s.rootStore.doMultiInvalidateCacheCluster(s.rootStore.profilesInChannelCache, toDelete, nil)
 }
 
 func (s *LocalCacheUserStore) InvalidateProfilesInChannelCache(channelID string) {