Merge branch 'master' into do_cmd_for_simple_conn

Author: ndyakov

.github/workflows/golangci-lint.yml

@@ -22,3 +22,6 @@ jobs:
       - uses: actions/checkout@v4
       - name: golangci-lint
         uses: golangci/[email protected]
+        with:
+          verify: false # disable verifying the configuration since golangci is currently introducing breaking changes in the configuration 
+

error.go

@@ -75,6 +75,9 @@ func shouldRetry(err error, retryTimeout bool) bool {
 	if strings.HasPrefix(s, "READONLY ") {
 		return true
 	}
+	if strings.HasPrefix(s, "MASTERDOWN ") {
+		return true
+	}
 	if strings.HasPrefix(s, "CLUSTERDOWN ") {
 		return true
 	}

internal/pool/bench_test.go

@@ -33,6 +33,7 @@ func BenchmarkPoolGetPut(b *testing.B) {
 				Dialer:          dummyDialer,
 				PoolSize:        bm.poolSize,
 				PoolTimeout:     time.Second,
+				DialTimeout:     1 * time.Second,
 				ConnMaxIdleTime: time.Hour,
 			})
 
@@ -76,6 +77,7 @@ func BenchmarkPoolGetRemove(b *testing.B) {
 				Dialer:          dummyDialer,
 				PoolSize:        bm.poolSize,
 				PoolTimeout:     time.Second,
+				DialTimeout:     1 * time.Second,
 				ConnMaxIdleTime: time.Hour,
 			})
 

internal/pool/pool.go

@@ -62,6 +62,7 @@ type Options struct {
 
 	PoolFIFO        bool
 	PoolSize        int
+	DialTimeout     time.Duration
 	PoolTimeout     time.Duration
 	MinIdleConns    int
 	MaxIdleConns    int
@@ -140,7 +141,10 @@ func (p *ConnPool) checkMinIdleConns() {
 }
 
 func (p *ConnPool) addIdleConn() error {
-	cn, err := p.dialConn(context.TODO(), true)
+	ctx, cancel := context.WithTimeout(context.Background(), p.cfg.DialTimeout)
+	defer cancel()
+
+	cn, err := p.dialConn(ctx, true)
 	if err != nil {
 		return err
 	}
@@ -230,15 +234,19 @@ func (p *ConnPool) tryDial() {
 			return
 		}
 
-		conn, err := p.cfg.Dialer(context.Background())
+		ctx, cancel := context.WithTimeout(context.Background(), p.cfg.DialTimeout)
+
+		conn, err := p.cfg.Dialer(ctx)
 		if err != nil {
 			p.setLastDialError(err)
 			time.Sleep(time.Second)
+			cancel()
 			continue
 		}
 
 		atomic.StoreUint32(&p.dialErrorsNum, 0)
 		_ = conn.Close()
+		cancel()
 		return
 	}
 }

internal/pool/pool_test.go

@@ -22,6 +22,7 @@ var _ = Describe("ConnPool", func() {
 			Dialer:          dummyDialer,
 			PoolSize:        10,
 			PoolTimeout:     time.Hour,
+			DialTimeout:     1 * time.Second,
 			ConnMaxIdleTime: time.Millisecond,
 		})
 	})
@@ -46,6 +47,7 @@ var _ = Describe("ConnPool", func() {
 			},
 			PoolSize:        10,
 			PoolTimeout:     time.Hour,
+			DialTimeout:     1 * time.Second,
 			ConnMaxIdleTime: time.Millisecond,
 			MinIdleConns:    minIdleConns,
 		})
@@ -129,6 +131,7 @@ var _ = Describe("MinIdleConns", func() {
 			PoolSize:        poolSize,
 			MinIdleConns:    minIdleConns,
 			PoolTimeout:     100 * time.Millisecond,
+			DialTimeout:     1 * time.Second,
 			ConnMaxIdleTime: -1,
 		})
 		Eventually(func() int {
@@ -306,6 +309,7 @@ var _ = Describe("race", func() {
 			Dialer:          dummyDialer,
 			PoolSize:        10,
 			PoolTimeout:     time.Minute,
+			DialTimeout:     1 * time.Second,
 			ConnMaxIdleTime: time.Millisecond,
 		})
 
@@ -336,6 +340,7 @@ var _ = Describe("race", func() {
 			PoolSize:     1000,
 			MinIdleConns: 50,
 			PoolTimeout:  3 * time.Second,
+			DialTimeout:  1 * time.Second,
 		}
 		p := pool.NewConnPool(opt)
 

options.go

@@ -531,6 +531,7 @@ func newConnPool(
 		PoolFIFO:        opt.PoolFIFO,
 		PoolSize:        opt.PoolSize,
 		PoolTimeout:     opt.PoolTimeout,
+		DialTimeout:     opt.DialTimeout,
 		MinIdleConns:    opt.MinIdleConns,
 		MaxIdleConns:    opt.MaxIdleConns,
 		MaxActiveConns:  opt.MaxActiveConns,

universal.go

@@ -154,6 +154,9 @@ func (o *UniversalOptions) Failover() *FailoverOptions {
 		SentinelUsername: o.SentinelUsername,
 		SentinelPassword: o.SentinelPassword,
 
+		RouteByLatency: o.RouteByLatency,
+		RouteRandomly:  o.RouteRandomly,
+
 		MaxRetries:      o.MaxRetries,
 		MinRetryBackoff: o.MinRetryBackoff,
 		MaxRetryBackoff: o.MaxRetryBackoff,
@@ -256,14 +259,22 @@ var (
 // NewUniversalClient returns a new multi client. The type of the returned client depends
 // on the following conditions:
 //
-// 1. If the MasterName option is specified, a sentinel-backed FailoverClient is returned.
-// 2. if the number of Addrs is two or more, a ClusterClient is returned.
-// 3. Otherwise, a single-node Client is returned.
+// 1. If the MasterName option is specified with RouteByLatency, RouteRandomly or IsClusterMode,
+//    a FailoverClusterClient is returned.
+// 2. If the MasterName option is specified without RouteByLatency, RouteRandomly or IsClusterMode,
+//    a sentinel-backed FailoverClient is returned.
+// 3. If the number of Addrs is two or more, or IsClusterMode option is specified,
+//    a ClusterClient is returned.
+// 4. Otherwise, a single-node Client is returned.
 func NewUniversalClient(opts *UniversalOptions) UniversalClient {
-	if opts.MasterName != "" {
+	switch {
+	case opts.MasterName != "" && (opts.RouteByLatency || opts.RouteRandomly || opts.IsClusterMode):
+		return NewFailoverClusterClient(opts.Failover())
+	case opts.MasterName != "":
 		return NewFailoverClient(opts.Failover())
-	} else if len(opts.Addrs) > 1 || opts.IsClusterMode {
+	case len(opts.Addrs) > 1 || opts.IsClusterMode:
 		return NewClusterClient(opts.Cluster())
+	default:
+		return NewClient(opts.Simple())
 	}
-	return NewClient(opts.Simple())
 }

universal_test.go

@@ -24,6 +24,16 @@ var _ = Describe("UniversalClient", func() {
 		Expect(client.Ping(ctx).Err()).NotTo(HaveOccurred())
 	})
 
+	It("should connect to failover cluster", Label("NonRedisEnterprise"), func() {
+		client = redis.NewUniversalClient(&redis.UniversalOptions{
+			MasterName:    sentinelName,
+			RouteRandomly: true,
+			Addrs:         sentinelAddrs,
+		})
+		_, ok := client.(*redis.ClusterClient)
+		Expect(ok).To(BeTrue(), "expected a ClusterClient")
+	})
+
 	It("should connect to simple servers", func() {
 		client = redis.NewUniversalClient(&redis.UniversalOptions{
 			Addrs: []string{redisAddr},
@@ -79,6 +89,7 @@ var _ = Describe("UniversalClient", func() {
 		err = client.Set(ctx, "somekey", "somevalue", 0).Err()
 		Expect(err).To(HaveOccurred())
 	})
+
 	It("should connect to clusters if IsClusterMode is set even if only a single address is provided", Label("NonRedisEnterprise"), func() {
 		client = redis.NewUniversalClient(&redis.UniversalOptions{
 			Addrs:         []string{cluster.addrs()[0]},
@@ -96,4 +107,3 @@ var _ = Describe("UniversalClient", func() {
 		Expect(client.ClusterSlots(ctx).Val()).To(HaveLen(3))
 	})
 })
-