Skip to content

Commit 27fc6fa

Browse files
LINKIWIndyakov
andauthored
Accumulate connection pool wait statistics across all nodes (#3809)
Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com>
1 parent c615d30 commit 27fc6fa

2 files changed

Lines changed: 55 additions & 0 deletions

File tree

osscluster.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1450,6 +1450,8 @@ func (c *ClusterClient) PoolStats() *PoolStats {
14501450
acc.Hits += s.Hits
14511451
acc.Misses += s.Misses
14521452
acc.Timeouts += s.Timeouts
1453+
acc.WaitCount += s.WaitCount
1454+
acc.WaitDurationNs += s.WaitDurationNs
14531455

14541456
acc.TotalConns += s.TotalConns
14551457
acc.IdleConns += s.IdleConns
@@ -1461,6 +1463,8 @@ func (c *ClusterClient) PoolStats() *PoolStats {
14611463
acc.Hits += s.Hits
14621464
acc.Misses += s.Misses
14631465
acc.Timeouts += s.Timeouts
1466+
acc.WaitCount += s.WaitCount
1467+
acc.WaitDurationNs += s.WaitDurationNs
14641468

14651469
acc.TotalConns += s.TotalConns
14661470
acc.IdleConns += s.IdleConns

osscluster_test.go

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -733,6 +733,57 @@ var _ = Describe("ClusterClient", func() {
733733
Expect(stats).To(BeAssignableToTypeOf(&redis.PoolStats{}))
734734
})
735735

736+
It("should sum pool wait statistics across all nodes", func() {
737+
// Set a unity pool size to force a wait on any concurrently dispatched requests.
738+
opt := redisClusterOptions()
739+
opt.PoolSize = 1
740+
opt.MinIdleConns = 0
741+
opt.PoolTimeout = 5 * time.Second
742+
client := cluster.newClusterClient(ctx, opt)
743+
744+
state, err := client.LoadState(ctx)
745+
Expect(err).NotTo(HaveOccurred())
746+
Expect(state.Masters).NotTo(BeEmpty())
747+
748+
// Deliberately contend connection acquisition on all per-node connection pools.
749+
for _, master := range state.Masters {
750+
nodePool := master.Client.Pool()
751+
cn, err := nodePool.Get(ctx)
752+
Expect(err).NotTo(HaveOccurred())
753+
time.AfterFunc(100*time.Millisecond, func() {
754+
nodePool.Put(ctx, cn)
755+
})
756+
757+
// Get will block until a connection is available; this synchronizes on the Put
758+
// of the connection obtained from the initial acquisition
759+
cn2, err := nodePool.Get(ctx)
760+
Expect(err).NotTo(HaveOccurred())
761+
nodePool.Put(ctx, cn2)
762+
}
763+
764+
var expectedWaits uint32
765+
var expectedWaitDuration int64
766+
767+
for _, master := range state.Masters {
768+
s := master.Client.Pool().Stats()
769+
expectedWaits += s.WaitCount
770+
expectedWaitDuration += s.WaitDurationNs
771+
}
772+
773+
for _, slave := range state.Slaves {
774+
s := slave.Client.Pool().Stats()
775+
expectedWaits += s.WaitCount
776+
expectedWaitDuration += s.WaitDurationNs
777+
}
778+
779+
Expect(expectedWaits).To(BeNumerically(">=", uint32(len(state.Masters))))
780+
Expect(expectedWaitDuration).To(BeNumerically(">", int64(0)))
781+
782+
stats := client.PoolStats()
783+
Expect(stats.WaitCount).To(Equal(expectedWaits))
784+
Expect(stats.WaitDurationNs).To(Equal(expectedWaitDuration))
785+
})
786+
736787
It("should return an error when there are no attempts left", func() {
737788
opt := redisClusterOptions()
738789
opt.MaxRedirects = -1

0 commit comments

Comments
 (0)