feat: add getClusterMetricsAsJSON to cluster registries (#470)

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

index.d.ts

@@ -194,6 +194,25 @@ export class ClusterRegistry<
 	 */
 	clusterMetrics(): Promise<string>;
 
+	/**
+	 * Gets aggregated metrics as objects for all workers.
+	 * @returns {Promise<MetricObjectWithValues<MetricValue<string>>[]>} Promise that resolves with the aggregated
+	 * metrics as objects.
+	 */
+	getClusterMetricsAsJSON(): Promise<
+		MetricObjectWithValues<MetricValue<string>>[]
+	>;
+
+	/**
+	 * Gets aggregated metrics as objects for all workers.
+	 * @param aggregator Filter by aggregator type
+	 * @returns {Promise<MetricObjectWithValues<MetricValue<string>>[]>} Promise that resolves with the aggregated
+	 * metrics as objects.
+	 */
+	getClusterMetricsAsJSON(
+		aggregator: string,
+	): Promise<MetricObjectWithValues<MetricValue<string>>[]>;
+
 	/**
 	 * 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<string>;
 
+	/**
+	 * Gets aggregated metrics as objects for all workers.
+	 * @returns {Promise<MetricObjectWithValues<MetricValue<string>>[]>} Promise that resolves with the aggregated
+	 * metrics as objects.
+	 */
+	getClusterMetricsAsJSON(): Promise<
+		MetricObjectWithValues<MetricValue<string>>[]
+	>;
+
+	/**
+	 * Gets aggregated metrics as objects for all workers.
+	 * @param aggregator Filter by aggregator type
+	 * @returns {Promise<MetricObjectWithValues<MetricValue<string>>[]>} Promise that resolves with the aggregated
+	 * metrics as objects.
+	 */
+	getClusterMetricsAsJSON(
+		aggregator: string,
+	): Promise<MetricObjectWithValues<MetricValue<string>>[]>;
+
 	/**
 	 * Orderly shutdown of the registry.
 	 *

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,10 @@ 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<string>} Promise that resolves with the aggregated
-	 *   metrics.
+	 * Executes an aggregated collection across all workers.
+	 * @returns {Promise<Registry>} the aggregated Registry
 	 */
-	async clusterMetrics() {
+	async #collect() {
 		const requestId = requestCtr++;
 		const orderedWorkers = [...workers.values()]
 			.filter(worker => worker.isConnected())
@@ -128,15 +126,34 @@ class AggregatorRegistry extends Registry {
 	 * @param requestId {number}
 	 * @param {any[]} historical - Previously collected values
 	 * @param promises {Promise[]}
-	 * @returns {Promise<string>}
+	 * @returns {Promise<Registry>}
 	 */
 	async #gather(requestId, historical, promises) {
 		const responses = await Promise.all(promises);
 		const metrics = responses.flatMap(response => response.metrics);
-		return Registry.aggregate(
-			[historical, ...metrics],
-			this.contentType,
-		).metrics();
+		return Registry.aggregate([historical, ...metrics], this.contentType);
+	}
+
+	/**
+	 * Gets aggregated metrics for all workers. The optional callback and
+	 * returned Promise resolve with the same value; either may be used.
+	 * @returns {Promise<string>} Promise that resolves with the aggregated
+	 *   metrics.
+	 */
+	async clusterMetrics() {
+		const registry = await this.#collect();
+		return 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) {
+		const registry = await this.#collect();
+		return registry.getMetricsAsJSON(aggregator);
 	}
 
 	get contentType() {

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;
@@ -273,9 +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));
+				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(shutdownCompleted).toBe(false);
+				expect(aggregateCompleted).toBe(false);
 
 				cluster.emit('message', worker, {
 					type: GET_METRICS_RES,
@@ -285,7 +437,9 @@ describe.each([
 
 				await Promise.all([promise, shutdown]);
 
-				expect(results).toEqual([1, 2]);
+				expect(shutdownCompleted).toBe(true);
+				expect(aggregateCompleted).toBe(true);
+				expect(aggregateSpy).toHaveBeenCalledTimes(1);
 			} finally {
 				cluster.emit('disconnect', worker);
 				cluster.workers = originalWorkers;

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<MetricValue<string>>[] =
+		await clusterRegistry.getClusterMetricsAsJSON();
+	void json;
+
+	const filteredJson: MetricObjectWithValues<MetricValue<string>>[] =
+		await clusterRegistry.getClusterMetricsAsJSON('sum');
+	void filteredJson;
+
+	const aggregatorRegistry = new AggregatorRegistry();
+	const aggJson: MetricObjectWithValues<MetricValue<string>>[] =
+		await aggregatorRegistry.getClusterMetricsAsJSON();
+	void aggJson;
+}
+void clusterMetricsAsJSONTypeCheck;