Add prometheus to client and begin implementing custom metrics (#3287)

* Add prometheus and txGuage for tx pool transaction count * Use import instead of require * Add cli option for enabling prometheus server * Fix test * Include typings for prometheus parameters * Update test timeouts * Update package files * Remove unneeded dep * Update packages/client/src/service/txpool.ts Co-authored-by: acolytec3 <[email protected]> * Update packages/client/src/service/txpool.ts Co-authored-by: acolytec3 <[email protected]> * Update packages/client/src/service/txpool.ts Co-authored-by: acolytec3 <[email protected]> * Track transaction in pool count by transaction type * Add test to verify tx count is incremented with prometheus gauge after transaction is added to pool * nits * Add prometheus port * Overhaul placement and management of metrics server * Fix typing * Generalize port number in comment --------- Co-authored-by: acolytec3 <[email protected]>
ethereumjs · Apr 17, 2024 · 78cebd5 · 78cebd5
1 parent e8f6ac5
commit 78cebd5
Show file tree

Hide file tree

Showing 14 changed files with 5,058 additions and 6,324 deletions.
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -55,8 +55,8 @@
     "vitest": "^1.2.2"
   },
   "peerDependencies": {
-    "@vitest/browser": "^1.4.0",
-    "webdriverio": "^8.35.1"
+    "@vitest/browser": "^1.5.0",
+    "webdriverio": "^8.36.0"
   },
   "peerDependenciesMeta": {
     "playwright": {

diff --git a/packages/client/README.md b/packages/client/README.md
@@ -305,6 +305,21 @@ dist/bin/cli.js --d
 --dnsNetworks        -- EIP-1459 ENR tree urls to query for peer discovery targets
 ```
 
+## Metrics
+
+The client can optionally collect metrics using the Prometheus metrics platform and expose them via an HTTP endpoint with the following CLI flags.  
+The current metrics that are reported by the client can be found [here](./src/util//metrics.ts).
+
+```sh
+# npm installation
+ethereumjs --prometheus
+
+# source installation
+npm run client:start:ts -- --prometheus --prometheusPort=9123
+```
+
+Note: The Prometheus endpoint runs on port 8000 by default
+
 ## API
 
 [API Reference](./docs/README.md)

diff --git a/packages/client/bin/cli.ts b/packages/client/bin/cli.ts
@@ -31,11 +31,15 @@ import { ecdsaRecover, ecdsaSign } from 'ethereum-cryptography/secp256k1-compat'
 import { sha256 } from 'ethereum-cryptography/sha256'
 import { existsSync, writeFileSync } from 'fs'
 import { ensureDirSync, readFileSync, removeSync } from 'fs-extra'
+import * as http from 'http'
+import { Server as RPCServer } from 'jayson/promise'
 import { loadKZG } from 'kzg-wasm'
 import { Level } from 'level'
 import { homedir } from 'os'
 import * as path from 'path'
+import * as promClient from 'prom-client'
 import * as readline from 'readline'
+import * as url from 'url'
 import * as yargs from 'yargs'
 import { hideBin } from 'yargs/helpers'
 
@@ -45,6 +49,7 @@ import { LevelDB } from '../src/execution/level'
 import { getLogger } from '../src/logging'
 import { Event } from '../src/types'
 import { parseMultiaddrs } from '../src/util'
+import { setupMetrics } from '../src/util/metrics'
 
 import { helprpc, startRPCServers } from './startRpc'
 
@@ -227,6 +232,16 @@ const args: ClientOpts = yargs
     number: true,
     default: 5,
   })
+  .option('prometheus', {
+    describe: 'Enable the Prometheus metrics server with HTTP endpoint',
+    boolean: true,
+    default: false,
+  })
+  .option('prometheusPort', {
+    describe: 'Enable the Prometheus metrics server with HTTP endpoint',
+    number: true,
+    default: 8000,
+  })
   .option('rpcDebug', {
     describe:
       'Additionally log truncated RPC calls filtered by name (prefix), e.g.: "eth,engine_getPayload" (use "all" for all methods). Truncated by default, add verbosity using "rpcDebugVerbose"',
@@ -833,11 +848,17 @@ function generateAccount(): Account {
  * @param config Client config object
  * @param clientStartPromise promise that returns a client and server object
  */
-const stopClient = async (config: Config, clientStartPromise: any) => {
+const stopClient = async (
+  config: Config,
+  clientStartPromise: Promise<{
+    client: EthereumClient
+    servers: (RPCServer | http.Server)[]
+  } | null>
+) => {
   config.logger.info('Caught interrupt signal. Obtaining client handle for clean shutdown...')
   config.logger.info('(This might take a little longer if client not yet fully started)')
   let timeoutHandle
-  if (clientStartPromise.toString().includes('Promise') === true)
+  if (clientStartPromise?.toString().includes('Promise') === true)
     // Client hasn't finished starting up so setting timeout to terminate process if not already shutdown gracefully
     timeoutHandle = setTimeout(() => {
       config.logger.warn('Client has become unresponsive while starting up.')
@@ -849,7 +870,7 @@ const stopClient = async (config: Config, clientStartPromise: any) => {
     config.logger.info('Shutting down the client and the servers...')
     const { client, servers } = clientHandle
     for (const s of servers) {
-      s.http().close()
+      s instanceof RPCServer ? (s as RPCServer).http().close() : (s as http.Server).close()
     }
     await client.stop()
     config.logger.info('Exiting.')
@@ -1027,6 +1048,41 @@ async function run() {
   const mine = args.mine !== undefined ? args.mine : args.dev !== undefined
   const isSingleNode = args.isSingleNode !== undefined ? args.isSingleNode : args.dev !== undefined
 
+  let prometheusMetrics = undefined
+  let metricsServer: http.Server | undefined
+  if (args.prometheus === true) {
+    // Create custom metrics
+    prometheusMetrics = setupMetrics()
+
+    const register = new promClient.Registry()
+    register.setDefaultLabels({
+      app: 'ethereumjs-client',
+    })
+    promClient.collectDefaultMetrics({ register })
+    for (const [_, metric] of Object.entries(prometheusMetrics)) {
+      register.registerMetric(metric)
+    }
+
+    metricsServer = http.createServer(async (req, res) => {
+      if (req.url === undefined) {
+        res.statusCode = 400
+        res.end('Bad Request: URL is missing')
+        return
+      }
+      const reqUrl = new url.URL(req.url, `http://${req.headers.host}`)
+      const route = reqUrl.pathname
+
+      if (route === '/metrics') {
+        // Return all metrics in the Prometheus exposition format
+        res.setHeader('Content-Type', register.contentType)
+        res.end(await register.metrics())
+      }
+    })
+    // Start the HTTP server which exposes the metrics on http://localhost:${args.prometheusPort}/metrics
+    logger.info(`Starting Metrics Server on port ${args.prometheusPort}`)
+    metricsServer.listen(args.prometheusPort)
+  }
+
   const config = new Config({
     accounts,
     bootnodes,
@@ -1072,6 +1128,7 @@ async function run() {
         ? 0
         : args.engineNewpayloadMaxExecute,
     ignoreStatelessInvalidExecs: args.ignoreStatelessInvalidExecs,
+    prometheusMetrics,
   })
   config.events.setMaxListeners(50)
   config.events.on(Event.SERVER_LISTENING, (details) => {
@@ -1097,7 +1154,7 @@ async function run() {
     genesisStateRoot: customGenesisStateRoot,
   })
     .then((client) => {
-      const servers =
+      const servers: (RPCServer | http.Server)[] =
         args.rpc === true || args.rpcEngine === true || args.ws === true
           ? startRPCServers(client, args as RPCArgs)
           : []
@@ -1107,6 +1164,7 @@ async function run() {
       ) {
         config.logger.warn(`Engine RPC endpoint not activated on a post-Merge HF setup.`)
       }
+      if (metricsServer !== undefined) servers.push(metricsServer)
       config.superMsg('Client started successfully')
       return { client, servers }
     })

diff --git a/packages/client/package.json b/packages/client/package.json
@@ -88,6 +88,7 @@
     "level": "^8.0.0",
     "memory-level": "^1.0.0",
     "multiaddr": "^10.0.1",
+    "prom-client": "^15.1.0",
     "qheap": "^1.4.0",
     "winston": "^3.3.3",
     "winston-daily-rotate-file": "^4.5.5",

diff --git a/packages/client/src/config.ts b/packages/client/src/config.ts
@@ -9,7 +9,7 @@ import { Event, EventBus } from './types'
 import { isBrowser, short } from './util'
 
 import type { Logger } from './logging'
-import type { EventBusType, MultiaddrLike } from './types'
+import type { EventBusType, MultiaddrLike, PrometheusMetrics } from './types'
 import type { BlockHeader } from '@ethereumjs/block'
 import type { VM, VMProfilerOpts } from '@ethereumjs/vm'
 import type { Multiaddr } from 'multiaddr'
@@ -338,6 +338,11 @@ export interface ConfigOptions {
   statelessVerkle?: boolean
   startExecution?: boolean
   ignoreStatelessInvalidExecs?: boolean | string
+
+  /**
+   * Enables Prometheus Metrics that can be collected for monitoring client health
+   */
+  prometheusMetrics?: PrometheusMetrics
 }
 
 export class Config {
@@ -461,6 +466,8 @@ export class Config {
 
   public readonly server: RlpxServer | undefined = undefined
 
+  public readonly metrics: PrometheusMetrics | undefined
+
   constructor(options: ConfigOptions = {}) {
     this.events = new EventBus() as EventBusType
 
@@ -536,6 +543,8 @@ export class Config {
     this.startExecution = options.startExecution ?? false
     this.ignoreStatelessInvalidExecs = options.ignoreStatelessInvalidExecs ?? false
 
+    this.metrics = options.prometheusMetrics
+
     // Start it off as synchronized if this is configured to mine or as single node
     this.synchronized = this.isSingleNode ?? this.mine
     this.lastSyncDate = 0

diff --git a/packages/client/src/miner/pendingBlock.ts b/packages/client/src/miner/pendingBlock.ts
@@ -361,7 +361,7 @@ export class PendingBlock {
         }
       } else if ((error as Error).message.includes('blobs missing')) {
         // Remove the blob tx which doesn't has blobs bundled
-        this.txPool.removeByHash(bytesToHex(tx.hash()))
+        this.txPool.removeByHash(bytesToHex(tx.hash()), tx)
         this.config.logger.error(
           `Pending: Removed from txPool a blob tx ${bytesToHex(tx.hash())} with missing blobs`
         )

diff --git a/packages/client/src/service/txpool.ts b/packages/client/src/service/txpool.ts
@@ -362,7 +362,21 @@ export class TxPool {
       add.push({ tx, added, hash })
       this.pool.set(address, add)
       this.handled.set(hash, { address, added })
+
       this.txsInPool++
+
+      if (isLegacyTx(tx)) {
+        this.config.metrics?.legacyTxGauge?.inc()
+      }
+      if (isAccessListEIP2930Tx(tx)) {
+        this.config.metrics?.accessListEIP2930TxGauge?.inc()
+      }
+      if (isFeeMarketEIP1559Tx(tx)) {
+        this.config.metrics?.feeMarketEIP1559TxGauge?.inc()
+      }
+      if (isBlobEIP4844Tx(tx)) {
+        this.config.metrics?.blobEIP4844TxGauge?.inc()
+      }
     } catch (e) {
       this.handled.set(hash, { address, added, error: e as Error })
       throw e
@@ -391,15 +405,30 @@ export class TxPool {
   /**
    * Removes the given tx from the pool
    * @param txHash Hash of the transaction
+   * @param tx Optional, the transaction object itself can be included for collecting metrics
    */
-  removeByHash(txHash: UnprefixedHash) {
+  removeByHash(txHash: UnprefixedHash, tx?: any) {
     const handled = this.handled.get(txHash)
     if (!handled) return
     const { address } = handled
     const poolObjects = this.pool.get(address)
     if (!poolObjects) return
     const newPoolObjects = poolObjects.filter((poolObj) => poolObj.hash !== txHash)
+
     this.txsInPool--
+    if (isLegacyTx(tx)) {
+      this.config.metrics?.legacyTxGauge?.dec()
+    }
+    if (isAccessListEIP2930Tx(tx)) {
+      this.config.metrics?.accessListEIP2930TxGauge?.dec()
+    }
+    if (isFeeMarketEIP1559Tx(tx)) {
+      this.config.metrics?.feeMarketEIP1559TxGauge?.dec()
+    }
+    if (isBlobEIP4844Tx(tx)) {
+      this.config.metrics?.blobEIP4844TxGauge?.dec()
+    }
+
     if (newPoolObjects.length === 0) {
       // List of txs for address is now empty, can delete
       this.pool.delete(address)
@@ -631,7 +660,7 @@ export class TxPool {
     for (const block of newBlocks) {
       for (const tx of block.transactions) {
         const txHash: UnprefixedHash = bytesToUnprefixedHex(tx.hash())
-        this.removeByHash(txHash)
+        this.removeByHash(txHash, tx)
       }
     }
   }
@@ -841,6 +870,12 @@ export class TxPool {
     this.pool.clear()
     this.handled.clear()
     this.txsInPool = 0
+    if (this.config.metrics !== undefined) {
+      // TODO: Only clear the metrics related to the transaction pool here
+      for (const [_, metric] of Object.entries(this.config.metrics)) {
+        metric.set(0)
+      }
+    }
     this.opened = false
   }
 

diff --git a/packages/client/src/types.ts b/packages/client/src/types.ts
@@ -7,6 +7,7 @@ import type { Block, BlockHeader } from '@ethereumjs/block'
 import type { DefaultStateManager } from '@ethereumjs/statemanager'
 import type { Address } from '@ethereumjs/util'
 import type { Multiaddr } from 'multiaddr'
+import type * as promClient from 'prom-client'
 
 /**
  * Types for the central event bus, emitted
@@ -129,6 +130,8 @@ export interface ClientOpts {
   logLevelFile?: string
   logRotate?: boolean
   logMaxFiles?: number
+  prometheus?: boolean
+  prometheusPort?: number
   rpcDebug?: string
   rpcDebugVerbose?: string
   rpcCors?: string
@@ -173,3 +176,10 @@ export interface ClientOpts {
   ignoreStatelessInvalidExecs?: string | boolean
   useJsCrypto?: boolean
 }
+
+export type PrometheusMetrics = {
+  legacyTxGauge: promClient.Gauge<string>
+  accessListEIP2930TxGauge: promClient.Gauge<string>
+  feeMarketEIP1559TxGauge: promClient.Gauge<string>
+  blobEIP4844TxGauge: promClient.Gauge<string>
+}
diff --git a/packages/client/src/util/metrics.ts b/packages/client/src/util/metrics.ts
@@ -0,0 +1,22 @@
+import * as promClient from 'prom-client'
+
+export const setupMetrics = () => {
+  return {
+    legacyTxGauge: new promClient.Gauge({
+      name: 'legacy_transactions_in_transaction_pool',
+      help: 'Number of legacy transactions in the client transaction pool',
+    }),
+    accessListEIP2930TxGauge: new promClient.Gauge({
+      name: 'access_list_eip2930_transactions_in_transaction_pool',
+      help: 'Number of access list EIP 2930 transactions in the client transaction pool',
+    }),
+    feeMarketEIP1559TxGauge: new promClient.Gauge({
+      name: 'fee_market_eip1559_transactions_in_transaction_pool',
+      help: 'Number of fee market EIP 1559 transactions in the client transaction pool',
+    }),
+    blobEIP4844TxGauge: new promClient.Gauge({
+      name: 'blob_eip_4844_transactions_in_transaction_pool',
+      help: 'Number of blob EIP 4844 transactions in the client transaction pool',
+    }),
+  }
+}
diff --git a/packages/client/test/miner/pendingBlock.spec.ts b/packages/client/test/miner/pendingBlock.spec.ts
@@ -55,12 +55,15 @@ common
   .map((hf) => {
     hf.timestamp = undefined
   })
-
+const txGauge: any = {
+  inc: () => {},
+}
 const config = new Config({
   common,
   accountCache: 10000,
   storageCache: 1000,
   logger: getLogger({ loglevel: 'debug' }),
+  prometheusMetrics: txGauge,
 })
 
 const setup = () => {

diff --git a/packages/client/test/rpc/eth/getBalance.spec.ts b/packages/client/test/rpc/eth/getBalance.spec.ts
@@ -107,5 +107,5 @@ describe(
       assert.ok(res.error.message.includes('"pending" is not yet supported'))
     })
   },
-  30000
+  40000
 )
diff --git a/packages/client/test/rpc/eth/getTransactionCount.spec.ts b/packages/client/test/rpc/eth/getTransactionCount.spec.ts
@@ -72,7 +72,7 @@ describe(method, () => {
     // call with nonexistent account
     res = await rpc.request(method, [`0x${'11'.repeat(20)}`, 'latest'])
     assert.equal(res.result, `0x0`, 'should return 0x0 for nonexistent account')
-  }, 30000)
+  }, 40000)
 
   it('call with unsupported block argument', async () => {
     const blockchain = await Blockchain.create()
@@ -84,5 +84,5 @@ describe(method, () => {
     const res = await rpc.request(method, ['0xccfd725760a68823ff1e062f4cc97e1360e8d997', 'pending'])
     assert.equal(res.error.code, INVALID_PARAMS)
     assert.ok(res.error.message.includes('"pending" is not yet supported'))
-  })
+  }, 40000)
 })