observe traffic and expose statistics

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,10 @@ module.exports = function connection (swarm) {
                 })
                 cb(null, peerInfo)
               }
-            ], cb)
+            ], (err, pi) => {
+              setPeerInfo.call(conn, pi)
+              cb(err, pi)
+            })
           }
 
           conn.getPeerInfo((err, peerInfo) => {
@@ -120,7 +121,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
+      }
+    })
+  }
 }

src/stats/index.js

@@ -0,0 +1,92 @@
+'use strict'
+
+const Stat = require('./stat')
+
+const defaultOptions = {
+  enabled: true,
+  computeThrottleMaxQueueSize: 1000,
+  computeThrottleTimeout: 2000,
+  movingAverageIntervals: [
+    60 * 1000, // 1 minute
+    5 * 60 * 1000, // 5 minutes
+    15 * 60 * 1000 // 15 minutes
+  ]
+}
+
+const initialCounters = [
+  'dataReceived',
+  'dataSent'
+]
+
+const directionToEvent = {
+  in: 'dataReceived',
+  out: 'dataSent'
+}
+
+module.exports = (observer, _options) => {
+  const options = Object.assign({}, defaultOptions, _options)
+  const globalStats = new Stat(initialCounters, options)
+  const peerStats = new Map()
+  const transportStats = new Map()
+  const protocolStats = new Map()
+
+  observer.on('message', (peerId, transportTag, protocolTag, direction, bufferLength) => {
+    // if (!peerId) {
+    //   throw new Error('message should have peer id')
+    // }
+    console.log('m', peerId, transportTag, protocolTag, direction, bufferLength)
+    const event = directionToEvent[direction]
+
+    // global stats
+    globalStats.push(event, bufferLength)
+
+    // peer stats
+    let peer = peerStats.get(peerId)
+    if (!peer) {
+      peer = new Stat(initialCounters, options)
+      peerStats.set(peerId, peer)
+    }
+    peer.push(event, bufferLength)
+
+    // transport stats
+    let transport = transportStats.get(transportTag)
+    if (!transport) {
+      transport = new Stat(initialCounters, options)
+      transportStats.set(transportTag, transport)
+    }
+    transport.push(event, bufferLength)
+
+    // protocol stats
+    let protocol = protocolStats.get(protocolTag)
+    if (!protocol) {
+      protocol = new Stat(initialCounters, options)
+      protocolStats.set(protocolTag, transport)
+    }
+    protocol.push(event, bufferLength)
+  })
+
+  observer.on('peer:closed', (peerId) => {
+    peerStats.delete(peerId)
+  })
+
+  return {
+    stop: stop,
+    global: globalStats,
+    peers: () => Array.from(peerStats.keys()),
+    forPeer: (peerId) => peerStats.get(peerId),
+    transports: () => Array.from(transportStats.keys()),
+    forTransport: (transport) => transportStats.get(transport),
+    protocols: () => Array.from(protocolStats.keys()),
+    forProtocol: (protocol) => protocolStats.get(protocol)
+  }
+
+  function stop () {
+    globalStats.stop()
+    for (let peerStat of peerStats.values()) {
+      peerStat.stop()
+    }
+    for (let transportStat of transportStats.values()) {
+      transportStat.stop()
+    }
+  }
+}