redis/cluster_test.go

1590 lines
41 KiB
Go
Raw Permalink Normal View History

2015-01-24 15:12:48 +03:00
package redis_test
import (
2019-07-04 11:18:06 +03:00
"context"
"crypto/tls"
"errors"
2015-11-14 16:54:16 +03:00
"fmt"
"net"
2015-12-16 17:11:52 +03:00
"strconv"
2015-11-14 16:54:16 +03:00
"strings"
2015-12-16 17:11:52 +03:00
"sync"
2015-03-18 13:41:24 +03:00
"time"
2015-01-24 15:12:48 +03:00
2023-01-27 18:00:49 +03:00
. "github.com/bsm/ginkgo/v2"
. "github.com/bsm/gomega"
2023-01-23 09:48:54 +03:00
"github.com/redis/go-redis/v9"
"github.com/redis/go-redis/v9/internal/hashtag"
2015-01-24 15:12:48 +03:00
)
type clusterScenario struct {
ports []string
2019-07-25 13:53:00 +03:00
nodeIDs []string
processes map[string]*redisProcess
clients map[string]*redis.Client
}
func (s *clusterScenario) masters() []*redis.Client {
result := make([]*redis.Client, 3)
for pos, port := range s.ports[:3] {
result[pos] = s.clients[port]
2015-01-24 15:12:48 +03:00
}
return result
}
2015-01-24 15:12:48 +03:00
func (s *clusterScenario) slaves() []*redis.Client {
result := make([]*redis.Client, 3)
for pos, port := range s.ports[3:] {
result[pos] = s.clients[port]
}
return result
}
2015-01-24 15:12:48 +03:00
2017-02-17 13:12:06 +03:00
func (s *clusterScenario) addrs() []string {
addrs := make([]string, len(s.ports))
for i, port := range s.ports {
addrs[i] = net.JoinHostPort("127.0.0.1", port)
}
2017-02-17 13:12:06 +03:00
return addrs
}
2020-09-05 11:34:37 +03:00
func (s *clusterScenario) newClusterClientUnstable(opt *redis.ClusterOptions) *redis.ClusterClient {
2017-02-17 13:12:06 +03:00
opt.Addrs = s.addrs()
return redis.NewClusterClient(opt)
}
2020-03-11 17:26:42 +03:00
func (s *clusterScenario) newClusterClient(
ctx context.Context, opt *redis.ClusterOptions,
) *redis.ClusterClient {
2020-09-05 11:34:37 +03:00
client := s.newClusterClientUnstable(opt)
2018-07-22 10:50:26 +03:00
err := eventually(func() error {
if opt.ClusterSlots != nil {
return nil
}
2020-03-11 17:26:42 +03:00
state, err := client.LoadState(ctx)
2018-05-17 16:09:56 +03:00
if err != nil {
2018-07-22 10:50:26 +03:00
return err
}
2020-03-11 17:26:42 +03:00
if !state.IsConsistent(ctx) {
2018-11-24 14:16:21 +03:00
return fmt.Errorf("cluster state is not consistent")
}
2018-07-22 10:50:26 +03:00
return nil
}, 30*time.Second)
if err != nil {
panic(err)
}
2018-05-17 16:09:56 +03:00
return client
}
2015-01-24 15:12:48 +03:00
2020-09-09 17:39:13 +03:00
func (s *clusterScenario) Close() error {
for _, port := range s.ports {
if process, ok := processes[port]; ok {
process.Close()
delete(processes, port)
}
2020-09-09 17:39:13 +03:00
}
return nil
}
2020-03-11 17:26:42 +03:00
func startCluster(ctx context.Context, scenario *clusterScenario) error {
2015-11-14 16:54:16 +03:00
// Start processes and collect node ids
for pos, port := range scenario.ports {
process, err := startRedis(port, "--cluster-enabled", "yes")
if err != nil {
return err
2015-01-24 15:12:48 +03:00
}
client := redis.NewClient(&redis.Options{
Addr: ":" + port,
})
2020-03-11 17:26:42 +03:00
info, err := client.ClusterNodes(ctx).Result()
if err != nil {
return err
2015-01-24 15:12:48 +03:00
}
scenario.processes[port] = process
scenario.clients[port] = client
2019-07-25 13:53:00 +03:00
scenario.nodeIDs[pos] = info[:40]
}
2017-07-09 13:10:07 +03:00
// Meet cluster nodes.
for _, client := range scenario.clients {
2020-03-11 17:26:42 +03:00
err := client.ClusterMeet(ctx, "127.0.0.1", scenario.ports[0]).Err()
if err != nil {
return err
2015-01-24 15:12:48 +03:00
}
}
2015-01-24 15:12:48 +03:00
2017-07-09 13:10:07 +03:00
// Bootstrap masters.
slots := []int{0, 5000, 10000, 16384}
2015-11-14 16:54:16 +03:00
for pos, master := range scenario.masters() {
2020-03-11 17:26:42 +03:00
err := master.ClusterAddSlotsRange(ctx, slots[pos], slots[pos+1]-1).Err()
if err != nil {
return err
}
}
2015-01-24 15:12:48 +03:00
2017-07-09 13:10:07 +03:00
// Bootstrap slaves.
2015-11-14 16:54:16 +03:00
for idx, slave := range scenario.slaves() {
2019-07-25 13:53:00 +03:00
masterID := scenario.nodeIDs[idx]
2015-11-14 16:54:16 +03:00
// Wait until master is available
err := eventually(func() error {
2020-03-11 17:26:42 +03:00
s := slave.ClusterNodes(ctx).Val()
2019-07-25 13:53:00 +03:00
wanted := masterID
2015-11-14 16:54:16 +03:00
if !strings.Contains(s, wanted) {
return fmt.Errorf("%q does not contain %q", s, wanted)
}
return nil
}, 10*time.Second)
if err != nil {
return err
2015-01-24 15:12:48 +03:00
}
2020-03-11 17:26:42 +03:00
err = slave.ClusterReplicate(ctx, masterID).Err()
if err != nil {
return err
2015-01-24 15:12:48 +03:00
}
}
2015-01-24 15:12:48 +03:00
2017-07-09 13:10:07 +03:00
// Wait until all nodes have consistent info.
2018-05-17 16:09:56 +03:00
wanted := []redis.ClusterSlot{{
Start: 0,
End: 4999,
Nodes: []redis.ClusterNode{{
2019-07-25 13:53:00 +03:00
ID: "",
2018-05-17 16:09:56 +03:00
Addr: "127.0.0.1:8220",
}, {
2019-07-25 13:53:00 +03:00
ID: "",
2018-05-17 16:09:56 +03:00
Addr: "127.0.0.1:8223",
}},
}, {
Start: 5000,
End: 9999,
Nodes: []redis.ClusterNode{{
2019-07-25 13:53:00 +03:00
ID: "",
2018-05-17 16:09:56 +03:00
Addr: "127.0.0.1:8221",
}, {
2019-07-25 13:53:00 +03:00
ID: "",
2018-05-17 16:09:56 +03:00
Addr: "127.0.0.1:8224",
}},
}, {
Start: 10000,
End: 16383,
Nodes: []redis.ClusterNode{{
2019-07-25 13:53:00 +03:00
ID: "",
2018-05-17 16:09:56 +03:00
Addr: "127.0.0.1:8222",
}, {
2019-07-25 13:53:00 +03:00
ID: "",
2018-05-17 16:09:56 +03:00
Addr: "127.0.0.1:8225",
}},
}}
for _, client := range scenario.clients {
2015-11-14 16:54:16 +03:00
err := eventually(func() error {
2020-03-11 17:26:42 +03:00
res, err := client.ClusterSlots(ctx).Result()
2015-11-22 15:44:38 +03:00
if err != nil {
return err
2015-11-21 14:16:13 +03:00
}
return assertSlotsEqual(res, wanted)
2016-03-14 17:51:46 +03:00
}, 30*time.Second)
if err != nil {
return err
}
}
return nil
}
func assertSlotsEqual(slots, wanted []redis.ClusterSlot) error {
2018-10-11 13:58:31 +03:00
outerLoop:
for _, s2 := range wanted {
for _, s1 := range slots {
if slotEqual(s1, s2) {
2018-10-11 13:58:31 +03:00
continue outerLoop
}
}
return fmt.Errorf("%v not found in %v", s2, slots)
}
return nil
}
func slotEqual(s1, s2 redis.ClusterSlot) bool {
if s1.Start != s2.Start {
return false
}
if s1.End != s2.End {
return false
}
2016-12-16 17:26:48 +03:00
if len(s1.Nodes) != len(s2.Nodes) {
return false
}
for i, n1 := range s1.Nodes {
if n1.Addr != s2.Nodes[i].Addr {
return false
}
}
return true
}
//------------------------------------------------------------------------------
var _ = Describe("ClusterClient", func() {
2018-05-17 16:09:56 +03:00
var failover bool
2016-12-16 17:26:48 +03:00
var opt *redis.ClusterOptions
var client *redis.ClusterClient
2015-01-24 15:12:48 +03:00
2016-12-16 17:26:48 +03:00
assertClusterClient := func() {
2015-01-24 15:12:48 +03:00
It("should GET/SET/DEL", func() {
2020-03-11 17:26:42 +03:00
err := client.Get(ctx, "A").Err()
2015-01-24 15:12:48 +03:00
Expect(err).To(Equal(redis.Nil))
2020-03-11 17:26:42 +03:00
err = client.Set(ctx, "A", "VALUE", 0).Err()
2015-01-24 15:12:48 +03:00
Expect(err).NotTo(HaveOccurred())
2017-07-09 13:10:07 +03:00
Eventually(func() string {
2020-03-11 17:26:42 +03:00
return client.Get(ctx, "A").Val()
2017-08-31 15:22:47 +03:00
}, 30*time.Second).Should(Equal("VALUE"))
2015-01-24 15:12:48 +03:00
2020-03-11 17:26:42 +03:00
cnt, err := client.Del(ctx, "A").Result()
2015-01-24 15:12:48 +03:00
Expect(err).NotTo(HaveOccurred())
Expect(cnt).To(Equal(int64(1)))
})
2018-05-17 16:09:56 +03:00
It("GET follows redirects", func() {
2020-03-11 17:26:42 +03:00
err := client.Set(ctx, "A", "VALUE", 0).Err()
2018-05-17 16:09:56 +03:00
Expect(err).NotTo(HaveOccurred())
2015-05-01 10:42:58 +03:00
2018-05-17 16:09:56 +03:00
if !failover {
Eventually(func() int64 {
2020-03-11 17:26:42 +03:00
nodes, err := client.Nodes(ctx, "A")
2018-05-17 16:09:56 +03:00
if err != nil {
return 0
}
2020-03-11 17:26:42 +03:00
return nodes[1].Client.DBSize(ctx).Val()
2018-05-17 16:09:56 +03:00
}, 30*time.Second).Should(Equal(int64(1)))
2015-01-24 15:12:48 +03:00
2018-05-17 16:09:56 +03:00
Eventually(func() error {
2020-03-11 17:26:42 +03:00
return client.SwapNodes(ctx, "A")
2018-05-17 16:09:56 +03:00
}, 30*time.Second).ShouldNot(HaveOccurred())
}
2020-03-11 17:26:42 +03:00
v, err := client.Get(ctx, "A").Result()
2018-05-17 16:09:56 +03:00
Expect(err).NotTo(HaveOccurred())
Expect(v).To(Equal("VALUE"))
})
It("SET follows redirects", func() {
if !failover {
Eventually(func() error {
2020-03-11 17:26:42 +03:00
return client.SwapNodes(ctx, "A")
2018-05-17 16:09:56 +03:00
}, 30*time.Second).ShouldNot(HaveOccurred())
}
2020-03-11 17:26:42 +03:00
err := client.Set(ctx, "A", "VALUE", 0).Err()
2018-05-17 16:09:56 +03:00
Expect(err).NotTo(HaveOccurred())
2020-03-11 17:26:42 +03:00
v, err := client.Get(ctx, "A").Result()
2018-05-17 16:09:56 +03:00
Expect(err).NotTo(HaveOccurred())
Expect(v).To(Equal("VALUE"))
})
2015-12-16 17:11:52 +03:00
2016-10-09 14:12:32 +03:00
It("distributes keys", func() {
for i := 0; i < 100; i++ {
2020-03-11 17:26:42 +03:00
err := client.Set(ctx, fmt.Sprintf("key%d", i), "value", 0).Err()
2016-10-09 14:12:32 +03:00
Expect(err).NotTo(HaveOccurred())
}
2020-03-11 17:26:42 +03:00
client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
2018-05-17 16:09:56 +03:00
defer GinkgoRecover()
2017-07-09 13:10:07 +03:00
Eventually(func() string {
2020-03-11 17:26:42 +03:00
return master.Info(ctx, "keyspace").Val()
2017-08-31 15:22:47 +03:00
}, 30*time.Second).Should(Or(
2017-07-09 13:10:07 +03:00
ContainSubstring("keys=31"),
ContainSubstring("keys=29"),
ContainSubstring("keys=40"),
))
2018-05-17 16:09:56 +03:00
return nil
})
2016-10-09 14:12:32 +03:00
})
It("distributes keys when using EVAL", func() {
script := redis.NewScript(`
local r = redis.call('SET', KEYS[1], ARGV[1])
return r
`)
var key string
for i := 0; i < 100; i++ {
key = fmt.Sprintf("key%d", i)
2020-03-11 17:26:42 +03:00
err := script.Run(ctx, client, []string{key}, "value").Err()
2016-10-09 14:12:32 +03:00
Expect(err).NotTo(HaveOccurred())
}
2020-03-11 17:26:42 +03:00
client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
2018-06-29 10:45:05 +03:00
defer GinkgoRecover()
2017-07-09 13:10:07 +03:00
Eventually(func() string {
2020-03-11 17:26:42 +03:00
return master.Info(ctx, "keyspace").Val()
2017-08-31 15:22:47 +03:00
}, 30*time.Second).Should(Or(
2017-07-09 13:10:07 +03:00
ContainSubstring("keys=31"),
ContainSubstring("keys=29"),
ContainSubstring("keys=40"),
))
2018-06-29 10:45:05 +03:00
return nil
})
2016-10-09 14:12:32 +03:00
})
It("distributes scripts when using Script Load", func() {
client.ScriptFlush(ctx)
script := redis.NewScript(`return 'Unique script'`)
script.Load(ctx, client)
client.ForEachShard(ctx, func(ctx context.Context, shard *redis.Client) error {
defer GinkgoRecover()
val, _ := script.Exists(ctx, shard).Result()
Expect(val[0]).To(Equal(true))
return nil
})
})
It("checks all shards when using Script Exists", func() {
client.ScriptFlush(ctx)
script := redis.NewScript(`return 'First script'`)
lostScriptSrc := `return 'Lost script'`
lostScript := redis.NewScript(lostScriptSrc)
script.Load(ctx, client)
client.Do(ctx, "script", "load", lostScriptSrc)
val, _ := client.ScriptExists(ctx, script.Hash(), lostScript.Hash()).Result()
Expect(val).To(Equal([]bool{true, false}))
})
It("flushes scripts from all shards when using ScriptFlush", func() {
script := redis.NewScript(`return 'Unnecessary script'`)
script.Load(ctx, client)
val, _ := client.ScriptExists(ctx, script.Hash()).Result()
Expect(val).To(Equal([]bool{true}))
client.ScriptFlush(ctx)
val, _ = client.ScriptExists(ctx, script.Hash()).Result()
Expect(val).To(Equal([]bool{false}))
})
2016-06-17 15:09:38 +03:00
It("supports Watch", func() {
2015-12-16 17:11:52 +03:00
var incr func(string) error
// Transactionally increments key using GET and SET commands.
incr = func(key string) error {
2020-03-11 17:26:42 +03:00
err := client.Watch(ctx, func(tx *redis.Tx) error {
n, err := tx.Get(ctx, key).Int64()
2016-05-02 15:54:15 +03:00
if err != nil && err != redis.Nil {
return err
}
2020-03-11 17:26:42 +03:00
_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
pipe.Set(ctx, key, strconv.FormatInt(n+1, 10), 0)
2016-05-02 15:54:15 +03:00
return nil
})
2015-12-16 17:11:52 +03:00
return err
2016-05-02 15:54:15 +03:00
}, key)
2015-12-16 17:11:52 +03:00
if err == redis.TxFailedErr {
return incr(key)
}
return err
}
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Add(1)
go func() {
2016-07-02 15:52:10 +03:00
defer GinkgoRecover()
2015-12-16 17:11:52 +03:00
defer wg.Done()
err := incr("key")
Expect(err).NotTo(HaveOccurred())
}()
}
wg.Wait()
2018-02-15 14:00:54 +03:00
Eventually(func() string {
2020-03-11 17:26:42 +03:00
return client.Get(ctx, "key").Val()
2018-02-15 14:00:54 +03:00
}, 30*time.Second).Should(Equal("100"))
2015-12-16 17:11:52 +03:00
})
2016-04-06 14:01:08 +03:00
2016-12-13 18:28:39 +03:00
Describe("pipelining", func() {
var pipe *redis.Pipeline
2016-04-06 14:01:08 +03:00
2016-12-13 18:28:39 +03:00
assertPipeline := func() {
2016-12-16 17:26:48 +03:00
keys := []string{"A", "B", "C", "D", "E", "F", "G"}
2016-04-06 14:01:08 +03:00
2016-12-16 17:26:48 +03:00
It("follows redirects", func() {
2018-05-17 16:09:56 +03:00
if !failover {
for _, key := range keys {
Eventually(func() error {
2020-03-11 17:26:42 +03:00
return client.SwapNodes(ctx, key)
2018-05-17 16:09:56 +03:00
}, 30*time.Second).ShouldNot(HaveOccurred())
}
2016-12-16 17:26:48 +03:00
}
2016-12-13 18:28:39 +03:00
for i, key := range keys {
2020-03-11 17:26:42 +03:00
pipe.Set(ctx, key, key+"_value", 0)
pipe.Expire(ctx, key, time.Duration(i+1)*time.Hour)
2016-12-13 18:28:39 +03:00
}
2020-03-11 17:26:42 +03:00
cmds, err := pipe.Exec(ctx)
2016-12-13 18:28:39 +03:00
Expect(err).NotTo(HaveOccurred())
Expect(cmds).To(HaveLen(14))
2020-06-10 15:04:12 +03:00
_ = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
defer GinkgoRecover()
Eventually(func() int64 {
2020-03-11 17:26:42 +03:00
return node.DBSize(ctx).Val()
}, 30*time.Second).ShouldNot(BeZero())
return nil
})
2018-05-17 16:09:56 +03:00
if !failover {
for _, key := range keys {
Eventually(func() error {
2020-03-11 17:26:42 +03:00
return client.SwapNodes(ctx, key)
2018-05-17 16:09:56 +03:00
}, 30*time.Second).ShouldNot(HaveOccurred())
}
2016-12-16 17:26:48 +03:00
}
2016-12-13 18:28:39 +03:00
for _, key := range keys {
2020-03-11 17:26:42 +03:00
pipe.Get(ctx, key)
pipe.TTL(ctx, key)
2016-12-13 18:28:39 +03:00
}
2020-03-11 17:26:42 +03:00
cmds, err = pipe.Exec(ctx)
2016-12-13 18:28:39 +03:00
Expect(err).NotTo(HaveOccurred())
Expect(cmds).To(HaveLen(14))
2016-12-16 17:26:48 +03:00
for i, key := range keys {
get := cmds[i*2].(*redis.StringCmd)
Expect(get.Val()).To(Equal(key + "_value"))
ttl := cmds[(i*2)+1].(*redis.DurationCmd)
2017-08-15 10:34:05 +03:00
dur := time.Duration(i+1) * time.Hour
2018-07-23 15:55:13 +03:00
Expect(ttl.Val()).To(BeNumerically("~", dur, 30*time.Second))
2016-12-16 17:26:48 +03:00
}
2016-12-13 18:28:39 +03:00
})
2016-04-06 14:01:08 +03:00
2016-12-13 18:28:39 +03:00
It("works with missing keys", func() {
2020-03-11 17:26:42 +03:00
pipe.Set(ctx, "A", "A_value", 0)
pipe.Set(ctx, "C", "C_value", 0)
_, err := pipe.Exec(ctx)
2016-12-16 17:26:48 +03:00
Expect(err).NotTo(HaveOccurred())
2020-03-11 17:26:42 +03:00
a := pipe.Get(ctx, "A")
b := pipe.Get(ctx, "B")
c := pipe.Get(ctx, "C")
cmds, err := pipe.Exec(ctx)
2016-12-13 18:28:39 +03:00
Expect(err).To(Equal(redis.Nil))
Expect(cmds).To(HaveLen(3))
Expect(a.Err()).NotTo(HaveOccurred())
Expect(a.Val()).To(Equal("A_value"))
Expect(b.Err()).To(Equal(redis.Nil))
Expect(b.Val()).To(Equal(""))
Expect(c.Err()).NotTo(HaveOccurred())
Expect(c.Val()).To(Equal("C_value"))
})
}
2017-07-09 13:10:07 +03:00
Describe("with Pipeline", func() {
2016-12-13 18:28:39 +03:00
BeforeEach(func() {
pipe = client.Pipeline().(*redis.Pipeline)
2016-10-09 14:12:32 +03:00
})
2016-04-06 14:01:08 +03:00
AfterEach(func() {})
2016-12-13 18:28:39 +03:00
assertPipeline()
})
2016-04-06 14:01:08 +03:00
2017-07-09 13:10:07 +03:00
Describe("with TxPipeline", func() {
2016-12-13 18:28:39 +03:00
BeforeEach(func() {
pipe = client.TxPipeline().(*redis.Pipeline)
2016-12-13 18:28:39 +03:00
})
AfterEach(func() {})
2016-04-06 14:01:08 +03:00
2016-12-13 18:28:39 +03:00
assertPipeline()
2016-10-09 14:12:32 +03:00
})
2016-04-06 14:01:08 +03:00
})
2016-06-17 15:09:38 +03:00
2017-07-09 10:07:20 +03:00
It("supports PubSub", func() {
2020-03-11 17:26:42 +03:00
pubsub := client.Subscribe(ctx, "mychannel")
2017-07-09 10:07:20 +03:00
defer pubsub.Close()
2017-07-09 13:10:07 +03:00
Eventually(func() error {
2020-03-11 17:26:42 +03:00
_, err := client.Publish(ctx, "mychannel", "hello").Result()
2017-07-09 13:10:07 +03:00
if err != nil {
return err
}
2017-07-09 10:07:20 +03:00
2020-03-11 17:26:42 +03:00
msg, err := pubsub.ReceiveTimeout(ctx, time.Second)
2017-07-09 13:10:07 +03:00
if err != nil {
return err
}
2017-07-09 10:07:20 +03:00
2017-07-09 13:10:07 +03:00
_, ok := msg.(*redis.Message)
if !ok {
return fmt.Errorf("got %T, wanted *redis.Message", msg)
}
2022-08-03 18:10:03 +03:00
return nil
}, 30*time.Second).ShouldNot(HaveOccurred())
})
It("supports sharded PubSub", func() {
pubsub := client.SSubscribe(ctx, "mychannel")
defer pubsub.Close()
Eventually(func() error {
_, err := client.SPublish(ctx, "mychannel", "hello").Result()
if err != nil {
return err
}
msg, err := pubsub.ReceiveTimeout(ctx, time.Second)
if err != nil {
return err
}
_, ok := msg.(*redis.Message)
if !ok {
return fmt.Errorf("got %T, wanted *redis.Message", msg)
}
2017-07-09 13:10:07 +03:00
return nil
}, 30*time.Second).ShouldNot(HaveOccurred())
})
It("supports PubSub.Ping without channels", func() {
2020-03-11 17:26:42 +03:00
pubsub := client.Subscribe(ctx)
defer pubsub.Close()
2020-03-11 17:26:42 +03:00
err := pubsub.Ping(ctx)
Expect(err).NotTo(HaveOccurred())
})
2020-09-11 12:46:38 +03:00
}
2023-05-16 17:02:22 +03:00
Describe("ClusterClient PROTO 2", func() {
BeforeEach(func() {
opt = redisClusterOptions()
opt.Protocol = 2
client = cluster.newClusterClient(ctx, opt)
err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
return master.FlushDB(ctx).Err()
})
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
_ = client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
return master.FlushDB(ctx).Err()
})
Expect(client.Close()).NotTo(HaveOccurred())
})
It("should CLUSTER PROTO 2", func() {
_ = client.ForEachShard(ctx, func(ctx context.Context, c *redis.Client) error {
val, err := c.Do(ctx, "HELLO").Result()
Expect(err).NotTo(HaveOccurred())
Expect(val).Should(ContainElements("proto", int64(2)))
return nil
})
})
})
2020-09-11 12:46:38 +03:00
Describe("ClusterClient", func() {
BeforeEach(func() {
opt = redisClusterOptions()
opt.ClientName = "cluster_hi"
2020-09-11 12:46:38 +03:00
client = cluster.newClusterClient(ctx, opt)
err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
return master.FlushDB(ctx).Err()
})
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
_ = client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
return master.FlushDB(ctx).Err()
})
Expect(client.Close()).NotTo(HaveOccurred())
})
It("returns pool stats", func() {
stats := client.PoolStats()
Expect(stats).To(BeAssignableToTypeOf(&redis.PoolStats{}))
})
It("returns an error when there are no attempts left", func() {
opt := redisClusterOptions()
opt.MaxRedirects = -1
client := cluster.newClusterClient(ctx, opt)
Eventually(func() error {
return client.SwapNodes(ctx, "A")
}, 30*time.Second).ShouldNot(HaveOccurred())
err := client.Get(ctx, "A").Err()
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("MOVED"))
Expect(client.Close()).NotTo(HaveOccurred())
})
It("calls fn for every master node", func() {
for i := 0; i < 10; i++ {
Expect(client.Set(ctx, strconv.Itoa(i), "", 0).Err()).NotTo(HaveOccurred())
}
err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
return master.FlushDB(ctx).Err()
})
Expect(err).NotTo(HaveOccurred())
size, err := client.DBSize(ctx).Result()
Expect(err).NotTo(HaveOccurred())
Expect(size).To(Equal(int64(0)))
})
It("should CLUSTER SLOTS", func() {
res, err := client.ClusterSlots(ctx).Result()
Expect(err).NotTo(HaveOccurred())
Expect(res).To(HaveLen(3))
wanted := []redis.ClusterSlot{{
Start: 0,
End: 4999,
Nodes: []redis.ClusterNode{{
ID: "",
Addr: "127.0.0.1:8220",
}, {
ID: "",
Addr: "127.0.0.1:8223",
}},
}, {
Start: 5000,
End: 9999,
Nodes: []redis.ClusterNode{{
ID: "",
Addr: "127.0.0.1:8221",
}, {
ID: "",
Addr: "127.0.0.1:8224",
}},
}, {
Start: 10000,
End: 16383,
Nodes: []redis.ClusterNode{{
ID: "",
Addr: "127.0.0.1:8222",
}, {
ID: "",
Addr: "127.0.0.1:8225",
}},
}}
Expect(assertSlotsEqual(res, wanted)).NotTo(HaveOccurred())
})
It("should CLUSTER SHARDS", func() {
res, err := client.ClusterShards(ctx).Result()
Expect(err).NotTo(HaveOccurred())
Expect(res).NotTo(BeEmpty())
// Iterate over the ClusterShard results and validate the fields.
for _, shard := range res {
Expect(shard.Slots).NotTo(BeEmpty())
for _, slotRange := range shard.Slots {
Expect(slotRange.Start).To(BeNumerically(">=", 0))
Expect(slotRange.End).To(BeNumerically(">=", slotRange.Start))
}
Expect(shard.Nodes).NotTo(BeEmpty())
for _, node := range shard.Nodes {
Expect(node.ID).NotTo(BeEmpty())
Expect(node.Endpoint).NotTo(BeEmpty())
Expect(node.IP).NotTo(BeEmpty())
Expect(node.Port).To(BeNumerically(">", 0))
validRoles := []string{"master", "slave", "replica"}
Expect(validRoles).To(ContainElement(node.Role))
Expect(node.ReplicationOffset).To(BeNumerically(">=", 0))
validHealthStatuses := []string{"online", "failed", "loading"}
Expect(validHealthStatuses).To(ContainElement(node.Health))
}
}
})
It("should CLUSTER LINKS", func() {
res, err := client.ClusterLinks(ctx).Result()
Expect(err).NotTo(HaveOccurred())
Expect(res).NotTo(BeEmpty())
// Iterate over the ClusterLink results and validate the map keys.
for _, link := range res {
Expect(link.Direction).NotTo(BeEmpty())
Expect([]string{"from", "to"}).To(ContainElement(link.Direction))
Expect(link.Node).NotTo(BeEmpty())
Expect(link.CreateTime).To(BeNumerically(">", 0))
Expect(link.Events).NotTo(BeEmpty())
validEventChars := []rune{'r', 'w'}
for _, eventChar := range link.Events {
Expect(validEventChars).To(ContainElement(eventChar))
}
Expect(link.SendBufferAllocated).To(BeNumerically(">=", 0))
Expect(link.SendBufferUsed).To(BeNumerically(">=", 0))
}
})
It("should cluster client setname", func() {
err := client.ForEachShard(ctx, func(ctx context.Context, c *redis.Client) error {
return c.Ping(ctx).Err()
})
Expect(err).NotTo(HaveOccurred())
_ = client.ForEachShard(ctx, func(ctx context.Context, c *redis.Client) error {
val, err := c.ClientList(ctx).Result()
Expect(err).NotTo(HaveOccurred())
Expect(val).Should(ContainSubstring("name=cluster_hi"))
return nil
})
})
2023-05-16 17:02:22 +03:00
It("should CLUSTER PROTO 3", func() {
_ = client.ForEachShard(ctx, func(ctx context.Context, c *redis.Client) error {
val, err := c.Do(ctx, "HELLO").Result()
Expect(err).NotTo(HaveOccurred())
Expect(val).Should(HaveKeyWithValue("proto", int64(3)))
return nil
})
})
It("should CLUSTER MYSHARDID", func() {
shardID, err := client.ClusterMyShardID(ctx).Result()
Expect(err).NotTo(HaveOccurred())
Expect(shardID).ToNot(BeEmpty())
})
2020-09-11 12:46:38 +03:00
It("should CLUSTER NODES", func() {
res, err := client.ClusterNodes(ctx).Result()
Expect(err).NotTo(HaveOccurred())
Expect(len(res)).To(BeNumerically(">", 400))
})
It("should CLUSTER INFO", func() {
res, err := client.ClusterInfo(ctx).Result()
Expect(err).NotTo(HaveOccurred())
Expect(res).To(ContainSubstring("cluster_known_nodes:6"))
})
It("should CLUSTER KEYSLOT", func() {
hashSlot, err := client.ClusterKeySlot(ctx, "somekey").Result()
Expect(err).NotTo(HaveOccurred())
Expect(hashSlot).To(Equal(int64(hashtag.Slot("somekey"))))
})
It("should CLUSTER GETKEYSINSLOT", func() {
keys, err := client.ClusterGetKeysInSlot(ctx, hashtag.Slot("somekey"), 1).Result()
Expect(err).NotTo(HaveOccurred())
Expect(len(keys)).To(Equal(0))
})
It("should CLUSTER COUNT-FAILURE-REPORTS", func() {
n, err := client.ClusterCountFailureReports(ctx, cluster.nodeIDs[0]).Result()
Expect(err).NotTo(HaveOccurred())
Expect(n).To(Equal(int64(0)))
})
It("should CLUSTER COUNTKEYSINSLOT", func() {
n, err := client.ClusterCountKeysInSlot(ctx, 10).Result()
Expect(err).NotTo(HaveOccurred())
Expect(n).To(Equal(int64(0)))
})
It("should CLUSTER SAVECONFIG", func() {
res, err := client.ClusterSaveConfig(ctx).Result()
Expect(err).NotTo(HaveOccurred())
Expect(res).To(Equal("OK"))
})
It("should CLUSTER SLAVES", func() {
nodesList, err := client.ClusterSlaves(ctx, cluster.nodeIDs[0]).Result()
Expect(err).NotTo(HaveOccurred())
Expect(nodesList).Should(ContainElement(ContainSubstring("slave")))
Expect(nodesList).Should(HaveLen(1))
})
It("should RANDOMKEY", func() {
const nkeys = 100
for i := 0; i < nkeys; i++ {
err := client.Set(ctx, fmt.Sprintf("key%d", i), "value", 0).Err()
Expect(err).NotTo(HaveOccurred())
}
var keys []string
addKey := func(key string) {
for _, k := range keys {
if k == key {
return
}
}
keys = append(keys, key)
}
for i := 0; i < nkeys*10; i++ {
key := client.RandomKey(ctx).Val()
addKey(key)
}
Expect(len(keys)).To(BeNumerically("~", nkeys, nkeys/10))
})
2020-02-14 15:30:07 +03:00
It("supports Process hook", func() {
testCtx, cancel := context.WithCancel(ctx)
defer cancel()
2020-03-11 17:26:42 +03:00
err := client.Ping(ctx).Err()
2020-02-14 15:30:07 +03:00
Expect(err).NotTo(HaveOccurred())
2020-06-10 15:04:12 +03:00
err = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
2020-03-11 17:26:42 +03:00
return node.Ping(ctx).Err()
2020-02-14 15:30:07 +03:00
})
Expect(err).NotTo(HaveOccurred())
var stack []string
clusterHook := &hook{
processHook: func(hook redis.ProcessHook) redis.ProcessHook {
return func(ctx context.Context, cmd redis.Cmder) error {
select {
case <-testCtx.Done():
return hook(ctx, cmd)
default:
}
Expect(cmd.String()).To(Equal("ping: "))
stack = append(stack, "cluster.BeforeProcess")
err := hook(ctx, cmd)
Expect(cmd.String()).To(Equal("ping: PONG"))
stack = append(stack, "cluster.AfterProcess")
return err
}
2020-02-14 15:30:07 +03:00
},
}
client.AddHook(clusterHook)
2020-02-14 16:37:35 +03:00
nodeHook := &hook{
processHook: func(hook redis.ProcessHook) redis.ProcessHook {
return func(ctx context.Context, cmd redis.Cmder) error {
select {
case <-testCtx.Done():
return hook(ctx, cmd)
default:
}
Expect(cmd.String()).To(Equal("ping: "))
stack = append(stack, "shard.BeforeProcess")
err := hook(ctx, cmd)
Expect(cmd.String()).To(Equal("ping: PONG"))
stack = append(stack, "shard.AfterProcess")
return err
}
2020-02-14 15:30:07 +03:00
},
}
2020-06-10 15:04:12 +03:00
_ = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
2020-02-14 16:37:35 +03:00
node.AddHook(nodeHook)
return nil
})
2020-02-14 15:30:07 +03:00
2020-03-11 17:26:42 +03:00
err = client.Ping(ctx).Err()
2020-02-14 15:30:07 +03:00
Expect(err).NotTo(HaveOccurred())
Expect(stack).To(Equal([]string{
"cluster.BeforeProcess",
"shard.BeforeProcess",
"shard.AfterProcess",
"cluster.AfterProcess",
}))
})
It("supports Pipeline hook", func() {
2020-03-11 17:26:42 +03:00
err := client.Ping(ctx).Err()
2020-02-14 15:30:07 +03:00
Expect(err).NotTo(HaveOccurred())
2020-06-10 15:04:12 +03:00
err = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
2020-03-11 17:26:42 +03:00
return node.Ping(ctx).Err()
2020-02-14 15:30:07 +03:00
})
Expect(err).NotTo(HaveOccurred())
var stack []string
client.AddHook(&hook{
processPipelineHook: func(hook redis.ProcessPipelineHook) redis.ProcessPipelineHook {
return func(ctx context.Context, cmds []redis.Cmder) error {
Expect(cmds).To(HaveLen(1))
Expect(cmds[0].String()).To(Equal("ping: "))
stack = append(stack, "cluster.BeforeProcessPipeline")
err := hook(ctx, cmds)
Expect(cmds).To(HaveLen(1))
Expect(cmds[0].String()).To(Equal("ping: PONG"))
stack = append(stack, "cluster.AfterProcessPipeline")
return err
}
2020-02-14 15:30:07 +03:00
},
})
2020-06-10 15:04:12 +03:00
_ = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
2020-02-14 16:37:35 +03:00
node.AddHook(&hook{
processPipelineHook: func(hook redis.ProcessPipelineHook) redis.ProcessPipelineHook {
return func(ctx context.Context, cmds []redis.Cmder) error {
Expect(cmds).To(HaveLen(1))
Expect(cmds[0].String()).To(Equal("ping: "))
stack = append(stack, "shard.BeforeProcessPipeline")
err := hook(ctx, cmds)
Expect(cmds).To(HaveLen(1))
Expect(cmds[0].String()).To(Equal("ping: PONG"))
stack = append(stack, "shard.AfterProcessPipeline")
return err
}
2020-02-14 15:30:07 +03:00
},
})
2020-02-14 16:37:35 +03:00
return nil
})
2020-02-14 15:30:07 +03:00
2020-03-11 17:26:42 +03:00
_, err = client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
pipe.Ping(ctx)
2020-02-14 15:30:07 +03:00
return nil
})
Expect(err).NotTo(HaveOccurred())
Expect(stack).To(Equal([]string{
"cluster.BeforeProcessPipeline",
"shard.BeforeProcessPipeline",
"shard.AfterProcessPipeline",
"cluster.AfterProcessPipeline",
}))
})
It("supports TxPipeline hook", func() {
2020-03-11 17:26:42 +03:00
err := client.Ping(ctx).Err()
2020-02-14 15:30:07 +03:00
Expect(err).NotTo(HaveOccurred())
2020-06-10 15:04:12 +03:00
err = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
2020-03-11 17:26:42 +03:00
return node.Ping(ctx).Err()
2020-02-14 15:30:07 +03:00
})
Expect(err).NotTo(HaveOccurred())
var stack []string
client.AddHook(&hook{
processPipelineHook: func(hook redis.ProcessPipelineHook) redis.ProcessPipelineHook {
return func(ctx context.Context, cmds []redis.Cmder) error {
Expect(cmds).To(HaveLen(3))
Expect(cmds[1].String()).To(Equal("ping: "))
stack = append(stack, "cluster.BeforeProcessPipeline")
err := hook(ctx, cmds)
Expect(cmds).To(HaveLen(3))
Expect(cmds[1].String()).To(Equal("ping: PONG"))
stack = append(stack, "cluster.AfterProcessPipeline")
return err
}
2020-02-14 15:30:07 +03:00
},
})
2020-06-10 15:04:12 +03:00
_ = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error {
2020-02-14 16:37:35 +03:00
node.AddHook(&hook{
processPipelineHook: func(hook redis.ProcessPipelineHook) redis.ProcessPipelineHook {
return func(ctx context.Context, cmds []redis.Cmder) error {
Expect(cmds).To(HaveLen(3))
Expect(cmds[1].String()).To(Equal("ping: "))
stack = append(stack, "shard.BeforeProcessPipeline")
err := hook(ctx, cmds)
Expect(cmds).To(HaveLen(3))
Expect(cmds[1].String()).To(Equal("ping: PONG"))
stack = append(stack, "shard.AfterProcessPipeline")
return err
}
2020-02-14 15:30:07 +03:00
},
})
2020-02-14 16:37:35 +03:00
return nil
})
2020-02-14 15:30:07 +03:00
2020-03-11 17:26:42 +03:00
_, err = client.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
pipe.Ping(ctx)
2020-02-14 15:30:07 +03:00
return nil
})
Expect(err).NotTo(HaveOccurred())
Expect(stack).To(Equal([]string{
"cluster.BeforeProcessPipeline",
"shard.BeforeProcessPipeline",
"shard.AfterProcessPipeline",
"cluster.AfterProcessPipeline",
}))
})
It("should return correct replica for key", func() {
2020-12-16 18:45:06 +03:00
client, err := client.SlaveForKey(ctx, "test")
Expect(err).ToNot(HaveOccurred())
info := client.Info(ctx, "server")
Expect(info.Val()).Should(ContainSubstring("tcp_port:8224"))
})
It("should return correct master for key", func() {
client, err := client.MasterForKey(ctx, "test")
Expect(err).ToNot(HaveOccurred())
info := client.Info(ctx, "server")
Expect(info.Val()).Should(ContainSubstring("tcp_port:8221"))
})
2017-07-09 13:10:07 +03:00
assertClusterClient()
})
Describe("ClusterClient with RouteByLatency", func() {
BeforeEach(func() {
2016-12-16 17:26:48 +03:00
opt = redisClusterOptions()
opt.RouteByLatency = true
2020-03-11 17:26:42 +03:00
client = cluster.newClusterClient(ctx, opt)
2016-06-17 15:09:38 +03:00
2020-03-11 17:26:42 +03:00
err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
return master.FlushDB(ctx).Err()
2016-06-17 15:09:38 +03:00
})
2018-05-17 16:09:56 +03:00
Expect(err).NotTo(HaveOccurred())
2017-08-31 15:22:47 +03:00
2020-03-11 17:26:42 +03:00
err = client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
2017-08-31 15:22:47 +03:00
Eventually(func() int64 {
2020-03-11 17:26:42 +03:00
return client.DBSize(ctx).Val()
2017-08-31 15:22:47 +03:00
}, 30*time.Second).Should(Equal(int64(0)))
return nil
})
2018-05-17 16:09:56 +03:00
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
2020-03-11 17:26:42 +03:00
err := client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
return slave.ReadWrite(ctx).Err()
2017-08-31 15:22:47 +03:00
})
2018-05-17 16:09:56 +03:00
Expect(err).NotTo(HaveOccurred())
err = client.Close()
Expect(err).NotTo(HaveOccurred())
})
2016-12-16 17:26:48 +03:00
assertClusterClient()
2016-04-06 14:01:08 +03:00
})
2018-06-29 10:45:05 +03:00
Describe("ClusterClient with ClusterSlots", func() {
BeforeEach(func() {
failover = true
opt = redisClusterOptions()
2020-09-09 15:27:17 +03:00
opt.ClusterSlots = func(ctx context.Context) ([]redis.ClusterSlot, error) {
2018-06-29 10:45:05 +03:00
slots := []redis.ClusterSlot{{
Start: 0,
End: 4999,
Nodes: []redis.ClusterNode{{
Addr: ":" + ringShard1Port,
}},
}, {
Start: 5000,
End: 9999,
Nodes: []redis.ClusterNode{{
Addr: ":" + ringShard2Port,
}},
}, {
Start: 10000,
End: 16383,
Nodes: []redis.ClusterNode{{
Addr: ":" + ringShard3Port,
}},
}}
return slots, nil
}
2020-03-11 17:26:42 +03:00
client = cluster.newClusterClient(ctx, opt)
2018-06-29 10:45:05 +03:00
2020-03-11 17:26:42 +03:00
err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
return master.FlushDB(ctx).Err()
2018-06-29 10:45:05 +03:00
})
Expect(err).NotTo(HaveOccurred())
2020-03-11 17:26:42 +03:00
err = client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
Eventually(func() int64 {
2020-03-11 17:26:42 +03:00
return client.DBSize(ctx).Val()
}, 30*time.Second).Should(Equal(int64(0)))
return nil
})
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
failover = false
err := client.Close()
Expect(err).NotTo(HaveOccurred())
})
assertClusterClient()
})
Describe("ClusterClient with RouteRandomly and ClusterSlots", func() {
BeforeEach(func() {
failover = true
opt = redisClusterOptions()
opt.RouteRandomly = true
2020-09-09 15:27:17 +03:00
opt.ClusterSlots = func(ctx context.Context) ([]redis.ClusterSlot, error) {
slots := []redis.ClusterSlot{{
Start: 0,
End: 4999,
Nodes: []redis.ClusterNode{{
Addr: ":" + ringShard1Port,
}},
}, {
Start: 5000,
End: 9999,
Nodes: []redis.ClusterNode{{
Addr: ":" + ringShard2Port,
}},
}, {
Start: 10000,
End: 16383,
Nodes: []redis.ClusterNode{{
Addr: ":" + ringShard3Port,
}},
}}
return slots, nil
}
2020-03-11 17:26:42 +03:00
client = cluster.newClusterClient(ctx, opt)
2020-03-11 17:26:42 +03:00
err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
return master.FlushDB(ctx).Err()
})
Expect(err).NotTo(HaveOccurred())
2020-03-11 17:26:42 +03:00
err = client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
2018-06-29 10:45:05 +03:00
Eventually(func() int64 {
2020-03-11 17:26:42 +03:00
return client.DBSize(ctx).Val()
2018-06-29 10:45:05 +03:00
}, 30*time.Second).Should(Equal(int64(0)))
return nil
})
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
failover = false
err := client.Close()
Expect(err).NotTo(HaveOccurred())
})
assertClusterClient()
})
Describe("ClusterClient with ClusterSlots with multiple nodes per slot", func() {
BeforeEach(func() {
failover = true
opt = redisClusterOptions()
opt.ReadOnly = true
opt.ClusterSlots = func(ctx context.Context) ([]redis.ClusterSlot, error) {
slots := []redis.ClusterSlot{{
Start: 0,
End: 4999,
Nodes: []redis.ClusterNode{{
Addr: ":8220",
}, {
Addr: ":8223",
}},
}, {
Start: 5000,
End: 9999,
Nodes: []redis.ClusterNode{{
Addr: ":8221",
}, {
Addr: ":8224",
}},
}, {
Start: 10000,
End: 16383,
Nodes: []redis.ClusterNode{{
Addr: ":8222",
}, {
Addr: ":8225",
}},
}}
return slots, nil
}
client = cluster.newClusterClient(ctx, opt)
err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
return master.FlushDB(ctx).Err()
})
Expect(err).NotTo(HaveOccurred())
err = client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
Eventually(func() int64 {
return client.DBSize(ctx).Val()
}, 30*time.Second).Should(Equal(int64(0)))
2018-06-29 10:45:05 +03:00
return nil
})
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
failover = false
err := client.Close()
Expect(err).NotTo(HaveOccurred())
})
assertClusterClient()
})
})
var _ = Describe("ClusterClient without nodes", func() {
var client *redis.ClusterClient
BeforeEach(func() {
client = redis.NewClusterClient(&redis.ClusterOptions{})
})
AfterEach(func() {
Expect(client.Close()).NotTo(HaveOccurred())
})
2017-08-31 15:22:47 +03:00
It("Ping returns an error", func() {
2020-03-11 17:26:42 +03:00
err := client.Ping(ctx).Err()
Expect(err).To(MatchError("redis: cluster has no nodes"))
})
It("pipeline returns an error", func() {
2020-03-11 17:26:42 +03:00
_, err := client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
pipe.Ping(ctx)
return nil
2016-10-09 14:12:32 +03:00
})
Expect(err).To(MatchError("redis: cluster has no nodes"))
})
})
var _ = Describe("ClusterClient without valid nodes", func() {
var client *redis.ClusterClient
BeforeEach(func() {
client = redis.NewClusterClient(&redis.ClusterOptions{
Addrs: []string{redisAddr},
})
})
AfterEach(func() {
Expect(client.Close()).NotTo(HaveOccurred())
})
It("returns an error", func() {
2020-03-11 17:26:42 +03:00
err := client.Ping(ctx).Err()
2018-02-15 14:00:54 +03:00
Expect(err).To(MatchError("ERR This instance has cluster support disabled"))
})
It("pipeline returns an error", func() {
2020-03-11 17:26:42 +03:00
_, err := client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
pipe.Ping(ctx)
return nil
})
2018-02-15 14:00:54 +03:00
Expect(err).To(MatchError("ERR This instance has cluster support disabled"))
})
})
var _ = Describe("ClusterClient with unavailable Cluster", func() {
var client *redis.ClusterClient
BeforeEach(func() {
opt := redisClusterOptions()
opt.ReadTimeout = 250 * time.Millisecond
opt.WriteTimeout = 250 * time.Millisecond
opt.MaxRedirects = 1
2020-09-05 11:34:37 +03:00
client = cluster.newClusterClientUnstable(opt)
Expect(client.Ping(ctx).Err()).NotTo(HaveOccurred())
for _, node := range cluster.clients {
err := node.ClientPause(ctx, 5*time.Second).Err()
Expect(err).NotTo(HaveOccurred())
}
})
AfterEach(func() {
Expect(client.Close()).NotTo(HaveOccurred())
})
It("recovers when Cluster recovers", func() {
2020-03-11 17:26:42 +03:00
err := client.Ping(ctx).Err()
Expect(err).To(HaveOccurred())
Eventually(func() error {
2020-03-11 17:26:42 +03:00
return client.Ping(ctx).Err()
}, "30s").ShouldNot(HaveOccurred())
})
})
var _ = Describe("ClusterClient timeout", func() {
var client *redis.ClusterClient
2016-10-09 14:12:32 +03:00
AfterEach(func() {
2017-03-04 14:04:27 +03:00
_ = client.Close()
})
testTimeout := func() {
It("Ping timeouts", func() {
2020-03-11 17:26:42 +03:00
err := client.Ping(ctx).Err()
Expect(err).To(HaveOccurred())
Expect(err.(net.Error).Timeout()).To(BeTrue())
2016-10-09 14:12:32 +03:00
})
It("Pipeline timeouts", func() {
2020-03-11 17:26:42 +03:00
_, err := client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
pipe.Ping(ctx)
2016-10-09 14:12:32 +03:00
return nil
})
Expect(err).To(HaveOccurred())
Expect(err.(net.Error).Timeout()).To(BeTrue())
2016-10-09 14:12:32 +03:00
})
It("Tx timeouts", func() {
2020-03-11 17:26:42 +03:00
err := client.Watch(ctx, func(tx *redis.Tx) error {
return tx.Ping(ctx).Err()
2017-08-31 15:22:47 +03:00
}, "foo")
Expect(err).To(HaveOccurred())
Expect(err.(net.Error).Timeout()).To(BeTrue())
})
It("Tx Pipeline timeouts", func() {
2020-03-11 17:26:42 +03:00
err := client.Watch(ctx, func(tx *redis.Tx) error {
_, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
pipe.Ping(ctx)
return nil
})
return err
2017-08-31 15:22:47 +03:00
}, "foo")
Expect(err).To(HaveOccurred())
Expect(err.(net.Error).Timeout()).To(BeTrue())
})
}
2016-10-09 14:12:32 +03:00
const pause = 5 * time.Second
2017-08-31 15:22:47 +03:00
Context("read/write timeout", func() {
BeforeEach(func() {
opt := redisClusterOptions()
2020-03-11 17:26:42 +03:00
client = cluster.newClusterClient(ctx, opt)
2020-06-10 15:04:12 +03:00
err := client.ForEachShard(ctx, func(ctx context.Context, client *redis.Client) error {
err := client.ClientPause(ctx, pause).Err()
opt := client.Options()
opt.ReadTimeout = time.Nanosecond
opt.WriteTimeout = time.Nanosecond
return err
})
Expect(err).NotTo(HaveOccurred())
// Overwrite timeouts after the client is initialized.
opt.ReadTimeout = time.Nanosecond
opt.WriteTimeout = time.Nanosecond
opt.MaxRedirects = 0
})
AfterEach(func() {
2020-06-10 15:04:12 +03:00
_ = client.ForEachShard(ctx, func(ctx context.Context, client *redis.Client) error {
2018-02-15 14:00:54 +03:00
defer GinkgoRecover()
opt := client.Options()
opt.ReadTimeout = time.Second
opt.WriteTimeout = time.Second
2017-08-15 10:34:05 +03:00
Eventually(func() error {
2020-03-11 17:26:42 +03:00
return client.Ping(ctx).Err()
2017-08-15 10:34:05 +03:00
}, 2*pause).ShouldNot(HaveOccurred())
return nil
})
2022-11-21 12:31:38 +03:00
err := client.Close()
Expect(err).NotTo(HaveOccurred())
})
testTimeout()
})
2015-01-24 15:12:48 +03:00
})
var _ = Describe("ClusterClient ParseURL", func() {
var cases = []struct {
test string
url string
o *redis.ClusterOptions // expected value
err error
}{
{
test: "ParseRedisURL",
url: "redis://localhost:123",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}},
}, {
test: "ParseRedissURL",
url: "rediss://localhost:123",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}, TLSConfig: &tls.Config{ServerName: "localhost"}},
}, {
test: "MissingRedisPort",
url: "redis://localhost",
o: &redis.ClusterOptions{Addrs: []string{"localhost:6379"}},
}, {
test: "MissingRedissPort",
url: "rediss://localhost",
o: &redis.ClusterOptions{Addrs: []string{"localhost:6379"}, TLSConfig: &tls.Config{ServerName: "localhost"}},
}, {
test: "MultipleRedisURLs",
url: "redis://localhost:123?addr=localhost:1234&addr=localhost:12345",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123", "localhost:1234", "localhost:12345"}},
}, {
test: "MultipleRedissURLs",
url: "rediss://localhost:123?addr=localhost:1234&addr=localhost:12345",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123", "localhost:1234", "localhost:12345"}, TLSConfig: &tls.Config{ServerName: "localhost"}},
}, {
test: "OnlyPassword",
url: "redis://:bar@localhost:123",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}, Password: "bar"},
}, {
test: "OnlyUser",
url: "redis://foo@localhost:123",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}, Username: "foo"},
}, {
test: "RedisUsernamePassword",
url: "redis://foo:bar@localhost:123",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}, Username: "foo", Password: "bar"},
}, {
test: "RedissUsernamePassword",
url: "rediss://foo:bar@localhost:123?addr=localhost:1234",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123", "localhost:1234"}, Username: "foo", Password: "bar", TLSConfig: &tls.Config{ServerName: "localhost"}},
}, {
test: "QueryParameters",
url: "redis://localhost:123?read_timeout=2&pool_fifo=true&addr=localhost:1234",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123", "localhost:1234"}, ReadTimeout: 2 * time.Second, PoolFIFO: true},
}, {
test: "DisabledTimeout",
url: "redis://localhost:123?conn_max_idle_time=0",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}, ConnMaxIdleTime: -1},
}, {
test: "DisabledTimeoutNeg",
url: "redis://localhost:123?conn_max_idle_time=-1",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}, ConnMaxIdleTime: -1},
}, {
test: "UseDefault",
url: "redis://localhost:123?conn_max_idle_time=",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}, ConnMaxIdleTime: 0},
}, {
test: "Protocol",
url: "redis://localhost:123?protocol=2",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}, Protocol: 2},
}, {
test: "ClientName",
url: "redis://localhost:123?client_name=cluster_hi",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}, ClientName: "cluster_hi"},
}, {
test: "UseDefaultMissing=",
url: "redis://localhost:123?conn_max_idle_time",
o: &redis.ClusterOptions{Addrs: []string{"localhost:123"}, ConnMaxIdleTime: 0},
}, {
test: "InvalidQueryAddr",
url: "rediss://foo:bar@localhost:123?addr=rediss://foo:barr@localhost:1234",
err: errors.New(`redis: unable to parse addr param: rediss://foo:barr@localhost:1234`),
}, {
test: "InvalidInt",
url: "redis://localhost?pool_size=five",
err: errors.New(`redis: invalid pool_size number: strconv.Atoi: parsing "five": invalid syntax`),
}, {
test: "InvalidBool",
url: "redis://localhost?pool_fifo=yes",
err: errors.New(`redis: invalid pool_fifo boolean: expected true/false/1/0 or an empty string, got "yes"`),
}, {
test: "UnknownParam",
url: "redis://localhost?abc=123",
err: errors.New("redis: unexpected option: abc"),
}, {
test: "InvalidScheme",
url: "https://google.com",
err: errors.New("redis: invalid URL scheme: https"),
},
}
It("match ParseClusterURL", func() {
for i := range cases {
tc := cases[i]
actual, err := redis.ParseClusterURL(tc.url)
if tc.err != nil {
Expect(err).Should(MatchError(tc.err))
} else {
Expect(err).NotTo(HaveOccurred())
}
if err == nil {
Expect(tc.o).NotTo(BeNil())
Expect(tc.o.Addrs).To(Equal(actual.Addrs))
Expect(tc.o.TLSConfig).To(Equal(actual.TLSConfig))
Expect(tc.o.Username).To(Equal(actual.Username))
Expect(tc.o.Password).To(Equal(actual.Password))
Expect(tc.o.MaxRetries).To(Equal(actual.MaxRetries))
Expect(tc.o.MinRetryBackoff).To(Equal(actual.MinRetryBackoff))
Expect(tc.o.MaxRetryBackoff).To(Equal(actual.MaxRetryBackoff))
Expect(tc.o.DialTimeout).To(Equal(actual.DialTimeout))
Expect(tc.o.ReadTimeout).To(Equal(actual.ReadTimeout))
Expect(tc.o.WriteTimeout).To(Equal(actual.WriteTimeout))
Expect(tc.o.PoolFIFO).To(Equal(actual.PoolFIFO))
Expect(tc.o.PoolSize).To(Equal(actual.PoolSize))
Expect(tc.o.MinIdleConns).To(Equal(actual.MinIdleConns))
Expect(tc.o.ConnMaxLifetime).To(Equal(actual.ConnMaxLifetime))
Expect(tc.o.ConnMaxIdleTime).To(Equal(actual.ConnMaxIdleTime))
Expect(tc.o.PoolTimeout).To(Equal(actual.PoolTimeout))
}
}
})
})