Merge pull request #33 from KyberNetwork/ft/redis-failovercluster

feat: redis client to route randomly for sentinel redis nodes by default

Author: NgoKimPhu

pkg/client/backoff.go

@@ -4,6 +4,7 @@ import (
 	"github.com/cenkalti/backoff/v4"
 )
 
+// BackoffCfg is a hotcfg to create a backoff.ExponentialBackOff
 type BackoffCfg struct {
 	backoff.ExponentialBackOff `mapstructure:",squash"`
 	MaxRetries                 uint64

pkg/client/eth.go

@@ -18,6 +18,8 @@ import (
 
 const EthCloseDelay = time.Minute
 
+// EthCfg is hotcfg for eth client. It creates a client that automatically choose to use
+// the provided full node rpc or archive node.
 type EthCfg struct {
 	Url        string
 	ArchiveUrl string

pkg/client/eth_batchable.go

@@ -12,6 +12,8 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
+// BatchableEthCfg is hotcfg for batchable eth client.
+// It batches eth_call's up to be sent together within 1 request to the rpc node.
 type BatchableEthCfg struct {
 	EthCfg    `mapstructure:",squash"`
 	BatchRate time.Duration

pkg/client/grpc.go

@@ -12,6 +12,9 @@ import (
 
 const GrpcCloseDelay = time.Minute
 
+// GrpcCfg is hotcfg for grpc client. On update, it
+// creates a new grpc client with the provided factory generated by grpc using service proto.
+// The client has interceptors for adding client id header, validating requests, adding timeout, metrics and tracing.
 type GrpcCfg[T any] struct {
 	grpcclient.Config `mapstructure:",squash"`
 	C                 T // the inner grpc client

pkg/client/grpcclient/client.go

@@ -28,7 +28,7 @@ type Config struct {
 	BaseURL           string
 	MinConnectTimeout time.Duration
 	ConnectBackoff    backoff.Config
-	IsBlockConnect    bool
+	IsBlockConnect    bool // deprecated: see grpc.WithBlock
 	GRPCCredentials   credentials.TransportCredentials
 	Insecure          bool
 	Compression       Compression
@@ -66,7 +66,7 @@ func New[T any](clientFactory func(grpc.ClientConnInterface) T, applyOptions ...
 	}
 
 	dialOpts := cfg.dialOptions()
-	grpcConn, err := grpc.Dial(cfg.BaseURL, dialOpts...)
+	grpcConn, err := grpc.NewClient(cfg.BaseURL, dialOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -123,10 +123,6 @@ func (c *Config) dialOptions() []grpc.DialOption {
 	}
 	dialOpts = append(dialOpts, grpc.WithConnectParams(connectParams))
 
-	if c.IsBlockConnect {
-		dialOpts = append(dialOpts, grpc.WithBlock())
-	}
-
 	requestHeaders := c.requestHeaders()
 	unaryInterceptors := []grpc.UnaryClientInterceptor{
 		validator.UnaryClientInterceptor(),

pkg/client/grpcclient/options.go

@@ -64,6 +64,8 @@ func WithCompression(compression Compression) ApplyOption {
 	}
 }
 
+// WithBlock is a no-op.
+// deprecated: see grpc.WithBlock
 func WithBlock() ApplyOption {
 	return func(c *Config) {
 		c.IsBlockConnect = true

pkg/client/http.go

@@ -7,6 +7,9 @@ import (
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
+// HttpCfg is hotcfg for a resty http client. On update, it
+// creates a new resty client with the new config
+// as well as instruments the client for metrics and tracing.
 type HttpCfg struct {
 	kutils.HttpCfg `mapstructure:",squash"`
 	C              *resty.Client

pkg/client/redis.go

@@ -13,8 +13,12 @@ import (
 
 const RedisCloseDelay = time.Minute
 
+// RedisCfg is hotcfg for redis client. On update, it
+// creates FailoverClusterClient with RouteRandomly for sentinel redis by default
+// as well as instrumenting the client for metrics and tracing.
 type RedisCfg struct {
 	redis.UniversalOptions `mapstructure:",squash"`
+	DisableRouteRandomly   bool
 	C                      redis.UniversalClient
 }
 
@@ -28,7 +32,8 @@ func (*RedisCfg) OnUpdate(old, new *RedisCfg) {
 			}
 		})
 	}
-	new.C = redis.NewUniversalClient(&new.UniversalOptions)
+	new.RouteRandomly = new.RouteRandomly || !new.DisableRouteRandomly
+	new.C = newRedisClient(&new.UniversalOptions)
 	if metric.Provider() != nil {
 		if err := redisotel.InstrumentMetrics(new.C); err != nil {
 			klog.Errorf(ctx, "RedisCfg.OnUpdate|redisotel.InstrumentMetrics failed|err=%v", err)
@@ -40,3 +45,13 @@ func (*RedisCfg) OnUpdate(old, new *RedisCfg) {
 		}
 	}
 }
+
+func newRedisClient(opts *redis.UniversalOptions) redis.UniversalClient {
+	if opts.MasterName == "" {
+		return redis.NewUniversalClient(opts)
+	}
+	failoverOpts := opts.Failover()
+	failoverOpts.RouteByLatency = opts.RouteByLatency
+	failoverOpts.RouteRandomly = opts.RouteRandomly
+	return redis.NewFailoverClusterClient(failoverOpts)
+}