feat: implement Redis cluster support #109

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://<user>:<password>@<host>:<port>/<db>?<query-params>`. Examples:
 
 - redis://localhost:6379/0
@@ -33,13 +38,48 @@ The URL follows the Go Redis library format: `redis://<user>:<password>@<host>:<
 
 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://<user>:<password>@<host>:<port>/<db>?<query-params>`
-- `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`

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 {

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 {

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) {

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

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

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 {

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,

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
 }