refactor: stats (#501)

* docs: add initial notes on stats

* feat: initial refactor of stats to metrics

* feat: add support for placeholder metrics

This is helpful for tracking metrics prior to knowing the remote peers id

* fix: add metrics tests and fix issues

* fix: always clear the dial timeout timer

* docs: add metrics to api doc

* chore: apply suggestions from code review

Co-Authored-By: Vasco Santos <vasco.santos@moxy.studio>

* docs: update metrics docs

* fix: call metrics.onDisconnect

* docs(config): add example headers so they appear in the TOC

* docs(config): add metrics configuration

* docs(relay): fix relay configuration docs

Author: jacobheun

doc/API.md

@@ -21,6 +21,13 @@
   * [`pubsub.getTopics`](#pubsub.getTopics)
   * [`pubsub.publish`](#pubsub.publish)
   * [`pubsub.subscribe`](#pubsub.subscribe)
+  * [`metrics.global`](#metrics.global)
+  * [`metrics.peers`](#metrics.peers)
+  * [`metrics.protocols`](#metrics.protocols)
+  * [`metrics.forPeer`](#metrics.forPeer)
+  * [`metrics.forProtocol`](#metrics.forProtocol)
+* [Types](#types)
+  * [`Stats`](#stats)
 
 ## Static Functions
 
@@ -117,7 +124,7 @@ unless they are performing a specific action. See [peer discovery and auto dial]
 
 </details>
 
-## Libp2p Instance Methods 
+## Libp2p Instance Methods
 
 ### start
 
@@ -635,3 +642,103 @@ const handler = (msg) => {
 
 libp2p.pubsub.unsubscribe(topic, handler)
 ```
+
+### metrics.global
+
+A [`Stats`](#stats) object of tracking the global bandwidth of the libp2p node.
+
+#### Example
+
+```js
+const peerIdStrings = libp2p.metrics.peers
+```
+
+### metrics.peers
+
+An array of `PeerId` strings of each peer currently being tracked.
+
+#### Example
+
+```js
+const peerIdStrings = libp2p.metrics.peers
+```
+
+### metrics.protocols
+
+An array of protocol strings that are currently being tracked.
+
+#### Example
+
+```js
+const protocols = libp2p.metrics.protocols
+```
+
+### metrics.forPeer
+
+Returns the [`Stats`](#stats) object for a given `PeerId` if it is being tracked.
+
+`libp2p.metrics.forPeer(peerId)`
+
+#### Parameters
+
+| Name | Type | Description |
+|------|------|-------------|
+| peerId | `PeerId` | The peer to get stats for |
+
+#### Returns
+
+| Type | Description |
+|------|-------------|
+| [`Stats`](#stats) | The bandwidth stats of the peer |
+
+#### Example
+
+```js
+const peerStats = libp2p.metrics.forPeer(peerInfo)
+console.log(peerStats.toJSON())
+```
+
+### metrics.forProtocol
+
+Returns the [`Stats`](#stats) object for a given protocol if it is being tracked.
+
+`libp2p.metrics.forProtocol(protocol)`
+
+#### Parameters
+
+| Name | Type | Description |
+|------|------|-------------|
+| protocol | `string` | The protocol to get stats for |
+
+#### Returns
+
+| Type | Description |
+|------|-------------|
+| [`Stats`](#stats) | The bandwidth stats of the protocol across all peers |
+
+#### Example
+
+```js
+const peerStats = libp2p.metrics.forProtocol('/meshsub/1.0.0')
+console.log(peerStats.toJSON())
+```
+
+## Types
+
+### Stats
+
+- `Stats`
+  - `toJSON<function()>`: Returns a JSON snapshot of the stats.
+    - `dataReceived<string>`: The stringified value of total incoming data for this stat.
+    - `dataSent<string>`: The stringified value of total outgoing data for this stat.
+    - `movingAverages<object>`: The properties are dependent on the configuration of the moving averages interval. Defaults are listed here.
+      - `['60000']<Number>`: The calculated moving average at a 1 minute interval.
+      - `['300000']<Number>`: The calculated moving average at a 5 minute interval.
+      - `['900000']<Number>`: The calculated moving average at a 15 minute interval.
+  - `snapshot<object>`: A getter that returns a clone of the raw stats.
+    - `dataReceived<BigNumber>`: A [`BigNumber`](https://github.com/MikeMcl/bignumber.js/) of the amount of incoming data
+    - `dataSent<BigNumber>`: A [`BigNumber`](https://github.com/MikeMcl/bignumber.js/) of the amount of outgoing data
+  - `movingAverages<object>`: A getter that returns a clone of the raw [moving averages](https://www.npmjs.com/package/moving-averages) stats. **Note**: The properties of this are dependent on configuration. The defaults are shown here.
+    - `['60000']<MovingAverage>`: The [MovingAverage](https://www.npmjs.com/package/moving-averages) at a 1 minute interval.
+    - `['300000']<MovingAverage>`: The [MovingAverage](https://www.npmjs.com/package/moving-averages) at a 5 minute interval.
+    - `['900000']<MovingAverage>`: The [MovingAverage](https://www.npmjs.com/package/moving-averages) at a 15 minute interval.

doc/CONFIGURATION.md

@@ -1,18 +1,26 @@
 # Configuration
 
-* [Overview](#overview)
-* [Modules](#modules)
-  * [Transport](#transport)
-  * [Stream Multiplexing](#stream-multiplexing)
-  * [Connection Encryption](#connection-encryption)
-  * [Peer Discovery](#peer-discovery)
-  * [Content Routing](#content-routing)
-  * [Peer Routing](#peer-routing)
-  * [DHT](#dht)
-  * [Pubsub](#pubsub)
-* [Customizing Libp2p](#customizing-libp2p)
-  * [Examples](#examples)
-* [Configuration examples](#configuration-examples)
+- [Configuration](#configuration)
+  - [Overview](#overview)
+  - [Modules](#modules)
+    - [Transport](#transport)
+    - [Stream Multiplexing](#stream-multiplexing)
+    - [Connection Encryption](#connection-encryption)
+    - [Peer Discovery](#peer-discovery)
+    - [Content Routing](#content-routing)
+    - [Peer Routing](#peer-routing)
+    - [DHT](#dht)
+    - [Pubsub](#pubsub)
+  - [Customizing libp2p](#customizing-libp2p)
+    - [Examples](#examples)
+      - [Basic setup](#basic-setup)
+      - [Customizing Peer Discovery](#customizing-peer-discovery)
+      - [Customizing Pubsub](#customizing-pubsub)
+      - [Customizing DHT](#customizing-dht)
+      - [Setup with Content and Peer Routing](#setup-with-content-and-peer-routing)
+      - [Setup with Relay](#setup-with-relay)
+      - [Configuring Metrics](#configuring-metrics)
+  - [Configuration examples](#configuration-examples)
 
 ## Overview
 
@@ -45,7 +53,7 @@ Bear in mind that only a **transport** and **connection encryption** are require
 
 Some available transports are:
 
-- [libp2p/js-libp2p-tcp](https://github.com/libp2p/js-libp2p-tcp) 
+- [libp2p/js-libp2p-tcp](https://github.com/libp2p/js-libp2p-tcp)
 - [libp2p/js-libp2p-webrtc-star](https://github.com/libp2p/js-libp2p-webrtc-star)
 - [libp2p/js-libp2p-webrtc-direct](https://github.com/libp2p/js-libp2p-webrtc-direct)
 - [libp2p/js-libp2p-websockets](https://github.com/libp2p/js-libp2p-websockets)
@@ -151,7 +159,7 @@ If you want to know more about libp2p DHT, you should read the following content
 - https://docs.libp2p.io/concepts/protocols/#kad-dht
 - https://github.com/libp2p/specs/pull/108
 
-#### Pubsub
+### Pubsub
 
 > Publish/Subscribe is a system where peers congregate around topics they are interested in. Peers interested in a topic are said to be subscribed to that topic and should receive the data published on it from other peers.
 
@@ -194,7 +202,7 @@ Besides the `modules` and `config`, libp2p allows other internal options and con
 
 ### Examples
 
-**1) Basic setup**
+#### Basic setup
 
 ```js
 // Creating a libp2p node with:
@@ -229,7 +237,7 @@ const node = await Libp2p.create({
 })
 ```
 
-**2) Customizing Peer Discovery**
+#### Customizing Peer Discovery
 
 ```js
 const Libp2p = require('libp2p')
@@ -248,7 +256,7 @@ const node = await Libp2p.create({
   config: {
     peerDiscovery: {
       autoDial: true,             // Auto connect to discovered peers (limited by ConnectionManager minPeers)
-      // The `tag` property will be searched when creating the instance of your Peer Discovery service. 
+      // The `tag` property will be searched when creating the instance of your Peer Discovery service.
       // The associated object, will be passed to the service when it is instantiated.
       [MulticastDNS.tag]: {
         interval: 1000,
@@ -260,7 +268,7 @@ const node = await Libp2p.create({
 })
 ```
 
-**3) Customizing Pubsub**
+#### Customizing Pubsub
 
 ```js
 const Libp2p = require('libp2p')
@@ -287,7 +295,7 @@ const node = await Libp2p.create({
 })
 ```
 
-**4) Customizing DHT**
+#### Customizing DHT
 
 ```js
 const Libp2p = require('libp2p')
@@ -317,7 +325,7 @@ const node = await Libp2p.create({
 })
 ```
 
-**5) Setup with Content and Peer Routing**
+#### Setup with Content and Peer Routing
 
 ```js
 const Libp2p = require('libp2p')
@@ -347,7 +355,7 @@ const node = await Libp2p.create({
 })
 ```
 
-**6) Setup with Relay**
+#### Setup with Relay
 
 ```js
 const Libp2p = require('libp2p')
@@ -361,13 +369,45 @@ const node = await Libp2p.create({
     streamMuxer: [MPLEX],
     connEncryption: [SECIO]
   },
-  relay: {                   // Circuit Relay options (this config is part of libp2p core configurations)
-    enabled: true,
-    hop: {
-      enabled: true,
-      active: true
+  config: {
+    relay: {                   // Circuit Relay options (this config is part of libp2p core configurations)
+      enabled: true,           // Allows you to dial and accept relayed connections. Does not make you a relay.
+      hop: {
+        enabled: true,         // Allows you to be a relay for other peers
+        active: true           // You will attempt to dial destination peers if you are not connected to them
+      }
     }
+  }
+})
+```
+
+#### Configuring Metrics
+
+Metrics are disabled in libp2p by default. You can enable and configure them as follows. Aside from enabled being `false` by default, the configuration options listed here are the current defaults.
+
+```js
+const Libp2p = require('libp2p')
+const TCP = require('libp2p-tcp')
+const MPLEX = require('libp2p-mplex')
+const SECIO = require('libp2p-secio')
+
+const node = await Libp2p.create({
+  modules: {
+    transport: [TCP],
+    streamMuxer: [MPLEX],
+    connEncryption: [SECIO]
   },
+  metrics: {
+    enabled: true,
+    computeThrottleMaxQueueSize: 1000,  // How many messages a stat will queue before processing
+    computeThrottleTimeout: 2000,       // Time in milliseconds a stat will wait, after the last item was added, before processing
+    movingAverageIntervals: [           // The moving averages that will be computed
+      60 * 1000, // 1 minute
+      5 * 60 * 1000, // 5 minutes
+      15 * 60 * 1000 // 15 minutes
+    ],
+    maxOldPeersRetention: 50            // How many disconnected peers we will retain stats for
+  }
 })
 ```
 

doc/METRICS.md

@@ -0,0 +1,28 @@
+# Bandwidth Metrics
+
+- Metrics gathering is optional, as there is a performance hit to using it.
+- Metrics do NOT currently contain OS level stats, only libp2p application level Metrics. For example, TCP messages (ACK, FIN, etc) are not accounted for.
+- See the [API](./API.md) for Metrics usage. Metrics in libp2p do not emit events, as such applications wishing to read Metrics will need to do so actively. This ensures that the system is not unnecessarily firing update notifications.
+
+## Tracking
+- When a transport hands off a connection for upgrading, Metrics are hooked up if enabled.
+- When a stream is created, Metrics will be tracked on that stream and associated to that streams protocol.
+- Tracked Metrics are associated to a specific peer, and count towards global bandwidth Metrics.
+
+### Metrics Processing
+- The main Metrics object consists of individual `Stats` objects
+- The following categories are tracked:
+  - Global stats; every byte in and out
+  - Peer stats; every byte in and out, per peer
+  - Protocol stats; every byte in and out, per protocol
+- When a message goes through Metrics:
+  - It is added to the global stat
+  - It is added to the stats for the remote peer
+  - It is added to the protocol stats if there is one
+- When data is pushed onto a `Stat` it is added to a queue
+  - The queue is processed at the earliest of either (configurable):
+    - every 2 seconds after the last item was added to the queue
+    - or once 1000 items have been queued
+  - When the queue is processed:
+    - The data length is added to either the `in` or `out` stat
+    - The moving averages is calculated since the last queue processing (based on most recently processed item timestamp)

package.json

@@ -63,6 +63,7 @@
     "moving-average": "^1.0.0",
     "multiaddr": "^7.2.1",
     "multistream-select": "^0.15.0",
+    "mutable-proxy": "^1.0.0",
     "p-any": "^2.1.0",
     "p-fifo": "^1.0.0",
     "p-settle": "^3.1.0",
@@ -82,7 +83,9 @@
     "cids": "^0.7.1",
     "delay": "^4.3.0",
     "dirty-chai": "^2.0.1",
+    "it-concat": "^1.0.0",
     "it-pair": "^1.0.0",
+    "it-pushable": "^1.4.0",
     "libp2p-bootstrap": "^0.10.3",
     "libp2p-delegated-content-routing": "^0.4.1",
     "libp2p-delegated-peer-routing": "^0.4.0",

src/config.js

@@ -6,6 +6,9 @@ const DefaultConfig = {
   connectionManager: {
     minPeers: 25
   },
+  metrics: {
+    enabled: false
+  },
   config: {
     dht: {
       enabled: false,

src/constants.js

@@ -9,5 +9,15 @@ module.exports = {
   MAX_PER_PEER_DIALS: 4, // Allowed parallel dials per DialRequest
   QUARTER_HOUR: 15 * 60e3,
   PRIORITY_HIGH: 10,
-  PRIORITY_LOW: 20
+  PRIORITY_LOW: 20,
+  METRICS: {
+    computeThrottleMaxQueueSize: 1000,
+    computeThrottleTimeout: 2000,
+    movingAverageIntervals: [
+      60 * 1000, // 1 minute
+      5 * 60 * 1000, // 5 minutes
+      15 * 60 * 1000 // 15 minutes
+    ],
+    maxOldPeersRetention: 50
+  }
 }

src/dialer/index.js

@@ -91,7 +91,6 @@ class Dialer {
 
     try {
       const dialResult = await dialRequest.run({ ...options, signal })
-      timeoutController.clear()
       log('dial succeeded to %s', dialResult.remoteAddr)
       return dialResult
     } catch (err) {
@@ -102,6 +101,7 @@ class Dialer {
       log.error(err)
       throw err
     } finally {
+      timeoutController.clear()
       this._pendingDials.delete(dial)
     }
   }