observe traffic and expose statistics

README.md

@@ -30,6 +30,7 @@ libp2p-switch is used by [libp2p](https://github.com/libp2p/js-libp2p) but it ca
   - [`switch.stop(callback)`](#swarmclosecallback)
   - [`switch.connection`](#connection)
   - [Internal Transports API](#transports)
+  - [`switch.stats`](#stats-api)
 - [Design Notes](#designnotes)
   - [Multitransport](#multitransport)
   - [Connection upgrades](#connection-upgrades)
@@ -51,7 +52,7 @@ libp2p-switch is used by [libp2p](https://github.com/libp2p/js-libp2p) but it ca
 ```JavaScript
 const switch = require('libp2p-switch')
 
-const sw = new switch(peerInfo [, peerBook])
+const sw = new switch(peerInfo [, peerBook [, options]])
 ```
 
 ## API
@@ -223,6 +224,106 @@ To avoid the confusion between connection, stream, transport, and other names th
 - connection - something that implements the transversal expectations of a stream between two peers, including the benefits of using a stream plus having a way to do half duplex, full duplex
 - transport - something that as a dial/listen interface and return objs that implement a connection interface
 
+### Stats API
+
+#### Global stats
+
+##### `switch.stats.global.snapshot`
+
+Should return a stats snapshot, which is an object containing the following keys and respective values:
+
+* dataSent: amount of bytes sent, [Big](https://github.com/MikeMcl/big.js#readme) number
+* dataReceived: amount of bytes received, [Big](https://github.com/MikeMcl/big.js#readme) number
+
+##### `switch.stats.global.movingAverages`
+
+Returns an object containing the following keys:
+
+* dataSent
+* dataReceived
+
+Each one of them contains an object that has a key for each interval (`60000`, `300000` and `900000` miliseconds).
+
+Each one of these values is [an exponential moving-average instance](https://github.com/pgte/moving-average#readme).
+
+#### Per-transport stats
+
+##### `switch.stats.transports()`
+
+Returns an array containing the tags (string) for each observed transport.
+
+##### `switch.stats.forTransports(transportTag).snapshot`
+
+Should return a stats snapshot, which is an object containing the following keys and respective values:
+
+* dataSent: amount of bytes sent, [Big](https://github.com/MikeMcl/big.js#readme) number
+* dataReceived: amount of bytes received, [Big](https://github.com/MikeMcl/big.js#readme) number
+
+##### `switch.stats.forTransports(transportTag).movingAverages`
+
+Returns an object containing the following keys:
+
+* dataSent
+* dataReceived
+
+Each one of them contains an object that has a key for each interval (`60000`, `300000` and `900000` miliseconds).
+
+Each one of these values is [an exponential moving-average instance](https://github.com/pgte/moving-average#readme).
+
+#### Per-protocol stats
+
+##### `switch.stats.protocols()`
+
+Returns an array containing the tags (string) for each observed protocol.
+
+##### `switch.stats.forProtocol(protocolTag).snapshot`
+
+Should return a stats snapshot, which is an object containing the following keys and respective values:
+
+* dataSent: amount of bytes sent, [Big](https://github.com/MikeMcl/big.js#readme) number
+* dataReceived: amount of bytes received, [Big](https://github.com/MikeMcl/big.js#readme) number
+
+
+##### `switch.stats.forProtocol(protocolTag).movingAverages`
+
+Returns an object containing the following keys:
+
+* dataSent
+* dataReceived
+
+Each one of them contains an object that has a key for each interval (`60000`, `300000` and `900000` miliseconds).
+
+Each one of these values is [an exponential moving-average instance](https://github.com/pgte/moving-average#readme).
+
+#### Per-peer stats
+
+##### `switch.stats.peers()`
+
+Returns an array containing the peerIDs (B58-encoded string) for each observed peer.
+
+##### `switch.stats.forPeer(peerId:String).snapshot`
+
+Should return a stats snapshot, which is an object containing the following keys and respective values:
+
+* dataSent: amount of bytes sent, [Big](https://github.com/MikeMcl/big.js#readme) number
+* dataReceived: amount of bytes received, [Big](https://github.com/MikeMcl/big.js#readme) number
+
+
+##### `switch.stats.forPeer(peerId:String).movingAverages`
+
+Returns an object containing the following keys:
+
+* dataSent
+* dataReceived
+
+Each one of them contains an object that has a key for each interval (`60000`, `300000` and `900000` miliseconds).
+
+Each one of these values is [an exponential moving-average instance](https://github.com/pgte/moving-average#readme).
+
+#### Stats update interval
+
+Stats are not updated in real-time. Instead, measurements are buffered and stats are updated at an interval. The maximum interval can be defined through the `Switch` constructor option `stats.computeThrottleTimeout`, defined in miliseconds.
+
 ### This module uses `pull-streams`
 
 We expose a streaming interface based on `pull-streams`, rather then on the Node.js core streams implementation (aka Node.js streams). `pull-streams` offers us a better mechanism for error handling and flow control guarantees. If you would like to know more about why we did this, see the discussion at this [issue](https://github.com/ipfs/js-ipfs/issues/362).

package.json

@@ -56,12 +56,14 @@
   },
   "dependencies": {
     "async": "^2.6.0",
+    "big.js": "^5.0.3",
     "debug": "^3.1.0",
     "interface-connection": "~0.3.2",
     "ip-address": "^5.8.9",
     "libp2p-circuit": "~0.1.4",
     "libp2p-identify": "~0.6.3",
     "lodash.includes": "^4.3.0",
+    "moving-average": "^1.0.0",
     "multiaddr": "^3.0.2",
     "multistream-select": "~0.14.1",
     "once": "^1.4.0",

src/connection.js

@@ -10,7 +10,6 @@ const setImmediate = require('async/setImmediate')
 
 const Circuit = require('libp2p-circuit')
 
-const protocolMuxer = require('./protocol-muxer')
 const plaintext = require('./plaintext')
 
 module.exports = function connection (swarm) {
@@ -25,16 +24,15 @@ module.exports = function connection (swarm) {
       swarm.handle(muxer.multicodec, (protocol, conn) => {
         const muxedConn = muxer.listener(conn)
 
-        muxedConn.on('stream', (conn) => {
-          protocolMuxer(swarm.protocols, conn)
-        })
+        muxedConn.on('stream', swarm.protocolMuxer(protocol))
 
         // If identify is enabled
         //   1. overload getPeerInfo
         //   2. call getPeerInfo
         //   3. add this conn to the pool
         if (swarm.identify) {
           // overload peerInfo to use Identify instead
+          const setPeerInfo = conn.setPeerInfo
           conn.getPeerInfo = (cb) => {
             const conn = muxedConn.newStream()
             const ms = new multistream.Dialer()
@@ -50,7 +48,12 @@ module.exports = function connection (swarm) {
                 })
                 cb(null, peerInfo)
               }
-            ], cb)
+            ], (err, pi) => {
+              if (pi) {
+                setPeerInfo.call(conn, pi)
+              }
+              cb(err, pi)
+            })
           }
 
           conn.getPeerInfo((err, peerInfo) => {
@@ -120,7 +123,7 @@ module.exports = function connection (swarm) {
       swarm.handle(tag, (protocol, conn) => {
         const myId = swarm._peerInfo.id
         const secure = encrypt(myId, conn, undefined, () => {
-          protocolMuxer(swarm.protocols, secure)
+          swarm.protocolMuxer(tag)(secure)
         })
       })
 

src/dial.js

@@ -9,8 +9,6 @@ const Circuit = require('libp2p-circuit')
 const debug = require('debug')
 const log = debug('libp2p:swarm:dial')
 
-const protocolMuxer = require('./protocol-muxer')
-
 function dial (swarm) {
   return (peer, protocol, callback) => {
     if (typeof protocol === 'function') {
@@ -22,6 +20,7 @@ function dial (swarm) {
     const pi = getPeerInfo(peer, swarm._peerBook)
 
     const proxyConn = new Connection()
+    proxyConn.setPeerInfo(pi)
 
     const b58Id = pi.id.toB58String()
     log('dialing %s', b58Id)
@@ -133,6 +132,7 @@ function dial (swarm) {
                   if (err) {
                     return cb(err)
                   }
+                  wrapped.setPeerInfo(pi)
                   cb(null, wrapped)
                 })
               })
@@ -189,7 +189,8 @@ function dial (swarm) {
 
           // For incoming streams, in case identify is on
           muxedConn.on('stream', (conn) => {
-            protocolMuxer(swarm.protocols, conn)
+            conn.setPeerInfo(pi)
+            swarm.protocolMuxer(key)(conn)
           })
 
           setImmediate(() => swarm.emit('peer-mux-established', pi))
@@ -213,6 +214,7 @@ function dial (swarm) {
           if (err) {
             return cb(err)
           }
+          proxyConn.setPeerInfo(pi)
           proxyConn.setInnerConn(conn)
           cb(null, proxyConn)
         })

src/index.js

@@ -7,18 +7,21 @@ const transport = require('./transport')
 const connection = require('./connection')
 const getPeerInfo = require('./get-peer-info')
 const dial = require('./dial')
-const protocolMuxer = require('./protocol-muxer')
+const ProtocolMuxer = require('./protocol-muxer')
 const plaintext = require('./plaintext')
+const Observer = require('./observer')
+const Stats = require('./stats')
 const assert = require('assert')
 
 class Switch extends EE {
-  constructor (peerInfo, peerBook) {
+  constructor (peerInfo, peerBook, options) {
     super()
     assert(peerInfo, 'You must provide a `peerInfo`')
     assert(peerBook, 'You must provide a `peerBook`')
 
     this._peerInfo = peerInfo
     this._peerBook = peerBook
+    this._options = options || {}
 
     this.setMaxListeners(Infinity)
     // transports --
@@ -69,10 +72,14 @@ class Switch extends EE {
         })
     }
 
+    const observer = Observer(this)
+    this.stats = Stats(observer, this._options.stats)
+    this.protocolMuxer = ProtocolMuxer(this.protocols, observer)
+
     this.handle(this.crypto.tag, (protocol, conn) => {
       const peerId = this._peerInfo.id
       const wrapped = this.crypto.encrypt(peerId, conn, undefined, () => {})
-      return protocolMuxer(this.protocols, wrapped)
+      return this.protocolMuxer(this.crypto.tag)(wrapped)
     })
 
     // higher level (public) API
@@ -88,6 +95,7 @@ class Switch extends EE {
   }
 
   stop (callback) {
+    this.stats.stop()
     series([
       (cb) => each(this.muxedConns, (conn, cb) => {
         conn.muxer.end((err) => {

src/observe-connection.js

@@ -0,0 +1,31 @@
+'use strict'
+
+const Connection = require('interface-connection').Connection
+const pull = require('pull-stream')
+
+module.exports = (transport, protocol, _conn, observer) => {
+  const peerInfo = new Promise((resolve, reject) => {
+    _conn.getPeerInfo((err, peerInfo) => {
+      if (!err && peerInfo) {
+        resolve(peerInfo)
+        return
+      }
+
+      const setPeerInfo = _conn.setPeerInfo
+      _conn.setPeerInfo = (pi) => {
+        setPeerInfo.call(_conn, pi)
+        resolve(pi)
+      }
+    })
+  })
+
+  const stream = {
+    source: pull(
+      _conn,
+      observer.incoming(transport, protocol, peerInfo)),
+    sink: pull(
+      observer.outgoing(transport, protocol, peerInfo),
+      _conn)
+  }
+  return new Connection(stream, _conn)
+}

src/observer.js

@@ -0,0 +1,37 @@
+'use strict'
+
+const pull = require('pull-stream')
+const EventEmitter = require('events')
+
+module.exports = (swarm) => {
+  const observer = Object.assign(new EventEmitter(), {
+    incoming: observe('in'),
+    outgoing: observe('out')
+  })
+
+  swarm.on('peer-mux-established', (peerInfo) => {
+    observer.emit('peer:connected', peerInfo.id.toB58String())
+  })
+
+  swarm.on('peer-mux-closed', (peerInfo) => {
+    observer.emit('peer:closed', peerInfo.id.toB58String())
+  })
+
+  return observer
+
+  function observe (direction) {
+    return (transport, protocol, peerInfo) => {
+      return pull.map((buffer) => {
+        willObserve(peerInfo, transport, protocol, direction, buffer.length)
+        return buffer
+      })
+    }
+  }
+
+  function willObserve (peerInfo, transport, protocol, direction, bufferLength) {
+    peerInfo.then((pi) => {
+      const peerId = pi.id.toB58String()
+      setImmediate(() => observer.emit('message', peerId, transport, protocol, direction, bufferLength))
+    })
+  }
+}

src/protocol-muxer.js

@@ -1,21 +1,29 @@
 'use strict'
 
 const multistream = require('multistream-select')
+const observeConn = require('./observe-connection')
 
-module.exports = function protocolMuxer (protocols, conn) {
-  const ms = new multistream.Listener()
+module.exports = function protocolMuxer (protocols, observer) {
+  return (transport) => (parentConn) => {
+    const ms = new multistream.Listener()
 
-  Object.keys(protocols).forEach((protocol) => {
-    if (!protocol) {
-      return
-    }
+    Object.keys(protocols).forEach((protocol) => {
+      if (!protocol) {
+        return
+      }
 
-    ms.addHandler(protocol, protocols[protocol].handlerFunc, protocols[protocol].matchFunc)
-  })
+      const handler = (protocol, _conn) => {
+        const conn = observeConn(transport, protocol, _conn, observer)
+        protocols[protocol].handlerFunc.call(null, protocol, conn)
+      }
 
-  ms.handle(conn, (err) => {
-    if (err) {
-      // the multistream handshake failed
-    }
-  })
+      ms.addHandler(protocol, handler, protocols[protocol].matchFunc)
+    })
+
+    ms.handle(parentConn, (err) => {
+      if (err) {
+        // the multistream handshake failed
+      }
+    })
+  }
 }