Cluster fixes

CHANGELOG.md

@@ -20,6 +20,7 @@ This release marks our first release under the Prometheus umbrella.
   runtime object. Value-style uses such as `MetricType.Counter` (which threw at runtime)
   no longer compile; compare against the string literals instead. Under
   `verbatimModuleSyntax`, import it with `import type`.
+- The cluster primary now reports metrics
 
 ### Changed
 
@@ -46,6 +47,8 @@ This release marks our first release under the Prometheus umbrella.
 - chore: Add copyright license headers and test
 - Make cluster and worker-thread metric aggregation order deterministic
 - Export `MetricObject`, `MetricObjectWithValues`, `MetricValue` and `MetricValueWithName` from the TypeScript definitions
+- Improve cluster support to allow workers to opt out
+- Abort cluster metric responses during process termination
 
 ### Added
 

example/cluster.js

@@ -22,6 +22,10 @@ const metricsServer = express();
 const clusterRegistry = new ClusterRegistry();
 
 if (cluster.isPrimary) {
+	require('../').collectDefaultMetrics({
+		gcDurationBuckets: [0.001, 0.01, 0.1, 1, 2, 5], // These are the default buckets.
+	});
+
 	for (let i = 1; i <= 4; i++) {
 		cluster.fork({ ...process.env, PORT: 3000 + i });
 	}
@@ -32,6 +36,7 @@ if (cluster.isPrimary) {
 			res.set('Content-Type', clusterRegistry.contentType);
 			res.send(metrics);
 		} catch (ex) {
+			console.log(ex);
 			res.statusCode = 500;
 			res.send(ex.message);
 		}

index.d.ts

@@ -206,7 +206,6 @@ export class WorkerRegistry<T extends RegistryContentType> extends Registry<T> {
 	 */
 	workerMetrics(): Promise<string>;
 
-	addWorker(worker: Worker): void;
 	/**
 	 * Sets the registry or registries to be aggregated. Call from workers to
 	 * use a registry/registries other than the default global registry.

lib/cluster.js

@@ -22,6 +22,7 @@
  * cluster master.
  */
 
+const { debuglog } = require('node:util');
 const Registry = require('./registry');
 // We need to lazy-load the 'cluster' module as some application servers -
 // namely Passenger - crash when it is imported.
@@ -31,17 +32,25 @@ let cluster = () => {
 	return data;
 };
 
+const debug = debuglog('prom:metrics:cluster');
+const ANNOUNCEMENT = '@prometheus-io/client:announcement';
 const GET_METRICS_REQ = '@prometheus-io/client:getMetricsReq';
 const GET_METRICS_RES = '@prometheus-io/client:getMetricsRes';
 
 let registries = [Registry.globalRegistry];
 let requestCtr = 0; // Concurrency control
 let listenersAdded = false;
 const requests = new Map(); // Pending requests for workers' local metrics.
+const workers = new Map();
 
 class AggregatorRegistry extends Registry {
+	/**
+	 * Create a Registry.
+	 * @param regContentType
+	 */
 	constructor(regContentType = Registry.PROMETHEUS_CONTENT_TYPE) {
 		super(regContentType);
+
 		addListeners();
 	}
 
@@ -53,9 +62,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);
+		const orderedWorkers = [...workers.values()].sort(
+			(left, right) => left.id - right.id,
+		);
 
 		return new Promise((resolve, reject) => {
 			let settled = false;
@@ -78,36 +87,43 @@ class AggregatorRegistry extends Registry {
 				responseHandlers,
 				done,
 				errorTimeout: setTimeout(() => {
-					const err = new Error('Operation timed out.');
+					const err = new Error(
+						`Operation timed out. ${request.responseHandlers.size} outstanding responses.`,
+					);
 					request.done(err);
-				}, 5000),
+				}, 5_000),
 			};
 			requests.set(requestId, request);
-
-			const message = {
-				type: GET_METRICS_REQ,
-				requestId,
-			};
-
-			if (workers.length === 0) {
-				// No workers were up
-				process.nextTick(() => done(undefined, ''));
-				return;
-			}
-
-			const responsePromises = workers.map(
+			const workerMetrics = orderedWorkers.map(
 				worker =>
 					new Promise((resolveResponse, rejectResponse) => {
 						responseHandlers.set(worker.id, {
 							resolve: resolveResponse,
 							reject: rejectResponse,
 						});
-						worker.send(message);
+
+						worker.send({
+							type: GET_METRICS_REQ,
+							requestId,
+						});
 					}),
 			);
 
-			Promise.all(responsePromises)
-				.then(metrics => Registry.aggregate(metrics.flat()).metrics())
+			const myMetrics = Promise.all(
+				registries.map(r => r.getMetricsAsJSON()),
+			).then(metrics => {
+				return { metrics };
+			});
+
+			if (workerMetrics.length === 0) {
+				debug('No workers found for requestId', requestId);
+			}
+
+			const allMetrics = [myMetrics, ...workerMetrics];
+
+			Promise.all(allMetrics)
+				.then(responses => responses.flatMap(response => response.metrics))
+				.then(metrics => Registry.aggregate(metrics).metrics())
 				.then(result => done(undefined, result), done);
 		});
 	}
@@ -158,54 +174,145 @@ class AggregatorRegistry extends Registry {
  * @returns {void}
  */
 function addListeners() {
-	if (listenersAdded) return;
+	if (listenersAdded) {
+		return;
+	}
+
 	listenersAdded = true;
 
 	if (cluster().isPrimary) {
-		// 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);
+		replaceListener('message', cluster(), primaryListener);
+		replaceListener('disconnect', cluster(), disconnect);
 
-				if (request === undefined) {
-					return;
-				}
+		announce();
+	} else {
+		replaceListener('message', process, workerListener);
 
-				const response = request.responseHandlers.get(worker.id);
-				if (response === undefined) {
-					return;
-				}
-				request.responseHandlers.delete(worker.id);
+		if (typeof process.send !== 'function') {
+			debug('worker has no process.send()');
+		} else if (!process.connected) {
+			debug('worker is not connected to parent process');
+		} else {
+			process.send({ type: ANNOUNCEMENT });
+		}
+	}
+}
 
-				if (message.error) {
-					response.reject(new Error(message.error));
-				} else {
-					response.resolve(message.metrics);
-				}
+/**
+ * Watch for metrics events and aggregator announcements
+ *
+ * Whereas clusters are a top-level activity, multiple modules may start their
+ * own workers and require telemetry collection.
+ * @param	message {MessageEvent}
+ */
+async function workerListener(message) {
+	if (message.type === ANNOUNCEMENT) {
+		process.send({ type: ANNOUNCEMENT });
+	} else if (message.type === GET_METRICS_REQ) {
+		try {
+			const metrics = await Promise.all(
+				registries.map(r => r.getMetricsAsJSON()),
+			);
+
+			if (!process.connected) {
+				debug('Connection to primary lost.');
+			} else {
+				process.send({
+					type: GET_METRICS_RES,
+					requestId: message.requestId,
+					metrics,
+				});
 			}
-		});
-	} else {
-		// Respond to master's requests for worker's local metrics.
-		process.on('message', message => {
-			if (message.type === GET_METRICS_REQ) {
-				Promise.all(registries.map(r => r.getMetricsAsJSON()))
-					.then(metrics => {
-						process.send({
-							type: GET_METRICS_RES,
-							requestId: message.requestId,
-							metrics,
-						});
-					})
-					.catch(error => {
-						process.send({
-							type: GET_METRICS_RES,
-							requestId: message.requestId,
-							error: error.message,
-						});
-					});
+		} catch (error) {
+			debug('Error sending to primary', error);
+			if (!process.connected) {
+				debug('Connection to primary lost.');
+			} else {
+				process.send({
+					type: GET_METRICS_RES,
+					requestId: message.requestId,
+					error: error.message,
+				});
 			}
-		});
+		}
+	}
+}
+
+/**
+ * Add workers to the aggregation list when they are announced.
+ *
+ * Whereas clusters are a top-level activity, multiple modules may start their
+ * own workers and require telemetry collection.
+ * @param	event {MessageEvent}
+ */
+
+async function primaryListener(worker, event) {
+	if (event.type === ANNOUNCEMENT) {
+		if (workers.has(worker.id)) {
+			debug('duplicate worker announcement', worker.id);
+			return;
+		}
+
+		workers.set(worker.id, worker);
+	} else if (event.type === GET_METRICS_RES) {
+		const request = requests.get(event.requestId);
+
+		if (request === undefined) {
+			debug('unexpected results from worker', worker.id);
+			return;
+		}
+
+		const response = request.responseHandlers.get(worker.id);
+		if (response === undefined) {
+			return;
+		}
+		request.responseHandlers.delete(worker.id);
+
+		if (event.error) {
+			response.reject(new Error(event.error));
+		} else {
+			response.resolve({
+				threadId: worker.id,
+				metrics: event.metrics,
+			});
+		}
+	}
+}
+
+function disconnect(event) {
+	debug('worker disconnected', event.id);
+	workers.delete(event.id);
+}
+
+function announce() {
+	for (const worker of Object.values(cluster().workers)) {
+		if (worker.isConnected()) {
+			worker.send({ type: ANNOUNCEMENT });
+		}
+	}
+}
+
+/**
+ * Replace any listeners with new ones.
+ *
+ * @param messageType
+ * @param emitter {EventEmitter}
+ * @param fn
+ */
+function replaceListener(messageType, emitter, fn) {
+	// Reloading a module creates a unique instance of each function, so the
+	// identity checks is cluster.off() will fail.
+	const functionString = fn.toString();
+
+	for (const listener of emitter.listeners(messageType)) {
+		// eslint-disable-next-line eqeqeq
+		if (functionString == listener) {
+			debug('removing duplicate listener', messageType);
+			emitter.off(messageType, listener);
+		}
 	}
+
+	emitter.on(messageType, fn);
 }
 
 module.exports = AggregatorRegistry;