Merge branch 'master' into jatin/add-custom-attrs

Author: urdarinda

error.go

@@ -38,6 +38,15 @@ type Error interface {
 
 var _ Error = proto.RedisError("")
 
+func isContextError(err error) bool {
+	switch err {
+	case context.Canceled, context.DeadlineExceeded:
+		return true
+	default:
+		return false
+	}
+}
+
 func shouldRetry(err error, retryTimeout bool) bool {
 	switch err {
 	case io.EOF, io.ErrUnexpectedEOF:

osscluster.go

@@ -1350,7 +1350,9 @@ func (c *ClusterClient) processPipelineNode(
 	_ = node.Client.withProcessPipelineHook(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
 		cn, err := node.Client.getConn(ctx)
 		if err != nil {
-			node.MarkAsFailing()
+			if !isContextError(err) {
+				node.MarkAsFailing()
+			}
 			_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 			setCmdsErr(cmds, err)
 			return err

osscluster_test.go

@@ -539,6 +539,39 @@ var _ = Describe("ClusterClient", func() {
 				AfterEach(func() {})
 
 				assertPipeline()
+
+				It("doesn't fail node with context.Canceled error", func() {
+					ctx, cancel := context.WithCancel(context.Background())
+					cancel()
+					pipe.Set(ctx, "A", "A_value", 0)
+					_, err := pipe.Exec(ctx)
+
+					Expect(err).To(HaveOccurred())
+					Expect(errors.Is(err, context.Canceled)).To(BeTrue())
+
+					clientNodes, _ := client.Nodes(ctx, "A")
+
+					for _, node := range clientNodes {
+						Expect(node.Failing()).To(BeFalse())
+					}
+				})
+
+				It("doesn't fail node with context.DeadlineExceeded error", func() {
+					ctx, cancel := context.WithTimeout(context.Background(), 1*time.Nanosecond)
+					defer cancel()
+
+					pipe.Set(ctx, "A", "A_value", 0)
+					_, err := pipe.Exec(ctx)
+
+					Expect(err).To(HaveOccurred())
+					Expect(errors.Is(err, context.DeadlineExceeded)).To(BeTrue())
+
+					clientNodes, _ := client.Nodes(ctx, "A")
+
+					for _, node := range clientNodes {
+						Expect(node.Failing()).To(BeFalse())
+					}
+				})
 			})
 
 			Describe("with TxPipeline", func() {

universal.go

@@ -163,6 +163,8 @@ func (o *UniversalOptions) Failover() *FailoverOptions {
 
 		TLSConfig: o.TLSConfig,
 
+		ReplicaOnly:     o.ReadOnly,
+
 		DisableIndentity: o.DisableIndentity,
 		IdentitySuffix:   o.IdentitySuffix,
 		UnstableResp3:    o.UnstableResp3,

universal_test.go

@@ -60,6 +60,25 @@ var _ = Describe("UniversalClient", func() {
 		Expect(a).ToNot(Panic())
 	})
 
+	It("should connect to failover servers on slaves when readonly Options is ok", Label("NonRedisEnterprise"), func() {
+		client = redis.NewUniversalClient(&redis.UniversalOptions{
+			MasterName: sentinelName,
+			Addrs:      sentinelAddrs,
+			ReadOnly:   true,
+		})
+		Expect(client.Ping(ctx).Err()).NotTo(HaveOccurred())
+
+		roleCmd := client.Do(ctx, "ROLE")
+		role, err := roleCmd.Result()
+		Expect(err).NotTo(HaveOccurred())
+
+		roleSlice, ok := role.([]interface{})
+		Expect(ok).To(BeTrue())
+		Expect(roleSlice[0]).To(Equal("slave"))
+
+		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]},
@@ -77,3 +96,4 @@ var _ = Describe("UniversalClient", func() {
 		Expect(client.ClusterSlots(ctx).Val()).To(HaveLen(3))
 	})
 })
+