feat: support aggregating metrics across workers in a Node.js worker_threads

Author: Rafal Augustyniak

example/workerThreads.js

@@ -0,0 +1,37 @@
+'use strict';
+
+const express = require('express');
+const metricsServer = express();
+const AggregatorRegistry = require('../').AggregatorRegistry;
+/* eslint-disable node/no-unsupported-features/node-builtins */
+const { Worker, isMainThread } = require('worker_threads');
+
+const aggregatorRegistry = new AggregatorRegistry();
+
+if (isMainThread) {
+	metricsServer.get('/metrics', async (req, res) => {
+		try {
+			const metrics = await aggregatorRegistry.clusterMetrics();
+			res.set('Content-Type', aggregatorRegistry.contentType);
+			res.send(metrics);
+		} catch (ex) {
+			res.statusCode = 500;
+			res.send(ex.message);
+		}
+	});
+
+	metricsServer.listen(3000);
+
+	const worker = new Worker(__filename);
+
+	aggregatorRegistry.attachWorkers([worker]);
+} else {
+	const Histogram = require('../').Histogram;
+	const h = new Histogram({
+		name: 'test_histogram',
+		help: 'Example of a histogram',
+		labelNames: ['code'],
+	});
+
+	h.labels('200').observe(Math.random());
+}

lib/cluster.js

@@ -11,6 +11,20 @@
 const Registry = require('./registry');
 const { Grouper } = require('./util');
 const { aggregators } = require('./metricAggregators');
+
+let parentPort, MessageChannel, isMainThread, Worker;
+try {
+	/* eslint-disable node/no-unsupported-features/node-builtins */
+	const worker_threads = require('worker_threads');
+
+	parentPort = worker_threads.parentPort;
+	MessageChannel = worker_threads.MessageChannel;
+	isMainThread = worker_threads.isMainThread;
+	Worker = worker_threads.Worker;
+} catch {
+	// node version is too old
+}
+
 // We need to lazy-load the 'cluster' module as some application servers -
 // namely Passenger - crash when it is imported.
 let cluster = () => {
@@ -25,6 +39,7 @@ const GET_METRICS_RES = 'prom-client:getMetricsRes';
 let registries = [Registry.globalRegistry];
 let requestCtr = 0; // Concurrency control
 let listenersAdded = false;
+const workersQueue = [];
 const requests = new Map(); // Pending requests for workers' local metrics.
 
 class AggregatorRegistry extends Registry {
@@ -33,6 +48,14 @@ class AggregatorRegistry extends Registry {
 		addListeners();
 	}
 
+	attachWorkers(workers = []) {
+		for (const worker in workers) {
+			if (worker instanceof Worker) {
+				workersQueue.push(worker);
+			}
+		}
+	}
+
 	/**
 	 * Gets aggregated metrics for all workers. The optional callback and
 	 * returned Promise resolve with the same value; either may be used.
@@ -76,6 +99,9 @@ class AggregatorRegistry extends Registry {
 				}
 			}
 
+			getWorkerThreadsMetrics(message);
+			request.pending += workersQueue.length || 0;
+
 			if (request.pending === 0) {
 				// No workers were up
 				clearTimeout(request.errorTimeout);
@@ -145,6 +171,31 @@ class AggregatorRegistry extends Registry {
 	}
 }
 
+function handleWorkerResponse(worker, message) {
+	if (message.type === GET_METRICS_RES) {
+		const request = requests.get(message.requestId);
+		request.pending += message.workerRequests || 0;
+
+		if (message.error) {
+			request.done(new Error(message.error));
+			return;
+		}
+
+		message.metrics.forEach(registry => request.responses.push(registry));
+		request.pending--;
+
+		if (request.pending === 0) {
+			// finalize
+			requests.delete(message.requestId);
+			clearTimeout(request.errorTimeout);
+
+			const registry = AggregatorRegistry.aggregate(request.responses);
+			const promString = registry.metrics();
+			request.done(null, promString);
+		}
+	}
+}
+
 /**
  * Adds event listeners for cluster aggregation. Idempotent (safe to call more
  * than once).
@@ -155,42 +206,53 @@ function addListeners() {
 	listenersAdded = true;
 
 	if (cluster().isMaster) {
-		// Listen for worker responses to requests for local metrics
-		cluster().on('message', (worker, message) => {
-			if (message.type === GET_METRICS_RES) {
-				const request = requests.get(message.requestId);
-
-				if (message.error) {
-					request.done(new Error(message.error));
-					return;
-				}
+		// Listen for cluster responses to requests for local metrics
+		cluster().on('message', handleWorkerResponse);
+	}
+}
 
-				message.metrics.forEach(registry => request.responses.push(registry));
-				request.pending--;
+function getWorkerThreadsMetrics(message) {
+	workersQueue.forEach(worker => {
+		if (worker && worker instanceof Worker) {
+			const metricsChannel = new MessageChannel();
 
-				if (request.pending === 0) {
-					// finalize
-					requests.delete(message.requestId);
-					clearTimeout(request.errorTimeout);
+			worker.postMessage(
+				{
+					...message,
+					port: metricsChannel.port1,
+				},
+				[metricsChannel.port1],
+			);
 
-					const registry = AggregatorRegistry.aggregate(request.responses);
-					const promString = registry.metrics();
-					request.done(null, promString);
+			metricsChannel.port2.on('message', response => {
+				if (response.type === GET_METRICS_RES) {
+					if (cluster().isWorker) {
+						process.send(response);
+					}
+
+					if (cluster().isMaster) {
+						handleWorkerResponse(worker, response);
+					}
+
+					metricsChannel.port2.close();
 				}
-			}
-		});
-	}
+			});
+		}
+	});
 }
 
 // Respond to master's requests for worker's local metrics.
 process.on('message', message => {
 	if (cluster().isWorker && message.type === GET_METRICS_REQ) {
+		getWorkerThreadsMetrics(message);
+
 		Promise.all(registries.map(r => r.getMetricsAsJSON()))
 			.then(metrics => {
 				process.send({
 					type: GET_METRICS_RES,
 					requestId: message.requestId,
 					metrics,
+					workerRequests: workersQueue.length,
 				});
 			})
 			.catch(error => {
@@ -203,4 +265,27 @@ process.on('message', message => {
 	}
 });
 
+// Respond to master's request for worker_threads worker local metrics
+if (!isMainThread) {
+	parentPort.on('message', ({ type, requestId, port } = {}) => {
+		if (type === GET_METRICS_REQ) {
+			Promise.all(registries.map(r => r.getMetricsAsJSON()))
+				.then(metrics => {
+					port.postMessage({
+						type: GET_METRICS_RES,
+						requestId,
+						metrics,
+					});
+				})
+				.catch(error => {
+					port.postMessage({
+						type: GET_METRICS_RES,
+						requestId,
+						error: error.message,
+					});
+				});
+		}
+	});
+}
+
 module.exports = AggregatorRegistry;