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..a0321084 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,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} Promise that resolves with the aggregated - * metrics. + * Executes an aggregated collection across all workers. + * @returns {Promise} 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} + * @returns {Promise} */ 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} 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() { diff --git a/test/clusterTest.js b/test/clusterTest.js index c475266c..75f7930e 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; @@ -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; 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;