From 0162ab553f8beb750bf4241f116b89257aef1078 Mon Sep 17 00:00:00 2001 From: Aysajan Eziz Date: Mon, 13 Jul 2026 18:31:38 -0400 Subject: [PATCH 1/2] fix(cluster): aggregate metrics in a deterministic order Collect cluster responses in numeric worker-ID order with per-worker promises, and worker-thread responses in numeric thread-ID order. Add reverse-arrival regression coverage for both paths. Fixes #539 Signed-off-by: Aysajan Eziz --- CHANGELOG.md | 1 + lib/cluster.js | 61 +++++++++++++++++++++++---------------- lib/worker.js | 16 +++++++---- test/clusterTest.js | 49 +++++++++++++++++++++++++++++++ test/workerTest.js | 70 ++++++++++++++++++++++++++++++++++++++++++++- 5 files changed, 165 insertions(+), 32 deletions(-) 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..65e18ae6 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -39,6 +39,16 @@ let requestCtr = 0; // Concurrency control let listenersAdded = false; const requests = new Map(); // Pending requests for workers' local metrics. +function deferred() { + let resolve; + let reject; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + class AggregatorRegistry extends Registry { constructor(regContentType = Registry.PROMETHEUS_CONTENT_TYPE) { super(regContentType); @@ -53,6 +63,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 +73,7 @@ class AggregatorRegistry extends Registry { if (settled) return; settled = true; + clearTimeout(request.errorTimeout); requests.delete(requestId); if (err !== undefined) { @@ -69,9 +83,14 @@ class AggregatorRegistry extends Registry { } } + const responses = new Map(); + const responsePromises = workers.map(worker => { + const response = deferred(); + responses.set(worker.id, response); + return response.promise; + }); const request = { - responses: [], - pending: 0, + responses, done, errorTimeout: setTimeout(() => { const err = new Error('Operation timed out.'); @@ -85,20 +104,17 @@ 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; } + + for (const worker of workers) worker.send(message); + + Promise.all(responsePromises) + .then(metrics => Registry.aggregate(metrics.flat()).metrics()) + .then(result => done(undefined, result), done); }); } @@ -161,21 +177,16 @@ function addListeners() { return; } - if (message.error) { - request.done(new Error(message.error)); + const response = request.responses.get(worker.id); + if (response === undefined) { return; } + request.responses.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..2281b94f 100644 --- a/lib/worker.js +++ b/lib/worker.js @@ -93,14 +93,18 @@ class WorkerRegistry extends Registry { return; } - message.metrics.forEach(metric => request.responses.push(metric)); + if (request.responses.has(message.threadId)) { + return; + } + request.responses.set(message.threadId, message.metrics); request.pending--; if (request.pending === 0) { // finalize - clearTimeout(request.errorTimeout); - - const registry = Registry.aggregate(request.responses); + const responses = [...request.responses.entries()] + .sort(([left], [right]) => left - right) + .flatMap(([, metrics]) => metrics); + const registry = Registry.aggregate(responses); const promString = registry.metrics(); request.done(undefined, promString); } @@ -126,6 +130,7 @@ class WorkerRegistry extends Registry { if (settled) return; settled = true; + clearTimeout(request.errorTimeout); requests.delete(requestId); if (err !== undefined) { @@ -136,7 +141,7 @@ class WorkerRegistry extends Registry { } const request = { - responses: [], + responses: new Map(), pending: this.channels.size, done, errorTimeout: setTimeout(() => { @@ -156,7 +161,6 @@ class WorkerRegistry extends Registry { if (request.pending === 0) { // No workers were up - clearTimeout(request.errorTimeout); process.nextTick(() => done(undefined, '')); } }); 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(); + } }); }); }); From b64dc651cb3a1a22ed5e51095bbea2296d2b57ad Mon Sep 17 00:00:00 2001 From: Aysajan Eziz Date: Mon, 13 Jul 2026 19:30:15 -0400 Subject: [PATCH 2/2] refactor(cluster): align response handling Inline per-peer response promises and use the same response-handler lifecycle for cluster workers and worker threads. Signed-off-by: Aysajan Eziz --- lib/cluster.js | 34 +++++++++++++------------------- lib/worker.js | 53 +++++++++++++++++++++++++++++++------------------- 2 files changed, 47 insertions(+), 40 deletions(-) diff --git a/lib/cluster.js b/lib/cluster.js index 65e18ae6..4e6f08b8 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -39,16 +39,6 @@ let requestCtr = 0; // Concurrency control let listenersAdded = false; const requests = new Map(); // Pending requests for workers' local metrics. -function deferred() { - let resolve; - let reject; - const promise = new Promise((resolvePromise, rejectPromise) => { - resolve = resolvePromise; - reject = rejectPromise; - }); - return { promise, resolve, reject }; -} - class AggregatorRegistry extends Registry { constructor(regContentType = Registry.PROMETHEUS_CONTENT_TYPE) { super(regContentType); @@ -83,14 +73,9 @@ class AggregatorRegistry extends Registry { } } - const responses = new Map(); - const responsePromises = workers.map(worker => { - const response = deferred(); - responses.set(worker.id, response); - return response.promise; - }); + const responseHandlers = new Map(); const request = { - responses, + responseHandlers, done, errorTimeout: setTimeout(() => { const err = new Error('Operation timed out.'); @@ -110,7 +95,16 @@ class AggregatorRegistry extends Registry { return; } - for (const worker of workers) worker.send(message); + 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()) @@ -177,11 +171,11 @@ function addListeners() { return; } - const response = request.responses.get(worker.id); + const response = request.responseHandlers.get(worker.id); if (response === undefined) { return; } - request.responses.delete(worker.id); + request.responseHandlers.delete(worker.id); if (message.error) { response.reject(new Error(message.error)); diff --git a/lib/worker.js b/lib/worker.js index 2281b94f..8f4e3b3a 100644 --- a/lib/worker.js +++ b/lib/worker.js @@ -88,25 +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); - if (request.responses.has(message.threadId)) { - return; - } - request.responses.set(message.threadId, message.metrics); - request.pending--; - - if (request.pending === 0) { - // finalize - const responses = [...request.responses.entries()] - .sort(([left], [right]) => left - right) - .flatMap(([, metrics]) => metrics); - const registry = Registry.aggregate(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, + }); } } }); @@ -140,18 +134,27 @@ class WorkerRegistry extends Registry { } } + const responseHandlers = new Map(); const request = { - responses: new Map(), - 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, @@ -159,10 +162,20 @@ class WorkerRegistry extends Registry { requestId, }); - if (request.pending === 0) { + if (responsePromises.length === 0) { // No workers were up 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); }); }