Merge pull request #3497 from filecoin-project/feat/bandwidth-tracking

add apis and command to inspect bandwidth usage

api/api_common.go

@@ -5,8 +5,10 @@ import (
  "fmt"
 
  "github.com/filecoin-project/go-jsonrpc/auth"
+ metrics "github.com/libp2p/go-libp2p-core/metrics"
  "github.com/libp2p/go-libp2p-core/network"
  "github.com/libp2p/go-libp2p-core/peer"
+ protocol "github.com/libp2p/go-libp2p-core/protocol"
 
  "github.com/filecoin-project/lotus/build"
 )
@@ -29,6 +31,18 @@ type Common interface {
  NetPubsubScores(context.Context) ([]PubsubScore, error)
  NetAutoNatStatus(context.Context) (NatInfo, error)
 
+ // NetBandwidthStats returns statistics about the nodes total bandwidth
+ // usage and current rate across all peers and protocols.
+ NetBandwidthStats(ctx context.Context) (metrics.Stats, error)
+
+ // NetBandwidthStatsByPeer returns statistics about the nodes bandwidth
+ // usage and current rate per peer
+ NetBandwidthStatsByPeer(ctx context.Context) (map[string]metrics.Stats, error)
+
+ // NetBandwidthStatsByProtocol returns statistics about the nodes bandwidth
+ // usage and current rate per protocol
+ NetBandwidthStatsByProtocol(ctx context.Context) (map[protocol.ID]metrics.Stats, error)
+
  // MethodGroup: Common
 
  // ID returns peerID of libp2p node backing this API

api/apistruct/struct.go

@@ -6,8 +6,10 @@ import (
  "time"
 
  "github.com/ipfs/go-cid"
+ metrics "github.com/libp2p/go-libp2p-core/metrics"
  "github.com/libp2p/go-libp2p-core/network"
  "github.com/libp2p/go-libp2p-core/peer"
+ protocol "github.com/libp2p/go-libp2p-core/protocol"
 
  "github.com/filecoin-project/go-address"
  "github.com/filecoin-project/go-fil-markets/piecestore"
@@ -42,14 +44,17 @@ type CommonStruct struct {
  AuthVerify func(ctx context.Context, token string) ([]auth.Permission, error) `perm:"read"`
  AuthNew func(ctx context.Context, perms []auth.Permission) ([]byte, error) `perm:"admin"`
 
- NetConnectedness func(context.Context, peer.ID) (network.Connectedness, error) `perm:"read"`
- NetPeers func(context.Context) ([]peer.AddrInfo, error) `perm:"read"`
- NetConnect func(context.Context, peer.AddrInfo) error `perm:"write"`
- NetAddrsListen func(context.Context) (peer.AddrInfo, error) `perm:"read"`
- NetDisconnect func(context.Context, peer.ID) error `perm:"write"`
- NetFindPeer func(context.Context, peer.ID) (peer.AddrInfo, error) `perm:"read"`
- NetPubsubScores func(context.Context) ([]api.PubsubScore, error) `perm:"read"`
- NetAutoNatStatus func(context.Context) (api.NatInfo, error) `perm:"read"`
+ NetConnectedness func(context.Context, peer.ID) (network.Connectedness, error) `perm:"read"`
+ NetPeers func(context.Context) ([]peer.AddrInfo, error) `perm:"read"`
+ NetConnect func(context.Context, peer.AddrInfo) error `perm:"write"`
+ NetAddrsListen func(context.Context) (peer.AddrInfo, error) `perm:"read"`
+ NetDisconnect func(context.Context, peer.ID) error `perm:"write"`
+ NetFindPeer func(context.Context, peer.ID) (peer.AddrInfo, error) `perm:"read"`
+ NetPubsubScores func(context.Context) ([]api.PubsubScore, error) `perm:"read"`
+ NetAutoNatStatus func(context.Context) (api.NatInfo, error) `perm:"read"`
+ NetBandwidthStats func(ctx context.Context) (metrics.Stats, error) `perm:"read"`
+ NetBandwidthStatsByPeer func(ctx context.Context) (map[string]metrics.Stats, error) `perm:"read"`
+ NetBandwidthStatsByProtocol func(ctx context.Context) (map[protocol.ID]metrics.Stats, error) `perm:"read"`
 
  ID func(context.Context) (peer.ID, error) `perm:"read"`
  Version func(context.Context) (api.Version, error) `perm:"read"`
@@ -371,6 +376,18 @@ func (c *CommonStruct) NetAutoNatStatus(ctx context.Context) (api.NatInfo, error
  return c.Internal.NetAutoNatStatus(ctx)
 }
 
+func (c *CommonStruct) NetBandwidthStats(ctx context.Context) (metrics.Stats, error) {
+ return c.Internal.NetBandwidthStats(ctx)
+}
+
+func (c *CommonStruct) NetBandwidthStatsByPeer(ctx context.Context) (map[string]metrics.Stats, error) {
+ return c.Internal.NetBandwidthStatsByPeer(ctx)
+}
+
+func (c *CommonStruct) NetBandwidthStatsByProtocol(ctx context.Context) (map[protocol.ID]metrics.Stats, error) {
+ return c.Internal.NetBandwidthStatsByProtocol(ctx)
+}
+
 // ID implements API.ID
 func (c *CommonStruct) ID(ctx context.Context) (peer.ID, error) {
  return c.Internal.ID(ctx)

api/docgen/docgen.go

@@ -14,8 +14,10 @@ import (
 
  "github.com/ipfs/go-cid"
  "github.com/ipfs/go-filestore"
+ metrics "github.com/libp2p/go-libp2p-core/metrics"
  "github.com/libp2p/go-libp2p-core/network"
  "github.com/libp2p/go-libp2p-core/peer"
+ protocol "github.com/libp2p/go-libp2p-core/protocol"
  pubsub "github.com/libp2p/go-libp2p-pubsub"
  "github.com/multiformats/go-multiaddr"
 
@@ -133,6 +135,22 @@ func init() {
  InvalidMessageDeliveries: 3,
  },
  })
+ addExample(map[string]metrics.Stats{
+ "12D3KooWSXmXLJmBR1M7i9RW9GQPNUhZSzXKzxDHWtAgNuJAbyEJ": {
+ RateIn: 100,
+ RateOut: 50,
+ TotalIn: 174000,
+ TotalOut: 12500,
+ },
+ })
+ addExample(map[protocol.ID]metrics.Stats{
+ "/fil/hello/1.0.0": {
+ RateIn: 100,
+ RateOut: 50,
+ TotalIn: 174000,
+ TotalOut: 12500,
+ },
+ })
 
  maddr, err := multiaddr.NewMultiaddr("/ip4/52.36.61.156/tcp/1347/p2p/12D3KooWFETiESTf1v4PGUvtnxMAcEFMzLZbJGg4tjWfGEimYior")
  if err != nil {

cli/net.go

@@ -6,9 +6,12 @@ import (
  "os"
  "sort"
  "strings"
+ "text/tabwriter"
 
  "github.com/libp2p/go-libp2p-core/peer"
+ protocol "github.com/libp2p/go-libp2p-core/protocol"
 
+ "github.com/dustin/go-humanize"
  "github.com/urfave/cli/v2"
 
  "github.com/filecoin-project/lotus/lib/addrutil"
@@ -25,6 +28,7 @@ var netCmd = &cli.Command{
  netFindPeer,
  netScores,
  NetReachability,
+ NetBandwidthCmd,
  },
 }
 
@@ -228,3 +232,88 @@ var NetReachability = &cli.Command{
  return nil
  },
 }
+
+var NetBandwidthCmd = &cli.Command{
+ Name: "bandwidth",
+ Usage: "Print bandwidth usage information",
+ Flags: []cli.Flag{
+ &cli.BoolFlag{
+ Name: "by-peer",
+ Usage: "list bandwidth usage by peer",
+ },
+ &cli.BoolFlag{
+ Name: "by-protocol",
+ Usage: "list bandwidth usage by protocol",
+ },
+ },
+ Action: func(cctx *cli.Context) error {
+ api, closer, err := GetAPI(cctx)
+ if err != nil {
+ return err
+ }
+ defer closer()
+
+ ctx := ReqContext(cctx)
+
+ bypeer := cctx.Bool("by-peer")
+ byproto := cctx.Bool("by-protocol")
+
+ tw := tabwriter.NewWriter(os.Stdout, 4, 4, 2, ' ', 0)
+
+ fmt.Fprintf(tw, "Segment\tTotalIn\tTotalOut\tRateIn\tRateOut\n")
+
+ if bypeer {
+ bw, err := api.NetBandwidthStatsByPeer(ctx)
+ if err != nil {
+ return err
+ }
+
+ var peers []string
+ for p := range bw {
+ peers = append(peers, p)
+ }
+
+ sort.Slice(peers, func(i, j int) bool {
+ return peers[i] < peers[j]
+ })
+
+ for _, p := range peers {
+ s := bw[p]
+ fmt.Fprintf(tw, "%s\t%s\t%s\t%s/s\t%s/s\n", p, humanize.Bytes(uint64(s.TotalIn)), humanize.Bytes(uint64(s.TotalOut)), humanize.Bytes(uint64(s.RateIn)), humanize.Bytes(uint64(s.RateOut)))
+ }
+ } else if byproto {
+ bw, err := api.NetBandwidthStatsByProtocol(ctx)
+ if err != nil {
+ return err
+ }
+
+ var protos []protocol.ID
+ for p := range bw {
+ protos = append(protos, p)
+ }
+
+ sort.Slice(protos, func(i, j int) bool {
+ return protos[i] < protos[j]
+ })
+
+ for _, p := range protos {
+ s := bw[p]
+ if p == "" {
+ p = "<unknown>"
+ }
+ fmt.Fprintf(tw, "%s\t%s\t%s\t%s/s\t%s/s\n", p, humanize.Bytes(uint64(s.TotalIn)), humanize.Bytes(uint64(s.TotalOut)), humanize.Bytes(uint64(s.RateIn)), humanize.Bytes(uint64(s.RateOut)))
+ }
+ } else {
+
+ s, err := api.NetBandwidthStats(ctx)
+ if err != nil {
+ return err
+ }
+
+ fmt.Fprintf(tw, "Total\t%s\t%s\t%s/s\t%s/s\n", humanize.Bytes(uint64(s.TotalIn)), humanize.Bytes(uint64(s.TotalOut)), humanize.Bytes(uint64(s.RateIn)), humanize.Bytes(uint64(s.RateOut)))
+ }
+
+ return tw.Flush()
+
+ },
+}

documentation/en/api-methods.md

@@ -85,6 +85,9 @@
 * [Net](#Net)
  * [NetAddrsListen](#NetAddrsListen)
  * [NetAutoNatStatus](#NetAutoNatStatus)
+ * [NetBandwidthStats](#NetBandwidthStats)
+ * [NetBandwidthStatsByPeer](#NetBandwidthStatsByPeer)
+ * [NetBandwidthStatsByProtocol](#NetBandwidthStatsByProtocol)
  * [NetConnect](#NetConnect)
  * [NetConnectedness](#NetConnectedness)
  * [NetDisconnect](#NetDisconnect)
@@ -2059,6 +2062,61 @@ Response:
 }
 ```
 
+### NetBandwidthStats
+
+
+Perms: read
+
+Inputs: `null`
+
+Response:
+```json
+{
+ "TotalIn": 9,
+ "TotalOut": 9,
+ "RateIn": 12.3,
+ "RateOut": 12.3
+}
+```
+
+### NetBandwidthStatsByPeer
+
+
+Perms: read
+
+Inputs: `null`
+
+Response:
+```json
+{
+ "12D3KooWSXmXLJmBR1M7i9RW9GQPNUhZSzXKzxDHWtAgNuJAbyEJ": {
+ "TotalIn": 174000,
+ "TotalOut": 12500,
+ "RateIn": 100,
+ "RateOut": 50
+ }
+}
+```
+
+### NetBandwidthStatsByProtocol
+
+
+Perms: read
+
+Inputs: `null`
+
+Response:
+```json
+{
+ "/fil/hello/1.0.0": {
+ "TotalIn": 174000,
+ "TotalOut": 12500,
+ "RateIn": 100,
+ "RateOut": 50
+ }
+}
+```
+
 ### NetConnect
 
 

go.mod

@@ -18,6 +18,7 @@ require (
  github.com/docker/go-units v0.4.0
  github.com/drand/drand v1.0.3-0.20200714175734-29705eaf09d4
  github.com/drand/kyber v1.1.1
+ github.com/dustin/go-humanize v1.0.0
  github.com/elastic/go-sysinfo v1.3.0
  github.com/fatih/color v1.8.0
  github.com/filecoin-project/chain-validation v0.0.6-0.20200813000554-40c22fe26eef

go.sum

@@ -829,6 +829,7 @@ github.com/libp2p/go-libp2p-core v0.6.1 h1:XS+Goh+QegCDojUZp00CaPMfiEADCrLjNZskW
 github.com/libp2p/go-libp2p-core v0.6.1/go.mod h1:FfewUH/YpvWbEB+ZY9AQRQ4TAD8sJBt/G1rVvhz5XT8=
 github.com/libp2p/go-libp2p-crypto v0.0.1/go.mod h1:yJkNyDmO341d5wwXxDUGO0LykUVT72ImHNUqh5D/dBE=
 github.com/libp2p/go-libp2p-crypto v0.0.2/go.mod h1:eETI5OUfBnvARGOHrJz2eWNyTUxEGZnBxMcbUjfIj4I=
+github.com/libp2p/go-libp2p-crypto v0.1.0 h1:k9MFy+o2zGDNGsaoZl0MA3iZ75qXxr9OOoAZF+sD5OQ=
 github.com/libp2p/go-libp2p-crypto v0.1.0/go.mod h1:sPUokVISZiy+nNuTTH/TY+leRSxnFj/2GLjtOTW90hI=
 github.com/libp2p/go-libp2p-daemon v0.2.2/go.mod h1:kyrpsLB2JeNYR2rvXSVWyY0iZuRIMhqzWR3im9BV6NQ=
 github.com/libp2p/go-libp2p-discovery v0.0.1/go.mod h1:ZkkF9xIFRLA1xCc7bstYFkd80gBGK8Fc1JqGoU2i+zI=
@@ -854,6 +855,7 @@ github.com/libp2p/go-libp2p-kbucket v0.4.2/go.mod h1:7sCeZx2GkNK1S6lQnGUW5JYZCFP
 github.com/libp2p/go-libp2p-loggables v0.0.1/go.mod h1:lDipDlBNYbpyqyPX/KcoO+eq0sJYEVR2JgOexcivchg=
 github.com/libp2p/go-libp2p-loggables v0.1.0 h1:h3w8QFfCt2UJl/0/NW4K829HX/0S4KD31PQ7m8UXXO8=
 github.com/libp2p/go-libp2p-loggables v0.1.0/go.mod h1:EyumB2Y6PrYjr55Q3/tiJ/o3xoDasoRYM7nOzEpoa90=
+github.com/libp2p/go-libp2p-metrics v0.0.1 h1:yumdPC/P2VzINdmcKZd0pciSUCpou+s0lwYCjBbzQZU=
 github.com/libp2p/go-libp2p-metrics v0.0.1/go.mod h1:jQJ95SXXA/K1VZi13h52WZMa9ja78zjyy5rspMsC/08=
 github.com/libp2p/go-libp2p-mplex v0.1.1/go.mod h1:KUQWpGkCzfV7UIpi8SKsAVxyBgz1c9R5EvxgnwLsb/I=
 github.com/libp2p/go-libp2p-mplex v0.2.0/go.mod h1:Ejl9IyjvXJ0T9iqUTE1jpYATQ9NM3g+OtR+EMMODbKo=
@@ -876,6 +878,7 @@ github.com/libp2p/go-libp2p-noise v0.1.1 h1:vqYQWvnIcHpIoWJKC7Al4D6Hgj0H012TuXRh
 github.com/libp2p/go-libp2p-noise v0.1.1/go.mod h1:QDFLdKX7nluB7DEnlVPbz7xlLHdwHFA9HiohJRr3vwM=
 github.com/libp2p/go-libp2p-peer v0.0.1/go.mod h1:nXQvOBbwVqoP+T5Y5nCjeH4sP9IX/J0AMzcDUVruVoo=
 github.com/libp2p/go-libp2p-peer v0.1.1/go.mod h1:jkF12jGB4Gk/IOo+yomm+7oLWxF278F7UnrYUQ1Q8es=
+github.com/libp2p/go-libp2p-peer v0.2.0 h1:EQ8kMjaCUwt/Y5uLgjT8iY2qg0mGUT0N1zUjer50DsY=
 github.com/libp2p/go-libp2p-peer v0.2.0/go.mod h1:RCffaCvUyW2CJmG2gAWVqwePwW7JMgxjsHm7+J5kjWY=
 github.com/libp2p/go-libp2p-peerstore v0.0.1/go.mod h1:RabLyPVJLuNQ+GFyoEkfi8H4Ti6k/HtZJ7YKgtSq+20=
 github.com/libp2p/go-libp2p-peerstore v0.0.6/go.mod h1:RabLyPVJLuNQ+GFyoEkfi8H4Ti6k/HtZJ7YKgtSq+20=
@@ -892,6 +895,7 @@ github.com/libp2p/go-libp2p-peerstore v0.2.6/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuD
 github.com/libp2p/go-libp2p-pnet v0.2.0 h1:J6htxttBipJujEjz1y0a5+eYoiPcFHhSYHH6na5f0/k=
 github.com/libp2p/go-libp2p-pnet v0.2.0/go.mod h1:Qqvq6JH/oMZGwqs3N1Fqhv8NVhrdYcO0BW4wssv21LA=
 github.com/libp2p/go-libp2p-protocol v0.0.1/go.mod h1:Af9n4PiruirSDjHycM1QuiMi/1VZNHYcK8cLgFJLZ4s=
+github.com/libp2p/go-libp2p-protocol v0.1.0 h1:HdqhEyhg0ToCaxgMhnOmUO8snQtt/kQlcjVk3UoJU3c=
 github.com/libp2p/go-libp2p-protocol v0.1.0/go.mod h1:KQPHpAabB57XQxGrXCNvbL6UEXfQqUgC/1adR2Xtflk=
 github.com/libp2p/go-libp2p-pubsub v0.1.1/go.mod h1:ZwlKzRSe1eGvSIdU5bD7+8RZN/Uzw0t1Bp9R1znpR/Q=
 github.com/libp2p/go-libp2p-pubsub v0.3.2-0.20200527132641-c0712c6e92cf/go.mod h1:TxPOBuo1FPdsTjFnv+FGZbNbWYsp74Culx+4ViQpato=

node/builder.go

@@ -86,6 +86,7 @@ var (
  NatPortMapKey = special{8} // Libp2p option
  ConnectionManagerKey = special{9} // Libp2p option
  AutoNATSvcKey = special{10} // Libp2p option
+ BandwidthReporterKey = special{11} // Libp2p option
 )
 
 type invoke int
@@ -181,6 +182,7 @@ func libp2p() Option {
  Override(new(routing.Routing), lp2p.Routing),
 
  Override(NatPortMapKey, lp2p.NatPortMap),
+ Override(BandwidthReporterKey, lp2p.BandwidthCounter),
 
  Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second, nil)),
  Override(AutoNATSvcKey, lp2p.AutoNATService),

node/impl/common/common.go

@@ -9,8 +9,10 @@ import (
 
  "github.com/gbrlsnchs/jwt/v3"
  "github.com/libp2p/go-libp2p-core/host"
+ metrics "github.com/libp2p/go-libp2p-core/metrics"
  "github.com/libp2p/go-libp2p-core/network"
  "github.com/libp2p/go-libp2p-core/peer"
+ protocol "github.com/libp2p/go-libp2p-core/protocol"
  swarm "github.com/libp2p/go-libp2p-swarm"
  basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
  ma "github.com/multiformats/go-multiaddr"
@@ -32,6 +34,7 @@ type CommonAPI struct {
  RawHost lp2p.RawHost
  Host host.Host
  Router lp2p.BaseIpfsRouting
+ Reporter metrics.Reporter
  Sk *dtypes.ScoreKeeper
  ShutdownChan dtypes.ShutdownChan
 }
@@ -133,6 +136,22 @@ func (a *CommonAPI) NetAutoNatStatus(ctx context.Context) (i api.NatInfo, err er
  }, nil
 }
 
+func (a *CommonAPI) NetBandwidthStats(ctx context.Context) (metrics.Stats, error) {
+ return a.Reporter.GetBandwidthTotals(), nil
+}
+
+func (a *CommonAPI) NetBandwidthStatsByPeer(ctx context.Context) (map[string]metrics.Stats, error) {
+ out := make(map[string]metrics.Stats)
+ for p, s := range a.Reporter.GetBandwidthByPeer() {
+ out[p.String()] = s
+ }
+ return out, nil
+}
+
+func (a *CommonAPI) NetBandwidthStatsByProtocol(ctx context.Context) (map[protocol.ID]metrics.Stats, error) {
+ return a.Reporter.GetBandwidthByProtocol(), nil
+}
+
 func (a *CommonAPI) ID(context.Context) (peer.ID, error) {
  return a.Host.ID(), nil
 }