[ADDED] RTT in routez's route info

Added the RTT field to each route reported in routez.
Ensure that when a route is accepted, we send a PING to compute
the first RTT and don't have to wait for the ping timer to fire.

Signed-off-by: Ivan Kozlovic <[email protected]>

Author: kozlovic

server/monitor.go

@@ -558,6 +558,7 @@ type RouteInfo struct {
 	Import       *SubjectPermission `json:"import,omitempty"`
 	Export       *SubjectPermission `json:"export,omitempty"`
 	Pending      int                `json:"pending_size"`
+	RTT          string             `json:"rtt,omitempty"`
 	InMsgs       int64              `json:"in_msgs"`
 	OutMsgs      int64              `json:"out_msgs"`
 	InBytes      int64              `json:"in_bytes"`
@@ -600,6 +601,7 @@ func (s *Server) Routez(routezOpts *RoutezOptions) (*Routez, error) {
 			NumSubs:      uint32(len(r.subs)),
 			Import:       r.opts.Import,
 			Export:       r.opts.Export,
+			RTT:          r.getRTT(),
 		}
 
 		if subs && len(r.subs) > 0 {

server/monitor_test.go

@@ -3123,3 +3123,39 @@ func TestMonitorGatewayzAccounts(t *testing.T) {
 		return nil
 	})
 }
+
+func TestMonitorRouteRTT(t *testing.T) {
+	// Do not change default PingInterval and expect RTT
+	// to still be reported
+
+	ob := DefaultOptions()
+	sb := RunServer(ob)
+	defer sb.Shutdown()
+
+	oa := DefaultOptions()
+	oa.Routes = RoutesFromStr(fmt.Sprintf("nats://%s:%d", ob.Cluster.Host, ob.Cluster.Port))
+	sa := RunServer(oa)
+	defer sa.Shutdown()
+
+	checkClusterFormed(t, sa, sb)
+
+	checkRouteInfo := func(t *testing.T, s *Server) {
+		t.Helper()
+		routezURL := fmt.Sprintf("http://127.0.0.1:%d/routez", s.MonitorAddr().Port)
+		for pollMode := 0; pollMode < 2; pollMode++ {
+			checkFor(t, time.Second, 15*time.Millisecond, func() error {
+				rz := pollRoutez(t, s, pollMode, routezURL, nil)
+				if len(rz.Routes) != 1 {
+					return fmt.Errorf("Expected 1 route, got %v", len(rz.Routes))
+				}
+				ri := rz.Routes[0]
+				if ri.RTT == _EMPTY_ {
+					return fmt.Errorf("Route's RTT not reported")
+				}
+				return nil
+			})
+		}
+	}
+	checkRouteInfo(t, sa)
+	checkRouteInfo(t, sb)
+}

server/route.go

@@ -485,6 +485,12 @@ func (c *client) processRouteInfo(info *Info) {
 		if !s.getOpts().Cluster.NoAdvertise {
 			s.addClientConnectURLsAndSendINFOToClients(info.ClientConnectURLs)
 		}
+
+		// This will allow us to determine the initial RTT without having to
+		// wait for first timer based PING.
+		c.mu.Lock()
+		c.sendPing()
+		c.mu.Unlock()
 	} else {
 		c.Debugf("Detected duplicate remote route %q", info.ID)
 		c.closeConnection(DuplicateRoute)

server/routes_test.go

@@ -1078,3 +1078,67 @@ func TestRouteNoCrashOnAddingSubToRoute(t *testing.T) {
 
 	waitCh(t, ch, "Did not get all messages")
 }
+
+func TestRouteRTT(t *testing.T) {
+	checkRTT := func(t *testing.T, s *Server, checkForUpdate bool) {
+		t.Helper()
+		var route *client
+		s.mu.Lock()
+		for _, r := range s.routes {
+			route = r
+			break
+		}
+		s.mu.Unlock()
+
+		prev := time.Duration(0)
+		check := func(t *testing.T) {
+			t.Helper()
+			checkFor(t, time.Second, 15*time.Millisecond, func() error {
+				route.mu.Lock()
+				rtt := route.rtt
+				route.mu.Unlock()
+				if rtt == 0 || rtt == prev {
+					return fmt.Errorf("RTT probably not tracked")
+				}
+				prev = rtt
+				return nil
+			})
+		}
+		check(t)
+		if checkForUpdate {
+			// Wait a bit and check that rtt is updated
+			time.Sleep(30 * time.Millisecond)
+			check(t)
+		}
+	}
+
+	ob := DefaultOptions()
+	ob.PingInterval = 15 * time.Millisecond
+	sb := RunServer(ob)
+	defer sb.Shutdown()
+
+	oa := DefaultOptions()
+	oa.PingInterval = 15 * time.Millisecond
+	oa.Routes = RoutesFromStr(fmt.Sprintf("nats://%s:%d", ob.Cluster.Host, ob.Cluster.Port))
+	sa := RunServer(oa)
+	defer sa.Shutdown()
+
+	checkClusterFormed(t, sa, sb)
+	checkRTT(t, sa, true)
+	checkRTT(t, sb, true)
+
+	sa.Shutdown()
+	sb.Shutdown()
+	// Now check that initial RTT is computed prior to first PingInterval
+	ob.PingInterval = time.Minute
+	sb = RunServer(ob)
+	defer sb.Shutdown()
+
+	oa.PingInterval = time.Minute
+	oa.Routes = RoutesFromStr(fmt.Sprintf("nats://%s:%d", ob.Cluster.Host, ob.Cluster.Port))
+	sa = RunServer(oa)
+	defer sa.Shutdown()
+
+	checkRTT(t, sa, false)
+	checkRTT(t, sb, false)
+}

test/new_routes_test.go

@@ -14,6 +14,7 @@
 package test
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -28,6 +29,14 @@ import (
 	"github.com/nats-io/nuid"
 )
 
+func routeSendInfo(b []byte, routeSend sendFun, routeExpect expectFun) {
+	routeSend(fmt.Sprintf("INFO %s\r\n", b))
+	// When a server receives an INFO from a route, it will send a PING
+	// to compute the RTT.
+	routeExpect(pingRe)
+	routeSend("PONG\r\n")
+}
+
 func runNewRouteServer(t *testing.T) (*server.Server, *server.Options) {
 	return RunServerWithConfig("./configs/new_cluster.conf")
 }
@@ -55,6 +64,18 @@ func TestNewRouteInfoOnConnect(t *testing.T) {
 	}
 }
 
+func getRSubBuffer(t *testing.T, routeExpect expectFun) (bool, []byte) {
+	t.Helper()
+	buf := routeExpect(rsubRe)
+	// See if PING has been read along with the RS+'s
+	gotPing := false
+	if idx := bytes.Index(buf, []byte("PING\r\n")); idx != -1 {
+		buf = buf[:idx]
+		gotPing = true
+	}
+	return gotPing, buf
+}
+
 func TestNewRouteConnectSubs(t *testing.T) {
 	s, opts := runNewRouteServer(t)
 	defer s.Shutdown()
@@ -89,7 +110,7 @@ func TestNewRouteConnectSubs(t *testing.T) {
 	}
 	routeSend(fmt.Sprintf("INFO %s\r\n", b))
 
-	buf := routeExpect(rsubRe)
+	gotPing, buf := getRSubBuffer(t, routeExpect)
 
 	matches := rsubRe.FindAllSubmatch(buf, -1)
 	if len(matches) != 2 {
@@ -116,6 +137,11 @@ func TestNewRouteConnectSubs(t *testing.T) {
 		}
 	}
 
+	// Wait for PING if we did not get it yet.
+	if !gotPing {
+		routeExpect(pingRe)
+	}
+
 	// Close the client connection, check the results.
 	c.Close()
 
@@ -163,7 +189,7 @@ func TestNewRouteConnectSubsWithAccount(t *testing.T) {
 	}
 	routeSend(fmt.Sprintf("INFO %s\r\n", b))
 
-	buf := routeExpect(rsubRe)
+	gotPing, buf := getRSubBuffer(t, routeExpect)
 
 	matches := rsubRe.FindAllSubmatch(buf, -1)
 	if len(matches) != 2 {
@@ -190,6 +216,11 @@ func TestNewRouteConnectSubsWithAccount(t *testing.T) {
 		}
 	}
 
+	// Wait for PING if we did not get it yet.
+	if !gotPing {
+		routeExpect(pingRe)
+	}
+
 	// Close the client connection, check the results.
 	c.Close()
 
@@ -240,7 +271,8 @@ func TestNewRouteRSubs(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Could not marshal test route info: %v", err)
 	}
-	routeSend(fmt.Sprintf("INFO %s\r\nPING\r\n", b))
+	routeSendInfo(b, routeSend, routeExpect)
+	routeSend("PING\r\n")
 	routeExpect(pongRe)
 
 	// Have the client listen on foo.
@@ -317,7 +349,7 @@ func TestNewRouteProgressiveNormalSubs(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Could not marshal test route info: %v", err)
 	}
-	routeSend(fmt.Sprintf("INFO %s\r\n", b))
+	routeSendInfo(b, routeSend, routeExpect)
 	routeSend("PING\r\n")
 	routeExpect(pongRe)
 
@@ -414,8 +446,7 @@ func TestNewRouteClientClosedWithNormalSubscriptions(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Could not marshal test route info: %v", err)
 	}
-	routeSend(fmt.Sprintf("INFO %s\r\n", b))
-
+	routeSendInfo(b, routeSend, routeExpect)
 	routeSend("PING\r\n")
 	routeExpect(pongRe)
 
@@ -464,8 +495,7 @@ func TestNewRouteClientClosedWithQueueSubscriptions(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Could not marshal test route info: %v", err)
 	}
-	routeSend(fmt.Sprintf("INFO %s\r\n", b))
-
+	routeSendInfo(b, routeSend, routeExpect)
 	routeSend("PING\r\n")
 	routeExpect(pongRe)
 
@@ -512,7 +542,7 @@ func TestNewRouteRUnsubAccountSpecific(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Could not marshal test route info: %v", err)
 	}
-	routeSend(fmt.Sprintf("INFO %s\r\n", b))
+	routeSendInfo(b, routeSend, routeExpect)
 
 	// Now create 500 subs on same subject but all different accounts.
 	for i := 0; i < 500; i++ {
@@ -568,7 +598,7 @@ func TestNewRouteRSubCleanupOnDisconnect(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Could not marshal test route info: %v", err)
 	}
-	routeSend(fmt.Sprintf("INFO %s\r\n", b))
+	routeSendInfo(b, routeSend, routeExpect)
 
 	// Now create 100 subs on 3 different accounts.
 	for i := 0; i < 100; i++ {
@@ -606,7 +636,8 @@ func TestNewRouteSendSubsAndMsgs(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Could not marshal test route info: %v", err)
 	}
-	routeSend(fmt.Sprintf("INFO %s\r\nPING\r\n", b))
+	routeSendInfo(b, routeSend, routeExpect)
+	routeSend("PING\r\n")
 	routeExpect(pongRe)
 
 	// Now let's send in interest from the new protocol.
@@ -733,7 +764,8 @@ func TestNewRouteProcessRoutedMsgs(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Could not marshal test route info: %v", err)
 	}
-	routeSend(fmt.Sprintf("INFO %s\r\nPING\r\n", b))
+	routeSendInfo(b, routeSend, routeExpect)
+	routeSend("PING\r\n")
 	routeExpect(pongRe)
 
 	// Create a client

test/norace_test.go

@@ -359,7 +359,7 @@ func TestQueueSubWeightOrderMultipleConnections(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Could not marshal test route info: %v", err)
 	}
-	routeSend(fmt.Sprintf("INFO %s\r\n", b))
+	routeSendInfo(b, routeSend, routeExpect)
 
 	start := make(chan bool)
 	for _, nc := range clients {

test/route_discovery_test.go

@@ -74,8 +74,7 @@ func TestSeedMultipleRouteInfo(t *testing.T) {
 	// register ourselves via INFO
 	r1Info := server.Info{ID: rc1ID, Host: rc1Host, Port: rc1Port}
 	b, _ := json.Marshal(r1Info)
-	infoJSON := fmt.Sprintf(server.InfoProto, b)
-	routeSend1(infoJSON)
+	routeSendInfo(b, routeSend1, route1Expect)
 	routeSend1("PING\r\n")
 	route1Expect(pongRe)
 
@@ -93,7 +92,7 @@ func TestSeedMultipleRouteInfo(t *testing.T) {
 	// register ourselves via INFO
 	r2Info := server.Info{ID: rc2ID, Host: rc2Host, Port: rc2Port}
 	b, _ = json.Marshal(r2Info)
-	infoJSON = fmt.Sprintf(server.InfoProto, b)
+	infoJSON := fmt.Sprintf(server.InfoProto, b)
 	routeSend2(infoJSON)
 
 	// Now read back the second INFO route1 should receive letting
@@ -556,8 +555,7 @@ func TestSeedReturnIPInInfo(t *testing.T) {
 	// register ourselves via INFO
 	r1Info := server.Info{ID: rc1ID, Host: rc1Host, Port: rc1Port}
 	b, _ := json.Marshal(r1Info)
-	infoJSON := fmt.Sprintf(server.InfoProto, b)
-	routeSend1(infoJSON)
+	routeSendInfo(b, routeSend1, route1Expect)
 	routeSend1("PING\r\n")
 	route1Expect(pongRe)
 
@@ -573,7 +571,7 @@ func TestSeedReturnIPInInfo(t *testing.T) {
 	// register ourselves via INFO
 	r2Info := server.Info{ID: rc2ID, Host: rc2Host, Port: rc2Port}
 	b, _ = json.Marshal(r2Info)
-	infoJSON = fmt.Sprintf(server.InfoProto, b)
+	infoJSON := fmt.Sprintf(server.InfoProto, b)
 	routeSend2(infoJSON)
 
 	// Now read info that route1 should have received from the seed
@@ -621,8 +619,7 @@ func TestImplicitRouteRetry(t *testing.T) {
 	// register ourselves via INFO
 	rbInfo := server.Info{ID: rcbID, Host: optsB.Cluster.Host, Port: optsB.Cluster.Port}
 	b, _ := json.Marshal(rbInfo)
-	infoJSON := fmt.Sprintf(server.InfoProto, b)
-	routeBSend(infoJSON)
+	routeSendInfo(b, routeBSend, routeBExpect)
 	routeBSend("PING\r\n")
 	routeBExpect(pongRe)
 
@@ -634,6 +631,7 @@ func TestImplicitRouteRetry(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Error during listen: %v", err)
 	}
+	defer rbListen.Close()
 	c, err := rbListen.Accept()
 	if err != nil {
 		t.Fatalf("Error during accept: %v", err)