diff --git a/CHANGELOG.md b/CHANGELOG.md index 41b3c037..d8cb5f0e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,7 @@ This release marks our first release under the Prometheus umbrella. - chore: update faceoff to 1.1 - perf: Stat aggregation uses similar strategy to collection. 60% faster aggregation - chore: Add copyright license headers and test +- Make cluster and worker-thread metric aggregation order deterministic ### Added diff --git a/lib/cluster.js b/lib/cluster.js index 697482a3..4e6f08b8 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -53,6 +53,9 @@ class AggregatorRegistry extends Registry { */ clusterMetrics() { const requestId = requestCtr++; + const workers = Object.values(cluster().workers) + .filter(worker => worker.isConnected()) + .sort((left, right) => left.id - right.id); return new Promise((resolve, reject) => { let settled = false; @@ -60,6 +63,7 @@ class AggregatorRegistry extends Registry { if (settled) return; settled = true; + clearTimeout(request.errorTimeout); requests.delete(requestId); if (err !== undefined) { @@ -69,9 +73,9 @@ class AggregatorRegistry extends Registry { } } + const responseHandlers = new Map(); const request = { - responses: [], - pending: 0, + responseHandlers, done, errorTimeout: setTimeout(() => { const err = new Error('Operation timed out.'); @@ -85,20 +89,26 @@ class AggregatorRegistry extends Registry { requestId, }; - for (const id in cluster().workers) { - // If the worker exits abruptly, it may still be in the workers - // list but not able to communicate. - if (cluster().workers[id].isConnected()) { - cluster().workers[id].send(message); - request.pending++; - } - } - - if (request.pending === 0) { + if (workers.length === 0) { // No workers were up - clearTimeout(request.errorTimeout); process.nextTick(() => done(undefined, '')); + return; } + + const responsePromises = workers.map( + worker => + new Promise((resolveResponse, rejectResponse) => { + responseHandlers.set(worker.id, { + resolve: resolveResponse, + reject: rejectResponse, + }); + worker.send(message); + }), + ); + + Promise.all(responsePromises) + .then(metrics => Registry.aggregate(metrics.flat()).metrics()) + .then(result => done(undefined, result), done); }); } @@ -161,21 +171,16 @@ function addListeners() { return; } - if (message.error) { - request.done(new Error(message.error)); + const response = request.responseHandlers.get(worker.id); + if (response === undefined) { return; } + request.responseHandlers.delete(worker.id); - message.metrics.forEach(metric => request.responses.push(metric)); - request.pending--; - - if (request.pending === 0) { - // finalize - clearTimeout(request.errorTimeout); - - const registry = Registry.aggregate(request.responses); - const promString = registry.metrics(); - request.done(undefined, promString); + if (message.error) { + response.reject(new Error(message.error)); + } else { + response.resolve(message.metrics); } } }); diff --git a/lib/worker.js b/lib/worker.js index 48c1409f..8f4e3b3a 100644 --- a/lib/worker.js +++ b/lib/worker.js @@ -88,21 +88,19 @@ class WorkerRegistry extends Registry { return; } - if (message.error) { - request.done(new Error(message.error)); + const response = request.responseHandlers.get(name); + if (response === undefined) { return; } + request.responseHandlers.delete(name); - message.metrics.forEach(metric => request.responses.push(metric)); - request.pending--; - - if (request.pending === 0) { - // finalize - clearTimeout(request.errorTimeout); - - const registry = Registry.aggregate(request.responses); - const promString = registry.metrics(); - request.done(undefined, promString); + if (message.error) { + response.reject(new Error(message.error)); + } else { + response.resolve({ + threadId: message.threadId, + metrics: message.metrics, + }); } } }); @@ -126,6 +124,7 @@ class WorkerRegistry extends Registry { if (settled) return; settled = true; + clearTimeout(request.errorTimeout); requests.delete(requestId); if (err !== undefined) { @@ -135,18 +134,27 @@ class WorkerRegistry extends Registry { } } + const responseHandlers = new Map(); const request = { - responses: [], - pending: this.channels.size, + responseHandlers, done, errorTimeout: setTimeout(() => { const err = new Error( - `Operation timed out. ${request.pending} outstanding responses.`, + `Operation timed out. ${request.responseHandlers.size} outstanding responses.`, ); request.done(err); }, 5_000), }; requests.set(requestId, request); + const responsePromises = [...this.channels.keys()].map( + name => + new Promise((resolveResponse, rejectResponse) => { + responseHandlers.set(name, { + resolve: resolveResponse, + reject: rejectResponse, + }); + }), + ); ANNOUNCEMENT_CHANNEL.postMessage({ type: GET_METRICS_REQ, @@ -154,11 +162,20 @@ class WorkerRegistry extends Registry { requestId, }); - if (request.pending === 0) { + if (responsePromises.length === 0) { // No workers were up - clearTimeout(request.errorTimeout); process.nextTick(() => done(undefined, '')); + return; } + + Promise.all(responsePromises) + .then(responses => + responses + .sort((left, right) => left.threadId - right.threadId) + .flatMap(response => response.metrics), + ) + .then(metrics => Registry.aggregate(metrics).metrics()) + .then(result => done(undefined, result), done); }); } diff --git a/test/clusterTest.js b/test/clusterTest.js index bd2141bc..70dcd081 100644 --- a/test/clusterTest.js +++ b/test/clusterTest.js @@ -18,6 +18,18 @@ const cluster = require('cluster'); const process = require('process'); const Registry = require('../lib/cluster'); +const GET_METRICS_RES = '@prometheus/client:getMetricsRes'; + +function metric(value) { + return { + help: 'test metric', + name: 'test_metric', + type: 'gauge', + values: [{ labels: {}, value }], + aggregator: 'sum', + }; +} + describe.each([ ['Prometheus', Registry.PROMETHEUS_CONTENT_TYPE], ['OpenMetrics', Registry.OPENMETRICS_CONTENT_TYPE], @@ -61,6 +73,43 @@ describe.each([ const metrics = await ar.clusterMetrics(); expect(metrics).toEqual(''); }); + + it('aggregates worker responses in worker id order', async () => { + const originalWorkers = cluster.workers; + const workers = Object.fromEntries( + [1, 2, 3].map(id => [ + id, + { + id, + isConnected: () => true, + send: jest.fn(), + }, + ]), + ); + cluster.workers = workers; + + try { + const registry = new Registry(regType); + const result = registry.clusterMetrics(); + const requestId = workers[1].send.mock.calls[0][0].requestId; + + for (const [id, value] of [ + [3, 0.3437699], + [1, 0.5848208], + [2, 0.5479198], + ]) { + cluster.emit('message', workers[id], { + type: GET_METRICS_RES, + requestId, + metrics: [[metric(value)]], + }); + } + + await expect(result).resolves.toContain('test_metric 1.4765105'); + } finally { + cluster.workers = originalWorkers; + } + }); }); describe('message handling', () => { diff --git a/test/workerTest.js b/test/workerTest.js index 9f28d467..76c3ad45 100644 --- a/test/workerTest.js +++ b/test/workerTest.js @@ -15,8 +15,23 @@ 'use strict'; const { EventEmitter } = require('events'); +const { setTimeout: delay } = require('timers/promises'); +const { BroadcastChannel } = require('worker_threads'); const Registry = require('../lib/worker'); +const GET_METRICS_REQ = '@prometheus/client:getMetricsReq'; +const GET_METRICS_RES = '@prometheus/client:getMetricsRes'; + +function metric(value) { + return { + help: 'test metric', + name: 'test_metric', + type: 'gauge', + values: [{ labels: {}, value }], + aggregator: 'sum', + }; +} + describe.each([ ['Prometheus', Registry.PROMETHEUS_CONTENT_TYPE], ['OpenMetrics', Registry.OPENMETRICS_CONTENT_TYPE], @@ -32,6 +47,55 @@ describe.each([ const metrics = await registry.workerMetrics(); expect(metrics).toEqual(''); }); + + if (tag === 'Prometheus') + it('aggregates worker responses in thread id order', async () => { + const registry = new Registry(regType); + const announcementChannel = new BroadcastChannel( + '@prometheus/client:announce', + ); + const responders = [1, 2, 3].map(threadId => { + const name = `@prometheus/client:test-worker:${threadId}`; + registry.addWorker(name); + return { + threadId, + channel: new BroadcastChannel(name), + }; + }); + + let finishSendingResponses; + const responsesSent = new Promise(resolve => { + finishSendingResponses = resolve; + }); + announcementChannel.addEventListener('message', async event => { + if (event.data.type !== GET_METRICS_REQ) return; + + for (const [threadId, value] of [ + [3, 0.3437699], + [1, 0.5848208], + [2, 0.5479198], + ]) { + responders[threadId - 1].channel.postMessage({ + type: GET_METRICS_RES, + requestId: event.data.requestId, + threadId, + metrics: [[metric(value)]], + }); + await delay(5); + } + finishSendingResponses(); + }); + + try { + const result = await registry.workerMetrics(); + await responsesSent; + expect(result).toContain('test_metric 1.4765105'); + } finally { + announcementChannel.close(); + for (const responder of responders) responder.channel.close(); + for (const channel of registry.channels.values()) channel.close(); + } + }); }); describe('message handling', () => { @@ -52,7 +116,11 @@ describe.each([ requestId: -3, }; - expect(() => emitter.emit('message', unexpected)).not.toThrow(); + try { + expect(() => emitter.emit('message', unexpected)).not.toThrow(); + } finally { + for (const channel of registry.channels.values()) channel.close(); + } }); }); });