From 7cbee9d33726d49b95fda0cdd047ad4f761a1e91 Mon Sep 17 00:00:00 2001 From: Vladimir Mihailenco Date: Sun, 2 Oct 2016 12:44:01 +0000 Subject: [PATCH] Enable reaper on ClusterClient and add tests. --- cluster.go | 39 ++++++---- cluster_test.go | 129 ++++++++----------------------- commands_test.go | 2 +- internal/hashtag/hashtag_test.go | 49 ++++++++++++ internal/pool/pool.go | 55 +++++++------ internal/pool/pool_single.go | 2 +- internal/pool/pool_sticky.go | 4 +- main_test.go | 32 +++++++- pool_test.go | 36 +++++++-- ring.go | 55 ++++++++++++- ring_test.go | 23 +++--- 11 files changed, 257 insertions(+), 169 deletions(-) diff --git a/cluster.go b/cluster.go index 402447e..ad56af7 100644 --- a/cluster.go +++ b/cluster.go @@ -45,20 +45,25 @@ var _ Cmdable = (*ClusterClient)(nil) // http://redis.io/topics/cluster-spec. func NewClusterClient(opt *ClusterOptions) *ClusterClient { opt.init() - client := &ClusterClient{ + + c := &ClusterClient{ opt: opt, nodes: make(map[string]*clusterNode), cmdsInfoOnce: new(sync.Once), } - client.cmdable.process = client.Process + c.cmdable.process = c.Process for _, addr := range opt.Addrs { - _, _ = client.nodeByAddr(addr) + _, _ = c.nodeByAddr(addr) } - client.reloadSlots() + c.reloadSlots() - return client + if opt.IdleCheckFrequency > 0 { + go c.reaper(opt.IdleCheckFrequency) + } + + return c } func (c *ClusterClient) cmdInfo(name string) *CommandInfo { @@ -333,11 +338,9 @@ func (c *ClusterClient) ForEachMaster(fn func(client *Client) error) error { slots := c.slots c.mu.RUnlock() - var retErr error - var mu sync.Mutex - var wg sync.WaitGroup visited := make(map[*clusterNode]struct{}) + errCh := make(chan error, 1) for _, nodes := range slots { if len(nodes) == 0 { continue @@ -351,20 +354,24 @@ func (c *ClusterClient) ForEachMaster(fn func(client *Client) error) error { wg.Add(1) go func(node *clusterNode) { + defer wg.Done() err := fn(node.Client) if err != nil { - mu.Lock() - if retErr == nil { - retErr = err + select { + case errCh <- err: + default: } - mu.Unlock() } - wg.Done() }(master) } wg.Wait() - return retErr + select { + case err := <-errCh: + return err + default: + return nil + } } // closeClients closes all clients and returns the first error if there are any. @@ -442,8 +449,8 @@ func (c *ClusterClient) setNodesLatency() { } // reaper closes idle connections to the cluster. -func (c *ClusterClient) reaper(frequency time.Duration) { - ticker := time.NewTicker(frequency) +func (c *ClusterClient) reaper(idleCheckFrequency time.Duration) { + ticker := time.NewTicker(idleCheckFrequency) defer ticker.Stop() for _ = range ticker.C { diff --git a/cluster_test.go b/cluster_test.go index 659e134..74b260f 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -2,7 +2,6 @@ package redis_test import ( "fmt" - "math/rand" "net" "strconv" "strings" @@ -50,17 +49,6 @@ func (s *clusterScenario) clusterClient(opt *redis.ClusterOptions) *redis.Cluste for i, port := range s.ports { addrs[i] = net.JoinHostPort("127.0.0.1", port) } - if opt == nil { - opt = &redis.ClusterOptions{ - DialTimeout: 10 * time.Second, - ReadTimeout: 30 * time.Second, - WriteTimeout: 30 * time.Second, - PoolSize: 10, - PoolTimeout: 30 * time.Second, - IdleTimeout: time.Second, - IdleCheckFrequency: time.Second, - } - } opt.Addrs = addrs return redis.NewClusterClient(opt) } @@ -193,52 +181,12 @@ func stopCluster(scenario *clusterScenario) error { //------------------------------------------------------------------------------ -var _ = Describe("Cluster", func() { - Describe("HashSlot", func() { - - It("should calculate hash slots", func() { - tests := []struct { - key string - slot int - }{ - {"123456789", 12739}, - {"{}foo", 9500}, - {"foo{}", 5542}, - {"foo{}{bar}", 8363}, - {"", 10503}, - {"", 5176}, - {string([]byte{83, 153, 134, 118, 229, 214, 244, 75, 140, 37, 215, 215}), 5463}, - } - // Empty keys receive random slot. - rand.Seed(100) - - for _, test := range tests { - Expect(hashtag.Slot(test.key)).To(Equal(test.slot), "for %s", test.key) - } - }) - - It("should extract keys from tags", func() { - tests := []struct { - one, two string - }{ - {"foo{bar}", "bar"}, - {"{foo}bar", "foo"}, - {"{user1000}.following", "{user1000}.followers"}, - {"foo{{bar}}zap", "{bar"}, - {"foo{bar}{zap}", "bar"}, - } - - for _, test := range tests { - Expect(hashtag.Slot(test.one)).To(Equal(hashtag.Slot(test.two)), "for %s <-> %s", test.one, test.two) - } - }) - - }) - - Describe("Commands", func() { +var _ = Describe("ClusterClient", func() { + var client *redis.ClusterClient + describeClusterClient := func() { It("should CLUSTER SLOTS", func() { - res, err := cluster.primary().ClusterSlots().Result() + res, err := client.ClusterSlots().Result() Expect(err).NotTo(HaveOccurred()) Expect(res).To(HaveLen(3)) @@ -251,73 +199,48 @@ var _ = Describe("Cluster", func() { }) It("should CLUSTER NODES", func() { - res, err := cluster.primary().ClusterNodes().Result() + res, err := client.ClusterNodes().Result() Expect(err).NotTo(HaveOccurred()) Expect(len(res)).To(BeNumerically(">", 400)) }) It("should CLUSTER INFO", func() { - res, err := cluster.primary().ClusterInfo().Result() + res, err := client.ClusterInfo().Result() Expect(err).NotTo(HaveOccurred()) Expect(res).To(ContainSubstring("cluster_known_nodes:6")) }) It("should CLUSTER KEYSLOT", func() { - hashSlot, err := cluster.primary().ClusterKeySlot("somekey").Result() + hashSlot, err := client.ClusterKeySlot("somekey").Result() Expect(err).NotTo(HaveOccurred()) Expect(hashSlot).To(Equal(int64(hashtag.Slot("somekey")))) }) It("should CLUSTER COUNT-FAILURE-REPORTS", func() { - n, err := cluster.primary().ClusterCountFailureReports(cluster.nodeIds[0]).Result() + n, err := client.ClusterCountFailureReports(cluster.nodeIds[0]).Result() Expect(err).NotTo(HaveOccurred()) Expect(n).To(Equal(int64(0))) }) It("should CLUSTER COUNTKEYSINSLOT", func() { - n, err := cluster.primary().ClusterCountKeysInSlot(10).Result() + n, err := client.ClusterCountKeysInSlot(10).Result() Expect(err).NotTo(HaveOccurred()) Expect(n).To(Equal(int64(0))) }) - It("should CLUSTER DELSLOTS", func() { - res, err := cluster.primary().ClusterDelSlotsRange(16000, 16384-1).Result() - Expect(err).NotTo(HaveOccurred()) - Expect(res).To(Equal("OK")) - cluster.primary().ClusterAddSlotsRange(16000, 16384-1) - }) - It("should CLUSTER SAVECONFIG", func() { - res, err := cluster.primary().ClusterSaveConfig().Result() + res, err := client.ClusterSaveConfig().Result() Expect(err).NotTo(HaveOccurred()) Expect(res).To(Equal("OK")) }) It("should CLUSTER SLAVES", func() { - nodesList, err := cluster.primary().ClusterSlaves(cluster.nodeIds[0]).Result() + nodesList, err := client.ClusterSlaves(cluster.nodeIds[0]).Result() Expect(err).NotTo(HaveOccurred()) Expect(nodesList).Should(ContainElement(ContainSubstring("slave"))) Expect(nodesList).Should(HaveLen(1)) }) - // It("should CLUSTER READONLY", func() { - // res, err := cluster.primary().ReadOnly().Result() - // Expect(err).NotTo(HaveOccurred()) - // Expect(res).To(Equal("OK")) - // }) - - // It("should CLUSTER READWRITE", func() { - // res, err := cluster.primary().ReadWrite().Result() - // Expect(err).NotTo(HaveOccurred()) - // Expect(res).To(Equal("OK")) - // }) - }) -}) - -var _ = Describe("ClusterClient", func() { - var client *redis.ClusterClient - - describeClusterClient := func() { It("should GET/SET/DEL", func() { val, err := client.Get("A").Result() Expect(err).To(Equal(redis.Nil)) @@ -340,6 +263,18 @@ var _ = Describe("ClusterClient", func() { Expect(client.PoolStats()).To(BeAssignableToTypeOf(&redis.PoolStats{})) }) + It("removes idle connections", func() { + stats := client.PoolStats() + Expect(stats.TotalConns).NotTo(BeZero()) + Expect(stats.FreeConns).NotTo(BeZero()) + + time.Sleep(2 * time.Second) + + stats = client.PoolStats() + Expect(stats.TotalConns).To(BeZero()) + Expect(stats.FreeConns).To(BeZero()) + }) + It("follows redirects", func() { Expect(client.Set("A", "VALUE", 0).Err()).NotTo(HaveOccurred()) @@ -352,9 +287,9 @@ var _ = Describe("ClusterClient", func() { }) It("returns an error when there are no attempts left", func() { - client := cluster.clusterClient(&redis.ClusterOptions{ - MaxRedirects: -1, - }) + opt := redisClusterOptions() + opt.MaxRedirects = -1 + client := cluster.clusterClient(opt) slot := hashtag.Slot("A") Expect(client.SwapSlotNodes(slot)).To(Equal([]string{"127.0.0.1:8224", "127.0.0.1:8221"})) @@ -483,7 +418,7 @@ var _ = Describe("ClusterClient", func() { Describe("default ClusterClient", func() { BeforeEach(func() { - client = cluster.clusterClient(nil) + client = cluster.clusterClient(redisClusterOptions()) _ = client.ForEachMaster(func(master *redis.Client) error { return master.FlushDb().Err() @@ -499,9 +434,9 @@ var _ = Describe("ClusterClient", func() { Describe("ClusterClient with RouteByLatency", func() { BeforeEach(func() { - client = cluster.clusterClient(&redis.ClusterOptions{ - RouteByLatency: true, - }) + opt := redisClusterOptions() + opt.RouteByLatency = true + client = cluster.clusterClient(opt) _ = client.ForEachMaster(func(master *redis.Client) error { return master.FlushDb().Err() @@ -543,11 +478,13 @@ func BenchmarkRedisClusterPing(b *testing.B) { processes: make(map[string]*redisProcess, 6), clients: make(map[string]*redis.Client, 6), } + if err := startCluster(cluster); err != nil { b.Fatal(err) } defer stopCluster(cluster) - client := cluster.clusterClient(nil) + + client := cluster.clusterClient(redisClusterOptions()) defer client.Close() b.ResetTimer() diff --git a/commands_test.go b/commands_test.go index ea33e1b..c89c333 100644 --- a/commands_test.go +++ b/commands_test.go @@ -1297,7 +1297,7 @@ var _ = Describe("Commands", func() { stats := client.PoolStats() Expect(stats.Requests).To(Equal(uint32(3))) - Expect(stats.Hits).To(Equal(uint32(2))) + Expect(stats.Hits).To(Equal(uint32(1))) Expect(stats.Timeouts).To(Equal(uint32(0))) }) diff --git a/internal/hashtag/hashtag_test.go b/internal/hashtag/hashtag_test.go index 8132878..7f0fedf 100644 --- a/internal/hashtag/hashtag_test.go +++ b/internal/hashtag/hashtag_test.go @@ -1,10 +1,18 @@ package hashtag import ( + "math/rand" + "testing" + . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" ) +func TestGinkgoSuite(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "hashtag") +} + var _ = Describe("CRC16", func() { // http://redis.io/topics/cluster-spec#keys-distribution-model @@ -23,3 +31,44 @@ var _ = Describe("CRC16", func() { }) }) + +var _ = Describe("HashSlot", func() { + + It("should calculate hash slots", func() { + tests := []struct { + key string + slot int + }{ + {"123456789", 12739}, + {"{}foo", 9500}, + {"foo{}", 5542}, + {"foo{}{bar}", 8363}, + {"", 10503}, + {"", 5176}, + {string([]byte{83, 153, 134, 118, 229, 214, 244, 75, 140, 37, 215, 215}), 5463}, + } + // Empty keys receive random slot. + rand.Seed(100) + + for _, test := range tests { + Expect(Slot(test.key)).To(Equal(test.slot), "for %s", test.key) + } + }) + + It("should extract keys from tags", func() { + tests := []struct { + one, two string + }{ + {"foo{bar}", "bar"}, + {"{foo}bar", "foo"}, + {"{user1000}.following", "{user1000}.followers"}, + {"foo{{bar}}zap", "{bar"}, + {"foo{bar}{zap}", "bar"}, + } + + for _, test := range tests { + Expect(Slot(test.one)).To(Equal(Slot(test.two)), "for %s <-> %s", test.one, test.two) + } + }) + +}) diff --git a/internal/pool/pool.go b/internal/pool/pool.go index 55c1f9f..bd1c413 100644 --- a/internal/pool/pool.go +++ b/internal/pool/pool.go @@ -25,8 +25,8 @@ var timers = sync.Pool{ }, } -// PoolStats contains pool state information and accumulated stats. -type PoolStats struct { +// Stats contains pool state information and accumulated stats. +type Stats struct { Requests uint32 // number of times a connection was requested by the pool Hits uint32 // number of times free connection was found in the pool Timeouts uint32 // number of times a wait timeout occurred @@ -41,7 +41,7 @@ type Pooler interface { Remove(*Conn, error) error Len() int FreeLen() int - Stats() *PoolStats + Stats() *Stats Close() error Closed() bool } @@ -64,7 +64,7 @@ type ConnPool struct { freeConnsMu sync.Mutex freeConns []*Conn - stats PoolStats + stats Stats _closed int32 // atomic lastErr atomic.Value @@ -173,16 +173,22 @@ func (p *ConnPool) Get() (*Conn, bool, error) { return nil, false, ErrPoolTimeout } - p.freeConnsMu.Lock() - cn := p.popFree() - p.freeConnsMu.Unlock() + for { + p.freeConnsMu.Lock() + cn := p.popFree() + p.freeConnsMu.Unlock() - if cn != nil { - atomic.AddUint32(&p.stats.Hits, 1) - if !cn.IsStale(p.idleTimeout) { - return cn, false, nil + if cn == nil { + break } - _ = p.closeConn(cn, errConnStale) + + if cn.IsStale(p.idleTimeout) { + p.remove(cn, errConnStale) + continue + } + + atomic.AddUint32(&p.stats.Hits, 1) + return cn, false, nil } newcn, err := p.NewConn() @@ -192,9 +198,6 @@ func (p *ConnPool) Get() (*Conn, bool, error) { } p.connsMu.Lock() - if cn != nil { - p.removeConn(cn) - } p.conns = append(p.conns, newcn) p.connsMu.Unlock() @@ -224,17 +227,13 @@ func (p *ConnPool) remove(cn *Conn, reason error) { _ = p.closeConn(cn, reason) p.connsMu.Lock() - p.removeConn(cn) - p.connsMu.Unlock() -} - -func (p *ConnPool) removeConn(cn *Conn) { for i, c := range p.conns { if c == cn { p.conns = append(p.conns[:i], p.conns[i+1:]...) break } } + p.connsMu.Unlock() } // Len returns total number of connections. @@ -253,14 +252,14 @@ func (p *ConnPool) FreeLen() int { return l } -func (p *ConnPool) Stats() *PoolStats { - stats := PoolStats{} - stats.Requests = atomic.LoadUint32(&p.stats.Requests) - stats.Hits = atomic.LoadUint32(&p.stats.Hits) - stats.Timeouts = atomic.LoadUint32(&p.stats.Timeouts) - stats.TotalConns = uint32(p.Len()) - stats.FreeConns = uint32(p.FreeLen()) - return &stats +func (p *ConnPool) Stats() *Stats { + return &Stats{ + Requests: atomic.LoadUint32(&p.stats.Requests), + Hits: atomic.LoadUint32(&p.stats.Hits), + Timeouts: atomic.LoadUint32(&p.stats.Timeouts), + TotalConns: uint32(p.Len()), + FreeConns: uint32(p.FreeLen()), + } } func (p *ConnPool) Closed() bool { diff --git a/internal/pool/pool_single.go b/internal/pool/pool_single.go index 0cf6c7c..18ca616 100644 --- a/internal/pool/pool_single.go +++ b/internal/pool/pool_single.go @@ -42,7 +42,7 @@ func (p *SingleConnPool) FreeLen() int { return 0 } -func (p *SingleConnPool) Stats() *PoolStats { +func (p *SingleConnPool) Stats() *Stats { return nil } diff --git a/internal/pool/pool_sticky.go b/internal/pool/pool_sticky.go index a2649e5..ce45f4b 100644 --- a/internal/pool/pool_sticky.go +++ b/internal/pool/pool_sticky.go @@ -106,7 +106,9 @@ func (p *StickyConnPool) FreeLen() int { return 0 } -func (p *StickyConnPool) Stats() *PoolStats { return nil } +func (p *StickyConnPool) Stats() *Stats { + return nil +} func (p *StickyConnPool) Close() error { defer p.mx.Unlock() diff --git a/main_test.go b/main_test.go index 5208903..a9afa82 100644 --- a/main_test.go +++ b/main_test.go @@ -108,8 +108,36 @@ func redisOptions() *redis.Options { WriteTimeout: 30 * time.Second, PoolSize: 10, PoolTimeout: 30 * time.Second, - IdleTimeout: time.Second, - IdleCheckFrequency: time.Second, + IdleTimeout: 500 * time.Millisecond, + IdleCheckFrequency: 500 * time.Millisecond, + } +} + +func redisClusterOptions() *redis.ClusterOptions { + return &redis.ClusterOptions{ + DialTimeout: 10 * time.Second, + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + PoolSize: 10, + PoolTimeout: 30 * time.Second, + IdleTimeout: 500 * time.Millisecond, + IdleCheckFrequency: 500 * time.Millisecond, + } +} + +func redisRingOptions() *redis.RingOptions { + return &redis.RingOptions{ + Addrs: map[string]string{ + "ringShardOne": ":" + ringShard1Port, + "ringShardTwo": ":" + ringShard2Port, + }, + DialTimeout: 10 * time.Second, + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + PoolSize: 10, + PoolTimeout: 30 * time.Second, + IdleTimeout: 500 * time.Millisecond, + IdleCheckFrequency: 500 * time.Millisecond, } } diff --git a/pool_test.go b/pool_test.go index 13edd70..2b0d2ad 100644 --- a/pool_test.go +++ b/pool_test.go @@ -1,6 +1,8 @@ package redis_test import ( + "time" + . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" @@ -20,7 +22,7 @@ var _ = Describe("pool", func() { Expect(client.Close()).NotTo(HaveOccurred()) }) - It("should respect max size", func() { + It("respects max size", func() { perform(1000, func(id int) { val, err := client.Ping().Result() Expect(err).NotTo(HaveOccurred()) @@ -33,7 +35,7 @@ var _ = Describe("pool", func() { Expect(pool.Len()).To(Equal(pool.FreeLen())) }) - It("should respect max on multi", func() { + It("srespect max size on multi", func() { perform(1000, func(id int) { var ping *redis.StatusCmd @@ -58,7 +60,7 @@ var _ = Describe("pool", func() { Expect(pool.Len()).To(Equal(pool.FreeLen())) }) - It("should respect max on pipelines", func() { + It("respects max size on pipelines", func() { perform(1000, func(id int) { pipe := client.Pipeline() ping := pipe.Ping() @@ -76,7 +78,7 @@ var _ = Describe("pool", func() { Expect(pool.Len()).To(Equal(pool.FreeLen())) }) - It("should respect max on pubsub", func() { + It("respects max size on pubsub", func() { connPool := client.Pool() connPool.(*pool.ConnPool).DialLimiter = nil @@ -90,7 +92,7 @@ var _ = Describe("pool", func() { Expect(connPool.Len()).To(BeNumerically("<=", 10)) }) - It("should remove broken connections", func() { + It("removes broken connections", func() { cn, _, err := client.Pool().Get() Expect(err).NotTo(HaveOccurred()) cn.NetConn = &badConn{} @@ -113,7 +115,7 @@ var _ = Describe("pool", func() { Expect(stats.Timeouts).To(Equal(uint32(0))) }) - It("should reuse connections", func() { + It("reuses connections", func() { for i := 0; i < 100; i++ { val, err := client.Ping().Result() Expect(err).NotTo(HaveOccurred()) @@ -129,4 +131,26 @@ var _ = Describe("pool", func() { Expect(stats.Hits).To(Equal(uint32(100))) Expect(stats.Timeouts).To(Equal(uint32(0))) }) + + It("removes idle connections", func() { + stats := client.PoolStats() + Expect(stats).To(Equal(&redis.PoolStats{ + Requests: 1, + Hits: 0, + Timeouts: 0, + TotalConns: 1, + FreeConns: 1, + })) + + time.Sleep(2 * time.Second) + + stats = client.PoolStats() + Expect(stats).To(Equal(&redis.PoolStats{ + Requests: 1, + Hits: 0, + Timeouts: 0, + TotalConns: 0, + FreeConns: 0, + })) + }) }) diff --git a/ring.go b/ring.go index 570c535..b0bba58 100644 --- a/ring.go +++ b/ring.go @@ -4,6 +4,7 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "time" "gopkg.in/redis.v4/internal" @@ -65,7 +66,7 @@ func (opt *RingOptions) clientOptions() *Options { type ringShard struct { Client *Client - down int + down int32 } func (shard *ringShard) String() string { @@ -80,7 +81,7 @@ func (shard *ringShard) String() string { func (shard *ringShard) IsDown() bool { const threshold = 3 - return shard.down >= threshold + return atomic.LoadInt32(&shard.down) >= threshold } func (shard *ringShard) IsUp() bool { @@ -91,7 +92,7 @@ func (shard *ringShard) IsUp() bool { func (shard *ringShard) Vote(up bool) bool { if up { changed := shard.IsDown() - shard.down = 0 + atomic.StoreInt32(&shard.down, 0) return changed } @@ -99,7 +100,7 @@ func (shard *ringShard) Vote(up bool) bool { return false } - shard.down++ + atomic.AddInt32(&shard.down, 1) return shard.IsDown() } @@ -157,6 +158,52 @@ func NewRing(opt *RingOptions) *Ring { return ring } +// PoolStats returns accumulated connection pool stats. +func (c *Ring) PoolStats() *PoolStats { + var acc PoolStats + for _, shard := range c.shards { + s := shard.Client.connPool.Stats() + acc.Requests += s.Requests + acc.Hits += s.Hits + acc.Timeouts += s.Timeouts + acc.TotalConns += s.TotalConns + acc.FreeConns += s.FreeConns + } + return &acc +} + +// ForEachShard concurrently calls the fn on each live shard in the ring. +// It returns the first error if any. +func (c *Ring) ForEachShard(fn func(client *Client) error) error { + var wg sync.WaitGroup + errCh := make(chan error, 1) + for _, shard := range c.shards { + if shard.IsDown() { + continue + } + + wg.Add(1) + go func(shard *ringShard) { + defer wg.Done() + err := fn(shard.Client) + if err != nil { + select { + case errCh <- err: + default: + } + } + }(shard) + } + wg.Wait() + + select { + case err := <-errCh: + return err + default: + return nil + } +} + func (c *Ring) cmdInfo(name string) *CommandInfo { c.cmdsInfoOnce.Do(func() { for _, shard := range c.shards { diff --git a/ring_test.go b/ring_test.go index a63f5c8..3c06686 100644 --- a/ring_test.go +++ b/ring_test.go @@ -11,8 +11,9 @@ import ( "gopkg.in/redis.v4" ) -var _ = Describe("Redis ring", func() { +var _ = Describe("Redis Ring", func() { const heartbeat = 100 * time.Millisecond + var ring *redis.Ring setRingKeys := func() { @@ -23,20 +24,14 @@ var _ = Describe("Redis ring", func() { } BeforeEach(func() { - ring = redis.NewRing(&redis.RingOptions{ - Addrs: map[string]string{ - "ringShardOne": ":" + ringShard1Port, - "ringShardTwo": ":" + ringShard2Port, - }, - HeartbeatFrequency: heartbeat, + opt := redisRingOptions() + opt.HeartbeatFrequency = heartbeat + ring = redis.NewRing(opt) + + err := ring.ForEachShard(func(cl *redis.Client) error { + return cl.FlushDb().Err() }) - - // Shards should not have any keys. - Expect(ringShard1.FlushDb().Err()).NotTo(HaveOccurred()) - Expect(ringShard1.Info().Val()).NotTo(ContainSubstring("keys=")) - - Expect(ringShard2.FlushDb().Err()).NotTo(HaveOccurred()) - Expect(ringShard2.Info().Val()).NotTo(ContainSubstring("keys=")) + Expect(err).NotTo(HaveOccurred()) }) AfterEach(func() {