fix: record query metrics properly (#2736)

Using a metric for running queries misses most queryies because
they complete within the scrape interval. Use a counter instead
along with another for errors and record outgoing RPC messages
in the same way as incoming ones.

Author: achingbrain

packages/kad-dht/src/network.ts

@@ -11,7 +11,7 @@ import {
   queryErrorEvent
 } from './query/events.js'
 import type { KadDHTComponents, QueryEvent } from './index.js'
-import type { AbortOptions, Logger, Stream, PeerId, PeerInfo, Startable, RoutingOptions } from '@libp2p/interface'
+import type { AbortOptions, Logger, Stream, PeerId, PeerInfo, Startable, RoutingOptions, CounterGroup } from '@libp2p/interface'
 
 export interface NetworkInit {
   protocol: string
@@ -32,6 +32,10 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
   private running: boolean
   private readonly components: KadDHTComponents
   private readonly timeout: AdaptiveTimeout
+  private readonly metrics: {
+    operations?: CounterGroup
+    errors?: CounterGroup
+  }
 
   /**
    * Create a new network
@@ -49,6 +53,10 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
       metrics: components.metrics,
       metricName: `${init.logPrefix.replaceAll(':', '_')}_network_message_send_times_milliseconds`
     })
+    this.metrics = {
+      operations: components.metrics?.registerCounterGroup(`${init.logPrefix.replaceAll(':', '_')}_outbound_rpc_requests_total`),
+      errors: components.metrics?.registerCounterGroup(`${init.logPrefix.replaceAll(':', '_')}_outbound_rpc_errors_total`)
+    }
   }
 
   /**
@@ -103,6 +111,8 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
     }
 
     try {
+      this.metrics.operations?.increment({ [type]: true })
+
       const connection = await this.components.connectionManager.openConnection(to, options)
       stream = await connection.newStream(this.protocol, options)
       const response = await this._writeReadMessage(stream, msg, options)
@@ -121,6 +131,8 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
         record: response.record == null ? undefined : Libp2pRecord.deserialize(response.record)
       }, options)
     } catch (err: any) {
+      this.metrics.errors?.increment({ [type]: true })
+
       stream?.abort(err)
 
       // only log if the incoming signal was not aborted - this means we were
@@ -162,6 +174,8 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
     }
 
     try {
+      this.metrics.operations?.increment({ [type]: true })
+
       const connection = await this.components.connectionManager.openConnection(to, options)
       stream = await connection.newStream(this.protocol, options)
 
@@ -175,6 +189,8 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
 
       yield peerResponseEvent({ from: to, messageType: type }, options)
     } catch (err: any) {
+      this.metrics.errors?.increment({ [type]: true })
+
       stream?.abort(err)
       yield queryErrorEvent({ from: to, error: err }, options)
     } finally {

packages/kad-dht/src/query/manager.ts

@@ -12,7 +12,7 @@ import { queryPath } from './query-path.js'
 import type { QueryFunc } from './types.js'
 import type { QueryEvent } from '../index.js'
 import type { RoutingTable } from '../routing-table/index.js'
-import type { ComponentLogger, Metric, Metrics, PeerId, RoutingOptions, Startable } from '@libp2p/interface'
+import type { ComponentLogger, Counter, Metric, Metrics, PeerId, RoutingOptions, Startable } from '@libp2p/interface'
 import type { ConnectionManager } from '@libp2p/interface-internal'
 import type { DeferredPromise } from 'p-defer'
 
@@ -52,16 +52,16 @@ export class QueryManager implements Startable {
   private readonly alpha: number
   private shutDownController: AbortController
   private running: boolean
-  private queries: number
   private readonly logger: ComponentLogger
   private readonly peerId: PeerId
   private readonly connectionManager: ConnectionManager
   private readonly routingTable: RoutingTable
   private initialQuerySelfHasRun?: DeferredPromise<void>
   private readonly logPrefix: string
-  private readonly metrics?: {
-    runningQueries: Metric
-    queryTime: Metric
+  private readonly metrics: {
+    queries?: Counter
+    errors?: Counter
+    queryTime?: Metric
   }
 
   constructor (components: QueryManagerComponents, init: QueryManagerInit) {
@@ -71,18 +71,16 @@ export class QueryManager implements Startable {
     this.disjointPaths = disjointPaths ?? K
     this.running = false
     this.alpha = alpha ?? ALPHA
-    this.queries = 0
     this.initialQuerySelfHasRun = init.initialQuerySelfHasRun
     this.routingTable = init.routingTable
     this.logger = components.logger
     this.peerId = components.peerId
     this.connectionManager = components.connectionManager
 
-    if (components.metrics != null) {
-      this.metrics = {
-        runningQueries: components.metrics.registerMetric(`${logPrefix.replaceAll(':', '_')}_running_queries`),
-        queryTime: components.metrics.registerMetric(`${logPrefix.replaceAll(':', '_')}_query_time_seconds`)
-      }
+    this.metrics = {
+      queries: components.metrics?.registerCounter(`${logPrefix.replaceAll(':', '_')}_queries_total`),
+      errors: components.metrics?.registerCounter(`${logPrefix.replaceAll(':', '_')}_query_errors_total`),
+      queryTime: components.metrics?.registerMetric(`${logPrefix.replaceAll(':', '_')}_query_time_seconds`)
     }
 
     // allow us to stop queries on shut down
@@ -121,7 +119,7 @@ export class QueryManager implements Startable {
       throw new Error('QueryManager not started')
     }
 
-    const stopQueryTimer = this.metrics?.queryTime.timer()
+    const stopQueryTimer = this.metrics.queryTime?.timer()
 
     if (options.signal == null) {
       // don't let queries run forever
@@ -167,8 +165,7 @@ export class QueryManager implements Startable {
       }
 
       log('query:start')
-      this.queries++
-      this.metrics?.runningQueries.update(this.queries)
+      this.metrics?.queries?.increment()
 
       const id = await convertBuffer(key)
       const peers = this.routingTable.closestPeers(id)
@@ -223,6 +220,10 @@ export class QueryManager implements Startable {
 
       queryFinished = true
     } catch (err: any) {
+      if (!queryFinished) {
+        this.metrics?.errors?.increment()
+      }
+
       if (!this.running && err.name === 'QueryAbortedError') {
         // ignore query aborted errors that were thrown during query manager shutdown
       } else {
@@ -236,13 +237,7 @@ export class QueryManager implements Startable {
 
       signal.clear()
 
-      this.queries--
-      this.metrics?.runningQueries.update(this.queries)
-
-      if (stopQueryTimer != null) {
-        stopQueryTimer()
-      }
-
+      stopQueryTimer?.()
       log('query:done in %dms', Date.now() - startTime)
     }
   }

packages/kad-dht/src/routing-table/index.ts

@@ -5,7 +5,7 @@ import { pbStream } from 'it-protobuf-stream'
 import { Message, MessageType } from '../message/dht.js'
 import * as utils from '../utils.js'
 import { KBucket, isLeafBucket, type Bucket, type PingEventDetails } from './k-bucket.js'
-import type { ComponentLogger, Logger, Metric, Metrics, PeerId, PeerStore, Startable, Stream } from '@libp2p/interface'
+import type { ComponentLogger, CounterGroup, Logger, Metric, Metrics, PeerId, PeerStore, Startable, Stream } from '@libp2p/interface'
 import type { ConnectionManager } from '@libp2p/interface-internal'
 
 export const KAD_CLOSE_TAG_NAME = 'kad-close'
@@ -64,6 +64,7 @@ export class RoutingTable extends TypedEventEmitter<RoutingTableEvents> implemen
     routingTableKadBucketTotal: Metric
     routingTableKadBucketAverageOccupancy: Metric
     routingTableKadBucketMaxDepth: Metric
+    kadBucketEvents: CounterGroup<'ping' | 'ping_error' | 'peer_added' | 'peer_removed'>
   }
 
   constructor (components: RoutingTableComponents, init: RoutingTableInit) {
@@ -95,7 +96,8 @@ export class RoutingTable extends TypedEventEmitter<RoutingTableEvents> implemen
         routingTableSize: this.components.metrics.registerMetric(`${init.logPrefix.replaceAll(':', '_')}_routing_table_size`),
         routingTableKadBucketTotal: this.components.metrics.registerMetric(`${init.logPrefix.replaceAll(':', '_')}_routing_table_kad_bucket_total`),
         routingTableKadBucketAverageOccupancy: this.components.metrics.registerMetric(`${init.logPrefix.replaceAll(':', '_')}_routing_table_kad_bucket_average_occupancy`),
-        routingTableKadBucketMaxDepth: this.components.metrics.registerMetric(`${init.logPrefix.replaceAll(':', '_')}_routing_table_kad_bucket_max_depth`)
+        routingTableKadBucketMaxDepth: this.components.metrics.registerMetric(`${init.logPrefix.replaceAll(':', '_')}_routing_table_kad_bucket_max_depth`),
+        kadBucketEvents: this.components.metrics.registerCounterGroup(`${init.logPrefix.replaceAll(':', '_')}_kad_bucket_events_total`)
       }
     }
   }
@@ -121,7 +123,10 @@ export class RoutingTable extends TypedEventEmitter<RoutingTableEvents> implemen
 
     // test whether to evict peers
     kBuck.addEventListener('ping', (evt) => {
+      this.metrics?.kadBucketEvents.increment({ ping: true })
+
       this._onPing(evt).catch(err => {
+        this.metrics?.kadBucketEvents.increment({ ping_error: true })
         this.log.error('could not process k-bucket ping event', err)
       })
     })
@@ -195,11 +200,13 @@ export class RoutingTable extends TypedEventEmitter<RoutingTableEvents> implemen
     kBuck.addEventListener('added', (evt) => {
       updatePeerTags()
 
+      this.metrics?.kadBucketEvents.increment({ peer_added: true })
       this.safeDispatchEvent('peer:add', { detail: evt.detail.peerId })
     })
     kBuck.addEventListener('removed', (evt) => {
       updatePeerTags()
 
+      this.metrics?.kadBucketEvents.increment({ peer_removed: true })
       this.safeDispatchEvent('peer:remove', { detail: evt.detail.peerId })
     })
   }
@@ -271,7 +278,7 @@ export class RoutingTable extends TypedEventEmitter<RoutingTableEvents> implemen
 
             return false
           } finally {
-            this.metrics?.routingTableSize.update(this.size)
+            this.updateMetrics()
           }
         }, {
           peerId: oldContact.peerId