per-peer stats

README.md

@@ -54,6 +54,94 @@ Loading this module through a script tag will make the `IpfsBitswap` object avai
 
 See https://ipfs.github.io/js-ipfs-bitswap
 
+### Stats
+
+```js
+const bitswapNode = // ...
+
+const stats = bitswapNode.stat()
+```
+
+Stats contains a snapshot accessor, a moving average acessor and a peer accessor.
+
+Besides that, it emits "update" events every time it is updated.
+
+```js
+stats.on('update', (stats) => {
+  console.log('latest stats snapshot: %j', stats)
+})
+```
+
+#### Peer accessor:
+
+You can get the stats for a specific peer by doing:
+
+```js
+const peerStats = stats.forPeer(peerId)
+```
+
+The returned object behaves like the root stats accessor (has a snapshot, a moving average accessors and is an event emitter).
+
+#### Global snapshot accessor:
+
+```js
+const snapshot = stats.snapshot
+console.log('stats: %j', snapshot)
+```
+
+the snapshot will contain the following keys, with the values being [Big.js](https://github.com/MikeMcl/big.js#readme) instances:
+
+```js
+// stats: {
+//   "dataReceived":"96",
+//   "blocksReceived":"2",
+//   "dataReceived":"96",
+//   "dupBlksReceived":"0",
+//   "dupDataReceived":"0",
+//   "blocksSent":"0",
+//   "dataSent":"0",
+//   "providesBufferLength":"0",
+//   "wantListLength":"0",
+//   "peerCount":"1"
+// }
+```
+
+#### Moving average accessor:
+
+```js
+const movingAverages = stats.movingAverages
+```
+
+This object contains these properties:
+
+* 'blocksReceived',
+* 'dataReceived',
+* 'dupBlksReceived',
+* 'dupDataReceived',
+* 'blocksSent',
+* 'dataSent',
+* 'providesBufferLength',
+* 'wantListLength',
+* 'peerCount'
+
+```js
+const dataReceivedMovingAverages = movingAverages.dataReceived
+```
+
+Each one of these will contain one key per interval (miliseconds), being the default intervals defined:
+
+* 60000 (1 minute)
+* 300000 (5 minutes)
+* 900000 (15 minutes)
+
+You can then select one of them
+
+```js
+const oneMinuteDataReceivedMovingAverages = dataReceivedMovingAverages[60000]
+```
+
+This object will be a [movingAverage](https://github.com/pgte/moving-average#readme) instance.
+
 ## Development
 
 ### Structure

src/decision-engine/index.js

@@ -269,7 +269,7 @@ class DecisionEngine {
 
     this.ledgerMap.set(peerIdStr, l)
     if (this._stats) {
-      this._stats.push('peerCount', 1)
+      this._stats.push(peerIdStr, 'peerCount', 1)
     }
 
     return l

src/index.js

@@ -106,7 +106,7 @@ class Bitswap {
     waterfall([
       (cb) => this.blockstore.has(block.cid, cb),
       (has, cb) => {
-        this._updateReceiveCounters(block, has)
+        this._updateReceiveCounters(peerId.toB58String(), block, has)
         if (has) {
           return cb()
         }
@@ -116,13 +116,13 @@ class Bitswap {
     ], callback)
   }
 
-  _updateReceiveCounters (block, exists) {
-    this._stats.push('blocksReceived', 1)
-    this._stats.push('dataReceived', block.data.length)
+  _updateReceiveCounters (peerId, block, exists) {
+    this._stats.push(peerId, 'blocksReceived', 1)
+    this._stats.push(peerId, 'dataReceived', block.data.length)
 
     if (exists) {
-      this._stats.push('dupBlksReceived', 1)
-      this._stats.push('dupDataReceived', block.data.length)
+      this._stats.push(peerId, 'dupBlksReceived', 1)
+      this._stats.push(peerId, 'dupDataReceived', block.data.length)
     }
   }
 
@@ -140,6 +140,7 @@ class Bitswap {
   _onPeerDisconnected (peerId) {
     this.wm.disconnected(peerId)
     this.engine.peerDisconnected(peerId)
+    this._stats.disconnected(peerId)
   }
 
   _putBlock (block, callback) {
@@ -389,6 +390,7 @@ class Bitswap {
    * @returns {void}
    */
   stop (callback) {
+    this._stats.stop()
     series([
       (cb) => this.wm.stop(cb),
       (cb) => this.network.stop(cb),

src/network.js

@@ -150,7 +150,7 @@ class Network {
         }
       })
       callback()
-      this._updateSentStats(msg.blocks)
+      this._updateSentStats(peer, msg.blocks)
     })
   }
 
@@ -179,10 +179,11 @@ class Network {
     })
   }
 
-  _updateSentStats (blocks) {
+  _updateSentStats (peer, blocks) {
+    const peerId = peer.toB58String()
     if (this._stats) {
-      blocks.forEach((block) => this._stats.push('dataSent', block.data.length))
-      this._stats.push('blocksSent', blocks.size)
+      blocks.forEach((block) => this._stats.push(peerId, 'dataSent', block.data.length))
+      this._stats.push(peerId, 'blocksSent', blocks.size)
     }
   }
 }

src/stats/index.js

@@ -0,0 +1,99 @@
+'use strict'
+
+const EventEmitter = require('events')
+const Stat = require('./stat')
+
+const defaultOptions = {
+  movingAverageIntervals: [
+    60 * 1000, // 1 minute
+    5 * 60 * 1000, // 5 minutes
+    15 * 60 * 1000 // 15 minutes
+  ]
+}
+
+class Stats extends EventEmitter {
+  constructor (initialCounters, _options) {
+    super()
+
+    const options = Object.assign({}, defaultOptions, _options)
+
+    if (typeof options.computeThrottleTimeout !== 'number') {
+      throw new Error('need computeThrottleTimeout')
+    }
+
+    if (typeof options.computeThrottleMaxQueueSize !== 'number') {
+      throw new Error('need computeThrottleMaxQueueSize')
+    }
+
+    this._initialCounters = initialCounters
+    this._options = options
+    this._enabled = this._options.enabled
+
+    this._global = new Stat(initialCounters, options)
+    this._global.on('update', (stats) => this.emit('update', stats))
+
+    this._peers = new Map()
+  }
+
+  enable () {
+    this._enabled = true
+    this._options.enabled = true
+    this._global.enable()
+  }
+
+  disable () {
+    this._enabled = false
+    this._options.enabled = false
+    this._global.disable()
+  }
+
+  stop () {
+    this._enabled = false
+    this._global.stop()
+    for (let peerStat of this._peers) {
+      peerStat[1].stop()
+    }
+  }
+
+  get snapshot () {
+    return this._global.snapshot
+  }
+
+  get movingAverages () {
+    return this._global.movingAverages
+  }
+
+  forPeer (peerId) {
+    if (peerId.toB58String) {
+      peerId = peerId.toB58String()
+    }
+    return this._peers.get(peerId)
+  }
+
+  push (peer, counter, inc) {
+    if (this._enabled) {
+      this._global.push(counter, inc)
+
+      if (peer) {
+        let peerStats = this._peers.get(peer)
+        if (!peerStats) {
+          peerStats = new Stat(this._initialCounters, this._options)
+          this._peers.set(peer, peerStats)
+        }
+
+        peerStats.push(counter, inc)
+      }
+    }
+  }
+
+  disconnected (peer) {
+    const peerId = peer.toB58String()
+    const peerStats = this._peers.get(peerId)
+    if (peerStats) {
+      peerStats.stop()
+      this._peers.delete(peerId)
+    }
+  }
+}
+
+module.exports = Stats

src/stats.js → src/stats/stat.js

@@ -4,28 +4,10 @@ const EventEmitter = require('events')
 const Big = require('big.js')
 const MovingAverage = require('moving-average')
 
-const defaultOptions = {
-  movingAverageIntervals: [
-    60 * 1000, // 1 minute
-    5 * 60 * 1000, // 5 minutes
-    15 * 60 * 1000 // 15 minutes
-  ]
-}
-
 class Stats extends EventEmitter {
-  constructor (initialCounters, _options) {
+  constructor (initialCounters, options) {
     super()
 
-    const options = Object.assign({}, defaultOptions, _options)
-
-    if (typeof options.computeThrottleTimeout !== 'number') {
-      throw new Error('need computeThrottleTimeout')
-    }
-
-    if (typeof options.computeThrottleMaxQueueSize !== 'number') {
-      throw new Error('need computeThrottleMaxQueueSize')
-    }
-
     this._options = options
     this._queue = []
     this._stats = {}
@@ -56,6 +38,12 @@ class Stats extends EventEmitter {
     this._disabled = true
   }
 
+  stop () {
+    if (this._timeout) {
+      clearTimeout(this._timeout)
+    }
+  }
+
   get snapshot () {
     return Object.assign({}, this._stats)
   }

src/types/wantlist/index.js

@@ -23,7 +23,7 @@ class Wantlist {
     } else {
       this.set.set(cidStr, new Entry(cid, priority))
       if (this._stats) {
-        this._stats.push('wantListSize', 1)
+        this._stats.push(null, 'wantListSize', 1)
       }
     }
   }
@@ -45,7 +45,7 @@ class Wantlist {
 
     this.set.delete(cidStr)
     if (this._stats) {
-      this._stats.push('wantListSize', -1)
+      this._stats.push(null, 'wantListSize', -1)
     }
   }
 

test/bitswap-stats.js

@@ -240,5 +240,27 @@ describe('bitswap stats', () => {
         finish()
       })
     })
+
+    it('has peer stats', (done) => {
+      const peerIds = libp2pNodes.map((node) => node.peerInfo.id.toB58String())
+      const peerStats = bs2.stat().forPeer(peerIds[0])
+      peerStats.once('update', (stats) => {
+        expect(stats.blocksReceived.eq(1)).to.be.true()
+        expect(stats.dataReceived.eq(48)).to.be.true()
+        expect(stats.dupBlksReceived.eq(0)).to.be.true()
+        expect(stats.dupDataReceived.eq(0)).to.be.true()
+        expect(stats.blocksSent.eq(0)).to.be.true()
+        expect(stats.dataSent.eq(0)).to.be.true()
+        expect(stats.providesBufferLength.eq(0)).to.be.true()
+        expect(stats.wantListLength.eq(0)).to.be.true()
+        expect(stats.peerCount.eq(1)).to.be.true()
+
+        const ma = peerStats.movingAverages.dataReceived[60000]
+        expect(ma.movingAverage()).to.be.above(0)
+        expect(ma.variance()).to.be.above(0)
+
+        done()
+      })
+    })
   })
 })