Repository: nats-io/nats-server
Pull request #2133 (2021-04-20 18:26:42)

Return not ready for connection reason
Currently, we use ReadyForConnections in server tests to wait for the
server to be ready. However, when this fails we don't get a clue about
why it failed.

This change adds a new unexported method called readyForConnections that
returns an error describing which checks failed. The exported
ReadyForConnections version works exactly as before. The unexported
version gets used in internal tests only.

Ulimit 256 too low error - before
```
panic: Unable to start NATS Server in Go Routine
```
Ulimit 256 too low error - with this change
```
panic: failed to be ready for connections after 10s: server(listen tcp 127.0.0.1:0: socket: too many open files)
```

At higher ulimit values, we still get the not ready panic, but for different
reasons.

Ulimit 65535 - before
```
panic: Unable to start NATS Server in Go Routine
```
Ulimit 65535 - with this change
```
panic: failed to be ready for connections after 10s: route(listen tcp 127.0.0.1:33230: bind: address already in use)
```

/cc @nats-io/core



Patch:
diff --git a/server/gateway.go b/server/gateway.go
index 69f02815..f0b94296 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -454,6 +454,7 @@ func (s *Server) startGatewayAcceptLoop() {
 	}
 	hp := net.JoinHostPort(opts.Gateway.Host, strconv.Itoa(port))
 	l, e := natsListen("tcp", hp)
+	s.gatewayListenerErr = e
 	if e != nil {
 		s.mu.Unlock()
 		s.Fatalf("Error listening on gateway port: %d - %v", opts.Gateway.Port, e)
diff --git a/server/leafnode.go b/server/leafnode.go
index 325a7933..2a697ee1 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -460,6 +460,7 @@ func (s *Server) startLeafNodeAcceptLoop() {
 	}
 	hp := net.JoinHostPort(opts.LeafNode.Host, strconv.Itoa(port))
 	l, e := natsListen("tcp", hp)
+	s.leafNodeListenerErr = e
 	if e != nil {
 		s.mu.Unlock()
 		s.Fatalf("Error listening on leafnode port: %d - %v", opts.LeafNode.Port, e)
diff --git a/server/mqtt.go b/server/mqtt.go
index 478f4f12..2a846aef 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -172,6 +172,7 @@ var (
 
 type srvMQTT struct {
 	listener     net.Listener
+	listenerErr  error
 	authOverride bool
 	sessmgr      mqttSessionManager
 }
@@ -345,6 +346,7 @@ func (s *Server) startMQTT() {
 	}
 	s.mqtt.sessmgr.sessions = make(map[string]*mqttAccountSessionManager)
 	hl, err = net.Listen("tcp", hp)
+	s.mqtt.listenerErr = err
 	if err != nil {
 		s.mu.Unlock()
 		s.Fatalf("Unable to listen for MQTT connections: %v", err)
diff --git a/server/route.go b/server/route.go
index cc6c0f19..edeb4b00 100644
--- a/server/route.go
+++ b/server/route.go
@@ -1653,6 +1653,7 @@ func (s *Server) startRouteAcceptLoop() {
 
 	hp := net.JoinHostPort(opts.Cluster.Host, strconv.Itoa(port))
 	l, e := natsListen("tcp", hp)
+	s.routeListenerErr = e
 	if e != nil {
 		s.mu.Unlock()
 		s.Fatalf("Error listening on router port: %d - %v", opts.Cluster.Port, e)
diff --git a/server/server.go b/server/server.go
index 4c5e3c05..10092191 100644
--- a/server/server.go
+++ b/server/server.go
@@ -105,50 +105,53 @@ type Info struct {
 type Server struct {
 	gcid uint64
 	stats
-	mu               sync.Mutex
-	kp               nkeys.KeyPair
-	prand            *rand.Rand
-	info             Info
-	configFile       string
-	optsMu           sync.RWMutex
-	opts             *Options
-	running          bool
-	shutdown         bool
-	reloading        bool
-	listener         net.Listener
-	gacc             *Account
-	sys              *internal
-	js               *jetStream
-	accounts         sync.Map
-	tmpAccounts      sync.Map // Temporarily stores accounts that are being built
-	activeAccounts   int32
-	accResolver      AccountResolver
-	clients          map[uint64]*client
-	routes           map[uint64]*client
-	routesByHash     sync.Map
-	remotes          map[string]*client
-	leafs            map[uint64]*client
-	users            map[string]*User
-	nkeys            map[string]*NkeyUser
-	totalClients     uint64
-	closed           *closedRingBuffer
-	done             chan bool
-	start            time.Time
-	http             net.Listener
-	httpHandler      http.Handler
-	httpBasePath     string
-	profiler         net.Listener
-	httpReqStats     map[string]uint64
-	routeListener    net.Listener
-	routeInfo        Info
-	routeInfoJSON    []byte
-	routeResolver    netResolver
-	routesToSelf     map[string]struct{}
-	leafNodeListener net.Listener
-	leafNodeInfo     Info
-	leafNodeInfoJSON []byte
-	leafURLsMap      refCountedUrlSet
-	leafNodeOpts     struct {
+	mu                  sync.Mutex
+	kp                  nkeys.KeyPair
+	prand               *rand.Rand
+	info                Info
+	configFile          string
+	optsMu              sync.RWMutex
+	opts                *Options
+	running             bool
+	shutdown            bool
+	reloading           bool
+	listener            net.Listener
+	listenerErr         error
+	gacc                *Account
+	sys                 *internal
+	js                  *jetStream
+	accounts            sync.Map
+	tmpAccounts         sync.Map // Temporarily stores accounts that are being built
+	activeAccounts      int32
+	accResolver         AccountResolver
+	clients             map[uint64]*client
+	routes              map[uint64]*client
+	routesByHash        sync.Map
+	remotes             map[string]*client
+	leafs               map[uint64]*client
+	users               map[string]*User
+	nkeys               map[string]*NkeyUser
+	totalClients        uint64
+	closed              *closedRingBuffer
+	done                chan bool
+	start               time.Time
+	http                net.Listener
+	httpHandler         http.Handler
+	httpBasePath        string
+	profiler            net.Listener
+	httpReqStats        map[string]uint64
+	routeListener       net.Listener
+	routeListenerErr    error
+	routeInfo           Info
+	routeInfoJSON       []byte
+	routeResolver       netResolver
+	routesToSelf        map[string]struct{}
+	leafNodeListener    net.Listener
+	leafNodeListenerErr error
+	leafNodeInfo        Info
+	leafNodeInfoJSON    []byte
+	leafURLsMap         refCountedUrlSet
+	leafNodeOpts        struct {
 		resolver    netResolver
 		dialTimeout time.Duration
 	}
@@ -182,8 +185,9 @@ type Server struct {
 	lastCURLsUpdate int64
 
 	// For Gateways
-	gatewayListener net.Listener // Accept listener
-	gateway         *srvGateway
+	gatewayListener    net.Listener // Accept listener
+	gatewayListenerErr error
+	gateway            *srvGateway
 
 	// Used by tests to check that http.Servers do
 	// not set any timeout.
@@ -1848,6 +1852,7 @@ func (s *Server) AcceptLoop(clr chan struct{}) {
 
 	hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port))
 	l, e := natsListen("tcp", hp)
+	s.listenerErr = e
 	if e != nil {
 		s.mu.Unlock()
 		s.Fatalf("Error listening on port: %s, %q", hp, e)
@@ -2682,28 +2687,63 @@ func (s *Server) ProfilerAddr() *net.TCPAddr {
 	return s.profiler.Addr().(*net.TCPAddr)
 }
 
-// ReadyForConnections returns `true` if the server is ready to accept clients
-// and, if routing is enabled, route connections. If after the duration
-// `dur` the server is still not ready, returns `false`.
-func (s *Server) ReadyForConnections(dur time.Duration) bool {
+func (s *Server) readyForConnections(d time.Duration) error {
 	// Snapshot server options.
 	opts := s.getOpts()
 
-	end := time.Now().Add(dur)
+	type info struct {
+		ok  bool
+		err error
+	}
+	chk := make(map[string]info)
+
+	end := time.Now().Add(d)
 	for time.Now().Before(end) {
 		s.mu.Lock()
-		ok := s.listener != nil &&
-			(opts.Cluster.Port == 0 || s.routeListener != nil) &&
-			(opts.Gateway.Name == "" || s.gatewayListener != nil) &&
-			(opts.LeafNode.Port == 0 || s.leafNodeListener != nil) &&
-			(opts.Websocket.Port == 0 || s.websocket.listener != nil)
+		chk["server"] = info{ok: s.listener != nil, err: s.listenerErr}
+		chk["route"] = info{ok: (opts.Cluster.Port == 0 || s.routeListener != nil), err: s.routeListenerErr}
+		chk["gateway"] = info{ok: (opts.Gateway.Name == "" || s.gatewayListener != nil), err: s.gatewayListenerErr}
+		chk["leafNode"] = info{ok: (opts.LeafNode.Port == 0 || s.leafNodeListener != nil), err: s.leafNodeListenerErr}
+		chk["websocket"] = info{ok: (opts.Websocket.Port == 0 || s.websocket.listener != nil), err: s.websocket.listenerErr}
+		chk["mqtt"] = info{ok: (opts.MQTT.Port == 0 || s.mqtt.listener != nil), err: s.mqtt.listenerErr}
 		s.mu.Unlock()
-		if ok {
-			return true
+
+		var numOK int
+		for _, inf := range chk {
+			if inf.ok {
+				numOK++
+			}
+		}
+		if numOK == len(chk) {
+			return nil
 		}
 		time.Sleep(25 * time.Millisecond)
 	}
-	return false
+
+	failed := make([]string, 0, len(chk))
+	for name, inf := range chk {
+		if inf.ok && inf.err != nil {
+			failed = append(failed, fmt.Sprintf("%s(ok, but %s)", name, inf.err))
+		}
+		if !inf.ok && inf.err == nil {
+			failed = append(failed, name)
+		}
+		if !inf.ok && inf.err != nil {
+			failed = append(failed, fmt.Sprintf("%s(%s)", name, inf.err))
+		}
+	}
+
+	return fmt.Errorf(
+		"failed to be ready for connections after %s: %s",
+		d, strings.Join(failed, ", "),
+	)
+}
+
+// ReadyForConnections returns `true` if the server is ready to accept clients
+// and, if routing is enabled, route connections. If after the duration
+// `dur` the server is still not ready, returns `false`.
+func (s *Server) ReadyForConnections(dur time.Duration) bool {
+	return s.readyForConnections(dur) == nil
 }
 
 // Quick utility to function to tell if the server supports headers.
diff --git a/server/websocket.go b/server/websocket.go
index 23de9f22..d22d95b8 100644
--- a/server/websocket.go
+++ b/server/websocket.go
@@ -116,6 +116,7 @@ type srvWebsocket struct {
 	mu             sync.RWMutex
 	server         *http.Server
 	listener       net.Listener
+	listenerErr    error
 	tls            bool
 	allowedOrigins map[string]*allowedOrigin // host will be the key
 	sameOrigin     bool
@@ -961,6 +962,7 @@ func (s *Server) startWebsocketServer() {
 		proto = wsSchemePrefix
 		hl, err = net.Listen("tcp", hp)
 	}
+	s.websocket.listenerErr = err
 	if err != nil {
 		s.mu.Unlock()
 		s.Fatalf("Unable to listen for websocket connections: %v", err)


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index 7d0d6438..64db8b0c 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -214,8 +214,8 @@ func TestAccountIsolationExportImport(t *testing.T) {
 			s := opTrustBasicSetup()
 			defer s.Shutdown()
 			go s.Start()
-			if !s.ReadyForConnections(5 * time.Second) {
-				t.Fatal("failed to be ready for connections")
+			if err := s.readyForConnections(5 * time.Second); err != nil {
+				t.Fatal(err)
 			}
 			buildMemAccResolver(s)
 
@@ -1689,8 +1689,8 @@ func TestAccountRequestReplyTrackLatency(t *testing.T) {
 	// Run server in Go routine. We need this one running for internal sending of msgs.
 	go s.Start()
 	// Wait for accept loop(s) to be started
-	if !s.ReadyForConnections(10 * time.Second) {
-		panic("Unable to start NATS Server in Go Routine")
+	if err := s.readyForConnections(10 * time.Second); err != nil {
+		t.Fatal(err)
 	}
 
 	cfoo, crFoo, _ := newClientForServer(s)
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index c2ad5469..2ede9b1d 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -216,8 +216,8 @@ func testMQTTRunServer(t testing.TB, o *Options) *Server {
 	l := &DummyLogger{}
 	s.SetLogger(l, true, true)
 	go s.Start()
-	if !s.ReadyForConnections(3 * time.Second) {
-		t.Fatal("Unable to start server")
+	if err := s.readyForConnections(3 * time.Second); err != nil {
+		t.Fatal(err)
 	}
 	return s
 }
diff --git a/server/routes_test.go b/server/routes_test.go
index f6fd1761..e5ffafee 100644
--- a/server/routes_test.go
+++ b/server/routes_test.go
@@ -1342,8 +1342,8 @@ func TestRouteIPResolutionAndRouteToSelf(t *testing.T) {
 	go func() {
 		s.Start()
 	}()
-	if !s.ReadyForConnections(time.Second) {
-		t.Fatalf("Failed to start server")
+	if err := s.readyForConnections(time.Second); err != nil {
+		t.Fatal(err)
 	}
 
 	select {
diff --git a/server/server_test.go b/server/server_test.go
index f4eb6f1b..dfe76673 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -84,8 +84,8 @@ func RunServer(opts *Options) *Server {
 	go s.Start()
 
 	// Wait for accept loop(s) to be started
-	if !s.ReadyForConnections(10 * time.Second) {
-		panic("Unable to start NATS Server in Go Routine")
+	if err := s.readyForConnections(10 * time.Second); err != nil {
+		panic(err)
 	}
 	return s
 }
@@ -1481,8 +1481,8 @@ func TestInsecureSkipVerifyWarning(t *testing.T) {
 			s.Start()
 			wg.Done()
 		}()
-		if !s.ReadyForConnections(time.Second) {
-			t.Fatal("Unable to start the server")
+		if err := s.readyForConnections(time.Second); err != nil {
+			t.Fatal(err)
 		}
 		select {
 		case w := <-l.warn:
diff --git a/server/service_test.go b/server/service_test.go
index 8811c185..93f2727e 100644
--- a/server/service_test.go
+++ b/server/service_test.go
@@ -16,7 +16,6 @@
 package server
 
 import (
-	"errors"
 	"testing"
 	"time"
 )
@@ -31,8 +30,8 @@ func TestRun(t *testing.T) {
 		errC <- Run(s)
 	}()
 	go func() {
-		if !s.ReadyForConnections(time.Second) {
-			started <- errors.New("failed to start in time")
+		if err := s.readyForConnections(time.Second); err != nil {
+			started <- err
 			return
 		}
 		s.Shutdown()