feat: Add metrics for common events

package.json

@@ -137,6 +137,7 @@
   "dependencies": {
     "@chainsafe/libp2p-noise": "^10.0.0",
     "@libp2p/interface-connection": "^3.0.2",
+    "@libp2p/interface-metrics": "^4.0.4",
     "@libp2p/interface-peer-id": "^1.0.5",
     "@libp2p/interface-stream-muxer": "^3.0.0",
     "@libp2p/interface-transport": "^2.0.0",

src/maconn.ts

@@ -1,4 +1,5 @@
 import type { MultiaddrConnection, MultiaddrConnectionTimeline } from '@libp2p/interface-connection'
+import type { CounterGroup } from '@libp2p/interface-metrics'
 import { logger } from '@libp2p/logger'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import type { Source, Sink } from 'it-stream-types'
@@ -22,6 +23,11 @@ interface WebRTCMultiaddrConnectionInit {
    * Holds the relevant events timestamps of the connection
    */
   timeline: MultiaddrConnectionTimeline
+
+  /**
+   * Optional metrics counter group for this connection
+   */
+  metrics: CounterGroup | null
 }
 
 export class WebRTCMultiaddrConnection implements MultiaddrConnection {
@@ -40,6 +46,11 @@ export class WebRTCMultiaddrConnection implements MultiaddrConnection {
    */
   timeline: MultiaddrConnectionTimeline;
 
+  /**
+   * Optional metrics counter group for this connection
+   */
+  metrics?: CounterGroup
+
   /**
    * The stream source, a no-op as the transport natively supports multiplexing
    */
@@ -62,6 +73,7 @@ export class WebRTCMultiaddrConnection implements MultiaddrConnection {
     }
 
     this.timeline.close = new Date().getTime()
+    this.metrics?.increment({ close: true })
     this.peerConnection.close()
   }
 }

src/muxer.ts

@@ -1,27 +1,42 @@
 import type { Stream } from '@libp2p/interface-connection'
+import type { CounterGroup } from '@libp2p/interface-metrics'
 import type { StreamMuxer, StreamMuxerFactory, StreamMuxerInit } from '@libp2p/interface-stream-muxer'
 import type { Source, Sink } from 'it-stream-types'
 
 import { WebRTCStream } from './stream.js'
 import { nopSink, nopSource } from './util.js'
 
+export interface DataChannelMuxerFactoryInit {
+  peerConnection: RTCPeerConnection
+  metrics?: CounterGroup
+}
+
 export class DataChannelMuxerFactory implements StreamMuxerFactory {
   /**
    * WebRTC Peer Connection
    */
   private readonly peerConnection: RTCPeerConnection
 
+  /**
+   * Optional metrics counter group for all incoming/outgoing mux events.
+   */
+  private readonly metrics?: CounterGroup
+
   /**
    * The string representation of the protocol, required by `StreamMuxerFactory`
    */
   protocol: string = '/webrtc'
 
-  constructor (peerConnection: RTCPeerConnection) {
+  constructor (init: DataChannelMuxerFactoryInit) {
+    const { metrics, peerConnection } = init
     this.peerConnection = peerConnection
+    if (metrics != null) {
+      this.metrics = metrics
+    }
   }
 
   createStreamMuxer (init?: StreamMuxerInit | undefined): StreamMuxer {
-    return new DataChannelMuxer(this.peerConnection, init)
+    return new DataChannelMuxer(this.peerConnection, this.metrics, init)
   }
 }
 
@@ -33,6 +48,12 @@ export class DataChannelMuxer implements StreamMuxer {
    * WebRTC Peer Connection
    */
   private readonly peerConnection: RTCPeerConnection
+
+  /**
+   * Optional metrics for this data channel muxer
+   */
+  private readonly metrics?: CounterGroup
+
   /**
    * The protocol as represented in the multiaddress
    */
@@ -63,11 +84,12 @@ export class DataChannelMuxer implements StreamMuxer {
    */
   sink: Sink<Uint8Array, Promise<void>> = nopSink;
 
-  constructor (peerConnection: RTCPeerConnection, init?: StreamMuxerInit) {
+  constructor (peerConnection: RTCPeerConnection, metrics: CounterGroup | undefined, init?: StreamMuxerInit) {
     /**
      * Initialized stream muxer
      */
     this.init = init
+    this.metrics = metrics
 
     /**
      * WebRTC Peer Connection
@@ -93,6 +115,7 @@ export class DataChannelMuxer implements StreamMuxer {
       })
 
       if ((init?.onIncomingStream) != null) {
+        this.metrics?.increment({ incoming_stream: true })
         init.onIncomingStream(stream)
       }
     }
@@ -104,6 +127,10 @@ export class DataChannelMuxer implements StreamMuxer {
    */
   newStream (name: string = ''): Stream {
     const channel = this.peerConnection.createDataChannel(name)
+    const closeCb = (stream: WebRTCStream) => {
+      this.metrics?.increment({ stream_end: true })
+      this.init?.onStreamEnd?.(stream)
+    }
     const stream = new WebRTCStream({
       channel,
       stat: {
@@ -112,7 +139,7 @@ export class DataChannelMuxer implements StreamMuxer {
           open: 0
         }
       },
-      closeCb: this.init?.onStreamEnd
+      closeCb
     })
 
     return stream

src/transport.ts

@@ -17,6 +17,7 @@ import { DataChannelMuxerFactory } from './muxer.js'
 import type { WebRTCDialOptions } from './options.js'
 import * as sdp from './sdp.js'
 import { WebRTCStream } from './stream.js'
+import type { CounterGroup, Metrics } from '@libp2p/interface-metrics'
 
 const log = logger('libp2p:webrtc:transport')
 
@@ -42,19 +43,33 @@ export const CERTHASH_CODE: number = 466
 /**
  * The peer for this transport
  */
-// @TODO(ddimaria): seems like an unnessary abstraction, consider removing
 export interface WebRTCTransportComponents {
   peerId: PeerId
+  metrics?: Metrics
+}
+
+export interface WebRTCMetrics {
+  dialerEvents: CounterGroup
 }
 
 export class WebRTCTransport implements Transport {
   /**
    * The peer for this transport
    */
   private readonly components: WebRTCTransportComponents
+  private readonly metrics?: WebRTCMetrics
 
   constructor (components: WebRTCTransportComponents) {
     this.components = components
+
+    if (components.metrics != null) {
+      this.metrics = {
+        dialerEvents: components.metrics.registerCounterGroup('libp2p_webrtc_dialer_events_total', {
+          label: 'event',
+          help: 'Total count of WebRTC dial events by type'
+        })
+      }
+    }
   }
 
   /**
@@ -125,6 +140,7 @@ export class WebRTCTransport implements Transport {
     const handhsakeTimeout = setTimeout(() => {
       const error = `Data channel was never opened: state: ${handshakeDataChannel.readyState}`
       log.error(error)
+      this.metrics?.dialerEvents.increment({ openError: true })
       dataChannelOpenPromise.reject(dataChannelError('data', error))
     }, HANDSHAKE_TIMEOUT_MS)
 
@@ -139,6 +155,8 @@ export class WebRTCTransport implements Transport {
       const errorTarget = event.target?.toString() ?? 'not specified'
       const error = `Error opening a data channel for handshaking: ${errorTarget}`
       log.error(error)
+      // NOTE: We use unknown error here but this could potentially be considered a reset by some standards.
+      this.metrics?.dialerEvents.increment({ unknownError: true })
       dataChannelOpenPromise.reject(dataChannelError('data', error))
     }
 
@@ -190,10 +208,16 @@ export class WebRTCTransport implements Transport {
       remoteAddr: ma,
       timeline: {
         open: (new Date()).getTime()
-      }
+      },
+      // Pass down dialer metrics to collect close and other events.
+      metrics: (this.metrics != null) ? this.metrics.dialerEvents : null
     })
 
-    const muxerFactory = new DataChannelMuxerFactory(peerConnection)
+    const muxerFactory = new DataChannelMuxerFactory({
+      peerConnection,
+      // we mass metrics here to count stream open & close.
+      metrics: this.metrics?.dialerEvents
+    })
 
     // For outbound connections, the remote is expected to start the noise handshake.
     // Therefore, we need to secure an inbound noise connection from the remote.

test/maconn.browser.spec.ts

@@ -11,6 +11,7 @@ describe('Multiaddr Connection', () => {
     const remoteAddr = multiaddr('/ip4/1.2.3.4/udp/1234/webrtc/certhash/uEiAUqV7kzvM1wI5DYDc1RbcekYVmXli_Qprlw3IkiEg6tQ')
     const maConn = new WebRTCMultiaddrConnection({
       peerConnection: peerConnection,
+      metrics: null,
       remoteAddr,
       timeline: {
         open: (new Date()).getTime()