From f4347d4b7d729e44ae2d2e0e08bb1c2a9c0e6052 Mon Sep 17 00:00:00 2001 From: sxwebdev Date: Thu, 15 Jan 2026 19:53:09 +0800 Subject: [PATCH] feat: implement Redis cluster support with configuration and validation updates --- docs/nodecore/07-app-storages.md | 46 +++++++++++- internal/caches/redis_connector.go | 2 +- internal/config/config.go | 8 ++ internal/config/config_test.go | 26 ++++++- .../cache-redis-cluster-and-address.yaml | 26 +++++++ .../configs/cache/cache-redis-cluster.yaml | 42 +++++++++++ internal/config/storages_validation.go | 10 ++- internal/ratelimiter/redis_engine.go | 4 +- internal/storages/redis_storage.go | 73 +++++++++++++++++-- 9 files changed, 222 insertions(+), 15 deletions(-) create mode 100644 internal/config/configs/cache/cache-redis-cluster-and-address.yaml create mode 100644 internal/config/configs/cache/cache-redis-cluster.yaml diff --git a/docs/nodecore/07-app-storages.md b/docs/nodecore/07-app-storages.md index dcab9f10..35efc1c2 100644 --- a/docs/nodecore/07-app-storages.md +++ b/docs/nodecore/07-app-storages.md @@ -24,7 +24,12 @@ app-storages: ## Redis Storage -All connection parameters can be specified using the `full-url` field. +Redis storage supports two modes: **single instance** and **cluster mode**. + +### Single Instance Mode + +For connecting to a single Redis server, use either `full-url` or `address`. + The URL follows the Go Redis library format: `redis://:@:/?`. Examples: - redis://localhost:6379/0 @@ -33,13 +38,48 @@ The URL follows the Go Redis library format: `redis://:@:< Any parameters defined explicitly under `redis` (e.g. `timeouts`, `pool`) will override the corresponding values in full-url. +### Cluster Mode + +For connecting to a Redis Cluster, use the `cluster` section with a list of cluster node addresses. + +```yaml +app-storages: + - name: redis-cluster + redis: + cluster: + addresses: + - node1.redis.example.com:6379 + - node2.redis.example.com:6379 + - node3.redis.example.com:6379 + route-by-latency: true + password: mypassword + timeouts: + connect-timeout: 1s + pool: + size: 50 +``` + +> **Note**: Cluster mode and single instance mode are mutually exclusive. You cannot use `cluster.addresses` together with `address` or `full-url`. + ### Fields +**Single Instance Mode:** + - `full-url` - Full connection URL in Go Redis format — `redis://:@:/?` -- `address` - Host and port of the Redis instance. Either `full-url` or `address` must be specified +- `address` - Host and port of the Redis instance. Either `full-url` or `address` must be specified for single instance mode +- `db` - Database index. **_Default_**: `0` + +**Cluster Mode:** + +- `cluster.addresses` - List of Redis cluster node addresses (host:port). At least one address is required for cluster mode +- `cluster.route-by-latency` - Route read commands to the node with the lowest latency. **_Default_**: `false` +- `cluster.route-randomly` - Route read commands to random nodes. **_Default_**: `false` +- `cluster.read-only` - Enable read-only mode for replica nodes. **_Default_**: `false` + +**Common Fields (both modes):** + - `username` - Optional username for Redis authentication - `password` - Password for Redis authentication -- `db` - Database index. **_Default_**: `0` - `timeouts.connect-timeout` - Maximum duration for establishing a connection to the Redis server. **_Default_**: `500ms` - `timeouts.read-timeout` - Timeout for reading a response from Redis. **_Default_**: `200ms` - `timeouts.write-timeout` - Timeout for writing data to Redis. **_Default_**: `200ms` diff --git a/internal/caches/redis_connector.go b/internal/caches/redis_connector.go index f02185e4..a5615218 100644 --- a/internal/caches/redis_connector.go +++ b/internal/caches/redis_connector.go @@ -15,7 +15,7 @@ const cacheKeyPrefix = "nodecore:entry:" type RedisConnector struct { id string - client *redis.Client + client redis.UniversalClient } func (r *RedisConnector) Initialize() error { diff --git a/internal/config/config.go b/internal/config/config.go index 22e09a6c..0fa94489 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -54,6 +54,14 @@ type RedisStorageConfig struct { DB *int `yaml:"db"` Timeouts *RedisStorageTimeoutsConfig `yaml:"timeouts"` Pool *RedisStoragePoolConfig `yaml:"pool"` + Cluster *RedisClusterConfig `yaml:"cluster"` +} + +type RedisClusterConfig struct { + Addresses []string `yaml:"addresses"` + RouteByLatency bool `yaml:"route-by-latency"` + RouteRandomly bool `yaml:"route-randomly"` + ReadOnly bool `yaml:"read-only"` } type RedisStorageTimeoutsConfig struct { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 994b1676..be5e7677 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -873,7 +873,31 @@ func TestRedisFullCustom(t *testing.T) { func TestRedisMissingAddressThenError(t *testing.T) { t.Setenv(config.ConfigPathVar, "configs/cache/cache-redis-missing-address.yaml") _, err := config.NewAppConfig() - assert.ErrorContains(t, err, "error during redis storage config validation, cause: either 'address' or 'full_url' must be specified") + assert.ErrorContains(t, err, "error during redis storage config validation, cause: either 'address', 'full-url', or 'cluster.addresses' must be specified") +} + +func TestRedisClusterConfig(t *testing.T) { + t.Setenv(config.ConfigPathVar, "configs/cache/cache-redis-cluster.yaml") + appCfg, err := config.NewAppConfig() + require.NoError(t, err) + + redisStorage := appCfg.AppStorages[0].Redis + require.NotNil(t, redisStorage.Cluster) + assert.Equal(t, []string{ + "node1.redis.local:6379", + "node2.redis.local:6379", + "node3.redis.local:6379", + }, redisStorage.Cluster.Addresses) + assert.True(t, redisStorage.Cluster.RouteByLatency) + assert.False(t, redisStorage.Cluster.RouteRandomly) + assert.False(t, redisStorage.Cluster.ReadOnly) + assert.Equal(t, "cluster-password", redisStorage.Password) +} + +func TestRedisClusterAndAddressThenError(t *testing.T) { + t.Setenv(config.ConfigPathVar, "configs/cache/cache-redis-cluster-and-address.yaml") + _, err := config.NewAppConfig() + assert.ErrorContains(t, err, "error during redis storage config validation, cause: cannot use both cluster mode (cluster.addresses) and single mode (address/full-url) at the same time") } func TestRedisNegativeReadTimeoutThenError(t *testing.T) { diff --git a/internal/config/configs/cache/cache-redis-cluster-and-address.yaml b/internal/config/configs/cache/cache-redis-cluster-and-address.yaml new file mode 100644 index 00000000..95c068f9 --- /dev/null +++ b/internal/config/configs/cache/cache-redis-cluster-and-address.yaml @@ -0,0 +1,26 @@ +server: + port: 9095 + +app-storages: + - name: redis-invalid + redis: + address: localhost:6379 + cluster: + addresses: + - node1.redis.local:6379 + - node2.redis.local:6379 + +cache: + connectors: + - driver: redis + id: redis1 + redis: + storage-name: redis-invalid + +upstream-config: + upstreams: + - id: eth-upstream + chain: ethereum + connectors: + - type: json-rpc + url: https://test.com diff --git a/internal/config/configs/cache/cache-redis-cluster.yaml b/internal/config/configs/cache/cache-redis-cluster.yaml new file mode 100644 index 00000000..cda0ad0f --- /dev/null +++ b/internal/config/configs/cache/cache-redis-cluster.yaml @@ -0,0 +1,42 @@ +server: + port: 9095 + +app-storages: + - name: redis-cluster-storage + redis: + cluster: + addresses: + - node1.redis.local:6379 + - node2.redis.local:6379 + - node3.redis.local:6379 + route-by-latency: true + route-randomly: false + read-only: false + password: cluster-password + timeouts: + connect-timeout: 1s + read-timeout: 500ms + write-timeout: 500ms + pool: + size: 100 + pool-timeout: 5s + min-idle-conns: 10 + max-idle-conns: 50 + max-active-conns: 200 + conn-max-idle-time: 10m + conn-max-life-time: 1h + +cache: + connectors: + - driver: redis + id: redis-cluster + redis: + storage-name: redis-cluster-storage + +upstream-config: + upstreams: + - id: eth-upstream + chain: ethereum + connectors: + - type: json-rpc + url: https://test.com diff --git a/internal/config/storages_validation.go b/internal/config/storages_validation.go index 155d1fff..41911ef7 100644 --- a/internal/config/storages_validation.go +++ b/internal/config/storages_validation.go @@ -26,8 +26,14 @@ func (a *AppStorageConfig) validate() (string, error) { } func (r *RedisStorageConfig) validate() error { - if r.FullUrl == "" && r.Address == "" { - return errors.New("either 'address' or 'full_url' must be specified") + isClusterMode := r.Cluster != nil && len(r.Cluster.Addresses) > 0 + isSingleMode := r.FullUrl != "" || r.Address != "" + + if !isClusterMode && !isSingleMode { + return errors.New("either 'address', 'full-url', or 'cluster.addresses' must be specified") + } + if isClusterMode && isSingleMode { + return errors.New("cannot use both cluster mode (cluster.addresses) and single mode (address/full-url) at the same time") } if r.Timeouts != nil { if r.Timeouts.ReadTimeout != nil && *r.Timeouts.ReadTimeout < 0 { diff --git a/internal/ratelimiter/redis_engine.go b/internal/ratelimiter/redis_engine.go index d76a6a7d..176f4398 100644 --- a/internal/ratelimiter/redis_engine.go +++ b/internal/ratelimiter/redis_engine.go @@ -8,10 +8,10 @@ import ( type RateLimitRedisEngine struct { name string - redis *redis.Client + redis redis.UniversalClient } -func NewRateLimitRedisEngine(name string, redis *redis.Client) *RateLimitRedisEngine { +func NewRateLimitRedisEngine(name string, redis redis.UniversalClient) *RateLimitRedisEngine { return &RateLimitRedisEngine{ name: name, redis: redis, diff --git a/internal/storages/redis_storage.go b/internal/storages/redis_storage.go index b23d9a82..571850a2 100644 --- a/internal/storages/redis_storage.go +++ b/internal/storages/redis_storage.go @@ -9,7 +9,7 @@ import ( ) type RedisStorage struct { - Redis *redis.Client + Redis redis.UniversalClient name string } @@ -18,6 +18,71 @@ func (r *RedisStorage) storage() string { } func NewRedisStorage(name string, redisConfig *config.RedisStorageConfig) (*RedisStorage, error) { + var client redis.UniversalClient + + if redisConfig.Cluster != nil && len(redisConfig.Cluster.Addresses) > 0 { + client = newRedisClusterClient(redisConfig) + } else { + var err error + client, err = newRedisSingleClient(name, redisConfig) + if err != nil { + return nil, err + } + } + + return &RedisStorage{ + Redis: client, + name: name, + }, nil +} + +func newRedisClusterClient(redisConfig *config.RedisStorageConfig) *redis.ClusterClient { + clusterOptions := &redis.ClusterOptions{ + Addrs: redisConfig.Cluster.Addresses, + RouteByLatency: redisConfig.Cluster.RouteByLatency, + RouteRandomly: redisConfig.Cluster.RouteRandomly, + ReadOnly: redisConfig.Cluster.ReadOnly, + } + + if redisConfig.Username != "" { + clusterOptions.Username = redisConfig.Username + } + if redisConfig.Password != "" { + clusterOptions.Password = redisConfig.Password + } + + if redisConfig.Timeouts != nil { + if redisConfig.Timeouts.ConnectTimeout != nil { + clusterOptions.DialTimeout = *redisConfig.Timeouts.ConnectTimeout + } + if redisConfig.Timeouts.ReadTimeout != nil { + clusterOptions.ReadTimeout = lo.Ternary(*redisConfig.Timeouts.ReadTimeout == 0, -1, *redisConfig.Timeouts.ReadTimeout) + } + if redisConfig.Timeouts.WriteTimeout != nil { + clusterOptions.WriteTimeout = lo.Ternary(*redisConfig.Timeouts.WriteTimeout == 0, -1, *redisConfig.Timeouts.WriteTimeout) + } + } + + if redisConfig.Pool != nil { + clusterOptions.PoolSize = redisConfig.Pool.Size + if redisConfig.Pool.PoolTimeout != nil { + clusterOptions.PoolTimeout = *redisConfig.Pool.PoolTimeout + } + clusterOptions.MinIdleConns = redisConfig.Pool.MinIdleConns + clusterOptions.MaxIdleConns = redisConfig.Pool.MaxIdleConns + clusterOptions.MaxActiveConns = redisConfig.Pool.MaxActiveConns + if redisConfig.Pool.ConnMaxIdleTime != nil { + clusterOptions.ConnMaxIdleTime = *redisConfig.Pool.ConnMaxIdleTime + } + if redisConfig.Pool.ConnMaxLifeTime != nil { + clusterOptions.ConnMaxLifetime = *redisConfig.Pool.ConnMaxLifeTime + } + } + + return redis.NewClusterClient(clusterOptions) +} + +func newRedisSingleClient(name string, redisConfig *config.RedisStorageConfig) (*redis.Client, error) { options := &redis.Options{} var err error if redisConfig.FullUrl != "" { @@ -76,9 +141,5 @@ func NewRedisStorage(name string, redisConfig *config.RedisStorageConfig) (*Redi } } - client := redis.NewClient(options) - return &RedisStorage{ - Redis: client, - name: name, - }, nil + return redis.NewClient(options), nil }