WIP: Update metrics

packages/portalnetwork/src/client/client.ts

@@ -12,6 +12,7 @@ import { HistoryNetwork } from '../networks/history/history.js'
 import {
   BeaconLightClientNetwork,
   NetworkId,
+  NetworkNames,
   StateNetwork,
   SyncStrategy,
 } from '../networks/index.js'
@@ -26,6 +27,7 @@ import { TransportLayer } from './types.js'
 import type { IDiscv5CreateOptions, SignableENRInput } from '@chainsafe/discv5'
 import type { ITalkReqMessage, ITalkRespMessage } from '@chainsafe/discv5/message'
 import type { Debugger } from 'debug'
+import type * as PromClient from 'prom-client'
 import type { BaseNetwork } from '../networks/network.js'
 import type {
   INodeAddress,
@@ -273,8 +275,6 @@ export class PortalNetwork extends EventEmitter<PortalNetworkEvents> {
     })
     if (opts.metrics) {
       this.metrics = opts.metrics
-      this.metrics.knownDiscv5Nodes.collect = () =>
-        this.metrics?.knownDiscv5Nodes.set(this.discv5.kadValues().length)
       this.metrics.currentDBSize.collect = async () => {
         this.metrics?.currentDBSize.set(await this.db.currentSize())
       }
@@ -303,6 +303,12 @@ export class PortalNetwork extends EventEmitter<PortalNetworkEvents> {
       }
       this.shouldRefresh && network.startRefresh()
       await network.prune()
+      if (this.metrics) {
+        network.on('ContentAdded', async () => {
+          const metric = (NetworkNames[network.networkId] + '_dbSize') as keyof PortalNetworkMetrics
+          ;(<PromClient.Gauge>this.metrics![metric]).set(await network.db.size())
+        })
+      }
     }
     void this.bootstrap()
   }
@@ -329,6 +335,7 @@ export class PortalNetwork extends EventEmitter<PortalNetworkEvents> {
     await this.db.close()
     for (const network of this.networks.values()) {
       network.stopRefresh()
+      network.removeAllListeners()
     }
   }
 
@@ -406,23 +413,31 @@ export class PortalNetwork extends EventEmitter<PortalNetworkEvents> {
     message: ITalkReqMessage,
   ) => {
     this.metrics?.totalBytesReceived.inc(message.request.length)
+    const network = this.networks.get(bytesToHex(message.protocol) as NetworkId)
+
     if (bytesToHex(message.protocol) === NetworkId.UTPNetwork) {
       await this.handleUTP(nodeAddress, message, message.request)
       return
     }
-    const network = this.networks.get(bytesToHex(message.protocol) as NetworkId)
     if (!network) {
       this.logger(`Received TALKREQ message on unsupported network ${bytesToHex(message.protocol)}`)
       await this.sendPortalNetworkResponse(nodeAddress, message.id, new Uint8Array())
 
       return
     }
 
+    if (this.metrics) {
+      const metric = (NetworkNames[bytesToHex(message.protocol) as NetworkId] +
+        '_talkReqReceived') as keyof PortalNetworkMetrics
+      this.metrics[metric].inc()
+    }
     await network.handle(message, nodeAddress)
   }
 
   private onTalkResp = (_: any, __: any, message: ITalkRespMessage) => {
-    this.metrics?.totalBytesReceived.inc(message.response.length)
+    if (this.metrics) {
+      this.metrics?.totalBytesReceived.inc(message.response.length)
+    }
   }
 
   /**

packages/portalnetwork/src/client/types.ts

@@ -71,8 +71,6 @@ interface Counter {
 }
 export interface PortalNetworkMetrics {
   totalContentLookups: Counter
-  knownHistoryNodes: Gauge
-  knownDiscv5Nodes: Gauge
   successfulContentLookups: Counter
   failedContentLookups: Counter
   offerMessagesSent: Counter

packages/portalnetwork/src/networks/network.ts

@@ -26,8 +26,9 @@ import type {
   OfferMessage,
   PingMessage,
   PongMessage,
+
   PortalNetwork,
-} from '../index.js'
+  PortalNetworkMetrics} from '../index.js'
 import {
   BasicRadius,
   ClientInfoAndCapabilities,
@@ -37,6 +38,7 @@ import {
   MAX_PACKET_SIZE,
   MessageCodes,
   NetworkId,
+  NetworkNames,
   NodeLookup,
   PingPongErrorCodes,
   PingPongPayloadExtensions,
@@ -49,8 +51,8 @@ import {
   encodeWithVariantPrefix,
   generateRandomNodeIdAtDistance,
   randUint16,
-  shortId,
-} from '../index.js'
+
+  shortId} from '../index.js'
 import { FoundContent } from '../wire/types.js'
 
 import { NetworkDB } from './networkDB.js'
@@ -110,11 +112,6 @@ export abstract class BaseNetwork extends EventEmitter {
       db,
       logger: this.logger,
     })
-    if (this.portal.metrics) {
-      this.portal.metrics.knownHistoryNodes.collect = () => {
-        this.portal.metrics?.knownHistoryNodes.set(this.routingTable.size)
-      }
-    }
     this.gossipManager = new GossipManager(this, gossipCount)
   }
 
@@ -142,6 +139,10 @@ export abstract class BaseNetwork extends EventEmitter {
     networkId: NetworkId,
     utpMessage?: boolean,
   ): Promise<Uint8Array> {
+    if (this.portal.metrics) {
+      const metric = (this.networkName + '_talkReqSent') as keyof PortalNetworkMetrics
+      this.portal.metrics[metric].inc()
+    }
     try {
       const res = await this.portal.sendPortalNetworkMessage(enr, payload, networkId, utpMessage)
       return res
@@ -938,6 +939,14 @@ export abstract class BaseNetwork extends EventEmitter {
     this.logger.extend('bucketRefresh')(
       `Finished bucket refresh with ${newSize} peers (${newSize - size} new peers)`,
     )
+    if (this.portal.metrics !== undefined) {
+      const metric = (NetworkNames[this.networkId] + '_peers') as keyof PortalNetworkMetrics
+      try {
+        (<PromClient.Gauge>this.portal.metrics[metric]).set(this.routingTable.size)
+      } catch (err) {
+        this.logger.extend('bucketRefresh')(`Error updating ${metric}:  ${(err as any).message}`)
+      }
+    }
   }
 
   /**

packages/portalnetwork/src/networks/types.ts

@@ -39,6 +39,10 @@ export enum NetworkId {
   UTPNetwork = '0x757470',
 }
 
+export const NetworkNames = Object.fromEntries(Object.entries(NetworkId).map(([n, i]) => [i, n]))
+
+export type NetworkName = keyof typeof NetworkNames
+
 export type SubNetwork<T extends NetworkId> = T extends '0x500a'
   ? HistoryNetwork
   : T extends '0x504a'

packages/portalnetwork/src/util/config.ts

@@ -106,7 +106,7 @@ export const cliConfig = async (args: PortalClientOpts) => {
       bootnodes.push(bootnode)
     }
   }
-  const metrics = setupMetrics()
+  const metrics = setupMetrics(networks.map((x) => x.networkId))
 
   const clientConfig: Partial<PortalNetworkOpts> = {
     config,

packages/portalnetwork/src/util/metrics.ts

@@ -1,86 +1,186 @@
-import * as PromClient from 'prom-client'
+import { Counter, Gauge, Histogram, Summary } from 'prom-client'
 
-export const setupMetrics = () => {
+import { NetworkId, NetworkNames } from '../networks/types.js'
+
+import type { PortalNetworkMetrics } from '../client/types.js'
+import type { Metric } from 'prom-client'
+
+export enum MetricType {
+  Counter,
+  Gauge,
+  Histogram,
+  Summary,
+}
+
+const metricTypes = {
+  [MetricType.Gauge]: Gauge,
+  [MetricType.Counter]: Counter,
+  [MetricType.Histogram]: Histogram,
+  [MetricType.Summary]: Summary,
+}
+
+interface MetricParams {
+  metric: MetricType
+  name: string
+  help: string
+}
+
+const createMetric = ({ metric, name, help }: MetricParams) => {
+  return (networks: NetworkId[]) => {
+    const metrics: Record<string, Metric<NetworkId>> = {}
+    for (const network of networks) {
+      const metricName = NetworkNames[network] + '_' + name
+      metrics[metricName] = new metricTypes[metric]({
+        name: 'ultralight_' + metricName,
+        help,
+      })
+    }
+    return metrics
+  }
+}
+
+const createMetrics = (metrics: MetricParams[], networks: NetworkId[]) => {
+  let m: Record<string, Metric<NetworkId>> = {}
+  const metricsFunctions = metrics.map(createMetric)
+  for (const metricFunction of metricsFunctions) {
+    m = { ...m, ...metricFunction(networks) }
+  }
+  return m as Record<keyof PortalNetworkMetrics, Metric<NetworkId>>
+}
+
+const ultralightMetrics = [
+  {
+    name: 'peers',
+    metric: MetricType.Gauge,
+    help: 'how many peers are in the routing table',
+  },
+  {
+    name: 'dbSize',
+    metric: MetricType.Gauge,
+    help: 'how many MBs are currently stored in the db',
+  },
+  {
+    name: 'talkReqSent',
+    metric: MetricType.Counter,
+    help: 'how many talk requests have been sent',
+  },
+  {
+    name: 'talkReqReceived',
+    metric: MetricType.Counter,
+    help: 'how many talk requests have been received',
+  },
+  {
+    name: 'utpPacketsSent',
+    metric: MetricType.Counter,
+    help: 'how many UTP packets have been sent',
+  },
+  {
+    name: 'utpPacketsReceived',
+    metric: MetricType.Counter,
+    help: 'how many UTP packets have been received',
+  },
+  {
+    name: 'utpStreamsTotal',
+    metric: MetricType.Gauge,
+    help: 'how many total UTP streams were opened',
+  },
+  {
+    name: 'utpWriteStreamsOpened',
+    metric: MetricType.Gauge,
+    help: 'how many UTP write streams were opened',
+  },
+  {
+    name: 'utpWriteStreamsCompleted',
+    metric: MetricType.Gauge,
+    help: 'how many UTP write streams were completed',
+  },
+  {
+    name: 'utpReadStreamsOpened',
+    metric: MetricType.Gauge,
+    help: 'how many UTP read streams were opened',
+  },
+  {
+    name: 'utpReadStreamsCompleted',
+    metric: MetricType.Gauge,
+    help: 'how many UTP read streams were completed',
+  },
+]
+
+export const setupMetrics = (
+  networks: NetworkId[] = [NetworkId.HistoryNetwork],
+): PortalNetworkMetrics => {
+  const metrics = createMetrics(ultralightMetrics, [...networks])
   return {
-    knownDiscv5Nodes: new PromClient.Gauge({
-      name: 'ultralight_known_discv5_peers',
-      help: 'how many peers are in discv5 routing table',
-      async collect() {},
-    }),
-    knownHistoryNodes: new PromClient.Gauge({
-      name: 'ultralight_known_history_peers',
-      help: 'how many peers are in discv5 routing table',
-      async collect() {},
-    }),
-    totalContentLookups: new PromClient.Gauge<string>({
+    ...metrics,
+    totalContentLookups: new Gauge<string>({
       name: 'ultralight_total_content_lookups',
       help: 'total number of content lookups initiated',
     }),
-    successfulContentLookups: new PromClient.Counter({
+    successfulContentLookups: new Counter({
       name: 'ultralight_successful_content_lookups',
       help: 'how many content lookups successfully returned content',
     }),
-    failedContentLookups: new PromClient.Counter({
+    failedContentLookups: new Counter({
       name: 'ultralight_failed_content_lookups',
       help: 'how many content lookups failed to return content',
     }),
-    offerMessagesSent: new PromClient.Counter({
+    offerMessagesSent: new Counter({
       name: 'ultralight_offer_messages_sent',
       help: 'how many offer messages have been sent',
     }),
-    offerMessagesReceived: new PromClient.Counter({
+    offerMessagesReceived: new Counter({
       name: 'ultralight_offer_messages_received',
       help: 'how many offer messages have been received',
     }),
-    acceptMessagesSent: new PromClient.Counter({
+    acceptMessagesSent: new Counter({
       name: 'ultralight_accept_messages_sent',
       help: 'how many accept messages have been sent',
     }),
-    acceptMessagesReceived: new PromClient.Counter({
+    acceptMessagesReceived: new Counter({
       name: 'ultralight_accept_messages_received',
       help: 'how many accept messages have been received',
     }),
-    findContentMessagesSent: new PromClient.Counter({
+    findContentMessagesSent: new Counter({
       name: 'ultralight_findContent_messages_sent',
       help: 'how many findContent messages have been sent',
     }),
-    findContentMessagesReceived: new PromClient.Counter({
+    findContentMessagesReceived: new Counter({
       name: 'ultralight_findContent_messages_received',
       help: 'how many findContent messages have been received',
     }),
-    contentMessagesSent: new PromClient.Counter({
+    contentMessagesSent: new Counter({
       name: 'ultralight_content_messages_sent',
       help: 'how many content messages have been sent',
     }),
-    contentMessagesReceived: new PromClient.Counter({
+    contentMessagesReceived: new Counter({
       name: 'ultralight_content_messages_received',
       help: 'how many content messages have been received',
     }),
-    findNodesMessagesSent: new PromClient.Counter({
+    findNodesMessagesSent: new Counter({
       name: 'ultralight_findNodes_messages_sent',
       help: 'how many findNodes messages have been sent',
     }),
-    findNodesMessagesReceived: new PromClient.Counter({
+    findNodesMessagesReceived: new Counter({
       name: 'ultralight_findNodes_messages_received',
       help: 'how many findNodes messages have been received',
     }),
-    nodesMessagesSent: new PromClient.Counter({
+    nodesMessagesSent: new Counter({
       name: 'ultralight_nodes_messages_sent',
       help: 'how many nodes messages have been sent',
     }),
-    nodesMessagesReceived: new PromClient.Counter({
+    nodesMessagesReceived: new Counter({
       name: 'ultralight_nodes_messages_received',
       help: 'how many nodes messages have been received',
     }),
-    totalBytesReceived: new PromClient.Counter({
+    totalBytesReceived: new Counter({
       name: 'ultralight_total_bytes_received',
       help: 'how many bytes have been received in Portal Network message payloads',
     }),
-    totalBytesSent: new PromClient.Counter({
+    totalBytesSent: new Counter({
       name: 'ultralight_total_bytes_sent',
       help: 'how many bytes have been sent in Portal Network message payloads',
     }),
-    currentDBSize: new PromClient.Gauge({
+    currentDBSize: new Gauge({
       name: 'ultralight_db_size',
       help: 'how many MBs are currently stored in the db',
     }),