Add p2p bandwidth metrics

p2p.go

@@ -16,6 +16,7 @@ import (
 	"github.com/multiformats/go-multiaddr"
 	"github.com/multiformats/go-multihash"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 	"golang.org/x/time/rate"
 
@@ -24,6 +25,7 @@ import (
 	connmgr "github.com/libp2p/go-libp2p-connmgr"
 	core "github.com/libp2p/go-libp2p-core"
 	"github.com/libp2p/go-libp2p-core/crypto"
+	"github.com/libp2p/go-libp2p-core/metrics"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/pnet"
 	"github.com/libp2p/go-libp2p-core/protocol"
@@ -118,6 +120,20 @@ var (
 	}
 )
 
+var (
+	_p2pBandwidthGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "iotex_p2p_bandwidth_gauge",
+			Help: "P2P bandwidth stats",
+		},
+		[]string{"protocol", "type"},
+	)
+)
+
+func init() {
+	prometheus.MustRegister(_p2pBandwidthGauge)
+}
+
 // Option defines the option function to modify the config for a host
 type Option func(cfg *Config) error
 
@@ -266,6 +282,7 @@ type Host struct {
 	peersLimiters  *lru.Cache
 	unicastLimiter *rate.Limiter
 	peerManager    *peerManager
+	bc             *metrics.BandwidthCounter
 }
 
 // NewHost constructs a host struct
@@ -308,6 +325,7 @@ func NewHost(ctx context.Context, options ...Option) (*Host, error) {
 	if err != nil {
 		return nil, err
 	}
+	bc := metrics.NewBandwidthCounter()
 	opts := []libp2p.Option{
 		libp2p.ListenAddrStrings(fmt.Sprintf("/ip4/%s/tcp/%d", ip, cfg.Port)),
 		libp2p.AddrsFactory(func(addrs []multiaddr.Multiaddr) []multiaddr.Multiaddr {
@@ -323,6 +341,7 @@ func NewHost(ctx context.Context, options ...Option) (*Host, error) {
 		}),
 		libp2p.Muxer("/yamux/2.0.0", yamux.DefaultTransport),
 		libp2p.ConnectionManager(connmgr.NewConnManager(cfg.ConnLowWater, cfg.ConnHighWater, cfg.ConnGracePeriod)),
+		libp2p.BandwidthReporter(bc),
 	}
 
 	if !cfg.SecureIO {
@@ -401,6 +420,7 @@ func NewHost(ctx context.Context, options ...Option) (*Host, error) {
 		unicastLimiter: rate.NewLimiter(rate.Limit(cfg.RateLimit.GlobalUnicastAvg), cfg.RateLimit.GlobalUnicastBurst),
 		peerManager: newPeerManager(host, discovery.NewRoutingDiscovery(kad), cfg.GroupID,
 			withMaxPeers(cfg.MaxPeer), withBlacklistTolerance(cfg.BlacklistTolerance), withBlacklistTimeout(cfg.BlackListTimeout)),
+		bc: bc,
 	}
 
 	addrs := make([]string, 0)
@@ -411,6 +431,18 @@ func NewHost(ctx context.Context, options ...Option) (*Host, error) {
 		zap.Strings("address", addrs),
 		zap.Bool("secureIO", myHost.cfg.SecureIO),
 		zap.Bool("gossip", myHost.cfg.Gossip))
+	// start metrics update
+	go func() {
+		ticker := time.NewTicker(500 * time.Millisecond)
+		for {
+			select {
+			case <-ticker.C:
+				myHost.updateMetrics()
+			case <-myHost.close:
+				return
+			}
+		}
+	}()
 	return &myHost, nil
 }
 
@@ -684,6 +716,19 @@ func (h *Host) Close() error {
 	return nil
 }
 
+func (h *Host) updateMetrics() {
+	if h.bc == nil {
+		return
+	}
+	for p, stats := range h.bc.GetBandwidthByProtocol() {
+		protocol := string(p)
+		_p2pBandwidthGauge.WithLabelValues(protocol, "in").Set(float64(stats.TotalIn))
+		_p2pBandwidthGauge.WithLabelValues(protocol, "out").Set(float64(stats.TotalOut))
+		_p2pBandwidthGauge.WithLabelValues(protocol, "rateIn").Set(float64(stats.RateIn))
+		_p2pBandwidthGauge.WithLabelValues(protocol, "rateOut").Set(float64(stats.RateOut))
+	}
+}
+
 func (h *Host) allowSource(src core.PeerID) (bool, error) {
 	if !h.cfg.EnableRateLimit {
 		return true, nil

peerManager.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/iotexproject/go-pkgs/cache/ttl"
 	core "github.com/libp2p/go-libp2p-core"
+	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	discovery "github.com/libp2p/go-libp2p-discovery"
 	"github.com/multiformats/go-multiaddr"
@@ -245,6 +246,9 @@ func (pm *peerManager) ConnectedPeers() []peer.AddrInfo {
 	connSet := make(map[string]bool, len(conns))
 	for _, conn := range conns {
 		remoteID := conn.RemotePeer()
+		if pm.host.Network().Connectedness(remoteID) != network.Connected {
+			continue
+		}
 		if connSet[remoteID.Pretty()] {
 			continue
 		}