feat: .stats.bw* - Bandwidth Stats

bw

@@ -0,0 +1 @@
+{"TotalIn":"0","TotalOut":"0","RateIn":"0","RateOut":"0"}

package.json

@@ -74,9 +74,8 @@
     "form-data": "^2.3.2",
     "go-ipfs-dep": "^0.4.13",
     "hat": "0.0.3",
-    "interface-ipfs-core": "~0.56.5",
     "ipfsd-ctl": "~0.30.1",
-    "left-pad": "^1.2.0",
+    "interface-ipfs-core": "~0.57.0",
     "lodash": "^4.17.5",
     "mocha": "^5.0.4",
     "ncp": "^2.0.0",
@@ -123,7 +122,11 @@
     "joi-browser": "^13.0.1",
     "joi-multiaddr": "^1.0.1",
     "libp2p": "~0.19.0",
+<<<<<<< HEAD
     "libp2p-circuit": "~0.1.5",
+=======
+    "libp2p-circuit": "~0.1.4",
+>>>>>>> feat: .stats.bw*
     "libp2p-floodsub": "~0.14.1",
     "libp2p-kad-dht": "~0.9.0",
     "libp2p-keychain": "~0.3.1",

src/cli/commands/stats/bw.js

@@ -0,0 +1,48 @@
+'use strict'
+
+const pull = require('pull-stream')
+
+module.exports = {
+  command: 'bw',
+
+  describe: 'Get bandwidth information.',
+
+  builder: {
+    peer: {
+      type: 'string',
+      default: ''
+    },
+    proto: {
+      type: 'string',
+      default: ''
+    },
+    poll: {
+      type: 'boolean',
+      default: false
+    },
+    interval: {
+      type: 'string',
+      default: '1s'
+    }
+  },
+
+  handler (argv) {
+    const stream = argv.ipfs.stats.bwPullStream({
+      peer: argv.peer,
+      proto: argv.proto,
+      poll: argv.poll,
+      interval: argv.interval
+    })
+
+    pull(
+      stream,
+      pull.drain((chunk) => {
+        console.log(`bandwidth status
+  total in: ${chunk.totalIn}B
+  total out: ${chunk.totalOut}B
+  rate in: ${chunk.rateIn}B/s
+  rate out: ${chunk.rateOut}B/s`)
+      })
+    )
+  }
+}

src/core/components/stats.js

@@ -1,8 +1,88 @@
 'use strict'
 
+const promisify = require('promisify-es6')
+const Big = require('big.js')
+const Pushable = require('pull-pushable')
+const human = require('human-to-milliseconds')
+const toStream = require('pull-stream-to-stream')
+
+function bandwidthStats (self, opts) {
+  return new Promise((resolve, reject) => {
+    let stats
+
+    if (opts.peer) {
+      stats = self._libp2pNode.stats.forPeer(opts.peer)
+    } else if (opts.proto) {
+      stats = self._libp2pNode.stats.forProtocol(opts.proto)
+    } else {
+      stats = self._libp2pNode.stats.global
+    }
+
+    if (!stats) {
+      resolve({
+        totalIn: new Big(0),
+        totalOut: new Big(0),
+        rateIn: new Big(0),
+        rateOut: new Big(0)
+      })
+      return
+    }
+
+    resolve({
+      totalIn: stats.snapshot.dataReceived,
+      totalOut: stats.snapshot.dataSent,
+      rateIn: new Big(stats.movingAverages.dataReceived['60000'].movingAverage() / 60),
+      rateOut: new Big(stats.movingAverages.dataSent['60000'].movingAverage() / 60)
+    })
+  })
+}
+
 module.exports = function stats (self) {
+  const _bwPullStream = (opts) => {
+    opts = opts || {}
+    let interval = null
+    let stream = Pushable(true, () => {
+      if (interval) {
+        clearInterval(interval)
+      }
+    })
+
+    if (opts.poll) {
+      human(opts.interval || '1s', (err, value) => {
+        if (err) throw err
+
+        interval = setInterval(() => {
+          bandwidthStats(self, opts)
+            .then((stats) => stream.push(stats))
+            .catch((err) => stream.end(err))
+        }, value)
+      })
+    } else {
+      bandwidthStats(self, opts)
+        .then((stats) => {
+          stream.push(stats)
+          stream.end()
+        })
+        .catch((err) => stream.end(err))
+    }
+
+    return stream.source
+  }
+
   return {
     bitswap: require('./bitswap')(self).stat,
-    repo: require('./repo')(self).stat
+    repo: require('./repo')(self).stat,
+    bw: promisify((opts, callback) => {
+      if (typeof opts === 'function') {
+        callback = opts
+        opts = {}
+      }
+
+      bandwidthStats(self, opts)
+        .then((stats) => callback(null, stats))
+        .catch((err) => callback(err))
+    }),
+    bwReadableStream: (opts) => toStream.source(_bwPullStream(opts)),
+    bwPullStream: _bwPullStream
   }
 }

src/core/components/swarm.js

@@ -3,6 +3,7 @@
 const multiaddr = require('multiaddr')
 const promisify = require('promisify-es6')
 const values = require('lodash.values')
+const PeerId = require('peer-id')
 
 const OFFLINE_ERROR = require('../utils').OFFLINE_ERROR
 

src/http/api/resources/stats.js

@@ -1,7 +1,46 @@
 'use strict'
 
+const { Transform } = require('readable-stream')
+
+const transformBandwidth = (stat) => {
+  return {
+    TotalIn: stat.totalIn,
+    TotalOut: stat.totalOut,
+    RateIn: stat.rateIn,
+    RateOut: stat.rateOut
+  }
+}
+
 exports = module.exports
 
 exports.bitswap = require('./bitswap').stat
 
 exports.repo = require('./repo').stat
+
+exports.bw = (request, reply) => {
+  const ipfs = request.server.app.ipfs
+  const options = {
+    peer: request.query.peer,
+    proto: request.query.proto,
+    poll: request.query.poll === 'true',
+    interval: request.query.interval || '1s'
+  }
+
+  const res = ipfs.stats.bwReadableStream(options)
+  const output = new Transform({
+    writableObjectMode: true,
+    transform (chunk, encoding, cb) {
+      this.push(JSON.stringify(transformBandwidth(chunk)) + '\n')
+      cb()
+    }
+  })
+
+  request.on('disconnect', () => {
+    res.destroy()
+  })
+
+  res.pipe(output)
+  reply(output)
+    .header('content-type', 'application/json')
+    .header('x-chunked-output', '1')
+}

src/http/api/routes/stats.js

@@ -20,4 +20,12 @@ module.exports = (server) => {
       handler: resources.stats.repo
     }
   })
+
+  api.route({
+    method: '*',
+    path: '/api/v0/stats/bw',
+    config: {
+      handler: resources.stats.bw
+    }
+  })
 }

test/cli/commands.js

@@ -4,7 +4,7 @@
 const expect = require('chai').expect
 const runOnAndOff = require('../utils/on-and-off')
 
-const commandCount = 72
+const commandCount = 73
 describe('commands', () => runOnAndOff((thing) => {
   let ipfs