diff --git a/rediscluster/cluster.go b/rediscluster/cluster.go index f28b530..8280db2 100644 --- a/rediscluster/cluster.go +++ b/rediscluster/cluster.go @@ -261,16 +261,29 @@ func NewCluster(ctx context.Context, initAddrs []string, opts Opts) (*Cluster, e cluster.nodeWait.promises = make(map[string]*[]connThen, 1) - var err error for _, addr := range initAddrs { - if _, ok := config.masters[addr]; !ok { - config.nodes[addr], err = cluster.newNode(addr, true) + connectionAddr, err := redisclusterutil.Resolve(addr) + if err != nil { + // skip unresolved node during startup, hoping for future availability + cluster.report(LogInitialNodeSkipped{Address: addr, Error: err}) + continue + } + if _, ok := config.nodes[connectionAddr]; ok { + continue + } + node, err := cluster.newNode(addr, connectionAddr, true) + if err != nil { // since we're connecting asynchronously, it can be only configuration error - if err != nil { - cluster.cancel() - return nil, err - } + cluster.cancel() + return nil, err } + config.nodes[connectionAddr] = node + } + + // case if all nodes are not resolved + if len(config.nodes) == 0 { + cluster.cancel() + return nil, ErrAddressNotResolved.New("for all of: %v", initAddrs) } // case if no nodes are accessible is handled here @@ -364,10 +377,11 @@ func (c *Cluster) control() { func (c *Cluster) reloadMapping() error { nodes, err := c.slotRangesAndInternalMasterOnly() - if err == nil { - c.updateMappings(nodes) + if err != nil { + return err } - return err + c.updateMappings(nodes) + return nil } // addWaitToMigrate schedules some actions to be executed after WaitToMigrate interval. diff --git a/rediscluster/cluster_test.go b/rediscluster/cluster_test.go index dab36f3..d340c6f 100644 --- a/rediscluster/cluster_test.go +++ b/rediscluster/cluster_test.go @@ -921,3 +921,34 @@ Loop: } s.Equal(N, cnt, "Not all goroutines finished") } + +func (s *Suite) TestConnectWithDeadAddresses() { + addrs := []string{ + "127.0.0.1:43200", // dead + "127.0.0.1:43210", // live + "127.0.0.1:43201", // dead + } + cl, err := NewCluster(s.ctx, addrs, clustopts) + s.Nil(err) + defer cl.Close() + + scl := redis.SyncCtx{cl} + key := slotkey("deadaddr", s.keys[0]) + s.Equal("OK", scl.Do(s.ctx, "SET", key, "val")) + s.Equal([]byte("val"), scl.Do(s.ctx, "GET", key)) +} + +func (s *Suite) TestConnectWithUnresolvableAddresses() { + addrs := []string{ + "no-such-host-xyzzy.invalid:6379", // guaranteed NXDOMAIN + "127.0.0.1:43210", // live + } + cl, err := NewCluster(s.ctx, addrs, clustopts) + s.Nil(err) + defer cl.Close() + + scl := redis.SyncCtx{cl} + key := slotkey("unresolvable", s.keys[0]) + s.Equal("OK", scl.Do(s.ctx, "SET", key, "val")) + s.Equal([]byte("val"), scl.Do(s.ctx, "GET", key)) +} diff --git a/rediscluster/error.go b/rediscluster/error.go index eac9abe..229c471 100644 --- a/rediscluster/error.go +++ b/rediscluster/error.go @@ -10,8 +10,8 @@ var ( ErrCluster = redis.Errors.NewSubNamespace("cluster") // ErrClusterSlots - fetching slots configuration failed ErrClusterSlots = ErrCluster.NewType("retrieve_slots") - // ErrAddressNotResolved - address could not be resolved - // Cluster resolves named hosts specified as start points. If this resolution fails, this error returned. + // ErrAddressNotResolved - address could not be resolved. + // Cluster resolves named hosts specified as start points. If this resolution failed at all, this error returned. ErrAddressNotResolved = ErrCluster.NewType("resolve_address") // ErrAddressHostname - hostname could not be extracted from address ErrAddressHostname = ErrCluster.NewType("address_hostname") diff --git a/rediscluster/logger.go b/rediscluster/logger.go index f5f89db..3a6964c 100644 --- a/rediscluster/logger.go +++ b/rediscluster/logger.go @@ -41,13 +41,20 @@ type LogClusterSlotsError struct { // LogSlotRangeError is logged when no host were able to respond to CLUSTER SLOTS. type LogSlotRangeError struct{} +// LogInitialNodeSkipped is logged when a bootstrap address is skipped because its hostname could not be resolved. +type LogInitialNodeSkipped struct { + Address string + Error error +} + // LogContextClosed is logged when cluster's context is closed. type LogContextClosed struct{ Error error } -func (LogHostEvent) logEvent() {} -func (LogClusterSlotsError) logEvent() {} -func (LogSlotRangeError) logEvent() {} -func (LogContextClosed) logEvent() {} +func (LogHostEvent) logEvent() {} +func (LogClusterSlotsError) logEvent() {} +func (LogSlotRangeError) logEvent() {} +func (LogContextClosed) logEvent() {} +func (LogInitialNodeSkipped) logEvent() {} // DefaultLogger is a default Logger implementation type DefaultLogger struct{} @@ -81,6 +88,9 @@ func (d DefaultLogger) Report(cluster *Cluster, event LogEvent) { case LogSlotRangeError: log.Printf("rediscluster %s: no alive nodes to request 'CLUSTER SLOTS'", cluster.Name()) + case LogInitialNodeSkipped: + log.Printf("rediscluster %s: skipping unresolvable bootstrap address %s: %s", + cluster.Name(), ev.Address, ev.Error.Error()) case LogContextClosed: log.Printf("rediscluster %s: shutting down (%s)", cluster.Name(), ev.Error) } diff --git a/rediscluster/mapping.go b/rediscluster/mapping.go index 1c13ce8..36d5cf5 100644 --- a/rediscluster/mapping.go +++ b/rediscluster/mapping.go @@ -34,17 +34,7 @@ type ClusterHandle struct { } // newNode creates handle for a connection, that will be established in a future. -func (c *Cluster) newNode(addr string, initial bool) (*node, error) { - var err error - connectionAddr := addr - - // If redis hosts are mentioned by names, a couple of connections will be established and closed shortly. - // Let's resolve them to ip addresses. - connectionAddr, err = redisclusterutil.Resolve(connectionAddr) - if err != nil { - return nil, ErrAddressNotResolved.WrapWithNoMessage(err) - } - +func (c *Cluster) newNode(addr string, connectionAddr string, initial bool) (*node, error) { nodeOpts, err := c.nodeOpts(addr) if err != nil { return nil, err @@ -180,7 +170,7 @@ func (c *Cluster) addNode(addr string) *node { if node, ok = c.prevNodes[addr]; ok { atomic.AddUint32(&node.refcnt, 1) } else { - node, _ = c.newNode(addr, false) + node, _ = c.newNode(addr, addr, false) } newConf.nodes[addr] = node diff --git a/rediscluster/redisclusterutil/cluster.go b/rediscluster/redisclusterutil/cluster.go index c6a7659..a6e6045 100644 --- a/rediscluster/redisclusterutil/cluster.go +++ b/rediscluster/redisclusterutil/cluster.go @@ -3,6 +3,7 @@ package redisclusterutil import ( "fmt" "hash/fnv" + "net" "sort" "strconv" "strings" @@ -22,9 +23,14 @@ const ( // SlotsRange represents slice of slots type SlotsRange struct { - From int - To int - Addrs []string // addresses of hosts hosting this range of slots. First address is a master, and other are slaves. + From int + To int + + // Addrs contains IP:port of hosts hosting this range of slots. First address is a master, and other are slaves. + Addrs []string + // AddrHostnames maps an item from Addrs to hostname:port for nodes that include hostname metadata in + // the CLUSTER SLOTS response. This is optional. + AddrHostnames map[string]string } // ParseSlotsInfo parses result of CLUSTER SLOTS command @@ -90,7 +96,20 @@ func ParseSlotsInfo(res interface{}) ([]SlotsRange, error) { return errf("address format mismatch: res[%d][%d] = %+v", i, j, rawaddr) } - r.Addrs = append(r.Addrs, string(host)+":"+strconv.Itoa(int(port))) + portStr := strconv.Itoa(int(port)) + addrStr := net.JoinHostPort(string(host), portStr) + r.Addrs = append(r.Addrs, addrStr) + if len(rawaddr) >= 4 { + if metadata, hasMetadata := rawaddr[3].([]interface{}); hasMetadata { + hostname := parseHostname(metadata) + if len(hostname) > 0 { + if r.AddrHostnames == nil { + r.AddrHostnames = make(map[string]string) + } + r.AddrHostnames[addrStr] = net.JoinHostPort(hostname, portStr) + } + } + } } sort.Strings(r.Addrs[1:]) ranges[i] = r @@ -101,6 +120,17 @@ func ParseSlotsInfo(res interface{}) ([]SlotsRange, error) { return ranges, nil } +func parseHostname(metadata []interface{}) string { + for i := 0; i < len(metadata)-1; i += 2 { + key, hasKey := metadata[i].([]byte) + value, hasValue := metadata[i+1].([]byte) + if hasKey && hasValue && string(key) == "hostname" { + return string(value) + } + } + return "" +} + // InstanceInfo represents line of CLUSTER NODES result. type InstanceInfo struct { Uuid string diff --git a/rediscluster/redisclusterutil/cluster_test.go b/rediscluster/redisclusterutil/cluster_test.go index c4e3eca..b7e7244 100644 --- a/rediscluster/redisclusterutil/cluster_test.go +++ b/rediscluster/redisclusterutil/cluster_test.go @@ -17,8 +17,8 @@ func TestParseSlotsInfo(t *testing.T) { int64(30001), []byte("09dbe9720cda62f7865eabc5fd8857c5d2678366"), []interface{}{ - "hostname", - "host-1.redis.example.com", + []byte("hostname"), + []byte("host-1.redis.example.com"), }, }, []interface{}{ @@ -26,8 +26,8 @@ func TestParseSlotsInfo(t *testing.T) { int64(30004), []byte("821d8ca00d7ccf931ed3ffc7e3db0599d2271abf"), []interface{}{ - "hostname", - "host-2.redis.example.com", + []byte("hostname"), + []byte("host-2.redis.example.com"), }, }, }, @@ -39,8 +39,8 @@ func TestParseSlotsInfo(t *testing.T) { int64(30002), []byte("c9d93d9f2c0c524ff34cc11838c2003d8c29e013"), []interface{}{ - "hostname", - "host-3.redis.example.com", + []byte("hostname"), + []byte("host-3.redis.example.com"), }, }, []interface{}{ @@ -48,8 +48,8 @@ func TestParseSlotsInfo(t *testing.T) { int64(30005), []byte("faadb3eb99009de4ab72ad6b6ed87634c7ee410f"), []interface{}{ - "hostname", - "host-4.redis.example.com", + []byte("hostname"), + []byte("host-4.redis.example.com"), }, }, }, @@ -61,8 +61,8 @@ func TestParseSlotsInfo(t *testing.T) { int64(30003), []byte("044ec91f325b7595e76dbcb18cc688b6a5b434a1"), []interface{}{ - "hostname", - "host-5.redis.example.com", + []byte("hostname"), + []byte("host-5.redis.example.com"), }, }, []interface{}{ @@ -70,8 +70,8 @@ func TestParseSlotsInfo(t *testing.T) { int64(30006), []byte("58e6e48d41228013e5d9c1c37c5060693925e97e"), []interface{}{ - "hostname", - "host-6.redis.example.com", + []byte("hostname"), + []byte("host-6.redis.example.com"), }, }, }, @@ -85,6 +85,10 @@ func TestParseSlotsInfo(t *testing.T) { "127.0.0.1:30001", "127.0.0.1:30004", }, + AddrHostnames: map[string]string{ + "127.0.0.1:30001": "host-1.redis.example.com:30001", + "127.0.0.1:30004": "host-2.redis.example.com:30004", + }, }, { From: 5461, @@ -93,6 +97,10 @@ func TestParseSlotsInfo(t *testing.T) { "127.0.0.1:30002", "127.0.0.1:30005", }, + AddrHostnames: map[string]string{ + "127.0.0.1:30002": "host-3.redis.example.com:30002", + "127.0.0.1:30005": "host-4.redis.example.com:30005", + }, }, { From: 10923, @@ -101,6 +109,10 @@ func TestParseSlotsInfo(t *testing.T) { "192.168.11.131:30003", "127.0.0.1:30006", }, + AddrHostnames: map[string]string{ + "192.168.11.131:30003": "host-5.redis.example.com:30003", + "127.0.0.1:30006": "host-6.redis.example.com:30006", + }, }, } @@ -120,8 +132,8 @@ func TestParseSlotsInfo_EmptyAddress(t *testing.T) { int64(30001), []byte("09dbe9720cda62f7865eabc5fd8857c5d2678366"), []interface{}{ - "hostname", - "host-1.redis.example.com", + []byte("hostname"), + []byte("host-1.redis.example.com"), }, }, []interface{}{ @@ -129,8 +141,8 @@ func TestParseSlotsInfo_EmptyAddress(t *testing.T) { int64(0), []byte("821d8ca00d7ccf931ed3ffc7e3db0599d2271abf"), []interface{}{ - "hostname", - "host-2.redis.example.com", + []byte("hostname"), + []byte("host-2.redis.example.com"), }, }, }, @@ -142,8 +154,8 @@ func TestParseSlotsInfo_EmptyAddress(t *testing.T) { int64(0), []byte("c9d93d9f2c0c524ff34cc11838c2003d8c29e013"), []interface{}{ - "hostname", - "host-3.redis.example.com", + []byte("hostname"), + []byte("host-3.redis.example.com"), }, }, []interface{}{ @@ -151,8 +163,8 @@ func TestParseSlotsInfo_EmptyAddress(t *testing.T) { int64(30005), []byte("faadb3eb99009de4ab72ad6b6ed87634c7ee410f"), []interface{}{ - "hostname", - "host-4.redis.example.com", + []byte("hostname"), + []byte("host-4.redis.example.com"), }, }, }, @@ -165,6 +177,9 @@ func TestParseSlotsInfo_EmptyAddress(t *testing.T) { Addrs: []string{ "127.0.0.1:30001", }, + AddrHostnames: map[string]string{ + "127.0.0.1:30001": "host-1.redis.example.com:30001", + }, }, { From: 5461, @@ -172,6 +187,80 @@ func TestParseSlotsInfo_EmptyAddress(t *testing.T) { Addrs: []string{ "127.0.0.1:30005", }, + AddrHostnames: map[string]string{ + "127.0.0.1:30005": "host-4.redis.example.com:30005", + }, + }, + } + + slots, err := ParseSlotsInfo(clusterSlotsResponse) + require.NoError(t, err) + + assert.Equal(t, expectedSlots, slots) +} + +func TestParseSlotsInfo_NoHostname(t *testing.T) { + clusterSlotsResponse := []interface{}{ + []interface{}{ + int64(0), + int64(5460), + []interface{}{ + []byte("127.0.0.1"), + int64(30001), + []byte("09dbe9720cda62f7865eabc5fd8857c5d2678366"), + []interface{}{ + []byte("hostname"), + []byte("host-1.redis.example.com"), + }, + }, + []interface{}{ + []byte("127.0.0.1"), + int64(30004), + []byte("821d8ca00d7ccf931ed3ffc7e3db0599d2271abf"), + }, + }, + []interface{}{ + int64(5461), + int64(16383), + []interface{}{ + []byte("127.0.0.1"), + int64(30002), + []byte("c9d93d9f2c0c524ff34cc11838c2003d8c29e013"), + }, + []interface{}{ + []byte("127.0.0.1"), + int64(30005), + []byte("faadb3eb99009de4ab72ad6b6ed87634c7ee410f"), + []interface{}{ + []byte("hostname"), + []byte("host-4.redis.example.com"), + }, + }, + }, + } + + expectedSlots := []SlotsRange{ + { + From: 0, + To: 5460, + Addrs: []string{ + "127.0.0.1:30001", + "127.0.0.1:30004", + }, + AddrHostnames: map[string]string{ + "127.0.0.1:30001": "host-1.redis.example.com:30001", + }, + }, + { + From: 5461, + To: 16383, + Addrs: []string{ + "127.0.0.1:30002", + "127.0.0.1:30005", + }, + AddrHostnames: map[string]string{ + "127.0.0.1:30005": "host-4.redis.example.com:30005", + }, }, } diff --git a/rediscluster/slotrange.go b/rediscluster/slotrange.go index a65891d..359a597 100644 --- a/rediscluster/slotrange.go +++ b/rediscluster/slotrange.go @@ -52,10 +52,14 @@ func (c *Cluster) updateMappings(slotRanges []redisclusterutil.SlotsRange) { } addrs := make(map[string]struct{}) + addrHostnames := make(map[string]string) for _, rng := range slotRanges { for _, addr := range rng.Addrs { addrs[addr] = struct{}{} } + for addr, hostname := range rng.AddrHostnames { + addrHostnames[addr] = hostname + } } c.m.Lock() @@ -76,7 +80,14 @@ func (c *Cluster) updateMappings(slotRanges []redisclusterutil.SlotsRange) { atomic.AddUint32(&node.refcnt, 1) newConfig.nodes[addr] = node } else { - node, _ = c.newNode(addr, false) + // For new nodes, addr is exposed with IP. + // To make TLS work, try to get the hostname from addrHostnames. + // This way, adding a discovered node behaves the same as in NewCluster. + hostname := addr + if h, ok := addrHostnames[addr]; ok { + hostname = h + } + node, _ = c.newNode(hostname, addr, false) newConfig.nodes[addr] = node } }