From 22f37a4b7b7416b19e4f968cdb7ea27ecb391661 Mon Sep 17 00:00:00 2001 From: amasen02 Date: Fri, 4 Sep 2026 15:21:41 +0530 Subject: [PATCH 1/4] feat: add getClusterMetricsAsJSON to cluster registries (#470) Signed-off-by: amasen02 --- README.md | 6 ++ index.d.ts | 38 +++++++++++++ lib/cluster.js | 47 ++++++++++++---- test/clusterTest.js | 132 ++++++++++++++++++++++++++++++++++++++++++++ test/typescript.ts | 19 +++++++ 5 files changed, 230 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index fdff3b7d..7da9a216 100644 --- a/README.md +++ b/README.md @@ -517,6 +517,12 @@ register }); ``` +To retrieve aggregated cluster metrics as JSON objects, use `await register.getClusterMetricsAsJSON()`: + +```js +const metricsJson = await register.getClusterMetricsAsJSON(); +``` + ### Pushgateway It is possible to push metrics via a diff --git a/index.d.ts b/index.d.ts index 6e60e8bf..7f9bd6d0 100644 --- a/index.d.ts +++ b/index.d.ts @@ -194,6 +194,25 @@ export class ClusterRegistry< */ clusterMetrics(): Promise; + /** + * Gets aggregated metrics as objects for all workers. + * @returns {Promise>[]>} Promise that resolves with the aggregated + * metrics as objects. + */ + getClusterMetricsAsJSON(): Promise< + MetricObjectWithValues>[] + >; + + /** + * Gets aggregated metrics as objects for all workers. + * @param aggregator Filter by aggregator type + * @returns {Promise>[]>} Promise that resolves with the aggregated + * metrics as objects. + */ + getClusterMetricsAsJSON( + aggregator: string, + ): Promise>[]>; + /** * Sets the registry or registries to be aggregated. Call from workers to * use a registry/registries other than the default global registry. @@ -260,6 +279,25 @@ export class AggregatorRegistry< */ clusterMetrics(): Promise; + /** + * Gets aggregated metrics as objects for all workers. + * @returns {Promise>[]>} Promise that resolves with the aggregated + * metrics as objects. + */ + getClusterMetricsAsJSON(): Promise< + MetricObjectWithValues>[] + >; + + /** + * Gets aggregated metrics as objects for all workers. + * @param aggregator Filter by aggregator type + * @returns {Promise>[]>} Promise that resolves with the aggregated + * metrics as objects. + */ + getClusterMetricsAsJSON( + aggregator: string, + ): Promise>[]>; + /** * Orderly shutdown of the registry. * diff --git a/lib/cluster.js b/lib/cluster.js index b3b04004..be452086 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -15,8 +15,8 @@ 'use strict'; /** - * Extends the Registry class with a `clusterMetrics` method that returns - * aggregated metrics for all workers. + * Extends the Registry class with `clusterMetrics` and `getClusterMetricsAsJSON` + * methods that return aggregated metrics for all workers. * * In cluster workers, listens for and responds to requests for metrics by the * cluster master. @@ -60,12 +60,11 @@ class AggregatorRegistry extends Registry { } /** - * Gets aggregated metrics for all workers. The optional callback and - * returned Promise resolve with the same value; either may be used. - * @returns {Promise} Promise that resolves with the aggregated - * metrics. + * Executes an aggregated collection across all workers and transforms the resulting Registry. + * @param {Function} transform - mapper function called with the aggregated Registry + * @returns {Promise} */ - async clusterMetrics() { + async #aggregateMetrics(transform) { const requestId = requestCtr++; const orderedWorkers = [...workers.values()] .filter(worker => worker.isConnected()) @@ -91,7 +90,7 @@ class AggregatorRegistry extends Registry { const request = { responseHandlers, promise: waitFor( - this.#gather(requestId, metricSnapshot, responsePromises), + this.#gather(requestId, metricSnapshot, responsePromises, transform), 5_000, ), }; @@ -128,15 +127,39 @@ class AggregatorRegistry extends Registry { * @param requestId {number} * @param {any[]} historical - Previously collected values * @param promises {Promise[]} - * @returns {Promise} + * @param {Function} transform + * @returns {Promise} */ - async #gather(requestId, historical, promises) { + async #gather(requestId, historical, promises, transform) { const responses = await Promise.all(promises); const metrics = responses.flatMap(response => response.metrics); - return Registry.aggregate( + const registry = Registry.aggregate( [historical, ...metrics], this.contentType, - ).metrics(); + ); + return transform(registry); + } + + /** + * Gets aggregated metrics for all workers. The optional callback and + * returned Promise resolve with the same value; either may be used. + * @returns {Promise} Promise that resolves with the aggregated + * metrics. + */ + async clusterMetrics() { + return this.#aggregateMetrics(registry => registry.metrics()); + } + + /** + * Gets aggregated metrics as JSON objects for all workers. + * @param {string} [aggregator] - filter by aggregator type, typically used for sum. + * @returns {Promise<*[]>} Promise that resolves with the aggregated + * metrics as objects. + */ + async getClusterMetricsAsJSON(aggregator) { + return this.#aggregateMetrics(registry => + registry.getMetricsAsJSON(aggregator), + ); } get contentType() { diff --git a/test/clusterTest.js b/test/clusterTest.js index c475266c..202b5c8f 100644 --- a/test/clusterTest.js +++ b/test/clusterTest.js @@ -226,6 +226,138 @@ describe.each([ }); }); + describe('aggregatorRegistry.getClusterMetricsAsJSON()', () => { + let AggregatorRegistry; + let listener; + let discovery; + + beforeEach(() => { + jest.resetModules(); + AggregatorRegistry = require('../lib/cluster'); + + discovery = new Promise(resolve => { + listener = message => { + resolve(message); + }; + + cluster.on('message', listener); + }); + }); + + afterEach(() => { + cluster.off('message', listener); + jest.restoreAllMocks(); + }); + + it('returns empty array if there are no cluster workers and no primary metrics', async () => { + const ar = new AggregatorRegistry(regType); + const metrics = await ar.getClusterMetricsAsJSON(); + expect(metrics).toEqual([]); + }); + + it('aggregates worker responses as JSON objects', async () => { + const originalWorkers = cluster.workers; + const registry = new AggregatorRegistry(regType); + const workers = Object.fromEntries( + [1, 2, 3].map(id => [ + id, + { + id, + isConnected: () => true, + send: jest.fn(), + }, + ]), + ); + cluster.workers = workers; + + Object.values(workers).forEach(worker => { + cluster.emit('message', worker, { type: ANNOUNCEMENT }); + }); + + try { + await discovery; + + const result = registry.getClusterMetricsAsJSON(); + for (const [id, value] of [ + [3, 0.3437699], + [1, 0.5848208], + [2, 0.5479198], + ]) { + cluster.emit('message', workers[id], { + type: GET_METRICS_RES, + requestId: 0, + metrics: [[metric(value)]], + }); + } + + const output = await result; + expect(output).toEqual([ + { + aggregator: 'sum', + help: 'test metric', + name: 'test_metric', + type: 'gauge', + values: [{ labels: {}, value: 1.4765105 }], + }, + ]); + } finally { + Object.values(workers).forEach(worker => { + cluster.emit('disconnect', worker); + }); + cluster.workers = originalWorkers; + } + }); + + it('supports aggregator parameter filtering', async () => { + const originalWorkers = cluster.workers; + const registry = new AggregatorRegistry(regType); + const workers = Object.fromEntries( + [1].map(id => [ + id, + { + id, + isConnected: () => true, + send: jest.fn(), + }, + ]), + ); + cluster.workers = workers; + + Object.values(workers).forEach(worker => { + cluster.emit('message', worker, { type: ANNOUNCEMENT }); + }); + + try { + await discovery; + + const result = registry.getClusterMetricsAsJSON('sum'); + cluster.emit('message', workers[1], { + type: GET_METRICS_RES, + requestId: 0, + metrics: [[metric(42)]], + }); + + const output = await result; + expect(output).toHaveLength(1); + expect(output[0].name).toBe('test_metric'); + + const omitResult = registry.getClusterMetricsAsJSON('omit'); + cluster.emit('message', workers[1], { + type: GET_METRICS_RES, + requestId: 1, + metrics: [[metric(42)]], + }); + const omitOutput = await omitResult; + expect(omitOutput).toEqual([]); + } finally { + Object.values(workers).forEach(worker => { + cluster.emit('disconnect', worker); + }); + cluster.workers = originalWorkers; + } + }); + }); + describe('shutdown()', () => { let AggregatorRegistry; let listener; diff --git a/test/typescript.ts b/test/typescript.ts index f55a15f4..f79c9914 100644 --- a/test/typescript.ts +++ b/test/typescript.ts @@ -13,6 +13,8 @@ // limitations under the License. import { + AggregatorRegistry, + ClusterRegistry, Counter, Pushgateway, Registry, @@ -107,3 +109,20 @@ async function metricTypeMatchesRuntimeStrings() { void MetricType.Counter; } void metricTypeMatchesRuntimeStrings; + +async function clusterMetricsAsJSONTypeCheck() { + const clusterRegistry = new ClusterRegistry(); + const json: MetricObjectWithValues>[] = + await clusterRegistry.getClusterMetricsAsJSON(); + void json; + + const filteredJson: MetricObjectWithValues>[] = + await clusterRegistry.getClusterMetricsAsJSON('sum'); + void filteredJson; + + const aggregatorRegistry = new AggregatorRegistry(); + const aggJson: MetricObjectWithValues>[] = + await aggregatorRegistry.getClusterMetricsAsJSON(); + void aggJson; +} +void clusterMetricsAsJSONTypeCheck; From 4d4424b9b8933e2e0af95a6b47d4ca5fc788290a Mon Sep 17 00:00:00 2001 From: amasen02 Date: Wed, 9 Sep 2026 00:05:30 +0530 Subject: [PATCH 2/4] refactor(cluster): return populated Registry from #gather directly (#470) * Remove transform parameter and indirection from #gather, returning populated Registry directly. * Address review feedback from @jdmarshall on #849. Signed-off-by: amasen02 --- lib/cluster.js | 25 +++++++++---------------- 1 file changed, 9 insertions(+), 16 deletions(-) diff --git a/lib/cluster.js b/lib/cluster.js index be452086..f6fb3eac 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -60,11 +60,11 @@ class AggregatorRegistry extends Registry { } /** - * Executes an aggregated collection across all workers and transforms the resulting Registry. - * @param {Function} transform - mapper function called with the aggregated Registry + * Executes an aggregated collection across all workers. + * @param {Function} fn - called with the aggregated Registry * @returns {Promise} */ - async #aggregateMetrics(transform) { + async #collect(fn) { const requestId = requestCtr++; const orderedWorkers = [...workers.values()] .filter(worker => worker.isConnected()) @@ -90,7 +90,7 @@ class AggregatorRegistry extends Registry { const request = { responseHandlers, promise: waitFor( - this.#gather(requestId, metricSnapshot, responsePromises, transform), + this.#gather(requestId, metricSnapshot, responsePromises).then(fn), 5_000, ), }; @@ -127,17 +127,12 @@ class AggregatorRegistry extends Registry { * @param requestId {number} * @param {any[]} historical - Previously collected values * @param promises {Promise[]} - * @param {Function} transform - * @returns {Promise} + * @returns {Promise} */ - async #gather(requestId, historical, promises, transform) { + async #gather(requestId, historical, promises) { const responses = await Promise.all(promises); const metrics = responses.flatMap(response => response.metrics); - const registry = Registry.aggregate( - [historical, ...metrics], - this.contentType, - ); - return transform(registry); + return Registry.aggregate([historical, ...metrics], this.contentType); } /** @@ -147,7 +142,7 @@ class AggregatorRegistry extends Registry { * metrics. */ async clusterMetrics() { - return this.#aggregateMetrics(registry => registry.metrics()); + return this.#collect(registry => registry.metrics()); } /** @@ -157,9 +152,7 @@ class AggregatorRegistry extends Registry { * metrics as objects. */ async getClusterMetricsAsJSON(aggregator) { - return this.#aggregateMetrics(registry => - registry.getMetricsAsJSON(aggregator), - ); + return this.#collect(registry => registry.getMetricsAsJSON(aggregator)); } get contentType() { From 7fffb06c887015bfd01f854f3782e937e23b3c39 Mon Sep 17 00:00:00 2001 From: amasen02 Date: Wed, 9 Sep 2026 13:41:44 +0530 Subject: [PATCH 3/4] refactor: drop fn indirection from #collect; callers operate on the returned Registry directly Signed-off-by: amasen02 --- lib/cluster.js | 13 +++++++------ test/clusterTest.js | 11 ++++++++++- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/lib/cluster.js b/lib/cluster.js index f6fb3eac..a0321084 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -61,10 +61,9 @@ class AggregatorRegistry extends Registry { /** * Executes an aggregated collection across all workers. - * @param {Function} fn - called with the aggregated Registry - * @returns {Promise} + * @returns {Promise} the aggregated Registry */ - async #collect(fn) { + async #collect() { const requestId = requestCtr++; const orderedWorkers = [...workers.values()] .filter(worker => worker.isConnected()) @@ -90,7 +89,7 @@ class AggregatorRegistry extends Registry { const request = { responseHandlers, promise: waitFor( - this.#gather(requestId, metricSnapshot, responsePromises).then(fn), + this.#gather(requestId, metricSnapshot, responsePromises), 5_000, ), }; @@ -142,7 +141,8 @@ class AggregatorRegistry extends Registry { * metrics. */ async clusterMetrics() { - return this.#collect(registry => registry.metrics()); + const registry = await this.#collect(); + return registry.metrics(); } /** @@ -152,7 +152,8 @@ class AggregatorRegistry extends Registry { * metrics as objects. */ async getClusterMetricsAsJSON(aggregator) { - return this.#collect(registry => registry.getMetricsAsJSON(aggregator)); + const registry = await this.#collect(); + return registry.getMetricsAsJSON(aggregator); } get contentType() { diff --git a/test/clusterTest.js b/test/clusterTest.js index 202b5c8f..adbb1d17 100644 --- a/test/clusterTest.js +++ b/test/clusterTest.js @@ -408,6 +408,15 @@ describe.each([ const results = []; const promise = registry.clusterMetrics().then(() => results.push(1)); const shutdown = registry.shutdown().then(() => results.push(2)); + let shutdownResolved = false; + shutdown.then(() => { + shutdownResolved = true; + }); + + // Drain the microtask queue: shutdown must still be waiting for + // the outstanding worker response at this point. + await new Promise(resolve => setImmediate(resolve)); + expect(shutdownResolved).toBe(false); cluster.emit('message', worker, { type: GET_METRICS_RES, @@ -417,7 +426,7 @@ describe.each([ await Promise.all([promise, shutdown]); - expect(results).toEqual([1, 2]); + expect(results.sort()).toEqual([1, 2]); } finally { cluster.emit('disconnect', worker); cluster.workers = originalWorkers; From ae3c0eaf5f1007aa6b47f54b67acfe8b4a4d7514 Mon Sep 17 00:00:00 2001 From: amasen02 Date: Fri, 11 Sep 2026 17:39:29 +0530 Subject: [PATCH 4/4] test(cluster): assert aggregate completes prior to shutdown without overlap Signed-off-by: amasen02 --- test/clusterTest.js | 29 +++++++++++++++++++++-------- 1 file changed, 21 insertions(+), 8 deletions(-) diff --git a/test/clusterTest.js b/test/clusterTest.js index adbb1d17..75f7930e 100644 --- a/test/clusterTest.js +++ b/test/clusterTest.js @@ -405,18 +405,29 @@ describe.each([ try { await discovery; - const results = []; - const promise = registry.clusterMetrics().then(() => results.push(1)); - const shutdown = registry.shutdown().then(() => results.push(2)); - let shutdownResolved = false; - shutdown.then(() => { - shutdownResolved = true; + const BaseRegistry = require('../lib/registry'); + let aggregateCompleted = false; + const origAggregate = BaseRegistry.aggregate; + const aggregateSpy = jest + .spyOn(BaseRegistry, 'aggregate') + .mockImplementation((...args) => { + const res = origAggregate.apply(BaseRegistry, args); + aggregateCompleted = true; + return res; + }); + + let shutdownCompleted = false; + const promise = registry.clusterMetrics(); + const shutdown = registry.shutdown().then(() => { + shutdownCompleted = true; + expect(aggregateCompleted).toBe(true); }); // Drain the microtask queue: shutdown must still be waiting for // the outstanding worker response at this point. await new Promise(resolve => setImmediate(resolve)); - expect(shutdownResolved).toBe(false); + expect(shutdownCompleted).toBe(false); + expect(aggregateCompleted).toBe(false); cluster.emit('message', worker, { type: GET_METRICS_RES, @@ -426,7 +437,9 @@ describe.each([ await Promise.all([promise, shutdown]); - expect(results.sort()).toEqual([1, 2]); + expect(shutdownCompleted).toBe(true); + expect(aggregateCompleted).toBe(true); + expect(aggregateSpy).toHaveBeenCalledTimes(1); } finally { cluster.emit('disconnect', worker); cluster.workers = originalWorkers;