MM-59934: Add Redis to CI and other improvements (#28164)

- Update library version.
- Added MaxFlush delay to help reduce CPU usage.
- Fall back to LRU cache for the caches which use SCAN.
- Added mattermost-redis and running for all api layer
tests in Postgres.

https://mattermost.atlassian.net/browse/MM-59934
```release-note
NONE
```

server/channels/api4/apitestlib.go

@@ -102,6 +102,17 @@ func setupTestHelper(dbStore store.Store, searchEngine *searchengine.Broker, ent
 	*memoryConfig.AnnouncementSettings.AdminNoticesEnabled = false
 	*memoryConfig.AnnouncementSettings.UserNoticesEnabled = false
 	*memoryConfig.PluginSettings.AutomaticPrepackagedPlugins = false
+	// Enabling Redis with Postgres.
+	if *memoryConfig.SqlSettings.DriverName == model.DatabaseDriverPostgres {
+		*memoryConfig.CacheSettings.CacheType = model.CacheTypeRedis
+		redisHost := "localhost"
+		if os.Getenv("IS_CI") == "true" {
+			redisHost = "redis"
+		}
+		*memoryConfig.CacheSettings.RedisAddress = redisHost + ":6379"
+		*memoryConfig.CacheSettings.DisableClientCache = true
+		*memoryConfig.CacheSettings.RedisDB = 0
+	}
 	if updateConfig != nil {
 		updateConfig(memoryConfig)
 	}

server/channels/api4/config_test.go

@@ -515,7 +515,7 @@ func TestUpdateConfigDiffInAuditRecord(t *testing.T) {
 	require.NotEmpty(t, data)
 
 	require.Contains(t, string(data),
-		fmt.Sprintf(`"config_diffs":[{"actual_val":%d,"base_val":%d,"path":"ServiceSettings.ReadTimeout"}]`,
+		fmt.Sprintf(`"config_diffs":[{"actual_val":%d,"base_val":%d,"path":"ServiceSettings.ReadTimeout"}`,
 			timeoutVal+1, timeoutVal))
 }
 

server/channels/api4/metrics_test.go

@@ -30,6 +30,7 @@ func setupMetricsMock() *mocks.MetricsInterface {
 	metricsMock.On("IncrementHTTPError").Return()
 	metricsMock.On("IncrementHTTPRequest").Return()
 	metricsMock.On("ObserveAPIEndpointDuration", mock.AnythingOfType("string"), mock.AnythingOfType("string"), mock.AnythingOfType("string"), mock.AnythingOfType("string"), mock.AnythingOfType("string"), mock.AnythingOfType("float64")).Return()
+	metricsMock.On("ObserveRedisEndpointDuration", mock.AnythingOfType("string"), mock.AnythingOfType("string"), mock.AnythingOfType("float64")).Return()
 	metricsMock.On("Register").Return()
 
 	return metricsMock

server/channels/app/platform/service.go

@@ -169,6 +169,7 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
 				RedisAddr:     *cacheConfig.RedisAddress,
 				RedisPassword: *cacheConfig.RedisPassword,
 				RedisDB:       *cacheConfig.RedisDB,
+				DisableCache:  *cacheConfig.DisableClientCache,
 			},
 		)
 	}
@@ -292,8 +293,13 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
 		return nil, fmt.Errorf("cannot create store: %w", err)
 	}
 
+	// Note: we hardcode the session and status cache to LRU because they lead
+	// to a lot of SCAN calls in case of Redis. We could potentially have a
+	// reverse mapping to avoid the scan, but this needs more complicated code.
+	// Leaving this for now.
+
 	// Needed before loading license
-	ps.statusCache, err = ps.cacheProvider.NewCache(&cache.CacheOptions{
+	ps.statusCache, err = cache.NewProvider().NewCache(&cache.CacheOptions{
 		Name:           "Status",
 		Size:           model.StatusCacheSize,
 		Striped:        true,
@@ -304,10 +310,6 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
 		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/store/localcachelayer/layer.go

@@ -340,7 +340,8 @@ func NewLocalCacheLayer(baseStore store.Store, metrics einterfaces.MetricsInterf
 	}); err != nil {
 		return
 	}
-	if localCacheStore.profilesInChannelCache, err = cacheProvider.NewCache(&cache.CacheOptions{
+	// Hardcoding this to LRU because of the volume of SCAN calls in case of Redis.
+	if localCacheStore.profilesInChannelCache, err = cache.NewProvider().NewCache(&cache.CacheOptions{
 		Size:                   ProfilesInChannelCacheSize,
 		Name:                   "ProfilesInChannel",
 		DefaultExpiry:          ProfilesInChannelCacheSec * time.Second,
@@ -526,7 +527,10 @@ func (s *LocalCacheStore) doMultiReadCache(cache cache.Cache, keys []string, val
 }
 
 func (s *LocalCacheStore) doClearCacheCluster(cache cache.Cache) {
-	cache.Purge()
+	err := cache.Purge()
+	if err != nil {
+		s.logger.Warn("Error while purging cache", mlog.Err(err), mlog.String("cache_name", cache.Name()))
+	}
 	if s.cluster != nil && s.cacheType == model.CacheTypeLRU {
 		msg := &model.ClusterMessage{
 			Event:    cache.GetInvalidateClusterEvent(),