Skip nodes with unresolved address during initialization

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.

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))
+}

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")

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)
 	}

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
 

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