Repository: nats-io/nats-server (go)

Issue:
Support DNS discovery of DNS names on routes
## Feature Request
Support resolving DNS names for routes and using resolved ips, excluding self ip.

#### Use Case:
Boostraping a NATS cluster using Consul 

#### Proposed Change:
Resolve DNS names instead of directly calling net.DialTCP() for routes

#### Who Benefits From The Change(s)?
Environments that use DNS as service discovery, including Consul clusters.




Pull request:
[ADDED] Support for route hostname resolution
We previously simply called DialTimeout() on a route's url when
soliciting. If it resolved to the IP of the host, it would create
a route to self, which server detects, but then would not try again
with other IPs that would have allowed to form a cluster with
other servers running on the other IPs.

This PR keeps track of local IPs + cluster port and exclude them
from the list of IPs returned by LookupHost API. This even prevent
solicitation of routes to self. Only non-local IPs will be tried.

Resolves #1586

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>


Patch:
diff --git a/server/client.go b/server/client.go
index 2e974bee..110c93e1 100644
--- a/server/client.go
+++ b/server/client.go
@@ -4207,7 +4207,7 @@ func (c *client) reconnect() {
 			srv.Debugf("Not attempting reconnect for solicited route, already connected to \"%s\"", rid)
 			return
 		} else if rid == srv.info.ID {
-			srv.Debugf("Detected route to self, ignoring \"%s\"", rurl)
+			srv.Debugf("Detected route to self, ignoring %q", rurl)
 			return
 		} else if rtype != Implicit || retryImplicit {
 			srv.Debugf("Attempting reconnect for solicited route \"%s\"", rurl)
diff --git a/server/gateway.go b/server/gateway.go
index e5ae82cd..6b0aff05 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -620,7 +620,7 @@ func (s *Server) solicitGateway(cfg *gatewayCfg, firstConnect bool) {
 		report := s.shouldReportConnectErr(firstConnect, attempts)
 		// Iteration is random
 		for _, u := range urls {
-			address, err := s.getRandomIP(s.gateway.resolver, u.Host)
+			address, err := s.getRandomIP(s.gateway.resolver, u.Host, nil)
 			if err != nil {
 				s.Errorf("Error getting IP for %s gateway %q (%s): %v", typeStr, cfg.Name, u.Host, err)
 				continue
diff --git a/server/leafnode.go b/server/leafnode.go
index 2f64792a..358d07c3 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -283,7 +283,7 @@ func (s *Server) connectToRemoteLeafNode(remote *leafNodeCfg, firstConnect bool)
 	attempts := 0
 	for s.isRunning() && s.remoteLeafNodeStillValid(remote) {
 		rURL := remote.pickNextURL()
-		url, err := s.getRandomIP(resolver, rURL.Host)
+		url, err := s.getRandomIP(resolver, rURL.Host, nil)
 		if err == nil {
 			var ipStr string
 			if url != rURL.Host {
diff --git a/server/opts.go b/server/opts.go
index c81f7a37..7015bb1a 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -71,6 +71,9 @@ type ClusterOpts struct {
 	Advertise      string            `json:"-"`
 	NoAdvertise    bool              `json:"-"`
 	ConnectRetries int               `json:"-"`
+
+	// Not exported (used in tests)
+	resolver netResolver
 }
 
 // GatewayOpts are options for gateways.
diff --git a/server/route.go b/server/route.go
index c47cb70b..8609f38f 100644
--- a/server/route.go
+++ b/server/route.go
@@ -1755,6 +1755,23 @@ func (s *Server) startRouteAcceptLoop() {
 		s.Warnf(clusterTLSInsecureWarning)
 	}
 
+	// Now that we have the port, keep track of all ip:port that resolve to this server.
+	if interfaceAddr, err := net.InterfaceAddrs(); err == nil {
+		var localIPs []string
+		for i := 0; i < len(interfaceAddr); i++ {
+			interfaceIP, _, _ := net.ParseCIDR(interfaceAddr[i].String())
+			ipStr := interfaceIP.String()
+			if net.ParseIP(ipStr) != nil {
+				localIPs = append(localIPs, ipStr)
+			}
+		}
+		var portStr = strconv.FormatInt(int64(s.routeInfo.Port), 10)
+		for _, ip := range localIPs {
+			ipPort := net.JoinHostPort(ip, portStr)
+			s.routesToSelf[ipPort] = struct{}{}
+		}
+	}
+
 	// Start the accept loop in a different go routine.
 	go s.acceptConnections(l, "Route", func(conn net.Conn) { s.createRoute(conn, nil) }, nil)
 
@@ -1836,13 +1853,26 @@ func (s *Server) connectToRoute(rURL *url.URL, tryForEver, firstConnect bool) {
 
 	const connErrFmt = "Error trying to connect to route (attempt %v): %v"
 
+	s.mu.Lock()
+	resolver := s.routeResolver
+	excludedAddresses := s.routesToSelf
+	s.mu.Unlock()
+
 	attempts := 0
 	for s.isRunning() && rURL != nil {
 		if tryForEver && !s.routeStillValid(rURL) {
 			return
 		}
-		s.Debugf("Trying to connect to route on %s", rURL.Host)
-		conn, err := natsDialTimeout("tcp", rURL.Host, DEFAULT_ROUTE_DIAL)
+		var conn net.Conn
+		address, err := s.getRandomIP(resolver, rURL.Host, excludedAddresses)
+		if err == errNoIPAvail {
+			// This is ok, we are done.
+			return
+		}
+		if err == nil {
+			s.Debugf("Trying to connect to route on %s (%s)", rURL.Host, address)
+			conn, err = natsDialTimeout("tcp", address, DEFAULT_ROUTE_DIAL)
+		}
 		if err != nil {
 			attempts++
 			if s.shouldReportConnectErr(firstConnect, attempts) {
diff --git a/server/server.go b/server/server.go
index aa38273a..93e624c5 100644
--- a/server/server.go
+++ b/server/server.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -140,6 +141,8 @@ type Server struct {
 	routeListener    net.Listener
 	routeInfo        Info
 	routeInfoJSON    []byte
+	routeResolver    netResolver
+	routesToSelf     map[string]struct{}
 	leafNodeListener net.Listener
 	leafNodeInfo     Info
 	leafNodeInfoJSON []byte
@@ -303,6 +306,7 @@ func NewServer(opts *Options) (*Server, error) {
 		gwLeafSubs:   NewSublistWithCache(),
 		httpBasePath: httpBasePath,
 		eventIds:     nuid.New(),
+		routesToSelf: make(map[string]struct{}),
 	}
 
 	// Trusted root operator keys.
@@ -313,6 +317,11 @@ func NewServer(opts *Options) (*Server, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	s.routeResolver = opts.Cluster.resolver
+	if s.routeResolver == nil {
+		s.routeResolver = net.DefaultResolver
+	}
+
 	// Used internally for quick look-ups.
 	s.clientConnectURLsMap = make(refCountedUrlSet)
 	s.websocket.connectURLsMap = make(refCountedUrlSet)
@@ -3122,7 +3131,9 @@ func (s *Server) acceptError(acceptName string, err error, tmpDelay time.Duratio
 	return tmpDelay
 }
 
-func (s *Server) getRandomIP(resolver netResolver, url string) (string, error) {
+var errNoIPAvail = errors.New("no IP available")
+
+func (s *Server) getRandomIP(resolver netResolver, url string, excludedAddresses map[string]struct{}) (string, error) {
 	host, port, err := net.SplitHostPort(url)
 	if err != nil {
 		return "", err
@@ -3135,6 +3146,24 @@ func (s *Server) getRandomIP(resolver netResolver, url string) (string, error) {
 	if err != nil {
 		return "", fmt.Errorf("lookup for host %q: %v", host, err)
 	}
+	if len(excludedAddresses) > 0 {
+		for i := 0; i < len(ips); i++ {
+			ip := ips[i]
+			addr := net.JoinHostPort(ip, port)
+			if _, excluded := excludedAddresses[addr]; excluded {
+				if len(ips) == 1 {
+					ips = nil
+					break
+				}
+				ips[i] = ips[len(ips)-1]
+				ips = ips[:len(ips)-1]
+				i--
+			}
+		}
+		if len(ips) == 0 {
+			return "", errNoIPAvail
+		}
+	}
 	var address string
 	if len(ips) == 0 {
 		s.Warnf("Unable to get IP for %s, will try with %s: %v", host, url, err)


Test patch:
diff --git a/server/routes_test.go b/server/routes_test.go
index 03be733f..e9b52720 100644
--- a/server/routes_test.go
+++ b/server/routes_test.go
@@ -14,6 +14,7 @@
 package server
 
 import (
+	"context"
 	"crypto/tls"
 	"fmt"
 	"net"
@@ -1281,3 +1282,60 @@ func TestRouteClusterNameConflictBetweenStaticAndDynamic(t *testing.T) {
 
 	checkClusterFormed(t, s1, s2)
 }
+
+type testRouteResolver struct{}
+
+func (r *testRouteResolver) LookupHost(ctx context.Context, host string) ([]string, error) {
+	return []string{"127.0.0.1", "other.host.in.cluster"}, nil
+}
+
+type routeHostLookupLogger struct {
+	DummyLogger
+	errCh chan string
+	ch    chan bool
+	count int
+}
+
+func (l *routeHostLookupLogger) Debugf(format string, v ...interface{}) {
+	l.Lock()
+	defer l.Unlock()
+	msg := fmt.Sprintf(format, v...)
+	if strings.Contains(msg, "127.0.0.1:1234") {
+		l.errCh <- msg
+	} else if strings.Contains(msg, "other.host.in.cluster") {
+		if l.count++; l.count == 10 {
+			l.ch <- true
+		}
+	}
+}
+
+func TestRouteIPResolutionAndRouteToSelf(t *testing.T) {
+	o := DefaultOptions()
+	o.Cluster.Port = 1234
+	r := &testRouteResolver{}
+	o.Cluster.resolver = r
+	o.Routes = RoutesFromStr("nats://routehost:1234")
+	o.Debug = true
+	o.NoLog = false
+	s, err := NewServer(o)
+	if err != nil {
+		t.Fatalf("Error creating server: %v", err)
+	}
+	defer s.Shutdown()
+	l := &routeHostLookupLogger{errCh: make(chan string, 1), ch: make(chan bool, 1)}
+	s.SetLogger(l, true, true)
+	go func() {
+		s.Start()
+	}()
+	if !s.ReadyForConnections(time.Second) {
+		t.Fatalf("Failed to start server")
+	}
+
+	select {
+	case e := <-l.errCh:
+		t.Fatalf("Unexpected trace: %q", e)
+	case <-l.ch:
+		// Ok
+		return
+	}
+}
diff --git a/server/server_test.go b/server/server_test.go
index bb8254e6..03fc6baa 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -1405,15 +1405,15 @@ func TestGetRandomIP(t *testing.T) {
 	s := &Server{}
 	resolver := &myDummyDNSResolver{}
 	// no port...
-	if _, err := s.getRandomIP(resolver, "noport"); err == nil || !strings.Contains(err.Error(), "port") {
+	if _, err := s.getRandomIP(resolver, "noport", nil); err == nil || !strings.Contains(err.Error(), "port") {
 		t.Fatalf("Expected error about port missing, got %v", err)
 	}
 	resolver.err = fmt.Errorf("on purpose")
-	if _, err := s.getRandomIP(resolver, "localhost:4222"); err == nil || !strings.Contains(err.Error(), "on purpose") {
+	if _, err := s.getRandomIP(resolver, "localhost:4222", nil); err == nil || !strings.Contains(err.Error(), "on purpose") {
 		t.Fatalf("Expected error about no port, got %v", err)
 	}
 	resolver.err = nil
-	a, err := s.getRandomIP(resolver, "localhost:4222")
+	a, err := s.getRandomIP(resolver, "localhost:4222", nil)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1421,7 +1421,7 @@ func TestGetRandomIP(t *testing.T) {
 		t.Fatalf("Expected address to be %q, got %q", "localhost:4222", a)
 	}
 	resolver.ips = []string{"1.2.3.4"}
-	a, err = s.getRandomIP(resolver, "localhost:4222")
+	a, err = s.getRandomIP(resolver, "localhost:4222", nil)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1432,7 +1432,7 @@ func TestGetRandomIP(t *testing.T) {
 	resolver.ips = []string{"1.2.3.4", "2.2.3.4", "3.2.3.4"}
 	dist := [3]int{}
 	for i := 0; i < 100; i++ {
-		ip, err := s.getRandomIP(resolver, "localhost:4222")
+		ip, err := s.getRandomIP(resolver, "localhost:4222", nil)
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -1448,6 +1448,43 @@ func TestGetRandomIP(t *testing.T) {
 			t.Logf("Warning: out of expected range [%v,%v] for ip %v, got %v", low, high, i, d)
 		}
 	}
+
+	// Check IP exclusions
+	excludedIPs := map[string]struct{}{"1.2.3.4:4222": struct{}{}}
+	for i := 0; i < 100; i++ {
+		ip, err := s.getRandomIP(resolver, "localhost:4222", excludedIPs)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if ip[0] == '1' {
+			t.Fatalf("Should not have returned this ip: %q", ip)
+		}
+	}
+	excludedIPs["2.2.3.4:4222"] = struct{}{}
+	for i := 0; i < 100; i++ {
+		ip, err := s.getRandomIP(resolver, "localhost:4222", excludedIPs)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if ip[0] != '3' {
+			t.Fatalf("Should only have returned '3.2.3.4', got returned %q", ip)
+		}
+	}
+	excludedIPs["3.2.3.4:4222"] = struct{}{}
+	for i := 0; i < 100; i++ {
+		if _, err := s.getRandomIP(resolver, "localhost:4222", excludedIPs); err != errNoIPAvail {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	}
+
+	// Now check that exclusion takes into account the port number.
+	resolver.ips = []string{"127.0.0.1"}
+	excludedIPs = map[string]struct{}{"127.0.0.1:4222": struct{}{}}
+	for i := 0; i < 100; i++ {
+		if _, err := s.getRandomIP(resolver, "localhost:4223", excludedIPs); err == errNoIPAvail {
+			t.Fatal("Should not have failed")
+		}
+	}
 }
 
 type slowWriteConn struct {