Merge pull request #2852 from thomas-gardner/bitswap/ledger

bitswap: add `ledger` subcommand

core/commands/bitswap.go

@@ -5,11 +5,12 @@ import (
 	"fmt"
 	"io"
 
-	"gx/ipfs/QmPSBJL4momYnE7DcUyk2DVhD6rH488ZmHBGLbxNdhU44K/go-humanize"
-
 	key "github.com/ipfs/go-ipfs/blocks/key"
 	cmds "github.com/ipfs/go-ipfs/commands"
 	bitswap "github.com/ipfs/go-ipfs/exchange/bitswap"
+	decision "github.com/ipfs/go-ipfs/exchange/bitswap/decision"
+
+	"gx/ipfs/QmPSBJL4momYnE7DcUyk2DVhD6rH488ZmHBGLbxNdhU44K/go-humanize"
 	peer "gx/ipfs/QmRBqJF7hb8ZSpRcMwUt8hNhydWcxGEhtk81HKq6oUwKvs/go-libp2p-peer"
 	u "gx/ipfs/QmZNVWh8LLjAavuQ2JXuFmuYH3C11xo988vSgp7UQrTRj1/go-ipfs-util"
 )
@@ -23,6 +24,7 @@ var BitswapCmd = &cmds.Command{
 		"wantlist": showWantlistCmd,
 		"stat":     bitswapStatCmd,
 		"unwant":   unwantCmd,
+		"ledger":   ledgerCmd,
 	},
 }
 
@@ -171,3 +173,60 @@ var bitswapStatCmd = &cmds.Command{
 		},
 	},
 }
+
+var ledgerCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Show the current ledger for a peer.",
+		ShortDescription: `
+The Bitswap decision engine tracks the number of bytes exchanged between IPFS
+nodes, and stores this information as a collection of ledgers. This command
+prints the ledger associated with a given peer.
+`,
+	},
+	Arguments: []cmds.Argument{
+		cmds.StringArg("peer", true, false, "The PeerID (B58) of the ledger to inspect."),
+	},
+	Type: decision.Receipt{},
+	Run: func(req cmds.Request, res cmds.Response) {
+		nd, err := req.InvocContext().GetNode()
+		if err != nil {
+			res.SetError(err, cmds.ErrNormal)
+			return
+		}
+
+		if !nd.OnlineMode() {
+			res.SetError(errNotOnline, cmds.ErrClient)
+			return
+		}
+
+		bs, ok := nd.Exchange.(*bitswap.Bitswap)
+		if !ok {
+			res.SetError(u.ErrCast(), cmds.ErrNormal)
+			return
+		}
+
+		partner, err := peer.IDB58Decode(req.Arguments()[0])
+		if err != nil {
+			res.SetError(err, cmds.ErrClient)
+			return
+		}
+		res.SetOutput(bs.LedgerForPeer(partner))
+	},
+	Marshalers: cmds.MarshalerMap{
+		cmds.Text: func(res cmds.Response) (io.Reader, error) {
+			out, ok := res.Output().(*decision.Receipt)
+			if !ok {
+				return nil, u.ErrCast()
+			}
+			buf := new(bytes.Buffer)
+			fmt.Fprintf(buf, "Ledger for %s\n"+
+				"Debt ratio:\t%f\n"+
+				"Exchanges:\t%d\n"+
+				"Bytes sent:\t%d\n"+
+				"Bytes received:\t%d\n\n",
+				out.Peer, out.Value, out.Exchanged,
+				out.Sent, out.Recv)
+			return buf, nil
+		},
+	},
+}

exchange/bitswap/bitswap.go

@@ -205,6 +205,10 @@ func (bs *Bitswap) WantlistForPeer(p peer.ID) []key.Key {
 	return out
 }
 
+func (bs *Bitswap) LedgerForPeer(p peer.ID) *decision.Receipt {
+	return bs.engine.LedgerForPeer(p)
+}
+
 // GetBlocks returns a channel where the caller may receive blocks that
 // correspond to the provided |keys|. Returns an error if BitSwap is unable to
 // begin this request within the deadline enforced by the context.

exchange/bitswap/decision/engine.go

@@ -114,6 +114,21 @@ func (e *Engine) WantlistForPeer(p peer.ID) (out []wl.Entry) {
 	return out
 }
 
+func (e *Engine) LedgerForPeer(p peer.ID) *Receipt {
+	ledger := e.findOrCreate(p)
+
+	ledger.lk.Lock()
+	defer ledger.lk.Unlock()
+
+	return &Receipt{
+		Peer:      ledger.Partner.String(),
+		Value:     ledger.Accounting.Value(),
+		Sent:      ledger.Accounting.BytesSent,
+		Recv:      ledger.Accounting.BytesRecv,
+		Exchanged: ledger.ExchangeCount(),
+	}
+}
+
 func (e *Engine) taskWorker(ctx context.Context) {
 	defer close(e.outbox) // because taskWorker uses the channel exclusively
 	for {

exchange/bitswap/decision/ledger.go

@@ -49,6 +49,14 @@ type ledger struct {
 	lk sync.Mutex
 }
 
+type Receipt struct {
+	Peer      string
+	Value     float64
+	Sent      uint64
+	Recv      uint64
+	Exchanged uint64
+}
+
 type debtRatio struct {
 	BytesSent uint64
 	BytesRecv uint64