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,14 @@
 
  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)
 
+ this.metrics?.dialerEvents.increment({ upgrade_start: true })
  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 +161,29 @@
  // 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'))
+ this.metrics?.dialerEvents.increment({ socket_open_start: true })
  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 })
+ }
  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,10 @@
  this.server = createServer({
  ...init,
  onConnection: (stream: DuplexWebSocket) => {
+ metrics?.increment({ socket_open_start: true })
  const maConn = socketToMaConn(stream, toMultiaddr(stream.remoteAddress ?? '', stream.remotePort ?? 0), {
- logger: components.logger
+ logger: components.logger,
+ metrics
  })
  this.log('new inbound connection %s', maConn.remoteAddr)
 
@@ -48,9 +52,11 @@
  })
 
  try {
+ metrics?.increment({ upgrade_start: true })
  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 +68,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
 }
 
 // Convert a stream into a MultiaddrConnection
 // https:/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
+
+ metrics?.increment({ maconn_open_start: true })
 
  const maConn: MultiaddrConnection = {
  log,
@@ -30,8 +34,10 @@
  })())
  } catch (err: any) {
  if (err.type !== 'aborted') {
+ metrics?.increment({ maconn_sink_error: true })
  log.error(err)
  }
+ metrics?.increment({ maconn_sink_abort: true })
  }
  },
 
@@ -43,6 +49,7 @@
 
  async close (options: AbortOptions = {}) {
  const start = Date.now()
+ metrics?.increment({ maconn_close_start: true })
 
  if (options.signal == null) {
  const signal = AbortSignal.timeout(CLOSE_TIMEOUT)
@@ -57,6 +64,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({ maconn_close_abort: true })
 
  this.abort(new CodeError('Socket close timeout', 'ERR_SOCKET_CLOSE_TIMEOUT'))
  }
@@ -65,8 +73,10 @@
 
  try {
  await stream.close()
+ metrics?.increment({ maconn_close_success: true })
  } catch (err: any) {
  log.error('error closing WebSocket gracefully', err)
+ metrics?.increment({ maconn_close_error: true })
  this.abort(err)
  } finally {
  options.signal?.removeEventListener('abort', listener)
@@ -79,6 +89,8 @@
  log('timeout closing stream to %s:%s due to error',
  host, port, err)
 
+ metrics?.increment({ maconn_abort: true })
+
  stream.destroy()
  maConn.timeline.close = Date.now()
  }
@@ -91,7 +103,10 @@
  if (maConn.timeline.close == null) {
  maConn.timeline.close = Date.now()
  }
+ metrics?.increment({ socket_close_success: true })
  }, { once: true })
 
+ metrics?.increment({ maconn_open_success: true })
+
  return maConn
 }