api: add ping info to handlers

Author: pymq

api/debug.go

@@ -52,6 +52,7 @@ func (h *Handler) GetP2pDebugInfo(c echo.Context) (err error) {
 			Total:      makeBandwidthInfo(h.p2p.NetworkStats()),
 			ByProtocol: bandwidthByProtocol,
 		},
+		KnownPeers: h.getKnownPeers(),
 	}
 
 	return c.JSONPretty(http.StatusOK, debugInfo, "    ")

api/peers.go

@@ -5,11 +5,12 @@ import (
 	"sort"
 	"strings"
 
+	"github.com/labstack/echo/v4"
+	"github.com/libp2p/go-libp2p/core/peer"
+
 	"github.com/anywherelan/awl/awldns"
 	"github.com/anywherelan/awl/config"
 	"github.com/anywherelan/awl/entity"
-	"github.com/labstack/echo/v4"
-	"github.com/libp2p/go-libp2p/core/peer"
 )
 
 const ErrorPeerAliasIsNotUniq = "peer name is not unique"
@@ -21,6 +22,11 @@ const ErrorPeerAliasIsNotUniq = "peer name is not unique"
 // @Success 200 {array} entity.KnownPeersResponse
 // @Router /peers/get_known [GET]
 func (h *Handler) GetKnownPeers(c echo.Context) (err error) {
+	result := h.getKnownPeers()
+	return c.JSON(http.StatusOK, result)
+}
+
+func (h *Handler) getKnownPeers() []entity.KnownPeersResponse {
 	h.conf.RLock()
 	result := make([]entity.KnownPeersResponse, 0, len(h.conf.KnownPeers))
 	peers := make([]string, 0, len(h.conf.KnownPeers))
@@ -52,6 +58,7 @@ func (h *Handler) GetKnownPeers(c echo.Context) (err error) {
 			Connections:            h.p2p.PeerConnectionsInfo(id),
 			NetworkStats:           netStats,
 			NetworkStatsInIECUnits: getStatsInIECUnits(netStats),
+			Ping:                   h.p2p.GetPeerLatency(id),
 		}
 		result = append(result, kpr)
 	}
@@ -61,7 +68,7 @@ func (h *Handler) GetKnownPeers(c echo.Context) (err error) {
 		return result[i].Connected && !result[j].Connected
 	})
 
-	return c.JSON(http.StatusOK, result)
+	return result
 }
 
 // @Tags Peers

application_test.go

@@ -56,6 +56,11 @@ func TestMakeFriends(t *testing.T) {
 	peer2 := ts.newTestPeer(false)
 
 	ts.makeFriends(peer2, peer1)
+
+	p1Ping := peer1.app.P2p.GetPeerLatency(peer2.app.P2p.PeerID())
+	ts.NotEmpty(p1Ping)
+	p2Ping := peer2.app.P2p.GetPeerLatency(peer1.app.P2p.PeerID())
+	ts.NotEmpty(p2Ping)
 }
 
 func TestRemovePeer(t *testing.T) {

entity/api.go

@@ -62,6 +62,7 @@ type (
 		Connections            []p2p.ConnectionInfo
 		NetworkStats           metrics.Stats
 		NetworkStatsInIECUnits StatsInUnits
+		Ping                   time.Duration `swaggertype:"primitive,integer"`
 	}
 
 	PeerInfo struct {
@@ -114,6 +115,7 @@ type (
 		DHT         DhtDebugInfo
 		Connections ConnectionsDebugInfo
 		Bandwidth   BandwidthDebugInfo
+		KnownPeers  []KnownPeersResponse
 	}
 
 	GeneralDebugInfo struct {

p2p/metrics.go

@@ -42,13 +42,12 @@ type ConnectionInfo struct {
 	Direction    string
 	Opened       time.Time
 	Transient    bool
-	Latency      Duration
 }
 
 type BootstrapPeerDebugInfo struct {
 	Error       string   `json:",omitempty"`
 	Connections []string `json:",omitempty"`
-	Latency     Duration `json:",omitempty"`
+	Ping        Duration `json:",omitempty"`
 }
 
 func (p *P2p) Uptime() time.Duration {
@@ -79,7 +78,6 @@ func (p *P2p) PeerConnectionsInfo(peerID peer.ID) []ConnectionInfo {
 		info.Direction = strings.ToLower(stat.Direction.String())
 		info.Opened = stat.Opened
 		info.Transient = stat.Limited
-		info.Latency = Duration(p.host.Peerstore().LatencyEWMA(peerID))
 		infos = append(infos, info)
 	}
 	return infos

p2p/p2p.go

@@ -280,6 +280,14 @@ func (p *P2p) UnprotectPeer(id peer.ID) {
 	p.host.ConnManager().Unprotect(id, protectedPeerTag)
 }
 
+func (p *P2p) RecordPeerLatency(id peer.ID, rtt time.Duration) {
+	p.host.Peerstore().RecordLatency(id, rtt)
+}
+
+func (p *P2p) GetPeerLatency(id peer.ID) time.Duration {
+	return p.host.Peerstore().LatencyEWMA(id)
+}
+
 func (p *P2p) SubscribeConnectionEvents(onConnected, onDisconnected func(network.Network, network.Conn)) {
 	notifyBundle := &network.NotifyBundle{
 		ConnectedF:    onConnected,
@@ -385,7 +393,7 @@ func (p *P2p) connectToKnownPeers(ctx context.Context, timeout time.Duration, pe
 			if err == nil {
 				p.pingPeer(ctx, peerAddr.ID)
 			}
-			info.Latency = Duration(p.host.Peerstore().LatencyEWMA(peerAddr.ID))
+			info.Ping = Duration(p.host.Peerstore().LatencyEWMA(peerAddr.ID))
 
 			mu.Lock()
 			bootstrapsInfo[peerAddr.ID.String()] = info
@@ -404,6 +412,7 @@ func (p *P2p) pingPeer(ctx context.Context, peerID peer.ID) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
+	// TODO: also calc jitter
 	ch := ping.Ping(ctx, p.host, peerID)
 	pingCount := 0
 	for range ch {

service/auth_status.go

@@ -31,6 +31,7 @@ type P2p interface {
 	NewStreamWithDedicatedConn(ctx context.Context, id peer.ID, proto libp2pProtocol.ID) (network.Stream, error)
 	SubscribeConnectionEvents(onConnected, onDisconnected func(network.Network, network.Conn))
 	ProtectPeer(id peer.ID)
+	RecordPeerLatency(id peer.ID, rtt time.Duration)
 }
 
 type AuthStatus struct {
@@ -125,6 +126,7 @@ func (s *AuthStatus) ExchangeNewStatusInfo(ctx context.Context, remotePeerID pee
 
 	_, isBlocked := s.conf.GetBlockedPeer(remotePeerID.String())
 	myPeerInfo := s.createPeerInfo(knownPeer, s.conf.P2pNode.Name, isBlocked)
+	timeStarted := time.Now()
 	err = protocol.SendStatus(stream, myPeerInfo)
 	if err != nil {
 		return fmt.Errorf("sending status info: %v", err)
@@ -134,6 +136,7 @@ func (s *AuthStatus) ExchangeNewStatusInfo(ctx context.Context, remotePeerID pee
 	if err != nil {
 		return fmt.Errorf("receiving status info: %v", err)
 	}
+	s.p2p.RecordPeerLatency(remotePeerID, time.Since(timeStarted))
 
 	s.logger.Infof("successfully exchanged status info (outbound) with %s (%s)", knownPeer.DisplayName(), remotePeerID.String())
 	if isBlocked {
@@ -270,6 +273,7 @@ func (s *AuthStatus) SendAuthRequest(ctx context.Context, peerID peer.ID, req pr
 		_ = stream.Close()
 	}()
 
+	timeStarted := time.Now()
 	err = protocol.SendAuth(stream, req)
 	if err != nil {
 		return fmt.Errorf("sending auth: %v", err)
@@ -279,6 +283,7 @@ func (s *AuthStatus) SendAuthRequest(ctx context.Context, peerID peer.ID, req pr
 	if err != nil {
 		return fmt.Errorf("receiving auth response from %s: %v", peerID, err)
 	}
+	s.p2p.RecordPeerLatency(peerID, time.Since(timeStarted))
 
 	if authResponse.Confirmed || authResponse.Declined {
 		s.authsLock.Lock()