feat: add WebSockets metrics

packages/transport-websockets/src/index.ts

@@ -67,7 +67,7 @@
 import * as filters from './filters.js'
 import { createListener } from './listener.js'
 import { socketToMaConn } from './socket-to-conn.js'
-import type { Transport, MultiaddrFilter, CreateListenerOptions, DialTransportOptions, Listener, AbortOptions, ComponentLogger, Logger, Connection, OutboundConnectionUpgradeEvents } from '@libp2p/interface'
+import type { Transport, MultiaddrFilter, CreateListenerOptions, DialTransportOptions, Listener, AbortOptions, ComponentLogger, Logger, Connection, OutboundConnectionUpgradeEvents, Metrics, CounterGroup } from '@libp2p/interface'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import type { Server } from 'http'
 import type { DuplexWebSocket } from 'it-ws/duplex'
@@ -82,6 +82,13 @@
 
 export interface WebSocketsComponents {
   logger: ComponentLogger
+  metrics?: Metrics
+}
+
+export interface WebSocketsMetrics {
+  dialerEvents: CounterGroup
+  listenerEvents: CounterGroup
+  component: Metrics
 }
 
 export type WebSocketsDialEvents =
@@ -92,11 +99,26 @@
   private readonly log: Logger
   private readonly init?: WebSocketsInit
   private readonly logger: ComponentLogger
+  private readonly metrics?: WebSocketsMetrics
 
   constructor (components: WebSocketsComponents, init?: WebSocketsInit) {
     this.log = components.logger.forComponent('libp2p:websockets')
     this.logger = components.logger
     this.init = init
+
+    if (components.metrics != null) {
+      this.metrics = {
+        component: components.metrics,
+        dialerEvents: components.metrics.registerCounterGroup('libp2p_websockets_dialer_events_total', {
+          label: 'event',
+          help: 'Total count of WebSockets dialer events by type'
+        }),
+        listenerEvents: components.metrics.registerCounterGroup('libp2p_websockets_listener_events_total', {
+          label: 'event',
+          help: 'Total count of WebSockets listener events by type'
+        })
+      }
+    }
   }
 
   readonly [transportSymbol] = true
@@ -113,11 +135,13 @@
 
     const socket = await this._connect(ma, options)
     const maConn = socketToMaConn(socket, ma, {
-      logger: this.logger
+      logger: this.logger,
+      metrics: this.metrics?.dialerEvents
     })
     this.log('new outbound connection %s', maConn.remoteAddr)
 
     const conn = await options.upgrader.upgradeOutbound(maConn, options)
+    this.metrics?.dialerEvents.increment({ upgrade_success: true })
     this.log('outbound connection %s upgraded', maConn.remoteAddr)
     return conn
   }
@@ -136,22 +160,30 @@
       // https://developer.mozilla.org/en-US/docs/Web/API/WebSocket/error_event
       const err = new CodeError(`Could not connect to ${ma.toString()}`, 'ERR_CONNECTION_FAILED')
       this.log.error('connection error:', err)
+      this.metrics?.dialerEvents.increment({ socket_open_error: true })
       errorPromise.reject(err)
     })
 
     try {
       options.onProgress?.(new CustomProgressEvent('websockets:open-connection'))
       await raceSignal(Promise.race([rawSocket.connected(), errorPromise.promise]), options.signal)
     } catch (err: any) {
+      if (options.signal?.aborted === true) {
+        this.metrics?.dialerEvents.increment({ socket_open_abort: true })
+      } else {
+        this.metrics?.dialerEvents.increment({ socket_open_error: true })
+      }
       rawSocket.close()
         .catch(err => {
           this.log.error('error closing raw socket', err)
+          this.metrics?.dialerEvents.increment({ socket_close_error: true })
         })
 
       throw err
     }
 
     this.log('connected %s', ma)
+    this.metrics?.dialerEvents.increment({ socket_open_success: true })
     return rawSocket
   }
 
@@ -162,7 +194,8 @@
    */
   createListener (options: CreateListenerOptions): Listener {
     return createListener({
-      logger: this.logger
+      logger: this.logger,
+      metrics: this.metrics?.listenerEvents
     }, {
       ...this.init,
       ...options

packages/transport-websockets/src/listener.ts

@@ -4,14 +4,15 @@
 import { multiaddr, protocols } from '@multiformats/multiaddr'
 import { createServer } from 'it-ws/server'
 import { socketToMaConn } from './socket-to-conn.js'
-import type { ComponentLogger, Logger, Connection, Listener, ListenerEvents, CreateListenerOptions } from '@libp2p/interface'
+import type { ComponentLogger, Logger, Connection, Listener, ListenerEvents, CreateListenerOptions, CounterGroup } from '@libp2p/interface'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import type { Server } from 'http'
 import type { DuplexWebSocket } from 'it-ws/duplex'
 import type { WebSocketServer } from 'it-ws/server'
 
 export interface WebSocketListenerComponents {
   logger: ComponentLogger
+  metrics?: CounterGroup
 }
 
 export interface WebSocketListenerInit extends CreateListenerOptions {
@@ -28,6 +29,7 @@
     super()
 
     this.log = components.logger.forComponent('libp2p:websockets:listener')
+    const metrics = components.metrics
     // Keep track of open connections to destroy when the listener is closed
     this.connections = new Set<DuplexWebSocket>()
 
@@ -36,8 +38,12 @@
     this.server = createServer({
       ...init,
       onConnection: (stream: DuplexWebSocket) => {
+        const listeningAddrDetails = this.listeningMultiaddr?.toOptions()
+
         const maConn = socketToMaConn(stream, toMultiaddr(stream.remoteAddress ?? '', stream.remotePort ?? 0), {
-          logger: components.logger
+          logger: components.logger,
+          metrics,
+          metricPrefix: `${listeningAddrDetails?.transport}_${listeningAddrDetails?.host}:${listeningAddrDetails?.port} `
         })
         this.log('new inbound connection %s', maConn.remoteAddr)
 
@@ -51,6 +57,7 @@
           void init.upgrader.upgradeInbound(maConn)
             .then((conn) => {
               this.log('inbound connection %s upgraded', maConn.remoteAddr)
+              metrics?.increment({ upgrade_success: true })
 
               if (init?.handler != null) {
                 init?.handler(conn)
@@ -62,11 +69,13 @@
             })
             .catch(async err => {
               this.log.error('inbound connection failed to upgrade', err)
+              metrics?.increment({ upgrade_error: true })
 
               await maConn.close().catch(err => {
                 this.log.error('inbound connection failed to close after upgrade failed', err)
               })
             })
+          metrics?.increment({ socket_open_success: true })
         } catch (err) {
           this.log.error('inbound connection failed to upgrade', err)
           maConn.close().catch(err => {

packages/transport-websockets/src/socket-to-conn.ts

@@ -1,18 +1,22 @@
 import { CodeError } from '@libp2p/interface'
 import { CLOSE_TIMEOUT } from './constants.js'
-import type { AbortOptions, ComponentLogger, MultiaddrConnection } from '@libp2p/interface'
+import type { AbortOptions, ComponentLogger, CounterGroup, MultiaddrConnection } from '@libp2p/interface'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import type { DuplexWebSocket } from 'it-ws/duplex'
 
 export interface SocketToConnOptions {
   localAddr?: Multiaddr
   logger: ComponentLogger
+  metrics?: CounterGroup
+  metricPrefix?: string
 }
 
 // Convert a stream into a MultiaddrConnection
 // https://github.com/libp2p/interface-transport#multiaddrconnection
 export function socketToMaConn (stream: DuplexWebSocket, remoteAddr: Multiaddr, options: SocketToConnOptions): MultiaddrConnection {
   const log = options.logger.forComponent('libp2p:websockets:maconn')
+  const metrics = options.metrics
+  const metricPrefix = options.metricPrefix ?? ''
 
   const maConn: MultiaddrConnection = {
     log,
@@ -30,8 +34,10 @@
         })())
       } catch (err: any) {
         if (err.type !== 'aborted') {
+          metrics?.increment({ [`${metricPrefix}maconn_sink_error`]: true })
           log.error(err)
         }
+        metrics?.increment({ [`${metricPrefix}maconn_sink_abort`]: true })
       }
     },
 
@@ -57,6 +63,7 @@
         const { host, port } = maConn.remoteAddr.toOptions()
         log('timeout closing stream to %s:%s after %dms, destroying it manually',
           host, port, Date.now() - start)
+        metrics?.increment({ [`${metricPrefix}maconn_close_abort`]: true })
 
         this.abort(new CodeError('Socket close timeout', 'ERR_SOCKET_CLOSE_TIMEOUT'))
       }
@@ -65,8 +72,10 @@
 
       try {
         await stream.close()
+        metrics?.increment({ [`${metricPrefix}maconn_close_success`]: true })
       } catch (err: any) {
         log.error('error closing WebSocket gracefully', err)
+        metrics?.increment({ [`${metricPrefix}maconn_close_error`]: true })
         this.abort(err)
       } finally {
         options.signal?.removeEventListener('abort', listener)
@@ -91,6 +100,7 @@
     if (maConn.timeline.close == null) {
       maConn.timeline.close = Date.now()
     }
+    metrics?.increment({ [`${metricPrefix}maconn_socket_close_success`]: true })
   }, { once: true })
 
   return maConn