mirror of
https://github.com/redis/go-redis.git
synced 2026-10-28 08:24:54 +08:00
fix(cluster): accept a null host in CLUSTER SLOTS (#4131)
* fix(cluster): accept a null host in CLUSTER SLOTS With cluster-preferred-endpoint-type unknown-endpoint the server sends a null host. ClusterSlotsCmd returned redis.Nil and left the rest of the reply unread on the connection. Read the null as an empty host and let newClusterState reuse the origin host, as redis.conf describes. * test(cluster): expect the origin host for an empty slot host An empty host in CLUSTER SLOTS, which a node sends when it does not know its own IP, now also takes the origin host, as redis-py does. Update the loopback spec that asserted the old ":1234" address. * chore(cluster): say why a null CLUSTER SLOTS host is accepted Explain the Nil guard so it is not removed later as a typo. * test(cluster): cover unknown-endpoint end to end Set cluster-preferred-endpoint-type unknown-endpoint on the 6-node test cluster, then check that a ClusterClient loads the topology, runs SET/GET on two shards and follows a MOVED redirect. * test(cluster): say the MOVED check relies on a loopback cluster With unknown-endpoint, Redis sends MOVED and ASK with an empty host, and the client dials ":port" on the local machine. The spec follows the MOVED only because the test cluster is on loopback. Say so in the spec comment and the It text, and leave taking the origin host for the redirect target to a follow-up.
This commit is contained in:
1 parent
1084ddeaad
commit
a179f00376
4 files changed
+108
-2
No files matched your search
+3
-1
@@ -4954,7 +4954,9 @@ func (cmd *ClusterSlotsCmd) readReply(rd *proto.Reader) error {
|
||||
}
|
||||
|
||||
ip, err := rd.ReadString()
|
||||
if err != nil {
|
||||
// A null host (cluster-preferred-endpoint-type unknown-endpoint)
|
||||
// means "reuse the origin host"; newClusterState fills it in.
|
||||
if err != nil && err != Nil {
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
+32
-1
@@ -100,7 +100,7 @@ var _ = Describe("newClusterState", func() {
|
||||
Expect(slotAddr(state.slots[0])).To(Equal("10.10.10.10:7001"))
|
||||
Expect(slotAddr(state.slots[1])).To(Equal("10.10.10.10:7002"))
|
||||
Expect(slotAddr(state.slots[2])).To(Equal("1.2.3.4:1234"))
|
||||
Expect(slotAddr(state.slots[3])).To(Equal(":1234"))
|
||||
Expect(slotAddr(state.slots[3])).To(Equal("10.10.10.10:1234"))
|
||||
})
|
||||
})
|
||||
|
||||
@@ -483,6 +483,37 @@ func TestNewClusterState_PortZero(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
// With `cluster-preferred-endpoint-type unknown-endpoint` the server sends a
|
||||
// null host in CLUSTER SLOTS, and the client is expected to reuse the host it
|
||||
// connected to.
|
||||
func TestNewClusterState_NullHost(t *testing.T) {
|
||||
for _, null := range []string{"_\r\n", "$-1\r\n"} {
|
||||
cmd := NewClusterSlotsCmd(context.Background())
|
||||
reply := "*1\r\n*3\r\n:0\r\n:5460\r\n*2\r\n" + null + ":6380\r\n"
|
||||
if err := cmd.readReply(proto.NewReader(strings.NewReader(reply))); err != nil {
|
||||
t.Fatalf("reply %q: readReply() = %v", null, err)
|
||||
}
|
||||
|
||||
opt := &ClusterOptions{}
|
||||
opt.init()
|
||||
nodes := newClusterNodes(opt)
|
||||
defer nodes.Close()
|
||||
|
||||
state, err := newClusterState(nodes, cmd.Val(), "172.30.0.10:6379")
|
||||
if err != nil {
|
||||
t.Fatalf("newClusterState failed: %v", err)
|
||||
}
|
||||
|
||||
client := state.slots[0].nodes[0].Client
|
||||
if got := client.Options().Addr; got != "172.30.0.10:6380" {
|
||||
t.Errorf("reply %q: Addr = %q, want %q", null, got, "172.30.0.10:6380")
|
||||
}
|
||||
if got := client.NodeAddress(); got != ":6380" {
|
||||
t.Errorf("reply %q: NodeAddress = %q, want %q", null, got, ":6380")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type fixedHash string
|
||||
|
||||
func (h fixedHash) Get(string) string {
|
||||
|
||||
@@ -1075,6 +1075,9 @@ func newClusterState(
|
||||
// reachable, since it is the port that returned this slot map.
|
||||
// See https://github.com/redis/go-redis/issues/3726.
|
||||
addr = replaceZeroPort(addr, originPort)
|
||||
// With cluster-preferred-endpoint-type unknown-endpoint the server
|
||||
// sends no host and expects the client to reuse the origin host.
|
||||
addr = replaceEmptyHost(addr, originHost)
|
||||
|
||||
node, err := c.nodes.GetOrCreateWithNodeAddress(addr, nodeAddress)
|
||||
if err != nil {
|
||||
@@ -1370,6 +1373,14 @@ func replaceZeroPort(nodeAddr, originPort string) string {
|
||||
return net.JoinHostPort(nodeHost, originPort)
|
||||
}
|
||||
|
||||
func replaceEmptyHost(nodeAddr, originHost string) string {
|
||||
nodeHost, nodePort, err := net.SplitHostPort(nodeAddr)
|
||||
if err != nil || nodeHost != "" {
|
||||
return nodeAddr
|
||||
}
|
||||
return net.JoinHostPort(originHost, nodePort)
|
||||
}
|
||||
|
||||
// isLoopback returns true if the host is a loopback address.
|
||||
// For IP addresses, it uses net.IP.IsLoopback().
|
||||
// For hostnames, it recognizes well-known loopback hostnames like "localhost"
|
||||
|
||||
@@ -2072,6 +2072,68 @@ var _ = Describe("ClusterClient", func() {
|
||||
})
|
||||
})
|
||||
|
||||
// With cluster-preferred-endpoint-type unknown-endpoint the servers send a
|
||||
// null host in CLUSTER SLOTS and an empty host in MOVED and ASK.
|
||||
// The MOVED check passes only because the test cluster is on loopback: the
|
||||
// client dials an empty-host target ":port" on the local machine, so a
|
||||
// redirect from a remote host is not covered. Taking the host of the node
|
||||
// that sent the redirect is left for a follow-up.
|
||||
var _ = Describe("ClusterClient with unknown-endpoint", func() {
|
||||
const param = "cluster-preferred-endpoint-type"
|
||||
|
||||
var client *redis.ClusterClient
|
||||
var saved map[string]string
|
||||
|
||||
BeforeEach(func() {
|
||||
saved = map[string]string{}
|
||||
for port, node := range cluster.clients {
|
||||
old, err := node.ConfigGet(ctx, param).Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
saved[port] = old[param]
|
||||
Expect(node.ConfigSet(ctx, param, "unknown-endpoint").Err()).NotTo(HaveOccurred())
|
||||
}
|
||||
client = cluster.newClusterClientUnstable(redisClusterOptions())
|
||||
})
|
||||
|
||||
AfterEach(func() {
|
||||
_ = client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
|
||||
return master.FlushDB(ctx).Err()
|
||||
})
|
||||
for port, value := range saved {
|
||||
Expect(cluster.clients[port].ConfigSet(ctx, param, value).Err()).NotTo(HaveOccurred())
|
||||
}
|
||||
Expect(client.Close()).NotTo(HaveOccurred())
|
||||
})
|
||||
|
||||
It("should load the topology and follow a MOVED redirect on loopback", func() {
|
||||
slots, err := cluster.masters()[0].ClusterSlots(ctx).Result()
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
for _, slot := range slots {
|
||||
for _, node := range slot.Nodes {
|
||||
Expect(node.Addr).To(HavePrefix(":"))
|
||||
}
|
||||
}
|
||||
|
||||
state, err := client.LoadState(ctx)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
for _, master := range state.Masters {
|
||||
Expect(master.Client.Options().Addr).To(HavePrefix("127.0.0.1:"))
|
||||
}
|
||||
|
||||
// A (slot 6373) and C (slot 14503) belong to different masters.
|
||||
for _, key := range []string{"A", "C"} {
|
||||
Expect(client.Set(ctx, key, "VALUE", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(client.Get(ctx, key).Val()).To(Equal("VALUE"))
|
||||
}
|
||||
|
||||
// Route the slot of A at its replica, which answers with
|
||||
// -MOVED <slot> :<port>; see the note above the Describe.
|
||||
Expect(client.SwapNodes(ctx, "A")).NotTo(HaveOccurred())
|
||||
Expect(client.Set(ctx, "A", "MOVED", 0).Err()).NotTo(HaveOccurred())
|
||||
Expect(client.Get(ctx, "A").Val()).To(Equal("MOVED"))
|
||||
})
|
||||
})
|
||||
|
||||
var _ = Describe("ClusterClient without nodes", func() {
|
||||
var client *redis.ClusterClient
|
||||
|
||||
|
||||
Reference in new issue
Block a user