feat: Add stats of client and pool to be accessible through agent (#4157)

Author: tdeekens

docs/docs/api/Agent.md

@@ -75,3 +75,9 @@ See [`Dispatcher.stream(options, factory[, callback])`](/docs/docs/api/Dispatche
 ### `Agent.upgrade(options[, callback])`
 
 See [`Dispatcher.upgrade(options[, callback])`](/docs/docs/api/Dispatcher.md#dispatcherupgradeoptions-callback).
+
+### `Agent.stats()`
+
+Returns an object of stats by origin in the format of `Record<string, TClientStats | TPoolStats>`
+
+See [`PoolStats`](/docs/docs/api/PoolStats.md) and [`ClientStats`](/docs/docs/api/ClientStats.md).

docs/docs/api/ClientStats.md

@@ -0,0 +1,27 @@
+# Class: ClientStats
+
+Stats for a [Client](/docs/docs/api/Client.md).
+
+## `new ClientStats(client)`
+
+Arguments:
+
+* **client** `Client` - Client from which to return stats.
+
+## Instance Properties
+
+### `ClientStats.connected`
+
+Boolean if socket as open connection by this client.
+
+### `ClientStats.pending`
+
+Number of pending requests of this client.
+
+### `ClientStats.running`
+
+Number of currently active requests across this client.
+
+### `ClientStats.size`
+
+Number of active, pending, or queued requests of this clients.

lib/dispatcher/agent.js

@@ -1,7 +1,7 @@
 'use strict'
 
 const { InvalidArgumentError } = require('../core/errors')
-const { kClients, kRunning, kClose, kDestroy, kDispatch } = require('../core/symbols')
+const { kClients, kRunning, kClose, kDestroy, kDispatch, kUrl } = require('../core/symbols')
 const DispatcherBase = require('./dispatcher-base')
 const Pool = require('./pool')
 const Client = require('./client')
@@ -110,6 +110,16 @@ class Agent extends DispatcherBase {
 
     await Promise.all(destroyPromises)
   }
+
+  get stats () {
+    const allClientStats = {}
+    for (const client of this[kClients].values()) {
+      if (client.stats) {
+        allClientStats[client[kUrl].origin] = client.stats
+      }
+    }
+    return allClientStats
+  }
 }
 
 module.exports = Agent

lib/dispatcher/client.js

@@ -4,6 +4,7 @@ const assert = require('node:assert')
 const net = require('node:net')
 const http = require('node:http')
 const util = require('../core/util.js')
+const { ClientStats } = require('../util/stats.js')
 const { channels } = require('../core/diagnostics.js')
 const Request = require('../core/request.js')
 const DispatcherBase = require('./dispatcher-base')
@@ -260,6 +261,10 @@ class Client extends DispatcherBase {
     this[kResume](true)
   }
 
+  get stats () {
+    return new ClientStats(this)
+  }
+
   get [kPending] () {
     return this[kQueue].length - this[kPendingIdx]
   }

lib/dispatcher/pool-base.js

@@ -1,9 +1,9 @@
 'use strict'
 
+const { PoolStats } = require('../util/stats.js')
 const DispatcherBase = require('./dispatcher-base')
 const FixedQueue = require('./fixed-queue')
 const { kConnected, kSize, kRunning, kPending, kQueued, kBusy, kFree, kUrl, kClose, kDestroy, kDispatch } = require('../core/symbols')
-const PoolStats = require('./pool-stats')
 
 const kClients = Symbol('clients')
 const kNeedDrain = Symbol('needDrain')
@@ -16,7 +16,6 @@ const kOnConnectionError = Symbol('onConnectionError')
 const kGetDispatcher = Symbol('get dispatcher')
 const kAddClient = Symbol('add client')
 const kRemoveClient = Symbol('remove client')
-const kStats = Symbol('stats')
 
 class PoolBase extends DispatcherBase {
   constructor () {
@@ -67,8 +66,6 @@ class PoolBase extends DispatcherBase {
     this[kOnConnectionError] = (origin, targets, err) => {
       pool.emit('connectionError', origin, [pool, ...targets], err)
     }
-
-    this[kStats] = new PoolStats(this)
   }
 
   get [kBusy] () {
@@ -108,7 +105,7 @@ class PoolBase extends DispatcherBase {
   }
 
   get stats () {
-    return this[kStats]
+    return new PoolStats(this)
   }
 
   async [kClose] () {

lib/dispatcher/pool-stats.js

@@ -0,0 +1,32 @@
+'use strict'
+
+const {
+  kConnected,
+  kPending,
+  kRunning,
+  kSize,
+  kFree,
+  kQueued
+} = require('../core/symbols')
+
+class ClientStats {
+  constructor (client) {
+    this.connected = client[kConnected]
+    this.pending = client[kPending]
+    this.running = client[kRunning]
+    this.size = client[kSize]
+  }
+}
+
+class PoolStats {
+  constructor (pool) {
+    this.connected = pool[kConnected]
+    this.free = pool[kFree]
+    this.pending = pool[kPending]
+    this.queued = pool[kQueued]
+    this.running = pool[kRunning]
+    this.size = pool[kSize]
+  }
+}
+
+module.exports = { ClientStats, PoolStats }

lib/util/stats.js

@@ -2166,3 +2166,33 @@ test('\\n in Method', async (t) => {
     t.strictEqual(err.message, 'invalid request method')
   })
 })
+
+test('stats', async (t) => {
+  t = tspl(t, { plan: 3 })
+
+  const server = createServer({ joinDuplicateHeaders: true }, (req, res) => {
+    t.strictEqual('/', req.url)
+    t.strictEqual('GET', req.method)
+    t.strictEqual(`localhost:${server.address().port}`, req.headers.host)
+    res.setHeader('Content-Type', 'text/plain')
+    res.end('hello')
+  })
+  after(() => server.close())
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    after(() => client.close())
+
+    client.request({
+      path: '/',
+      method: 'GET'
+    }, (err, data) => {
+      t.ifError(err)
+      t.strictEqual(client.stats.connected, true)
+      t.strictEqual(client.stats.pending, 1)
+      t.strictEqual(client.stats.running, 1)
+    })
+  })
+
+  await t.completed
+})

test/client.js

@@ -808,3 +808,40 @@ test('the dispatcher is truly global', t => {
   assert.ok(require.resolve('../../index.js') in require.cache)
   assert.strictEqual(agent, undiciFresh.getGlobalDispatcher())
 })
+
+test('stats', async t => {
+  const p = tspl(t, { plan: 7 })
+  const wanted = 'payload'
+
+  const server = http.createServer({ joinDuplicateHeaders: true }, (req, res) => {
+    p.strictEqual('/', req.url)
+    p.strictEqual('GET', req.method)
+    res.end(wanted)
+  })
+
+  t.after(closeServerAsPromise(server))
+
+  const dispatcher = new Agent({
+    connect: {
+      servername: 'agent1'
+    }
+  })
+
+  server.listen(0, () => {
+    request(`http://localhost:${server.address().port}`, { dispatcher })
+      .then(({ statusCode, headers, body }) => {
+        p.strictEqual(statusCode, 200)
+        const originForStats = `http://localhost:${server.address().port}`
+        const agentStats = dispatcher.stats[originForStats]
+        p.strictEqual(agentStats.connected, 1)
+        p.strictEqual(agentStats.pending, 0)
+        p.strictEqual(agentStats.running, 0)
+        p.strictEqual(agentStats.size, 0)
+      })
+      .catch(err => {
+        p.fail(err)
+      })
+  })
+
+  await p.completed
+})

test/node-test/agent.js

@@ -1150,3 +1150,35 @@ test('pool destroy fails queued requests', async (t) => {
   })
   await t.completed
 })
+
+test('stats', async (t) => {
+  t = tspl(t, { plan: 11 })
+
+  const server = createServer({ joinDuplicateHeaders: true }, (req, res) => {
+    t.strictEqual('/', req.url)
+    t.strictEqual('GET', req.method)
+    res.setHeader('content-type', 'text/plain')
+    res.end('hello')
+  })
+  after(() => server.close())
+
+  server.listen(0, async () => {
+    const client = new Pool(`http://localhost:${server.address().port}`)
+    after(() => client.destroy())
+
+    t.strictEqual(client[kUrl].origin, `http://localhost:${server.address().port}`)
+
+    client.request({ path: '/', method: 'GET' }, (err, { statusCode, headers, body }) => {
+      t.ifError(err)
+      t.strictEqual(statusCode, 200)
+      t.strictEqual(client.stats.connected, 1)
+      t.strictEqual(client.stats.free, 0)
+      t.strictEqual(client.stats.pending, 0)
+      t.strictEqual(client.stats.queued, 0)
+      t.strictEqual(client.stats.running, 1)
+      t.strictEqual(client.stats.size, 1)
+    })
+  })
+
+  await t.completed
+})