Merge pull request #393 from go-redis/fix/cluster-testss

More cluster tests.
This commit is contained in:
Vladimir Mihailenco 2016-10-09 14:37:19 +03:00 committed by GitHub
commit 97e6ccb3d8
3 changed files with 158 additions and 77 deletions

View File

@ -12,6 +12,8 @@ import (
"gopkg.in/redis.v5/internal/pool" "gopkg.in/redis.v5/internal/pool"
) )
var errClusterNoNodes = errors.RedisError("redis: cluster has no nodes")
// ClusterOptions are used to configure a cluster client and should be // ClusterOptions are used to configure a cluster client and should be
// passed to NewClusterClient. // passed to NewClusterClient.
type ClusterOptions struct { type ClusterOptions struct {
@ -155,14 +157,14 @@ func (c *ClusterClient) cmdInfo(name string) *CommandInfo {
func (c *ClusterClient) getNodes() map[string]*clusterNode { func (c *ClusterClient) getNodes() map[string]*clusterNode {
var nodes map[string]*clusterNode var nodes map[string]*clusterNode
c.mu.RLock()
if !c.closed { if !c.closed {
nodes = make(map[string]*clusterNode, len(c.nodes)) nodes = make(map[string]*clusterNode, len(c.nodes))
c.mu.RLock()
for addr, node := range c.nodes { for addr, node := range c.nodes {
nodes[addr] = node nodes[addr] = node
} }
c.mu.RUnlock()
} }
c.mu.RUnlock()
return nodes return nodes
} }
@ -261,6 +263,9 @@ func (c *ClusterClient) randomNode() (*clusterNode, error) {
return nil, pool.ErrClosed return nil, pool.ErrClosed
} }
if len(addrs) == 0 {
return nil, errClusterNoNodes
}
n := rand.Intn(len(addrs)) n := rand.Intn(len(addrs))
node, err := c.nodeByAddr(addrs[n]) node, err := c.nodeByAddr(addrs[n])

View File

@ -301,6 +301,37 @@ var _ = Describe("ClusterClient", func() {
Expect(client.Close()).NotTo(HaveOccurred()) Expect(client.Close()).NotTo(HaveOccurred())
}) })
It("distributes keys", func() {
for i := 0; i < 100; i++ {
err := client.Set(fmt.Sprintf("key%d", i), "value", 0).Err()
Expect(err).NotTo(HaveOccurred())
}
wanted := []string{"keys=31", "keys=29", "keys=40"}
for i, master := range cluster.masters() {
Expect(master.Info().Val()).To(ContainSubstring(wanted[i]))
}
})
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)
err := script.Run(client, []string{key}, "value").Err()
Expect(err).NotTo(HaveOccurred())
}
wanted := []string{"keys=31", "keys=29", "keys=40"}
for i, master := range cluster.masters() {
Expect(master.Info().Val()).To(ContainSubstring(wanted[i]))
}
})
It("supports Watch", func() { It("supports Watch", func() {
var incr func(string) error var incr func(string) error
@ -342,60 +373,62 @@ var _ = Describe("ClusterClient", func() {
Expect(n).To(Equal(int64(100))) Expect(n).To(Equal(int64(100)))
}) })
It("supports pipeline", func() { Describe("pipeline", func() {
slot := hashtag.Slot("A") It("follows redirects", func() {
Expect(client.SwapSlotNodes(slot)).To(Equal([]string{"127.0.0.1:8224", "127.0.0.1:8221"})) slot := hashtag.Slot("A")
Expect(client.SwapSlotNodes(slot)).To(Equal([]string{"127.0.0.1:8224", "127.0.0.1:8221"}))
pipe := client.Pipeline() pipe := client.Pipeline()
defer pipe.Close() defer pipe.Close()
keys := []string{"A", "B", "C", "D", "E", "F", "G"} keys := []string{"A", "B", "C", "D", "E", "F", "G"}
for i, key := range keys { for i, key := range keys {
pipe.Set(key, key+"_value", 0) pipe.Set(key, key+"_value", 0)
pipe.Expire(key, time.Duration(i+1)*time.Hour) pipe.Expire(key, time.Duration(i+1)*time.Hour)
} }
cmds, err := pipe.Exec() cmds, err := pipe.Exec()
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
Expect(cmds).To(HaveLen(14)) Expect(cmds).To(HaveLen(14))
for _, key := range keys { for _, key := range keys {
pipe.Get(key) pipe.Get(key)
pipe.TTL(key) pipe.TTL(key)
} }
cmds, err = pipe.Exec() cmds, err = pipe.Exec()
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
Expect(cmds).To(HaveLen(14)) Expect(cmds).To(HaveLen(14))
Expect(cmds[0].(*redis.StringCmd).Val()).To(Equal("A_value")) Expect(cmds[0].(*redis.StringCmd).Val()).To(Equal("A_value"))
Expect(cmds[1].(*redis.DurationCmd).Val()).To(BeNumerically("~", 1*time.Hour, time.Second)) Expect(cmds[1].(*redis.DurationCmd).Val()).To(BeNumerically("~", 1*time.Hour, time.Second))
Expect(cmds[6].(*redis.StringCmd).Val()).To(Equal("D_value")) Expect(cmds[6].(*redis.StringCmd).Val()).To(Equal("D_value"))
Expect(cmds[7].(*redis.DurationCmd).Val()).To(BeNumerically("~", 4*time.Hour, time.Second)) Expect(cmds[7].(*redis.DurationCmd).Val()).To(BeNumerically("~", 4*time.Hour, time.Second))
Expect(cmds[12].(*redis.StringCmd).Val()).To(Equal("G_value")) Expect(cmds[12].(*redis.StringCmd).Val()).To(Equal("G_value"))
Expect(cmds[13].(*redis.DurationCmd).Val()).To(BeNumerically("~", 7*time.Hour, time.Second)) Expect(cmds[13].(*redis.DurationCmd).Val()).To(BeNumerically("~", 7*time.Hour, time.Second))
})
It("supports pipeline with missing keys", func() {
Expect(client.Set("A", "A_value", 0).Err()).NotTo(HaveOccurred())
Expect(client.Set("C", "C_value", 0).Err()).NotTo(HaveOccurred())
var a, b, c *redis.StringCmd
cmds, err := client.Pipelined(func(pipe *redis.Pipeline) error {
a = pipe.Get("A")
b = pipe.Get("B")
c = pipe.Get("C")
return nil
}) })
Expect(err).To(Equal(redis.Nil))
Expect(cmds).To(HaveLen(3))
Expect(a.Err()).NotTo(HaveOccurred()) It("works with missing keys", func() {
Expect(a.Val()).To(Equal("A_value")) Expect(client.Set("A", "A_value", 0).Err()).NotTo(HaveOccurred())
Expect(client.Set("C", "C_value", 0).Err()).NotTo(HaveOccurred())
Expect(b.Err()).To(Equal(redis.Nil)) var a, b, c *redis.StringCmd
Expect(b.Val()).To(Equal("")) cmds, err := client.Pipelined(func(pipe *redis.Pipeline) error {
a = pipe.Get("A")
b = pipe.Get("B")
c = pipe.Get("C")
return nil
})
Expect(err).To(Equal(redis.Nil))
Expect(cmds).To(HaveLen(3))
Expect(c.Err()).NotTo(HaveOccurred()) Expect(a.Err()).NotTo(HaveOccurred())
Expect(c.Val()).To(Equal("C_value")) 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"))
})
}) })
It("calls fn for every master node", func() { It("calls fn for every master node", func() {
@ -451,6 +484,25 @@ var _ = Describe("ClusterClient", func() {
describeClusterClient() describeClusterClient()
}) })
Describe("ClusterClient without nodes", func() {
BeforeEach(func() {
client = redis.NewClusterClient(&redis.ClusterOptions{})
})
It("returns an error", func() {
err := client.Ping().Err()
Expect(err).To(MatchError("redis: cluster has no nodes"))
})
It("pipeline returns an error", func() {
_, err := client.Pipelined(func(pipe *redis.Pipeline) error {
pipe.Ping()
return nil
})
Expect(err).To(MatchError("redis: cluster has no nodes"))
})
})
Describe("ClusterClient without valid nodes", func() { Describe("ClusterClient without valid nodes", func() {
BeforeEach(func() { BeforeEach(func() {
client = redis.NewClusterClient(&redis.ClusterOptions{ client = redis.NewClusterClient(&redis.ClusterOptions{
@ -462,6 +514,14 @@ var _ = Describe("ClusterClient", func() {
err := client.Ping().Err() err := client.Ping().Err()
Expect(err).To(MatchError("ERR This instance has cluster support disabled")) Expect(err).To(MatchError("ERR This instance has cluster support disabled"))
}) })
It("pipeline returns an error", func() {
_, err := client.Pipelined(func(pipe *redis.Pipeline) error {
pipe.Ping()
return nil
})
Expect(err).To(MatchError("ERR This instance has cluster support disabled"))
})
}) })
}) })

View File

@ -38,7 +38,7 @@ var _ = Describe("Redis Ring", func() {
Expect(ring.Close()).NotTo(HaveOccurred()) Expect(ring.Close()).NotTo(HaveOccurred())
}) })
It("uses both shards", func() { It("distributes keys", func() {
setRingKeys() setRingKeys()
// Both shards should have some keys now. // Both shards should have some keys now.
@ -46,6 +46,23 @@ var _ = Describe("Redis Ring", func() {
Expect(ringShard2.Info().Val()).To(ContainSubstring("keys=43")) Expect(ringShard2.Info().Val()).To(ContainSubstring("keys=43"))
}) })
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)
err := script.Run(ring, []string{key}, "value").Err()
Expect(err).NotTo(HaveOccurred())
}
Expect(ringShard1.Info().Val()).To(ContainSubstring("keys=57"))
Expect(ringShard2.Info().Val()).To(ContainSubstring("keys=43"))
})
It("uses single shard when one of the shards is down", func() { It("uses single shard when one of the shards is down", func() {
// Stop ringShard2. // Stop ringShard2.
Expect(ringShard2.Close()).NotTo(HaveOccurred()) Expect(ringShard2.Close()).NotTo(HaveOccurred())
@ -89,34 +106,8 @@ var _ = Describe("Redis Ring", func() {
Expect(ringShard2.Info().Val()).To(ContainSubstring("keys=100")) Expect(ringShard2.Info().Val()).To(ContainSubstring("keys=100"))
}) })
It("supports eval key search", func() { Describe("pipeline", func() {
script := redis.NewScript(` It("distributes keys", func() {
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)
err := script.Run(ring, []string{key}, "value").Err()
Expect(err).NotTo(HaveOccurred())
}
Expect(ringShard1.Info().Val()).To(ContainSubstring("keys=52"))
Expect(ringShard2.Info().Val()).To(ContainSubstring("keys=48"))
})
Describe("pipelining", func() {
It("returns an error when all shards are down", func() {
ring := redis.NewRing(&redis.RingOptions{})
_, err := ring.Pipelined(func(pipe *redis.Pipeline) error {
pipe.Ping()
return nil
})
Expect(err).To(MatchError("redis: all ring shards are down"))
})
It("uses both shards", func() {
pipe := ring.Pipeline() pipe := ring.Pipeline()
for i := 0; i < 100; i++ { for i := 0; i < 100; i++ {
err := pipe.Set(fmt.Sprintf("key%d", i), "value", 0).Err() err := pipe.Set(fmt.Sprintf("key%d", i), "value", 0).Err()
@ -175,3 +166,28 @@ var _ = Describe("Redis Ring", func() {
}) })
}) })
}) })
var _ = Describe("empty Redis Ring", func() {
var ring *redis.Ring
BeforeEach(func() {
ring = redis.NewRing(&redis.RingOptions{})
})
AfterEach(func() {
Expect(ring.Close()).NotTo(HaveOccurred())
})
It("returns an error", func() {
err := ring.Ping().Err()
Expect(err).To(MatchError("redis: all ring shards are down"))
})
It("pipeline returns an error", func() {
_, err := ring.Pipelined(func(pipe *redis.Pipeline) error {
pipe.Ping()
return nil
})
Expect(err).To(MatchError("redis: all ring shards are down"))
})
})