diff --git a/server/channels/app/admin.go b/server/channels/app/admin.go index 634c8d0b0a..f768b9097e 100644 --- a/server/channels/app/admin.go +++ b/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) } diff --git a/server/channels/app/platform/cluster_handlers.go b/server/channels/app/platform/cluster_handlers.go index 780fe6546c..af90c2783a 100644 --- a/server/channels/app/platform/cluster_handlers.go +++ b/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, diff --git a/server/channels/app/platform/config_test.go b/server/channels/app/platform/config_test.go index bdc33aea47..83cde4ee0d 100644 --- a/server/channels/app/platform/config_test.go +++ b/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) { diff --git a/server/channels/app/platform/service.go b/server/channels/app/platform/service.go index b393824760..eca176cad7 100644 --- a/server/channels/app/platform/service.go +++ b/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, diff --git a/server/channels/app/platform/session.go b/server/channels/app/platform/session.go index d019f93606..0c4d46b244 100644 --- a/server/channels/app/platform/session.go +++ b/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 } } diff --git a/server/channels/app/platform/session_test.go b/server/channels/app/platform/session_test.go index 99e4fd488e..aa85c87d94 100644 --- a/server/channels/app/platform/session_test.go +++ b/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) { diff --git a/server/channels/app/platform/status.go b/server/channels/app/platform/status.go index e3a7d34848..6014ba62e7 100644 --- a/server/channels/app/platform/status.go +++ b/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 diff --git a/server/channels/store/localcachelayer/layer.go b/server/channels/store/localcachelayer/layer.go index a5b79d2e72..ed98561455 100644 --- a/server/channels/store/localcachelayer/layer.go +++ b/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 } diff --git a/server/channels/store/localcachelayer/role_layer.go b/server/channels/store/localcachelayer/role_layer.go index 1583654484..fdd65cd88d 100644 --- a/server/channels/store/localcachelayer/role_layer.go +++ b/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") } } diff --git a/server/channels/store/localcachelayer/user_layer.go b/server/channels/store/localcachelayer/user_layer.go index 84c0d801ab..5dd5808414 100644 --- a/server/channels/store/localcachelayer/user_layer.go +++ b/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) { diff --git a/server/platform/services/cache/cache.go b/server/platform/services/cache/cache.go index 7baaf4eb72..63b94a8a8b 100644 --- a/server/platform/services/cache/cache.go +++ b/server/platform/services/cache/cache.go @@ -18,10 +18,6 @@ type Cache interface { // Purge is used to completely clear the cache. Purge() error - // Set adds the given key and value to the store without an expiry. If the key already exists, - // it will overwrite the previous value. - Set(key string, value any) error - // SetWithDefaultExpiry adds the given key and value to the store with the default expiry. If // the key already exists, it will overwrite the previous value SetWithDefaultExpiry(key string, value any) error @@ -31,16 +27,24 @@ type Cache interface { SetWithExpiry(key string, value any, ttl time.Duration) error // Get the content stored in the cache for the given key, and decode it into the value interface. - // Return ErrKeyNotFound if the key is missing from the cache + // Returns ErrKeyNotFound if the key is missing from the cache Get(key string, value any) error + // GetMulti returns values for multiple keys in a single operation. + // Returns ErrKeyNotFound if the key is missing from the cache. GetMulti(keys []string, values []any) []error // Remove deletes the value for a given key. Remove(key string) error - // Keys returns a slice of the keys in the cache. - Keys() ([]string, error) + // RemoveMulti deletes multiple keys in a single operation. + RemoveMulti(keys []string) error + + // Scan allows incremental iteration over the entire key-space + // in a performant manner. It provides a callback that consumers + // can use to process the keys. If the callback returns an error, + // the scan stops, returning the same error. + Scan(f func([]string) error) error // GetInvalidateClusterEvent returns the cluster event configured when this cache was created. GetInvalidateClusterEvent() model.ClusterEvent diff --git a/server/platform/services/cache/lru.go b/server/platform/services/cache/lru.go index 9d8bd5edd1..ee7c07bfda 100644 --- a/server/platform/services/cache/lru.go +++ b/server/platform/services/cache/lru.go @@ -57,12 +57,6 @@ func (l *LRU) Purge() error { return nil } -// Set adds the given key and value to the store without an expiry. If the key already exists, -// it will overwrite the previous value. -func (l *LRU) Set(key string, value any) error { - return l.SetWithExpiry(key, value, 0) -} - // SetWithDefaultExpiry adds the given key and value to the store with the default expiry. If // the key already exists, it will overwrite the previous value func (l *LRU) SetWithDefaultExpiry(key string, value any) error { @@ -101,11 +95,27 @@ func (l *LRU) Remove(key string) error { return nil } -// Keys returns a slice of the keys in the cache. -func (l *LRU) Keys() ([]string, error) { - l.lock.RLock() - defer l.lock.RUnlock() +func (l *LRU) RemoveMulti(keys []string) error { + l.lock.Lock() + defer l.lock.Unlock() + // Note, this is a copy of l.Remove. But we want to avoid + // fine-grained locking for every single removal. Therefore, + // we copy a bit of code for simplicity. + for _, key := range keys { + if ent, ok := l.items[key]; ok { + l.removeElement(ent) + } + } + + return nil +} + +// Scan passes the whole slice of keys to the callback in LRU mode. +// We don't need this callback style for LRU, but since we share +// the same interface with Redis, we maintain parity. +func (l *LRU) Scan(f func([]string) error) error { + l.lock.RLock() keys := make([]string, l.len) i := 0 for ent := l.evictList.Back(); ent != nil; ent = ent.Prev() { @@ -115,7 +125,9 @@ func (l *LRU) Keys() ([]string, error) { i++ } } - return keys, nil + l.lock.RUnlock() + + return f(keys) } // Len returns the number of items in the cache. diff --git a/server/platform/services/cache/lru_striped.go b/server/platform/services/cache/lru_striped.go index b4251f1b6d..1a502dedfe 100644 --- a/server/platform/services/cache/lru_striped.go +++ b/server/platform/services/cache/lru_striped.go @@ -4,6 +4,7 @@ package cache import ( + "errors" "fmt" "math" "time" @@ -62,11 +63,6 @@ func (L LRUStriped) Purge() error { return nil } -// Set does the same as LRU.Set -func (L LRUStriped) Set(key string, value any) error { - return L.keyBucket(key).Set(key, value) -} - // SetWithDefaultExpiry does the same as LRU.SetWithDefaultExpiry func (L LRUStriped) SetWithDefaultExpiry(key string, value any) error { return L.keyBucket(key).SetWithDefaultExpiry(key, value) @@ -95,16 +91,22 @@ func (L LRUStriped) Remove(key string) error { return L.keyBucket(key).Remove(key) } -// Keys does the same as LRU.Keys. However, because this is lock-free, keys might be -// inserted or removed from a previously scanned LRU cache. -// This is not as precise as using a single LRU instance. -func (L LRUStriped) Keys() ([]string, error) { - var keys []string - for _, lru := range L.buckets { - k, _ := lru.Keys() // Keys never returns any error - keys = append(keys, k...) +// RemoveMulti does the same as LRU.RemoveMulti +func (L LRUStriped) RemoveMulti(keys []string) error { + var err error + for _, key := range keys { + err = errors.Join(err, L.keyBucket(key).Remove(key)) } - return keys, nil + return err +} + +// Scan is basically a copy of Keys in LRU mode. +// See comment in LRU.Scan. +func (L LRUStriped) Scan(f func([]string) error) error { + for _, lru := range L.buckets { + lru.Scan(f) + } + return nil } // Len does the same as LRU.Len. As for LRUStriped.Keys, this call cannot be precise. diff --git a/server/platform/services/cache/lru_striped_bench_test.go b/server/platform/services/cache/lru_striped_bench_test.go index 31a39d66fe..e51c689d6d 100644 --- a/server/platform/services/cache/lru_striped_bench_test.go +++ b/server/platform/services/cache/lru_striped_bench_test.go @@ -43,7 +43,7 @@ func BenchmarkLRUStriped(b *testing.B) { bucketKeys[bucketKey] = append(bucketKeys[bucketKey], key) } for i := 0; i < opts.Size; i++ { - cache.Set(keys[i], "preflight") + cache.SetWithDefaultExpiry(keys[i], "preflight") } wgGet := &sync.WaitGroup{} @@ -58,7 +58,7 @@ func BenchmarkLRUStriped(b *testing.B) { case <-stopSet: return default: - _ = cache.Set(keys[i], "ignored") + _ = cache.SetWithDefaultExpiry(keys[i], "ignored") } } } diff --git a/server/platform/services/cache/lru_striped_test.go b/server/platform/services/cache/lru_striped_test.go index 3cb125f34b..0e1c1971e6 100644 --- a/server/platform/services/cache/lru_striped_test.go +++ b/server/platform/services/cache/lru_striped_test.go @@ -41,11 +41,15 @@ func TestNewLRUStriped(t *testing.T) { func TestLRUStripedKeyDistribution(t *testing.T) { dataset := makeLRUPredictableTestData(100) - scache, err := NewLRUStriped(&CacheOptions{StripedBuckets: 4, Size: len(dataset)}) + scache, err := NewLRUStriped(&CacheOptions{ + StripedBuckets: 4, + Size: len(dataset), + DefaultExpiry: 0, + }) require.NoError(t, err) cache := scache.(LRUStriped) for _, kv := range dataset { - require.NoError(t, cache.Set(kv[0], kv[1])) + require.NoError(t, cache.SetWithDefaultExpiry(kv[0], kv[1])) var out string require.NoError(t, cache.Get(kv[0], &out)) require.Equal(t, kv[1], out) @@ -87,13 +91,17 @@ func TestLRUStriped_HashKey(t *testing.T) { } func TestLRUStriped_Get(t *testing.T) { - cache, err := NewLRUStriped(&CacheOptions{StripedBuckets: 4, Size: 128}) + cache, err := NewLRUStriped(&CacheOptions{ + StripedBuckets: 4, + Size: 128, + DefaultExpiry: 0, + }) require.NoError(t, err) var out string require.Equal(t, ErrKeyNotFound, cache.Get("key", &out)) require.Zero(t, out) - require.NoError(t, cache.Set("key", "value")) + require.NoError(t, cache.SetWithDefaultExpiry("key", "value")) require.NoError(t, cache.Get("key", &out)) require.Equal(t, "value", out) } diff --git a/server/platform/services/cache/lru_test.go b/server/platform/services/cache/lru_test.go index eed675754f..a109511bb6 100644 --- a/server/platform/services/cache/lru_test.go +++ b/server/platform/services/cache/lru_test.go @@ -23,7 +23,7 @@ func TestLRU(t *testing.T) { }) for i := 0; i < 256; i++ { - err := l.Set(fmt.Sprintf("%d", i), i) + err := l.SetWithDefaultExpiry(fmt.Sprintf("%d", i), i) require.NoError(t, err) } @@ -32,25 +32,40 @@ func TestLRU(t *testing.T) { size := lru.len require.Equalf(t, size, 128, "bad len: %v", size) - keys, err := l.Keys() - require.NoError(t, err) - for i, k := range keys { - var v int - err = l.Get(k, &v) - require.NoError(t, err, "bad key: %v", k) - require.Equalf(t, fmt.Sprintf("%d", v), k, "bad key: %v", k) - require.Equalf(t, i+128, v, "bad value: %v", k) - } + l.Scan(func(keys []string) error { + for i, k := range keys { + var v int + err := l.Get(k, &v) + require.NoError(t, err, "bad key: %v", k) + require.Equalf(t, fmt.Sprintf("%d", v), k, "bad key: %v", k) + require.Equalf(t, i+128, v, "bad value: %v", k) + } + return nil + }) + for i := 0; i < 128; i++ { var v int - err = l.Get(fmt.Sprintf("%d", i), &v) + err := l.Get(fmt.Sprintf("%d", i), &v) require.Equal(t, ErrKeyNotFound, err, "should be evicted %v: %v", i, err) } for i := 128; i < 256; i++ { var v int - err = l.Get(fmt.Sprintf("%d", i), &v) + err := l.Get(fmt.Sprintf("%d", i), &v) require.NoError(t, err, "should not be evicted %v: %v", i, err) } + var v1, v2 int + var values = []any{&v1, &v2} + errs := l.GetMulti([]string{"128", "129"}, values) + for _, err := range errs { + require.NoError(t, err) + } + err := l.RemoveMulti([]string{"128", "129"}) + require.NoError(t, err) + errs = l.GetMulti([]string{"128", "129"}, values) + for i, err := range errs { + require.Equal(t, ErrKeyNotFound, err, "should be deleted %v: %v", i, err) + } + for i := 128; i < 192; i++ { l.Remove(fmt.Sprintf("%d", i)) var v int @@ -63,12 +78,13 @@ func TestLRU(t *testing.T) { require.NoError(t, err, "should exist") require.Equalf(t, 192, v, "bad value: %v", v) - keys, err = l.Keys() - require.NoError(t, err) - for i, k := range keys { - require.Falsef(t, i < 63 && k != fmt.Sprintf("%d", i+193), "out of order key: %v", k) - require.Falsef(t, i == 63 && k != "192", "out of order key: %v", k) - } + l.Scan(func(keys []string) error { + for i, k := range keys { + require.Falsef(t, i < 63 && k != fmt.Sprintf("%d", i+193), "out of order key: %v", k) + require.Falsef(t, i == 63 && k != "192", "out of order key: %v", k) + } + return nil + }) l.Purge() size = lru.len @@ -76,7 +92,7 @@ func TestLRU(t *testing.T) { err = l.Get("200", &v) require.Equal(t, err, ErrKeyNotFound, "should contain nothing") - err = l.Set("201", 301) + err = l.SetWithDefaultExpiry("201", 301) require.NoError(t, err) err = l.Get("201", &v) require.NoError(t, err) @@ -116,7 +132,7 @@ func TestLRUMarshalUnMarshal(t *testing.T) { "key1": 1, "key2": "value2", } - err := l.Set("test", value1) + err := l.SetWithDefaultExpiry("test", value1) require.NoError(t, err) @@ -196,7 +212,7 @@ func TestLRUMarshalUnMarshal(t *testing.T) { }, }, } - err = l.Set("post", post.Clone()) + err = l.SetWithDefaultExpiry("post", post.Clone()) require.NoError(t, err) var p model.Post @@ -224,7 +240,7 @@ func TestLRUMarshalUnMarshal(t *testing.T) { }, } - err = l.Set("session", session) + err = l.SetWithDefaultExpiry("session", session) require.NoError(t, err) var s = &model.Session{} err = l.Get("session", s) @@ -267,7 +283,7 @@ func TestLRUMarshalUnMarshal(t *testing.T) { TermsOfServiceCreateAt: 111111, } - err = l.Set("user", user) + err = l.SetWithDefaultExpiry("user", user) require.NoError(t, err) var u *model.User @@ -280,7 +296,7 @@ func TestLRUMarshalUnMarshal(t *testing.T) { tt := make(map[string]*model.User) tt["1"] = u - err = l.Set("mm", model.UserMap(tt)) + err = l.SetWithDefaultExpiry("mm", model.UserMap(tt)) require.NoError(t, err) var out map[string]*model.User @@ -300,7 +316,7 @@ func BenchmarkLRU(b *testing.B) { DefaultExpiry: 0, InvalidateClusterEvent: "", }) - err := l2.Set("test", value1) + err := l2.SetWithDefaultExpiry("test", value1) require.NoError(b, err) var val string @@ -350,7 +366,7 @@ func BenchmarkLRU(b *testing.B) { DefaultExpiry: 0, InvalidateClusterEvent: "", }) - err := l2.Set("test", value2) + err := l2.SetWithDefaultExpiry("test", value2) require.NoError(b, err) var val obj @@ -433,7 +449,7 @@ func BenchmarkLRU(b *testing.B) { DefaultExpiry: 0, InvalidateClusterEvent: "", }) - err := l2.Set("test", user) + err := l2.SetWithDefaultExpiry("test", user) require.NoError(b, err) var val model.User @@ -466,7 +482,7 @@ func BenchmarkLRU(b *testing.B) { DefaultExpiry: 0, InvalidateClusterEvent: "", }) - err := l2.Set("test", model.UserMap(uMap)) + err := l2.SetWithDefaultExpiry("test", model.UserMap(uMap)) require.NoError(b, err) var val map[string]*model.User @@ -545,7 +561,7 @@ func BenchmarkLRU(b *testing.B) { DefaultExpiry: 0, InvalidateClusterEvent: "", }) - err := l2.Set("test", post) + err := l2.SetWithDefaultExpiry("test", post) require.NoError(b, err) var val model.Post @@ -569,7 +585,7 @@ func BenchmarkLRU(b *testing.B) { DefaultExpiry: 0, InvalidateClusterEvent: "", }) - err := l2.Set("test", status) + err := l2.SetWithDefaultExpiry("test", status) require.NoError(b, err) var val *model.Status @@ -605,7 +621,7 @@ func BenchmarkLRU(b *testing.B) { DefaultExpiry: 0, InvalidateClusterEvent: "", }) - err := l2.Set("test", &session) + err := l2.SetWithDefaultExpiry("test", &session) require.NoError(b, err) var val *model.Session @@ -622,14 +638,14 @@ func TestLRURace(t *testing.T) { InvalidateClusterEvent: "", }) var wg sync.WaitGroup - l2.Set("test", "value1") + l2.SetWithDefaultExpiry("test", "value1") wg.Add(2) go func() { defer wg.Done() value1 := "simplestring" - err := l2.Set("test", value1) + err := l2.SetWithDefaultExpiry("test", value1) require.NoError(t, err) }() diff --git a/server/platform/services/cache/mocks/Cache.go b/server/platform/services/cache/mocks/Cache.go index 7ad868ec9e..c3d02eb78c 100644 --- a/server/platform/services/cache/mocks/Cache.go +++ b/server/platform/services/cache/mocks/Cache.go @@ -72,36 +72,6 @@ func (_m *Cache) GetMulti(keys []string, values []interface{}) []error { return r0 } -// Keys provides a mock function with given fields: -func (_m *Cache) Keys() ([]string, error) { - ret := _m.Called() - - if len(ret) == 0 { - panic("no return value specified for Keys") - } - - var r0 []string - var r1 error - if rf, ok := ret.Get(0).(func() ([]string, error)); ok { - return rf() - } - if rf, ok := ret.Get(0).(func() []string); ok { - r0 = rf() - } else { - if ret.Get(0) != nil { - r0 = ret.Get(0).([]string) - } - } - - if rf, ok := ret.Get(1).(func() error); ok { - r1 = rf() - } else { - r1 = ret.Error(1) - } - - return r0, r1 -} - // Name provides a mock function with given fields: func (_m *Cache) Name() string { ret := _m.Called() @@ -156,17 +126,35 @@ func (_m *Cache) Remove(key string) error { return r0 } -// Set provides a mock function with given fields: key, value -func (_m *Cache) Set(key string, value interface{}) error { - ret := _m.Called(key, value) +// RemoveMulti provides a mock function with given fields: keys +func (_m *Cache) RemoveMulti(keys []string) error { + ret := _m.Called(keys) if len(ret) == 0 { - panic("no return value specified for Set") + panic("no return value specified for RemoveMulti") } var r0 error - if rf, ok := ret.Get(0).(func(string, interface{}) error); ok { - r0 = rf(key, value) + if rf, ok := ret.Get(0).(func([]string) error); ok { + r0 = rf(keys) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// Scan provides a mock function with given fields: f +func (_m *Cache) Scan(f func([]string) error) error { + ret := _m.Called(f) + + if len(ret) == 0 { + panic("no return value specified for Scan") + } + + var r0 error + if rf, ok := ret.Get(0).(func(func([]string) error) error); ok { + r0 = rf(f) } else { r0 = ret.Error(0) } diff --git a/server/platform/services/cache/provider_test.go b/server/platform/services/cache/provider_test.go index d0c7fafdfc..1aa2be53c3 100644 --- a/server/platform/services/cache/provider_test.go +++ b/server/platform/services/cache/provider_test.go @@ -18,15 +18,16 @@ func TestNewCache(t *testing.T) { size := 1 c, err := p.NewCache(&CacheOptions{ - Size: size, + Size: size, + DefaultExpiry: 0, }) require.NoError(t, err) - err = c.Set("key1", "val1") + err = c.SetWithDefaultExpiry("key1", "val1") require.NoError(t, err) - err = c.Set("key2", "val2") + err = c.SetWithDefaultExpiry("key2", "val2") require.NoError(t, err) - err = c.Set("key3", "val3") + err = c.SetWithDefaultExpiry("key3", "val3") require.NoError(t, err) }) @@ -35,15 +36,16 @@ func TestNewCache(t *testing.T) { size := 1 c, err := p.NewCache(&CacheOptions{ - Size: size, + Size: size, + DefaultExpiry: 0, }) require.NoError(t, err) - err = c.Set("key1", "val1") + err = c.SetWithDefaultExpiry("key1", "val1") require.NoError(t, err) - err = c.Set("key2", "val2") + err = c.SetWithDefaultExpiry("key2", "val2") require.NoError(t, err) - err = c.Set("key3", "val3") + err = c.SetWithDefaultExpiry("key3", "val3") require.NoError(t, err) }) @@ -91,14 +93,15 @@ func TestNewCache_Striped(t *testing.T) { Size: size, Striped: true, StripedBuckets: 1, + DefaultExpiry: 0, }) require.NoError(t, err) - err = c.Set("key1", "val1") + err = c.SetWithDefaultExpiry("key1", "val1") require.NoError(t, err) - err = c.Set("key2", "val2") + err = c.SetWithDefaultExpiry("key2", "val2") require.NoError(t, err) - err = c.Set("key3", "val3") + err = c.SetWithDefaultExpiry("key3", "val3") require.NoError(t, err) }) @@ -110,14 +113,15 @@ func TestNewCache_Striped(t *testing.T) { Size: size, Striped: true, StripedBuckets: 1, + DefaultExpiry: 0, }) require.NoError(t, err) - err = c.Set("key1", "val1") + err = c.SetWithDefaultExpiry("key1", "val1") require.NoError(t, err) - err = c.Set("key2", "val2") + err = c.SetWithDefaultExpiry("key2", "val2") require.NoError(t, err) - err = c.Set("key3", "val3") + err = c.SetWithDefaultExpiry("key3", "val3") require.NoError(t, err) }) diff --git a/server/platform/services/cache/redis.go b/server/platform/services/cache/redis.go index 672a89d6fb..38b72d2c5c 100644 --- a/server/platform/services/cache/redis.go +++ b/server/platform/services/cache/redis.go @@ -7,6 +7,7 @@ import ( "context" "errors" "fmt" + "strings" "time" "github.com/mattermost/mattermost/server/public/model" @@ -38,20 +39,7 @@ func NewRedis(opts *CacheOptions, client rueidis.Client) (*Redis, error) { } func (r *Redis) Purge() error { - // TODO: move to scan - keys, err := r.Keys() - if err != nil { - return err - } - return r.client.Do(context.Background(), - r.client.B().Del(). - Key(keys...). - Build(), - ).Error() -} - -func (r *Redis) Set(key string, value any) error { - return r.SetWithExpiry(key, value, 0) + return r.Scan(r.RemoveMulti) } // SetWithDefaultExpiry adds the given key and value to the store with the default expiry. If @@ -156,10 +144,9 @@ func (r *Redis) GetMulti(keys []string, values []any) []error { }() errs := make([]error, len(keys)) - newKeys := make([]string, len(keys)) - for i := range keys { - newKeys[i] = r.name + ":" + keys[i] - } + newKeys := sliceMapper(keys, func(elem string) string { + return r.name + ":" + elem + }) vals, err := r.client.DoCache(context.Background(), r.client.B().Mget(). Key(newKeys...). @@ -227,7 +214,7 @@ func (r *Redis) Remove(key string) error { defer func() { if r.metrics != nil { elapsed := time.Since(now).Seconds() - r.metrics.ObserveRedisEndpointDuration(r.name, "Del", elapsed) + r.metrics.ObserveRedisEndpointDuration(r.name, "Remove", elapsed) } }() @@ -238,22 +225,61 @@ func (r *Redis) Remove(key string) error { ).Error() } -// Keys returns a slice of the keys in the cache. -func (r *Redis) Keys() ([]string, error) { +func (r *Redis) RemoveMulti(keys []string) error { now := time.Now() defer func() { if r.metrics != nil { elapsed := time.Since(now).Seconds() - r.metrics.ObserveRedisEndpointDuration(r.name, "Keys", elapsed) + r.metrics.ObserveRedisEndpointDuration(r.name, "RemoveMulti", elapsed) } }() - // TODO: migrate to a function that works on a batch of keys. + if len(keys) == 0 { + return nil + } + + newKeys := sliceMapper(keys, func(elem string) string { + return r.name + ":" + elem + }) + return r.client.Do(context.Background(), - r.client.B().Keys(). - Pattern(r.name+":*"). + r.client.B().Del(). + Key(newKeys...). Build(), - ).AsStrSlice() + ).Error() +} + +func (r *Redis) Scan(f func([]string) error) error { + now := time.Now() + defer func() { + if r.metrics != nil { + elapsed := time.Since(now).Seconds() + r.metrics.ObserveRedisEndpointDuration(r.name, "Scan", elapsed) + } + }() + + var scan rueidis.ScanEntry + var err error + for more := true; more; more = scan.Cursor != 0 { + scan, err = r.client.Do(context.Background(), + r.client.B().Scan(). + Cursor(scan.Cursor). + Match(r.name+":*"). + Count(100). + Build()).AsScanEntry() + if err != nil { + return err + } + + removed := sliceMapper(scan.Elements, func(elem string) string { + return strings.TrimPrefix(elem, r.name+":") + }) + err = f(removed) + if err != nil { + return err + } + } + return nil } // Len returns the number of items in the cache. @@ -285,3 +311,11 @@ func (r *Redis) GetInvalidateClusterEvent() model.ClusterEvent { func (r *Redis) Name() string { return r.name } + +func sliceMapper[S ~[]E, E, R any](slice S, mapper func(E) R) []R { + newSlice := make([]R, len(slice)) + for i, v := range slice { + newSlice[i] = mapper(v) + } + return newSlice +}