fix(cluster): aggregate metrics in a deterministic order

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
 

lib/cluster.js

@@ -53,13 +53,17 @@ 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;
 			function done(err, result) {
 				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);
 				}
 			}
 		});

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,30 +134,48 @@ 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,
 				threadId,
 				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);
 		});
 	}
 

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', () => {

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();
+			}
 		});
 	});
 });