Add latency checks to peer connection and status output (#1725)

* adding peer healthcheck * generate proto file * fix return in udp mux and replace with continue * use ice agent for latency checks * fix status output * remove some logs * fix status test * revert bind and ebpf code * fix error handling on binding response callback * extend error handling on binding response callback --------- Co-authored-by: Maycon Santos <mlsmaycon@gmail.com>
netbirdio · Mar 20, 2024 · 8468719 · 8468719
1 parent 6cba9c0
commit 8468719
Show file tree

Hide file tree

Showing 9 changed files with 252 additions and 185 deletions.
diff --git a/client/cmd/status.go b/client/cmd/status.go
@@ -34,6 +34,7 @@ type peerStateDetailOutput struct {
 	LastWireguardHandshake time.Time        `json:"lastWireguardHandshake" yaml:"lastWireguardHandshake"`
 	TransferReceived       int64            `json:"transferReceived" yaml:"transferReceived"`
 	TransferSent           int64            `json:"transferSent" yaml:"transferSent"`
+	Latency                time.Duration    `json:"latency" yaml:"latency"`
 	RosenpassEnabled       bool             `json:"quantumResistance" yaml:"quantumResistance"`
 	Routes                 []string         `json:"routes" yaml:"routes"`
 }
@@ -376,6 +377,7 @@ func mapPeers(peers []*proto.PeerState) peersStateOutput {
 			LastWireguardHandshake: lastHandshake,
 			TransferReceived:       transferReceived,
 			TransferSent:           transferSent,
+			Latency:                pbPeerState.GetLatency().AsDuration(),
 			RosenpassEnabled:       pbPeerState.GetRosenpassEnabled(),
 			Routes:                 pbPeerState.GetRoutes(),
 		}
@@ -638,7 +640,8 @@ func parsePeers(peers peersStateOutput, rosenpassEnabled, rosenpassPermissive bo
 				"  Last WireGuard handshake: %s\n"+
 				"  Transfer status (received/sent) %s/%s\n"+
 				"  Quantum resistance: %s\n"+
-				"  Routes: %s\n",
+				"  Routes: %s\n"+
+				"  Latency: %s\n",
 			peerState.FQDN,
 			peerState.IP,
 			peerState.PubKey,
@@ -655,6 +658,7 @@ func parsePeers(peers peersStateOutput, rosenpassEnabled, rosenpassPermissive bo
 			toIEC(peerState.TransferSent),
 			rosenpassEnabledStatus,
 			routes,
+			peerState.Latency.String(),
 		)
 
 		peersString += peerString

diff --git a/client/cmd/status_test.go b/client/cmd/status_test.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/netbirdio/netbird/client/proto"
@@ -45,6 +46,7 @@ var resp = &proto.StatusResponse{
 				Routes: []string{
 					"10.1.0.0/24",
 				},
+				Latency: durationpb.New(time.Duration(10000000)),
 			},
 			{
 				IP:                         "192.168.178.102",
@@ -61,6 +63,7 @@ var resp = &proto.StatusResponse{
 				LastWireguardHandshake:     timestamppb.New(time.Date(2002, time.Month(2), 2, 2, 2, 3, 0, time.UTC)),
 				BytesRx:                    2000,
 				BytesTx:                    1000,
+				Latency:                    durationpb.New(time.Duration(10000000)),
 			},
 		},
 		ManagementState: &proto.ManagementState{
@@ -147,6 +150,7 @@ var overview = statusOutputOverview{
 				Routes: []string{
 					"10.1.0.0/24",
 				},
+				Latency: time.Duration(10000000),
 			},
 			{
 				IP:               "192.168.178.102",
@@ -167,6 +171,7 @@ var overview = statusOutputOverview{
 				LastWireguardHandshake: time.Date(2002, 2, 2, 2, 2, 3, 0, time.UTC),
 				TransferReceived:       2000,
 				TransferSent:           1000,
+				Latency:                time.Duration(10000000),
 			},
 		},
 	},
@@ -288,6 +293,7 @@ func TestParsingToJSON(t *testing.T) {
                 "lastWireguardHandshake": "2001-01-01T01:01:02Z",
                 "transferReceived": 200,
                 "transferSent": 100,
+				"latency": 10000000,
                 "quantumResistance": false,
                 "routes": [
                   "10.1.0.0/24"
@@ -312,6 +318,7 @@ func TestParsingToJSON(t *testing.T) {
                 "lastWireguardHandshake": "2002-02-02T02:02:03Z",
                 "transferReceived": 2000,
                 "transferSent": 1000,
+				"latency": 10000000,
                 "quantumResistance": false,
                 "routes": null
               }
@@ -409,6 +416,7 @@ func TestParsingToYAML(t *testing.T) {
           lastWireguardHandshake: 2001-01-01T01:01:02Z
           transferReceived: 200
           transferSent: 100
+          latency: 10ms
           quantumResistance: false
           routes:
             - 10.1.0.0/24
@@ -428,6 +436,7 @@ func TestParsingToYAML(t *testing.T) {
           lastWireguardHandshake: 2002-02-02T02:02:03Z
           transferReceived: 2000
           transferSent: 1000
+          latency: 10ms
           quantumResistance: false
           routes: []
 cliVersion: development
@@ -496,6 +505,7 @@ func TestParsingToDetail(t *testing.T) {
   Transfer status (received/sent) 200 B/100 B
   Quantum resistance: false
   Routes: 10.1.0.0/24
+  Latency: 10ms
 
  peer-2.awesome-domain.com:
   NetBird IP: 192.168.178.102
@@ -511,6 +521,7 @@ func TestParsingToDetail(t *testing.T) {
   Transfer status (received/sent) 2.0 KiB/1000 B
   Quantum resistance: false
   Routes: -
+  Latency: 10ms
 
 Daemon version: 0.14.1
 CLI version: development

diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go
@@ -133,6 +133,9 @@ type Conn struct {
 	adapter        iface.TunAdapter
 	iFaceDiscover  stdnet.ExternalIFaceDiscover
 	sentExtraSrflx bool
+
+	remoteEndpoint *net.UDPAddr
+	remoteConn     *ice.Conn
 }
 
 // meta holds meta information about a connection
@@ -234,6 +237,17 @@ func (conn *Conn) reCreateAgent() error {
 		return err
 	}
 
+	err = conn.agent.OnSuccessfulSelectedPairBindingResponse(func(p *ice.CandidatePair) {
+		err := conn.statusRecorder.UpdateLatency(conn.config.Key, p.Latency())
+		if err != nil {
+			log.Debugf("failed to update latency for peer %s: %s", conn.config.Key, err)
+			return
+		}
+	})
+	if err != nil {
+		return fmt.Errorf("failed setting binding response callback: %w", err)
+	}
+
 	return nil
 }
 
@@ -348,6 +362,9 @@ func (conn *Conn) Open() error {
 	if remoteOfferAnswer.WgListenPort != 0 {
 		remoteWgPort = remoteOfferAnswer.WgListenPort
 	}
+
+	conn.remoteConn = remoteConn
+
 	// the ice connection has been established successfully so we are ready to start the proxy
 	remoteAddr, err := conn.configureConnection(remoteConn, remoteWgPort, remoteOfferAnswer.RosenpassPubKey,
 		remoteOfferAnswer.RosenpassAddr)
@@ -397,6 +414,7 @@ func (conn *Conn) configureConnection(remoteConn net.Conn, remoteWgPort int, rem
 	}
 
 	endpointUdpAddr, _ := net.ResolveUDPAddr(endpoint.Network(), endpoint.String())
+	conn.remoteEndpoint = endpointUdpAddr
 
 	err = conn.config.WgConfig.WgInterface.UpdatePeer(conn.config.WgConfig.RemoteKey, conn.config.WgConfig.AllowedIps, defaultWgKeepAlive, endpointUdpAddr, conn.config.WgConfig.PreSharedKey)
 	if err != nil {

diff --git a/client/internal/peer/status.go b/client/internal/peer/status.go
@@ -28,6 +28,7 @@ type State struct {
 	LastWireguardHandshake     time.Time
 	BytesTx                    int64
 	BytesRx                    int64
+	Latency                    time.Duration
 	RosenpassEnabled           bool
 	Routes                     map[string]struct{}
 }
@@ -410,6 +411,22 @@ func (d *Status) GetManagementState() ManagementState {
 	}
 }
 
+func (d *Status) UpdateLatency(pubKey string, latency time.Duration) error {
+	if latency <= 0 {
+		return nil
+	}
+
+	d.mux.Lock()
+	defer d.mux.Unlock()
+	peerState, ok := d.peers[pubKey]
+	if !ok {
+		return errors.New("peer doesn't exist")
+	}
+	peerState.Latency = latency
+	d.peers[pubKey] = peerState
+	return nil
+}
+
 // IsLoginRequired determines if a peer's login has expired.
 func (d *Status) IsLoginRequired() bool {
 	d.mux.Lock()