mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-20 06:07:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

filer: make the redis connection settings configurable (#10441)

Browse Source 
The sentinel stores hardcoded a 30s read timeout and a 1m retry backoff.
After a sentinel failover every request that picked a pooled connection to
the old master sat there for 30s before the connection was retired, and the
pool timeout derived from it (read timeout + 1s) queued the rest behind
them. The other redis stores took the go-redis defaults with no way to tune
anything.

Read the dial, timeout and pool knobs from each redis store section instead,
keeping the go-redis default for every key left unset.
This commit is contained in:
Chris Lu
2026-07-25 19:55:39 -07:00
committed by GitHub
parent 5cac980b32
commit 83f754763e
11 changed files with 301 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/command/scaffold/filer.toml
+18
View File
@@ -230,6 +230,23 @@ client_cert_path = ""
client_key_path = ""
# This changes the data layout. Only add new directories. Removing/Updating will cause data loss.
superLargeDirectories = []
# Connection tuning, accepted by every redis filer store section. Each key keeps the go-redis
# default when left at 0, and a timeout cannot be turned off - a filer that never times out a read
# is what stalls a failover. Shorter read timeouts make the filer notice a dead redis sooner, which
# matters most for the sentinel store: every pooled connection to the old master has to time out
# before a failover is absorbed.
# max_retries = 0 # 0 is three retries, -1 disables retrying
# min_retry_backoff_millisecond = 0 # default 8
# max_retry_backoff_millisecond = 0 # default 512
# dial_timeout_millisecond = 0 # default 5000
# read_timeout_millisecond = 0 # default 3000
# write_timeout_millisecond = 0 # default 3000, follows read_timeout_millisecond
# pool_size = 0 # default 10 per CPU
# pool_timeout_millisecond = 0 # default read_timeout_millisecond + 1000
# min_idle_conns = 0
# max_idle_conns = 0
# conn_max_idle_time_seconds = 0 # default 1800
# conn_max_lifetime_seconds = 0 # default unlimited
[redis2_sentinel]
enabled = false
@@ -247,6 +264,7 @@ enable_tls = false
ca_cert_path = ""
client_cert_path = ""
client_key_path = ""
# see [redis2] for the connection tuning keys
[redis_cluster2]
enabled = false
weed/filer/redis/redis_cluster_store.go
+7 -3
View File
@@ -3,6 +3,7 @@ package redis
import (
"github.com/redis/go-redis/v9"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_conf"
"github.com/seaweedfs/seaweedfs/weed/util"
)
@@ -28,15 +29,18 @@ func (store *RedisClusterStore) Initialize(configuration util.Configuration, pre
configuration.GetString(prefix+"password"),
configuration.GetBool(prefix+"useReadOnly"),
configuration.GetBool(prefix+"routeByLatency"),
redis_conf.Read(configuration, prefix),
)
}
func (store *RedisClusterStore) initialize(addresses []string, password string, readOnly, routeByLatency bool) (err error) {
store.Client = redis.NewClusterClient(&redis.ClusterOptions{
func (store *RedisClusterStore) initialize(addresses []string, password string, readOnly, routeByLatency bool, settings redis_conf.Settings) (err error) {
options := &redis.ClusterOptions{
Addrs: addresses,
Password: password,
ReadOnly: readOnly,
RouteByLatency: routeByLatency,
})
}
settings.ApplyToCluster(options)
store.Client = redis.NewClusterClient(options)
return
}
weed/filer/redis/redis_store.go
+7 -3
View File
@@ -3,6 +3,7 @@ package redis
import (
"github.com/redis/go-redis/v9"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_conf"
"github.com/seaweedfs/seaweedfs/weed/util"
)
@@ -23,14 +24,17 @@ func (store *RedisStore) Initialize(configuration util.Configuration, prefix str
configuration.GetString(prefix+"address"),
configuration.GetString(prefix+"password"),
configuration.GetInt(prefix+"database"),
redis_conf.Read(configuration, prefix),
)
}
func (store *RedisStore) initialize(hostPort string, password string, database int) (err error) {
store.Client = redis.NewClient(&redis.Options{
func (store *RedisStore) initialize(hostPort string, password string, database int, settings redis_conf.Settings) (err error) {
options := &redis.Options{
Addr: hostPort,
Password: password,
DB: database,
})
}
settings.ApplyTo(options)
store.Client = redis.NewClient(options)
return
}
weed/filer/redis2/redis_cluster_store.go
+7 -3
View File
@@ -5,6 +5,7 @@ import (
"github.com/redis/go-redis/v9"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_conf"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_tls"
"github.com/seaweedfs/seaweedfs/weed/util"
)
@@ -40,18 +41,21 @@ func (store *RedisCluster2Store) Initialize(configuration util.Configuration, pr
configuration.GetBool(prefix+"routeByLatency"),
configuration.GetStringSlice(prefix+"superLargeDirectories"),
tlsConfig,
redis_conf.Read(configuration, prefix),
)
}
func (store *RedisCluster2Store) initialize(addresses []string, username string, password string, keyPrefix string, readOnly, routeByLatency bool, superLargeDirectories []string, tlsConfig *tls.Config) (err error) {
store.Client = redis.NewClusterClient(&redis.ClusterOptions{
func (store *RedisCluster2Store) initialize(addresses []string, username string, password string, keyPrefix string, readOnly, routeByLatency bool, superLargeDirectories []string, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
options := &redis.ClusterOptions{
Addrs: addresses,
Username: username,
Password: password,
ReadOnly: readOnly,
RouteByLatency: routeByLatency,
TLSConfig: tlsConfig,
})
}
settings.ApplyToCluster(options)
store.Client = redis.NewClusterClient(options)
store.keyPrefix = keyPrefix
store.loadSuperLargeDirectories(superLargeDirectories)
return
weed/filer/redis2/redis_sentinel_store.go
+7 -8
View File
@@ -2,10 +2,10 @@ package redis2
import (
"crypto/tls"
"time"
"github.com/redis/go-redis/v9"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_conf"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_tls"
"github.com/seaweedfs/seaweedfs/weed/util"
)
@@ -37,11 +37,12 @@ func (store *Redis2SentinelStore) Initialize(configuration util.Configuration, p
configuration.GetInt(prefix+"database"),
configuration.GetString(prefix+"keyPrefix"),
tlsConfig,
redis_conf.Read(configuration, prefix),
)
}
func (store *Redis2SentinelStore) initialize(addresses []string, masterName string, username string, password string, sentinelUsername string, sentinelPassword string, database int, keyPrefix string, tlsConfig *tls.Config) (err error) {
store.Client = redis.NewFailoverClient(&redis.FailoverOptions{
func (store *Redis2SentinelStore) initialize(addresses []string, masterName string, username string, password string, sentinelUsername string, sentinelPassword string, database int, keyPrefix string, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
options := &redis.FailoverOptions{
MasterName: masterName,
SentinelAddrs: addresses,
Username: username,
@@ -50,11 +51,9 @@ func (store *Redis2SentinelStore) initialize(addresses []string, masterName stri
SentinelPassword: sentinelPassword,
DB: database,
TLSConfig: tlsConfig,
MinRetryBackoff: time.Millisecond * 100,
MaxRetryBackoff: time.Minute * 1,
ReadTimeout: time.Second * 30,
WriteTimeout: time.Second * 5,
})
}
settings.ApplyToFailover(options)
store.Client = redis.NewFailoverClient(options)
store.keyPrefix = keyPrefix
return
}
weed/filer/redis2/redis_store.go
+7 -3
View File
@@ -5,6 +5,7 @@ import (
"github.com/redis/go-redis/v9"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_conf"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_tls"
"github.com/seaweedfs/seaweedfs/weed/util"
)
@@ -34,17 +35,20 @@ func (store *Redis2Store) Initialize(configuration util.Configuration, prefix st
configuration.GetString(prefix+"keyPrefix"),
configuration.GetStringSlice(prefix+"superLargeDirectories"),
tlsConfig,
redis_conf.Read(configuration, prefix),
)
}
func (store *Redis2Store) initialize(hostPort string, username string, password string, database int, keyPrefix string, superLargeDirectories []string, tlsConfig *tls.Config) (err error) {
store.Client = redis.NewClient(&redis.Options{
func (store *Redis2Store) initialize(hostPort string, username string, password string, database int, keyPrefix string, superLargeDirectories []string, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
options := &redis.Options{
Addr: hostPort,
Username: username,
Password: password,
DB: database,
TLSConfig: tlsConfig,
})
}
settings.ApplyTo(options)
store.Client = redis.NewClient(options)
store.keyPrefix = keyPrefix
store.loadSuperLargeDirectories(superLargeDirectories)
return
weed/filer/redis3/redis_cluster_store.go
+7 -3
View File
@@ -7,6 +7,7 @@ import (
"github.com/go-redsync/redsync/v4/redis/goredis/v9"
"github.com/redis/go-redis/v9"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_conf"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_tls"
"github.com/seaweedfs/seaweedfs/weed/util"
)
@@ -39,17 +40,20 @@ func (store *RedisCluster3Store) Initialize(configuration util.Configuration, pr
configuration.GetBool(prefix+"useReadOnly"),
configuration.GetBool(prefix+"routeByLatency"),
tlsConfig,
redis_conf.Read(configuration, prefix),
)
}
func (store *RedisCluster3Store) initialize(addresses []string, password string, readOnly, routeByLatency bool, tlsConfig *tls.Config) (err error) {
store.Client = redis.NewClusterClient(&redis.ClusterOptions{
func (store *RedisCluster3Store) initialize(addresses []string, password string, readOnly, routeByLatency bool, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
options := &redis.ClusterOptions{
Addrs: addresses,
Password: password,
ReadOnly: readOnly,
RouteByLatency: routeByLatency,
TLSConfig: tlsConfig,
})
}
settings.ApplyToCluster(options)
store.Client = redis.NewClusterClient(options)
store.redsync = redsync.New(goredis.NewPool(store.Client))
return
}
weed/filer/redis3/redis_sentinel_store.go
+7 -8
View File
@@ -2,12 +2,12 @@ package redis3
import (
"crypto/tls"
"time"
"github.com/go-redsync/redsync/v4"
"github.com/go-redsync/redsync/v4/redis/goredis/v9"
"github.com/redis/go-redis/v9"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_conf"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_tls"
"github.com/seaweedfs/seaweedfs/weed/util"
)
@@ -38,11 +38,12 @@ func (store *Redis3SentinelStore) Initialize(configuration util.Configuration, p
configuration.GetString(prefix+"sentinel_password"),
configuration.GetInt(prefix+"database"),
tlsConfig,
redis_conf.Read(configuration, prefix),
)
}
func (store *Redis3SentinelStore) initialize(addresses []string, masterName string, username string, password string, sentinelUsername string, sentinelPassword string, database int, tlsConfig *tls.Config) (err error) {
store.Client = redis.NewFailoverClient(&redis.FailoverOptions{
func (store *Redis3SentinelStore) initialize(addresses []string, masterName string, username string, password string, sentinelUsername string, sentinelPassword string, database int, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
options := &redis.FailoverOptions{
MasterName: masterName,
SentinelAddrs: addresses,
Username: username,
@@ -51,11 +52,9 @@ func (store *Redis3SentinelStore) initialize(addresses []string, masterName stri
SentinelPassword: sentinelPassword,
DB: database,
TLSConfig: tlsConfig,
MinRetryBackoff: time.Millisecond * 100,
MaxRetryBackoff: time.Minute * 1,
ReadTimeout: time.Second * 30,
WriteTimeout: time.Second * 5,
})
}
settings.ApplyToFailover(options)
store.Client = redis.NewFailoverClient(options)
store.redsync = redsync.New(goredis.NewPool(store.Client))
return
}
weed/filer/redis3/redis_store.go
+7 -3
View File
@@ -7,6 +7,7 @@ import (
"github.com/go-redsync/redsync/v4/redis/goredis/v9"
"github.com/redis/go-redis/v9"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_conf"
"github.com/seaweedfs/seaweedfs/weed/filer/redis_tls"
"github.com/seaweedfs/seaweedfs/weed/util"
)
@@ -33,16 +34,19 @@ func (store *Redis3Store) Initialize(configuration util.Configuration, prefix st
configuration.GetString(prefix+"password"),
configuration.GetInt(prefix+"database"),
tlsConfig,
redis_conf.Read(configuration, prefix),
)
}
func (store *Redis3Store) initialize(hostPort string, password string, database int, tlsConfig *tls.Config) (err error) {
store.Client = redis.NewClient(&redis.Options{
func (store *Redis3Store) initialize(hostPort string, password string, database int, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
options := &redis.Options{
Addr: hostPort,
Password: password,
DB: database,
TLSConfig: tlsConfig,
})
}
settings.ApplyTo(options)
store.Client = redis.NewClient(options)
store.redsync = redsync.New(goredis.NewPool(store.Client))
return
}
weed/filer/redis_conf/redis_conf.go
+108
View File
@@ -0,0 +1,108 @@
// Package redis_conf reads the dial, timeout and connection pool settings shared by the redis
// filer stores.
package redis_conf
import (
"time"
"github.com/redis/go-redis/v9"
"github.com/seaweedfs/seaweedfs/weed/util"
)
// Settings carries the go-redis tuning a filer store section may override. A zero field keeps the
// go-redis default, so an untouched configuration behaves the way it did before these keys existed.
type Settings struct {
MaxRetries int
MinRetryBackoff time.Duration
MaxRetryBackoff time.Duration
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
PoolSize int
PoolTimeout time.Duration
MinIdleConns int
MaxIdleConns int
ConnMaxIdleTime time.Duration
ConnMaxLifetime time.Duration
}
// Read parses the tuning keys of a redis filer store section.
func Read(configuration util.Configuration, prefix string) Settings {
return Settings{
// go-redis reads 0 as three retries and -1 as no retry, so hand the value over untouched
MaxRetries: configuration.GetInt(prefix + "max_retries"),
MinRetryBackoff: milliseconds(configuration, prefix+"min_retry_backoff_millisecond"),
MaxRetryBackoff: milliseconds(configuration, prefix+"max_retry_backoff_millisecond"),
DialTimeout: milliseconds(configuration, prefix+"dial_timeout_millisecond"),
ReadTimeout: milliseconds(configuration, prefix+"read_timeout_millisecond"),
WriteTimeout: milliseconds(configuration, prefix+"write_timeout_millisecond"),
PoolSize: configuration.GetInt(prefix + "pool_size"),
PoolTimeout: milliseconds(configuration, prefix+"pool_timeout_millisecond"),
MinIdleConns: configuration.GetInt(prefix + "min_idle_conns"),
MaxIdleConns: configuration.GetInt(prefix + "max_idle_conns"),
ConnMaxIdleTime: seconds(configuration, prefix+"conn_max_idle_time_seconds"),
ConnMaxLifetime: seconds(configuration, prefix+"conn_max_lifetime_seconds"),
}
}
func (settings Settings) ApplyTo(options *redis.Options) {
options.MaxRetries = settings.MaxRetries
options.MinRetryBackoff = settings.MinRetryBackoff
options.MaxRetryBackoff = settings.MaxRetryBackoff
options.DialTimeout = settings.DialTimeout
options.ReadTimeout = settings.ReadTimeout
options.WriteTimeout = settings.WriteTimeout
options.PoolSize = settings.PoolSize
options.PoolTimeout = settings.PoolTimeout
options.MinIdleConns = settings.MinIdleConns
options.MaxIdleConns = settings.MaxIdleConns
options.ConnMaxIdleTime = settings.ConnMaxIdleTime
options.ConnMaxLifetime = settings.ConnMaxLifetime
}
func (settings Settings) ApplyToCluster(options *redis.ClusterOptions) {
options.MaxRetries = settings.MaxRetries
options.MinRetryBackoff = settings.MinRetryBackoff
options.MaxRetryBackoff = settings.MaxRetryBackoff
options.DialTimeout = settings.DialTimeout
options.ReadTimeout = settings.ReadTimeout
options.WriteTimeout = settings.WriteTimeout
options.PoolSize = settings.PoolSize
options.PoolTimeout = settings.PoolTimeout
options.MinIdleConns = settings.MinIdleConns
options.MaxIdleConns = settings.MaxIdleConns
options.ConnMaxIdleTime = settings.ConnMaxIdleTime
options.ConnMaxLifetime = settings.ConnMaxLifetime
}
func (settings Settings) ApplyToFailover(options *redis.FailoverOptions) {
options.MaxRetries = settings.MaxRetries
options.MinRetryBackoff = settings.MinRetryBackoff
options.MaxRetryBackoff = settings.MaxRetryBackoff
options.DialTimeout = settings.DialTimeout
options.ReadTimeout = settings.ReadTimeout
options.WriteTimeout = settings.WriteTimeout
options.PoolSize = settings.PoolSize
options.PoolTimeout = settings.PoolTimeout
options.MinIdleConns = settings.MinIdleConns
options.MaxIdleConns = settings.MaxIdleConns
options.ConnMaxIdleTime = settings.ConnMaxIdleTime
options.ConnMaxLifetime = settings.ConnMaxLifetime
}
// milliseconds and seconds keep the go-redis default for anything non-positive. go-redis spells
// "no timeout" as a negative nanosecond count, which these keys deliberately cannot reach: a filer
// that never times out a read is what leaves a failover hanging.
func milliseconds(configuration util.Configuration, key string) time.Duration {
if value := configuration.GetInt(key); value > 0 {
return time.Duration(value) * time.Millisecond
}
return 0
}
func seconds(configuration util.Configuration, key string) time.Duration {
if value := configuration.GetInt(key); value > 0 {
return time.Duration(value) * time.Second
}
return 0
}
weed/filer/redis_conf/redis_conf_test.go
+119
View File
@@ -0,0 +1,119 @@
package redis_conf
import (
"testing"
"time"
"github.com/redis/go-redis/v9"
)
type fakeConfiguration map[string]interface{}
func (c fakeConfiguration) GetString(key string) string {
value, _ := c[key].(string)
return value
}
func (c fakeConfiguration) GetBool(key string) bool {
value, _ := c[key].(bool)
return value
}
func (c fakeConfiguration) GetInt(key string) int {
value, _ := c[key].(int)
return value
}
func (c fakeConfiguration) GetStringSlice(key string) []string {
value, _ := c[key].([]string)
return value
}
func (c fakeConfiguration) SetDefault(key string, value interface{}) {
if _, found := c[key]; !found {
c[key] = value
}
}
func TestUnsetKeepsGoRedisDefaults(t *testing.T) {
settings := Read(fakeConfiguration{}, "redis2_sentinel.")
if settings != (Settings{}) {
t.Fatalf("expected zero settings, got %+v", settings)
}
options := &redis.FailoverOptions{MasterName: "master"}
settings.ApplyToFailover(options)
client := redis.NewFailoverClient(options)
defer client.Close()
if got := client.Options().ReadTimeout; got != 3*time.Second {
t.Fatalf("read timeout %v, want the go-redis default of 3s", got)
}
if got := client.Options().MaxRetries; got != 3 {
t.Fatalf("max retries %d, want the go-redis default of 3", got)
}
}
func TestRead(t *testing.T) {
settings := Read(fakeConfiguration{
"redis2.max_retries": 5,
"redis2.min_retry_backoff_millisecond": 10,
"redis2.max_retry_backoff_millisecond": 500,
"redis2.dial_timeout_millisecond": 2000,
"redis2.read_timeout_millisecond": 1500,
"redis2.write_timeout_millisecond": 1500,
"redis2.pool_size": 64,
"redis2.pool_timeout_millisecond": 2500,
"redis2.min_idle_conns": 4,
"redis2.max_idle_conns": 16,
"redis2.conn_max_idle_time_seconds": 300,
"redis2.conn_max_lifetime_seconds": 900,
}, "redis2.")
want := Settings{
MaxRetries: 5,
MinRetryBackoff: 10 * time.Millisecond,
MaxRetryBackoff: 500 * time.Millisecond,
DialTimeout: 2 * time.Second,
ReadTimeout: 1500 * time.Millisecond,
WriteTimeout: 1500 * time.Millisecond,
PoolSize: 64,
PoolTimeout: 2500 * time.Millisecond,
MinIdleConns: 4,
MaxIdleConns: 16,
ConnMaxIdleTime: 5 * time.Minute,
ConnMaxLifetime: 15 * time.Minute,
}
if settings != want {
t.Fatalf("got %+v, want %+v", settings, want)
}
options := &redis.Options{Addr: "localhost:6379"}
settings.ApplyTo(options)
if options.ReadTimeout != want.ReadTimeout || options.PoolSize != want.PoolSize || options.ConnMaxLifetime != want.ConnMaxLifetime {
t.Fatalf("options not applied: %+v", options)
}
clusterOptions := &redis.ClusterOptions{Addrs: []string{"localhost:6379"}}
settings.ApplyToCluster(clusterOptions)
if clusterOptions.ReadTimeout != want.ReadTimeout || clusterOptions.PoolSize != want.PoolSize {
t.Fatalf("cluster options not applied: %+v", clusterOptions)
}
}
func TestNegativeDurationIsIgnored(t *testing.T) {
settings := Read(fakeConfiguration{
"redis2.read_timeout_millisecond": -1,
"redis2.conn_max_lifetime_seconds": -1,
"redis2.max_retries": -1,
}, "redis2.")
if settings.ReadTimeout != 0 || settings.ConnMaxLifetime != 0 {
t.Fatalf("negative durations should keep the go-redis default, got %+v", settings)
}
// -1 is how go-redis spells "no retry", so it has to survive
if settings.MaxRetries != -1 {
t.Fatalf("max retries %d, want -1", settings.MaxRetries)
}
}