fix: add error metrics (#39)

Adds error event metrics similar to the TCP transport

Author: achingbrain

src/connection.ts

@@ -11,6 +11,7 @@ interface QuicConnectionInit {
   logger: ComponentLogger
   direction: Direction
   metrics?: CounterGroup
+  metricsPrefix?: string
 }
 
 interface QuicConnectionEvents {
@@ -23,6 +24,9 @@ export class QuicConnection extends TypedEventEmitter<QuicConnectionEvents> impl
   readonly log: Logger
   readonly remoteAddr: Multiaddr
   readonly metrics?: CounterGroup
+  readonly metricsPrefix: string
+
+  private remoteClosed?: boolean
 
   timeline: MultiaddrConnectionTimeline = {
     open: Date.now()
@@ -38,9 +42,12 @@ export class QuicConnection extends TypedEventEmitter<QuicConnectionEvents> impl
     this.log = init.logger.forComponent(`libp2p:quic:connection:${this.#connection.id()}:${init.direction}`)
     this.remoteAddr = multiaddr(this.#connection.remoteMultiaddr())
     this.metrics = init.metrics
+    this.metricsPrefix = init.metricsPrefix ?? ''
 
     // close maconn when connection is closed by remote
     this.#connection.closed().then(() => {
+      this.remoteClosed = true
+      this.metrics?.increment({ [`${this.metricsPrefix}end`]: true })
       this.close()
         .catch(err => {
           this.abort(err)
@@ -59,7 +66,11 @@ export class QuicConnection extends TypedEventEmitter<QuicConnectionEvents> impl
 
     this.timeline.close = Date.now()
     this.log('closed')
-    this.metrics?.increment({ close: true })
+
+    if (this.remoteClosed !== true) {
+      this.metrics?.increment({ [`${this.metricsPrefix}close`]: true })
+    }
+
     this.safeDispatchEvent('close')
   }
 
@@ -72,7 +83,7 @@ export class QuicConnection extends TypedEventEmitter<QuicConnectionEvents> impl
 
     this.timeline.close = Date.now()
     this.log('aborted - %e', err)
-    this.metrics?.increment({ abort: true })
+    this.metrics?.increment({ [`${this.metricsPrefix}abort`]: true })
     this.safeDispatchEvent('close')
   }
 }

src/listener.ts

@@ -13,7 +13,8 @@ export interface QuicCreateListenerOptions extends CreateListenerOptions {
 }
 
 export interface QuicListenerMetrics {
-  events: CounterGroup
+  events?: CounterGroup
+  errors?: CounterGroup
 }
 
 interface QuicListenerInit {
@@ -40,8 +41,9 @@ export class QuicListener extends TypedEventEmitter<ListenerEvents> implements L
   readonly init: QuicListenerInit
   readonly options: QuicCreateListenerOptions
   readonly log: Logger
-  readonly metrics?: QuicListenerMetrics
+  readonly metrics: QuicListenerMetrics
   private readonly shutdownController: AbortController
+  private addr: string
 
   state: QuicListenerState = { status: 'ready' }
 
@@ -51,17 +53,36 @@ export class QuicListener extends TypedEventEmitter<ListenerEvents> implements L
     this.init = init
     this.options = init.options
     this.log = init.logger.forComponent('libp2p:quic:listener')
+    this.addr = 'unknown'
 
     this.shutdownController = new AbortController()
     setMaxListeners(Infinity, this.shutdownController.signal)
 
-    if (init.metrics != null) {
-      this.metrics = {
-        events: init.metrics.registerMetricGroup('libp2p_quic_listener_events_total', {
-          label: 'address',
-          help: 'Total count of QUIC listener events by type'
-        })
+    init.metrics?.registerMetricGroup('libp2p_quic_inbound_connections_total', {
+      label: 'address',
+      help: 'Current active connections in QUIC listener',
+      calculate: () => {
+        if (this.state.status !== 'listening') {
+          return {
+            [this.addr]: 0
+          }
+        }
+
+        return {
+          [this.addr]: this.state.connections.size
+        }
       }
+    })
+
+    this.metrics = {
+      events: init.metrics?.registerMetricGroup('libp2p_quic_listener_events_total', {
+        label: 'address',
+        help: 'Total count of QUIC listener events by type'
+      }),
+      errors: init.metrics?.registerMetricGroup('libp2p_quic_listener_errors_total', {
+        label: 'address',
+        help: 'Total count of QUIC listener errors by type'
+      })
     }
 
     this.log('new')
@@ -91,6 +112,7 @@ export class QuicListener extends TypedEventEmitter<ListenerEvents> implements L
     const addr = ma.nodeAddress()
     const controller = new AbortController()
     const listener = new napi.Server(this.#config, addr.address, addr.port)
+    this.addr = `${addr.address}:${addr.port === 0 ? listener.port() : addr.port}`
 
     // replace wildcard port with actual listening port
     if (addr.port === 0) {
@@ -143,15 +165,18 @@ export class QuicListener extends TypedEventEmitter<ListenerEvents> implements L
         try {
           const listenerPromise = this.state.listener.inboundConnection()
           listenerPromise
-            .then(() => this.metrics?.events.increment({ connect: true }))
-            .catch(() => this.metrics?.events.increment({ error: true }))
+            .then(() => this.metrics.events?.increment({ [`${this.addr} connect`]: true }))
+            .catch((err) => {
+              this.log.error('%a error awaiting inbound connection - %e', listenAddr, err)
+              this.metrics.events?.increment({ [`${this.addr} error`]: true })
+            })
 
           const connection = await raceSignal(listenerPromise, signal)
           this.onInboundConnection(connection).catch((e) => {
-            this.log.error('%s error handling inbound connection', listenAddr.toString(), e)
+            this.log.error('%a error handling inbound connection - %e', listenAddr, e)
           })
         } catch (e) {
-          this.log.error('%s error accepting connection', listenAddr.toString(), e)
+          this.log.error('%a error accepting connection - %e', listenAddr, e)
 
           if (signal.aborted) {
             break
@@ -170,12 +195,19 @@ export class QuicListener extends TypedEventEmitter<ListenerEvents> implements L
       return
     }
 
-    const maConn = new QuicConnection({
-      connection,
-      logger: this.init.logger,
-      direction: 'inbound',
-      metrics: this.metrics?.events
-    })
+    let maConn: QuicConnection
+    try {
+      maConn = new QuicConnection({
+        connection,
+        logger: this.init.logger,
+        direction: 'inbound',
+        metrics: this.metrics?.events,
+        metricsPrefix: `${this.addr} `
+      })
+    } catch (err) {
+      this.metrics.errors?.increment({ [`${this.addr} inbound_to_connection`]: true })
+      throw err
+    }
 
     try {
       await this.options.upgrader.upgradeInbound(maConn, {
@@ -196,6 +228,7 @@ export class QuicListener extends TypedEventEmitter<ListenerEvents> implements L
       }, { once: true })
     } catch (err) {
       this.log.error('%s error handling inbound connection', this.state.listenAddr.toString(), err)
+      this.metrics.errors?.increment({ [`${this.addr} inbound_upgrade`]: true })
       maConn.abort(err as Error)
     }
   }

src/transport.ts

@@ -11,7 +11,8 @@ import type { Connection, CounterGroup, Listener, Logger, MultiaddrFilter, Trans
 import type { Multiaddr } from '@multiformats/multiaddr'
 
 interface QuicTransportMetrics {
-  events: CounterGroup
+  events?: CounterGroup
+  errors?: CounterGroup
 }
 
 export class QuicTransport implements Transport {
@@ -23,7 +24,7 @@ export class QuicTransport implements Transport {
 
   readonly log: Logger
   readonly components: QuicComponents
-  readonly metrics?: QuicTransportMetrics
+  readonly metrics: QuicTransportMetrics
 
   readonly #config: napi.QuinnConfig
 
@@ -48,13 +49,15 @@ export class QuicTransport implements Transport {
       ip6: new napi.Client(this.#config, 1)
     }
 
-    if (this.components.metrics != null) {
-      this.metrics = {
-        events: this.components.metrics?.registerCounterGroup('libp2p_quic_dialer_events_total', {
-          label: 'event',
-          help: 'Total count of QUIC dialer events by type'
-        })
-      }
+    this.metrics = {
+      events: this.components.metrics?.registerCounterGroup('libp2p_quic_dialer_events_total', {
+        label: 'event',
+        help: 'Total count of QUIC dialer events by type'
+      }),
+      errors: this.components.metrics?.registerCounterGroup('libp2p_quic_dialer_errors_total', {
+        label: 'event',
+        help: 'Total count of QUIC dialer errors by type'
+      })
     }
 
     this.listenFilter = listenFilter
@@ -74,25 +77,41 @@ export class QuicTransport implements Transport {
 
     const dialPromise = dialer.outboundConnection(addr.address, addr.port)
     dialPromise
-      .then(() => this.metrics?.events.increment({ connect: true }))
-      .catch(() => this.metrics?.events.increment({ error: true }))
+      .then(() => this.metrics.events?.increment({ connect: true }))
+      .catch(() => this.metrics.events?.increment({ error: true }))
     const connection = await dialPromise
 
-    const maConn = new QuicConnection({
-      connection,
-      logger: this.components.logger,
-      direction: 'outbound',
-      metrics: this.metrics?.events
-    })
-    return options.upgrader.upgradeOutbound(maConn, {
-      skipEncryption: true,
-      skipProtection: true,
-      muxerFactory: new QuicStreamMuxerFactory({
+    let maConn: QuicConnection
+
+    try {
+      maConn = new QuicConnection({
         connection,
-        logger: this.components.logger
-      }),
-      signal: options.signal
-    })
+        logger: this.components.logger,
+        direction: 'outbound',
+        metrics: this.metrics?.events
+      })
+    } catch (err) {
+      this.metrics.errors?.increment({ outbound_to_connection: true })
+      throw err
+    }
+
+    try {
+      this.log('new outbound connection %a', maConn.remoteAddr)
+      return await options.upgrader.upgradeOutbound(maConn, {
+        skipEncryption: true,
+        skipProtection: true,
+        muxerFactory: new QuicStreamMuxerFactory({
+          connection,
+          logger: this.components.logger
+        }),
+        signal: options.signal
+      })
+    } catch (err: any) {
+      this.metrics.errors?.increment({ outbound_upgrade: true })
+      this.log.error('error upgrading outbound connection - %e', err)
+      maConn.abort(err)
+      throw err
+    }
   }
 
   createListener (options: QuicCreateListenerOptions): Listener {