diff --git a/osscluster.go b/osscluster.go index e2b9f973..cf6fba9b 100644 --- a/osscluster.go +++ b/osscluster.go @@ -62,6 +62,17 @@ type ClusterOptions struct { // Allows routing read-only commands to the closest master or slave node. // It automatically enables ReadOnly. RouteByLatency bool + // RouteByLatencyTolerance widens RouteByLatency from "the single fastest node" to + // "any node within this much of the fastest", round-robining between them. + // + // RouteByLatency alone takes a strict minimum, so when several nodes are equally + // close - replicas sharing an availability zone, say - every client picks the same + // one and the others take no read traffic. Latency is estimated from ten pings and + // refreshed at most every 10s, so a difference well inside the noise can decide the + // whole read load for the next interval. + // + // Zero keeps the strict-minimum behaviour. Has no effect unless RouteByLatency is set. + RouteByLatencyTolerance time.Duration // Allows routing read-only commands to the random master or slave node. // It automatically enables ReadOnly. RouteRandomly bool @@ -379,6 +390,7 @@ func setupClusterQueryParams(u *url.URL, o *ClusterOptions) (*ClusterOptions, er o.MaxRedirects = q.int("max_redirects") o.ReadOnly = q.bool("read_only") o.RouteByLatency = q.bool("route_by_latency") + o.RouteByLatencyTolerance = q.duration("route_by_latency_tolerance") o.RouteRandomly = q.bool("route_randomly") o.MaxRetries = q.int("max_retries") o.MinRetryBackoff = q.duration("min_retry_backoff") @@ -518,7 +530,7 @@ func newClusterNodeWithNodeAddress(clOpt *ClusterOptions, addr, nodeAddress stri Client: clOpt.NewClient(opt), } - node.latency.Store(math.MaxUint32) + node.latency.Store(unmeasuredNodeLatencyMicros) if clOpt.RouteByLatency { go node.updateLatency() } @@ -536,6 +548,14 @@ func (n *clusterNode) Close() error { const maximumNodeLatency = 1 * time.Minute +// Latency held by a node from creation until its first probe completes. Deliberately distinct +// from maximumNodeLatency, which marks a node whose pings all failed - the two must stay +// separable, so this is compared for equality rather than as a ">= huge" threshold. +const ( + unmeasuredNodeLatencyMicros uint32 = math.MaxUint32 + unmeasuredNodeLatency = time.Duration(unmeasuredNodeLatencyMicros) * time.Microsecond +) + func (n *clusterNode) updateLatency() { const numProbe = 10 var dur uint64 @@ -817,6 +837,13 @@ type clusterSlot struct { start int end int nodes []*clusterNode + + // Round-robin cursor over the nodes inside the RouteByLatencyTolerance band. Latency + // decides only which nodes are in the band; within it they are treated as equally close + // and picked in turn, so this does not order them. Per slot on purpose: a counter shared + // across slots is advanced by the other slots between two visits to this one, so with a + // regular interleaving each slot keeps landing on the same candidate index. + latencyBandNodeCursor atomic.Uint32 } type clusterState struct { @@ -1027,6 +1054,147 @@ func (c *clusterState) slotClosestNode(slot int) (*clusterNode, error) { return c.nodes.Random() } +// slotNodeWithinLatency picks from the healthy nodes whose latency is within tolerance of the +// fastest one, rotating across them so equally-close nodes share the read traffic. Used only +// when RouteByLatencyTolerance is set; slotClosestNode keeps the strict-minimum behaviour and +// is left untouched for everyone else. +func (c *clusterState) slotNodeWithinLatency(slot int, tolerance time.Duration) (*clusterNode, error) { + entry := c.slotEntry(slot) + if entry == nil || len(entry.nodes) == 0 { + return c.nodes.Random() + } + nodes := entry.nodes + + // Latency and health are sampled once per node. The background probe updates them + // concurrently, so re-reading would let the candidate set disagree with the minimum it + // is compared against - and could leave that set empty. + type sample struct { + node *clusterNode + latency time.Duration + } + + var ( + healthy = make([]sample, 0, len(nodes)) + closestNode *clusterNode + closestHealthyNode *clusterNode + minLatency = time.Duration(math.MaxInt64) + minHealthyLatency = time.Duration(math.MaxInt64) + anyHealthyMeasured bool + ) + + for _, n := range nodes { + latency := n.Latency() + if latency < minLatency { + closestNode, minLatency = n, latency + } + + if n.Failing() { + continue + } + + healthy = append(healthy, sample{node: n, latency: latency}) + + // Derived from the latency just captured, not from a second read of the node. + // updateLatency publishes the latency and its timestamp as two separate stores, so + // reading the timestamp here could observe a probe that landed after the latency + // above was sampled - marking the set measured while every sampled latency is still + // a sentinel, which is exactly the case this guards. + if latency != unmeasuredNodeLatency { + anyHealthyMeasured = true + } + + // Tracked separately: a healthy node that is not the outright fastest must still be + // preferred over a failing one. + if latency < minHealthyLatency { + closestHealthyNode, minHealthyLatency = n, latency + } + } + + if closestHealthyNode != nil { + // Until the first probe lands, every node still holds the sentinel latency stored by + // newClusterNodeWithNodeAddress, so every difference is zero and any positive tolerance + // would admit the whole slot - scattering startup reads across distant zones instead of + // preserving locality. Keep strict selection until at least one measurement exists; a + // mix needs no special case, since an unmeasured node's sentinel latency puts it far + // outside the band of any measured one. + if !anyHealthyMeasured { + return closestHealthyNode, nil + } + + candidates := make([]*clusterNode, 0, len(healthy)) + for _, s := range healthy { + // Subtraction rather than minHealthyLatency+tolerance, which would overflow for + // a very large tolerance. Always true for closestHealthyNode, so candidates is + // never empty here. + if s.latency-minHealthyLatency <= tolerance { + candidates = append(candidates, s.node) + } + } + + // Drop nodes that are still loading, matching slotSlaveNode. Checked here rather than + // in the pass above so Loading() - which can cost a Ping when the node is not known + // loaded - is only paid for nodes actually eligible for this read. Widening the + // candidate set is what makes this matter: under a strict minimum a loading replica + // has to be the fastest to be picked, within a tolerance band it merely has to be + // close, and a replica is loading precisely after the full resync that a too-small + // replication backlog causes. + ready := candidates[:0] + for _, n := range candidates { + if !n.Loading() { + ready = append(ready, n) + } + } + + if len(ready) == 0 { + // Every node inside the band is loading. closestHealthyNode is itself in the band, + // so returning it hands back a node we just observed loading. Widen the search to + // the healthy nodes outside the band and take the closest ready one - which in a + // multi-AZ layout is typically the master, still serving while the local replicas + // reload together after a resync. Only reached in this exceptional case, so the + // extra Loading() calls are off the hot path, and the latency comparison is + // evaluated first so most of them are skipped. + var ( + fallback *clusterNode + fallbackLatency = time.Duration(math.MaxInt64) + ) + for _, s := range healthy { + if s.latency-minHealthyLatency <= tolerance { + continue // in the band, already known to be loading + } + if s.latency < fallbackLatency && !s.node.Loading() { + fallback, fallbackLatency = s.node, s.latency + } + } + if fallback != nil { + return fallback, nil + } + + // Nothing is ready anywhere. Return the closest healthy node so the caller still + // gets a node to retry against, which is what this path did before the filter. + return closestHealthyNode, nil + } + + if len(ready) == 1 { + return ready[0], nil + } + + // Reduced in the unsigned domain: the cursor wraps, and on 32-bit builds converting a + // value past 2^31 to int before the modulo would make the index negative. + return ready[(entry.latencyBandNodeCursor.Add(1)-1)%uint32(len(ready))], nil + } + + // Every node is failing. Fall back to the least-slow one, so a transient failure does + // not take the slot down. + if minLatency < maximumNodeLatency && closestNode != nil { + internal.Logger.Printf(context.TODO(), "redis: all nodes are marked as failed, picking the temporarily failing node with lowest latency") + return closestNode, nil + } + + // If all nodes are having the maximum latency(all pings are failing) - return a random node across the cluster + internal.Logger.Printf(context.TODO(), "redis: pings to all nodes are failing, picking a random node across the cluster") + return c.nodes.Random() +} + func (c *clusterState) slotRandomNode(slot int) (*clusterNode, error) { nodes := c.slotNodes(slot) if len(nodes) == 0 { @@ -1067,7 +1235,7 @@ func (c *clusterState) slotShardPickerSlaveNode(slot int, shardPicker routing.Sh return nodes[0], nil } -func (c *clusterState) slotNodes(slot int) []*clusterNode { +func (c *clusterState) slotEntry(slot int) *clusterSlot { i := sort.Search(len(c.slots), func(i int) bool { return c.slots[i].end >= slot }) @@ -1076,6 +1244,13 @@ func (c *clusterState) slotNodes(slot int) []*clusterNode { } x := c.slots[i] if slot >= x.start && slot <= x.end { + return x + } + return nil +} + +func (c *clusterState) slotNodes(slot int) []*clusterNode { + if x := c.slotEntry(slot); x != nil { return x.nodes } return nil @@ -2982,6 +3157,9 @@ func (c *ClusterClient) cmdNodeWithShardPicker( func (c *ClusterClient) slotReadOnlyNode(state *clusterState, slot int) (*clusterNode, error) { if c.opt.RouteByLatency { + if c.opt.RouteByLatencyTolerance > 0 { + return state.slotNodeWithinLatency(slot, c.opt.RouteByLatencyTolerance) + } return state.slotClosestNode(slot) } if c.opt.RouteRandomly { diff --git a/osscluster_latency_tolerance_test.go b/osscluster_latency_tolerance_test.go new file mode 100644 index 00000000..1be2c0cc --- /dev/null +++ b/osscluster_latency_tolerance_test.go @@ -0,0 +1,414 @@ +package redis + +import ( + "math" + "testing" + "time" +) + +func toleranceTestNode(t *testing.T, latency time.Duration, failing bool) *clusterNode { + t.Helper() + + n := &clusterNode{Client: NewClient(&Options{Addr: "127.0.0.1:0"})} + t.Cleanup(func() { _ = n.Client.Close() }) + + // Steady state: the node has answered a PING at least once, so Loading() takes its + // cached path instead of dialling a dead address for 100ms per candidate. + n.loaded.Store(1) + n.latency.Store(uint32(latency / time.Microsecond)) + if failing { + n.MarkAsFailing() + } + + return n +} + +func toleranceTestState(nodes ...*clusterNode) *clusterState { + return &clusterState{slots: []*clusterSlot{{start: 0, end: 16383, nodes: nodes}}} +} + +func TestRouteByLatencyToleranceZeroKeepsStrictMinimum(t *testing.T) { + fastest := toleranceTestNode(t, 100*time.Microsecond, false) + state := toleranceTestState( + fastest, + toleranceTestNode(t, 120*time.Microsecond, false), + toleranceTestNode(t, 900*time.Microsecond, false), + ) + + for i := 0; i < 20; i++ { + got, err := state.slotClosestNode(0) + if err != nil { + t.Fatalf("slotClosestNode: %v", err) + } + if got != fastest { + t.Fatalf("iteration %d: expected the fastest node with zero tolerance", i) + } + } +} + +func TestSlotNodeWithinLatencySpreadsAcrossCloseNodes(t *testing.T) { + a := toleranceTestNode(t, 100*time.Microsecond, false) + b := toleranceTestNode(t, 120*time.Microsecond, false) + far := toleranceTestNode(t, 900*time.Microsecond, false) + state := toleranceTestState(a, b, far) + + counts := map[*clusterNode]int{} + + for i := 0; i < 100; i++ { + got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + counts[got]++ + } + + if counts[far] != 0 { + t.Fatalf("node outside the tolerance received %d requests", counts[far]) + } + if counts[a] == 0 || counts[b] == 0 { + t.Fatalf("expected both close nodes to be used, got a=%d b=%d", counts[a], counts[b]) + } + if diff := counts[a] - counts[b]; diff > 1 || diff < -1 { + t.Fatalf("expected an even split across close nodes, got a=%d b=%d", counts[a], counts[b]) + } +} + +func TestSlotNodeWithinLatencySkipsFailingNodes(t *testing.T) { + healthy := toleranceTestNode(t, 120*time.Microsecond, false) + state := toleranceTestState(toleranceTestNode(t, 100*time.Microsecond, true), healthy) + + for i := 0; i < 10; i++ { + got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + if got != healthy { + t.Fatal("a failing node was selected") + } + } +} + +// The fastest node being unhealthy must not stop a slower healthy one from being chosen. +// Previously the healthy candidate was only recorded while scanning for a new minimum, so +// a healthy-but-slower node could be skipped and selection fell through to the +// all-nodes-failing path. +// Exercises slotNodeWithinLatency rather than slotClosestNode: preferring a slower healthy +// node over a faster failing one is behaviour of the new path. slotClosestNode is left as it +// was on master, latent bug included - see the PR discussion. +func TestSlotNodeWithinLatencyPrefersSlowerHealthyNode(t *testing.T) { + healthy := toleranceTestNode(t, 800*time.Microsecond, false) + state := toleranceTestState(toleranceTestNode(t, 100*time.Microsecond, true), healthy) + + got, err := state.slotNodeWithinLatency(0, 0) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + if got != healthy { + t.Fatal("expected the slower healthy node rather than the failing fastest one") + } +} + +func TestSlotNodeWithinLatencyNeverReturnsNilWithHealthyNode(t *testing.T) { + nodes := []*clusterNode{ + toleranceTestNode(t, 100*time.Microsecond, false), + toleranceTestNode(t, 120*time.Microsecond, false), + toleranceTestNode(t, 900*time.Microsecond, false), + } + state := toleranceTestState(nodes...) + + stop := make(chan struct{}) + defer close(stop) + + // Churn the latencies while selection runs, including pushing the fastest node above + // the band it established. + go func() { + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + nodes[i%len(nodes)].latency.Store(uint32(100 + (i*37)%5000)) + } + }() + + for i := 0; i < 5000; i++ { + got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + if got == nil { + t.Fatal("returned nil despite healthy nodes being available") + } + } +} + +func TestSlotNodeWithinLatencyHandlesHugeTolerance(t *testing.T) { + a := toleranceTestNode(t, 100*time.Microsecond, false) + state := toleranceTestState(a, toleranceTestNode(t, 900*time.Microsecond, false)) + + // A tolerance near the maximum must not overflow into a negative comparison. + got, err := state.slotNodeWithinLatency(0, time.Duration(math.MaxInt64)) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + if got == nil { + t.Fatal("returned nil for a very large tolerance") + } +} + +func TestRouteByLatencyToleranceFromClusterURL(t *testing.T) { + o, err := ParseClusterURL("redis://localhost:6379?route_by_latency=true&route_by_latency_tolerance=250us") + if err != nil { + t.Fatalf("ParseClusterURL: %v", err) + } + if !o.RouteByLatency { + t.Fatal("RouteByLatency not parsed") + } + if o.RouteByLatencyTolerance != 250*time.Microsecond { + t.Fatalf("tolerance = %v, want 250us", o.RouteByLatencyTolerance) + } +} + +func TestRouteByLatencyToleranceFromFailoverURL(t *testing.T) { + o, err := ParseFailoverURL("redis://localhost:26379?master_name=mymaster&route_by_latency=true&route_by_latency_tolerance=1ms") + if err != nil { + t.Fatalf("ParseFailoverURL: %v", err) + } + if o.RouteByLatencyTolerance != time.Millisecond { + t.Fatalf("tolerance = %v, want 1ms", o.RouteByLatencyTolerance) + } +} + +func TestRouteByLatencyToleranceFromUniversalOptions(t *testing.T) { + o := &UniversalOptions{ + Addrs: []string{"localhost:6379"}, + RouteByLatency: true, + RouteByLatencyTolerance: 300 * time.Microsecond, + } + + if got := o.Cluster().RouteByLatencyTolerance; got != 300*time.Microsecond { + t.Fatalf("Cluster() tolerance = %v, want 300us", got) + } + + o.MasterName = "mymaster" + if got := o.Failover().RouteByLatencyTolerance; got != 300*time.Microsecond { + t.Fatalf("Failover() tolerance = %v, want 300us", got) + } +} + +// With the option unset, selection must stay on the pre-existing hot path: no candidate +// slice, no allocation per command. +func TestSlotClosestNodeStillDoesNotAllocate(t *testing.T) { + state := toleranceTestState( + toleranceTestNode(t, 100*time.Microsecond, false), + toleranceTestNode(t, 120*time.Microsecond, false), + toleranceTestNode(t, 900*time.Microsecond, false), + ) + + allocs := testing.AllocsPerRun(200, func() { + if _, err := state.slotClosestNode(0); err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + }) + + if allocs != 0 { + t.Fatalf("zero tolerance allocated %.0f times per call, want 0", allocs) + } +} + +// Rotation state is per slot. A counter shared across slots is advanced by the other slots +// between two visits to this one, so with a regular interleaving each slot keeps landing on +// the same candidate index and one replica per shard takes all of that shard's reads. +func TestSlotNodeWithinLatencyRotatesPerSlot(t *testing.T) { + nodesFor := func() []*clusterNode { + return []*clusterNode{ + toleranceTestNode(t, 100*time.Microsecond, false), + toleranceTestNode(t, 120*time.Microsecond, false), + } + } + a, b := nodesFor(), nodesFor() + state := &clusterState{slots: []*clusterSlot{ + {start: 0, end: 100, nodes: a}, + {start: 101, end: 200, nodes: b}, + }} + + counts := map[*clusterNode]int{} + for i := 0; i < 100; i++ { + for _, slot := range []int{50, 150} { // alternate slots, the interleaving that breaks a shared counter + got, err := state.slotNodeWithinLatency(slot, 50*time.Microsecond) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + counts[got]++ + } + } + + for name, nodes := range map[string][]*clusterNode{"slot 0-100": a, "slot 101-200": b} { + if counts[nodes[0]] == 0 || counts[nodes[1]] == 0 { + t.Fatalf("%s did not rotate: %d / %d", name, counts[nodes[0]], counts[nodes[1]]) + } + } +} + +// The rotation cursor is a uint32 and wraps. Reducing it to an index must stay in the unsigned +// domain: on 32-bit builds int is 32 bits, so converting a cursor past 2^31 before the modulo +// yields a negative index and panics. Run with GOARCH=386 to exercise that. +func TestSlotNodeWithinLatencyRotationPastIntMax(t *testing.T) { + first := toleranceTestNode(t, 100*time.Microsecond, false) + second := toleranceTestNode(t, 110*time.Microsecond, false) + state := toleranceTestState(first, second) + state.slots[0].latencyBandNodeCursor.Store(math.MaxInt32) + + seen := map[*clusterNode]int{} + for i := 0; i < 8; i++ { + got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + if got != first && got != second { + t.Fatalf("unexpected node %p", got) + } + seen[got]++ + } + + if len(seen) != 2 { + t.Fatalf("rotation stalled across the uint32 boundary: %v", seen) + } +} + +// A node that has not yet been confirmed loaded must still be selectable: Loading() only +// reports true for an actual LOADING reply, and a connection error is not one. Guards +// against the tolerance filter over-excluding and shrinking the rotation to nothing. +func TestSlotNodeWithinLatencyKeepsUnconfirmedNode(t *testing.T) { + fast := toleranceTestNode(t, 100*time.Microsecond, false) + unconfirmed := toleranceTestNode(t, 110*time.Microsecond, false) + unconfirmed.loaded.Store(0) + + state := toleranceTestState(fast, unconfirmed) + + seen := map[*clusterNode]bool{} + for i := 0; i < 8; i++ { + got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + if got == nil { + t.Fatal("slotNodeWithinLatency returned nil") + } + seen[got] = true + } + + if !seen[unconfirmed] { + t.Fatal("node with loaded=0 was excluded from the rotation; a connection error is not a LOADING reply") + } +} + +// Before the background probes finish, every node still carries the math.MaxUint32 sentinel +// from newClusterNodeWithNodeAddress. All differences are then zero, so a positive tolerance +// would admit the entire slot and round-robin startup reads across distant zones. Until one +// measurement lands, the tolerance path must behave exactly like the strict one. +func TestSlotNodeWithinLatencyKeepsStrictOrderBeforeFirstProbe(t *testing.T) { + unprobed := func() *clusterNode { + n := &clusterNode{Client: NewClient(&Options{Addr: "127.0.0.1:0"})} + t.Cleanup(func() { _ = n.Client.Close() }) + n.loaded.Store(1) + n.latency.Store(unmeasuredNodeLatencyMicros) // as newClusterNodeWithNodeAddress does + return n + } + + first, second, third := unprobed(), unprobed(), unprobed() + state := toleranceTestState(first, second, third) + + for i := 0; i < 50; i++ { + got, err := state.slotNodeWithinLatency(0, time.Second) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + if got != first { + t.Fatalf("iteration %d: rotated across unmeasured nodes; want the strict pick", i) + } + } + + strict, err := state.slotClosestNode(0) + if err != nil { + t.Fatalf("slotClosestNode: %v", err) + } + if strict != first { + t.Fatalf("slotClosestNode disagrees with the tolerance path before the first probe") + } +} + +// Once any healthy node has a real measurement the band is meaningful again, and an +// unmeasured node must stay out of it - its sentinel latency is ~4295s, far beyond any +// sane tolerance. +func TestSlotNodeWithinLatencyExcludesUnmeasuredOnceProbed(t *testing.T) { + measured := toleranceTestNode(t, 100*time.Microsecond, false) + alsoMeasured := toleranceTestNode(t, 130*time.Microsecond, false) + + unprobed := &clusterNode{Client: NewClient(&Options{Addr: "127.0.0.1:0"})} + t.Cleanup(func() { _ = unprobed.Client.Close() }) + unprobed.loaded.Store(1) + unprobed.latency.Store(unmeasuredNodeLatencyMicros) + + state := toleranceTestState(measured, alsoMeasured, unprobed) + + for i := 0; i < 60; i++ { + got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + if got == unprobed { + t.Fatalf("iteration %d: unmeasured node entered the tolerance band", i) + } + } +} + +// updateLatency publishes the latency and its timestamp as two separate stores, so a probe can +// land between selection sampling a node's latency and reading its measurement timestamp. That +// interleaving is observable as "sentinel latency, timestamp set" - and must still count as +// unmeasured, or the band widens to the whole slot while every sampled latency is a sentinel. +func TestSlotNodeWithinLatencyIgnoresTimestampWithoutLatency(t *testing.T) { + interleaved := func() *clusterNode { + n := &clusterNode{Client: NewClient(&Options{Addr: "127.0.0.1:0"})} + t.Cleanup(func() { _ = n.Client.Close() }) + n.loaded.Store(1) + n.latency.Store(unmeasuredNodeLatencyMicros) + // The half-published state: timestamp visible, latency not yet stored. + n.SetLastLatencyMeasurement(time.Now()) + return n + } + + first, second, third := interleaved(), interleaved(), interleaved() + state := toleranceTestState(first, second, third) + + for i := 0; i < 50; i++ { + got, err := state.slotNodeWithinLatency(0, time.Second) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + if got != first { + t.Fatalf("iteration %d: a set timestamp widened the band while latencies were sentinels", i) + } + } +} + +// maximumNodeLatency (all pings failed) must not be mistaken for the unmeasured sentinel: such a +// node has a real measurement, so the band applies normally rather than collapsing to strict. +func TestSlotNodeWithinLatencyTreatsFailedPingsAsMeasured(t *testing.T) { + a := toleranceTestNode(t, maximumNodeLatency, false) + b := toleranceTestNode(t, maximumNodeLatency, false) + state := toleranceTestState(a, b) + + counts := map[*clusterNode]int{} + for i := 0; i < 80; i++ { + got, err := state.slotNodeWithinLatency(0, time.Millisecond) + if err != nil { + t.Fatalf("slotNodeWithinLatency: %v", err) + } + counts[got]++ + } + if counts[a] == 0 || counts[b] == 0 { + t.Fatalf("expected rotation across both measured nodes, got a=%d b=%d", counts[a], counts[b]) + } +} diff --git a/osscluster_test.go b/osscluster_test.go index 979017d8..cdc3debb 100644 --- a/osscluster_test.go +++ b/osscluster_test.go @@ -1784,6 +1784,93 @@ var _ = Describe("ClusterClient", func() { assertClusterClient() }) + Describe("ClusterClient with RouteByLatencyTolerance", func() { + BeforeEach(func() { + opt = redisClusterOptions() + opt.RouteByLatency = true + // Far wider than any intra-cluster spread, so every healthy node for a slot is a + // candidate and the per-slot rotation is the thing under test. + opt.RouteByLatencyTolerance = time.Second + 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))) + return nil + }) + Expect(err).NotTo(HaveOccurred()) + }) + + AfterEach(func() { + err := client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error { + return slave.ReadWrite(ctx).Err() + }) + Expect(err).NotTo(HaveOccurred()) + + err = client.Close() + Expect(err).NotTo(HaveOccurred()) + }) + + It("should read keys from distinct shards", func() { + // Discovered rather than hardcoded: keyN does not spread over the three masters + // (key2 and key3 both hash into 0-5460), and hardcoded keys would silently stop + // covering multiple shards if the slot ranges ever moved. + keys := make([]string, 0, 3) + slots := map[int64]string{} + masters := map[string]string{} + + for i := 0; len(masters) < 3 && i < 100; i++ { + key := fmt.Sprintf("tolerance-key-%d", i) + + nodes, err := client.Nodes(ctx, key) + Expect(err).NotTo(HaveOccurred()) + Expect(nodes).NotTo(BeEmpty()) + + master := nodes[0].Client.Options().Addr + if _, seen := masters[master]; seen { + continue + } + masters[master] = key + + slot := client.ClusterKeySlot(ctx, key).Val() + Expect(slots).NotTo(HaveKey(slot)) + slots[slot] = key + keys = append(keys, key) + } + Expect(masters).To(HaveLen(3), "could not find keys on three distinct masters: %v", masters) + + for _, key := range keys { + Expect(client.Set(ctx, key, "VALUE-"+key, 0).Err()).NotTo(HaveOccurred()) + } + + // Reads go to whichever in-band node the rotation picks, including replicas that + // may not have caught up yet, so poll rather than failing on the first attempt. + for _, key := range keys { + Eventually(func() string { + return client.Get(ctx, key).Val() + }, 30*time.Second).Should(Equal("VALUE-" + key)) + } + + // Repeated reads must keep resolving to a usable node for every slot. This is what + // regressed when the rotation cursor was shared across slots. + for i := 0; i < 50; i++ { + for _, key := range keys { + Eventually(func() string { + return client.Get(ctx, key).Val() + }, 30*time.Second).Should(Equal("VALUE-" + key)) + } + } + }) + + assertClusterClient() + }) + Describe("ClusterClient with ClusterSlots", func() { BeforeEach(func() { failover = true diff --git a/sentinel.go b/sentinel.go index 126ed2d2..eeb0b318 100644 --- a/sentinel.go +++ b/sentinel.go @@ -47,6 +47,8 @@ type FailoverOptions struct { // Allows routing read-only commands to the closest master or replica node. // This option only works with NewFailoverClusterClient. RouteByLatency bool + // RouteByLatencyTolerance is passed through to ClusterOptions; see its documentation. + RouteByLatencyTolerance time.Duration // Allows routing read-only commands to the random master or replica node. // This option only works with NewFailoverClusterClient. RouteRandomly bool @@ -328,9 +330,10 @@ func (opt *FailoverOptions) clusterOptions() *ClusterOptions { MaxRedirects: opt.MaxRetries, - ReadOnly: opt.ReplicaOnly, - RouteByLatency: opt.RouteByLatency, - RouteRandomly: opt.RouteRandomly, + ReadOnly: opt.ReplicaOnly, + RouteByLatency: opt.RouteByLatency, + RouteByLatencyTolerance: opt.RouteByLatencyTolerance, + RouteRandomly: opt.RouteRandomly, MinRetryBackoff: opt.MinRetryBackoff, MaxRetryBackoff: opt.MaxRetryBackoff, @@ -462,6 +465,7 @@ func setupFailoverConnParams(u *url.URL, o *FailoverOptions) (*FailoverOptions, o.MasterName = q.string("master_name") o.ClientName = q.string("client_name") o.RouteByLatency = q.bool("route_by_latency") + o.RouteByLatencyTolerance = q.duration("route_by_latency_tolerance") o.RouteRandomly = q.bool("route_randomly") o.ReplicaOnly = q.bool("replica_only") o.UseDisconnectedReplicas = q.bool("use_disconnected_replicas") diff --git a/universal.go b/universal.go index ae2ad999..ec6c0f22 100644 --- a/universal.go +++ b/universal.go @@ -120,7 +120,10 @@ type UniversalOptions struct { MaxRedirects int ReadOnly bool RouteByLatency bool - RouteRandomly bool + // RouteByLatencyTolerance is passed through to ClusterOptions and FailoverOptions; + // see ClusterOptions.RouteByLatencyTolerance. + RouteByLatencyTolerance time.Duration + RouteRandomly bool // MasterName is the sentinel master name. // Only for failover clients. @@ -203,10 +206,11 @@ func (o *UniversalOptions) Cluster() *ClusterOptions { CredentialsProviderContext: o.CredentialsProviderContext, StreamingCredentialsProvider: o.StreamingCredentialsProvider, - MaxRedirects: o.MaxRedirects, - ReadOnly: o.ReadOnly, - RouteByLatency: o.RouteByLatency, - RouteRandomly: o.RouteRandomly, + MaxRedirects: o.MaxRedirects, + ReadOnly: o.ReadOnly, + RouteByLatency: o.RouteByLatency, + RouteByLatencyTolerance: o.RouteByLatencyTolerance, + RouteRandomly: o.RouteRandomly, MaxRetries: o.MaxRetries, MinRetryBackoff: o.MinRetryBackoff, @@ -276,8 +280,9 @@ func (o *UniversalOptions) Failover() *FailoverOptions { SentinelUsername: o.SentinelUsername, SentinelPassword: o.SentinelPassword, - RouteByLatency: o.RouteByLatency, - RouteRandomly: o.RouteRandomly, + RouteByLatency: o.RouteByLatency, + RouteByLatencyTolerance: o.RouteByLatencyTolerance, + RouteRandomly: o.RouteRandomly, MaxRetries: o.MaxRetries, MinRetryBackoff: o.MinRetryBackoff,