Record cluster worker scrape failures

CHANGELOG.md

@@ -21,6 +21,9 @@ project adheres to [Semantic Versioning](http://semver.org/).
 
 ### Added
 
+- Record cluster and worker thread scrape failures in internal histograms,
+  including timeouts, worker-reported errors, and failures with no known worker errors.
+
 ## [0.16.0] - 2026-08-24
 
 This release marks our first release as a Prometheus subproject.

README.md

@@ -630,6 +630,24 @@ aggregation method, set the `aggregator` property in the metric config to one of
 'sum', 'first', 'min', 'max', 'average' or 'omit'. (See `lib/metrics/version.js`
 for an example.)
 
+Failed cluster collections are recorded in the
+`prom_client_cluster_worker_scrape_failures` histogram. Worker thread collections
+use `prom_client_worker_scrape_failures`. Each failed collection records one
+observation: the number of outstanding worker responses on a timeout, or the
+number of worker-reported errors received before the collection rejects. Other
+collection errors record zero when no worker failures are known. Successful
+collections do not add observations.
+
+Failed collections reject without returning partial metrics. Their observations
+are exposed by subsequent successful calls to `clusterMetrics()` or
+`workerMetrics()`, respectively, even if there are no workers left. These internal
+histograms are registered in the global registry when their corresponding
+`ClusterRegistry` or `WorkerRegistry` is constructed. Aggregation also includes
+the coordinating process or thread's metrics. If custom registries selected with
+`setRegistries()` do not contain the internal histogram, it is included separately
+so failure observations remain available without duplicating it in the default
+registry path.
+
 If you need to expose metrics about an individual worker, you can include a
 value that is unique to the worker (such as the worker ID or process ID) in a
 label. (See `example/server.js` for an example using

lib/cluster.js

@@ -23,6 +23,7 @@
  */
 
 const { debuglog } = require('node:util');
+const Histogram = require('./histogram');
 const Registry = require('./registry');
 const { waitFor } = require('./util');
 
@@ -41,6 +42,14 @@ const GET_METRICS_REQ = '@prometheus-io/client:getMetricsReq';
 const GET_METRICS_RES = '@prometheus-io/client:getMetricsRes';
 const GOODBYE = '@prometheus-io/client:goodbye';
 
+const clusterWorkerScrapeFailures = new Histogram({
+	name: 'prom_client_cluster_worker_scrape_failures',
+	help: 'Number of workers that failed to return metrics during a failed cluster scrape.',
+	buckets: [0, 1, 2, 4, 8, 16, 32],
+	// Register on construction, so importing the module does not add metrics.
+	registers: [],
+});
+
 let registries = [Registry.globalRegistry];
 let listenersAdded = false;
 let requestCtr = 0; // Concurrency control
@@ -55,6 +64,7 @@ class AggregatorRegistry extends Registry {
 	 */
 	constructor(regContentType = Registry.PROMETHEUS_CONTENT_TYPE) {
 		super(regContentType);
+		Registry.globalRegistry.registerMetric(clusterWorkerScrapeFailures);
 
 		addListeners();
 	}
@@ -88,11 +98,14 @@ class AggregatorRegistry extends Registry {
 		);
 
 		const responsePromises = [this.#selfMetrics(), ...workerMetrics];
+		const timeoutError = new Error('Timeout');
 		const request = {
 			responseHandlers,
+			workerFailures: 0,
 			promise: waitFor(
 				this.#gather(requestId, metricSnapshot, responsePromises),
 				5_000,
+				timeoutError,
 			),
 		};
 
@@ -105,7 +118,22 @@ class AggregatorRegistry extends Registry {
 
 			return await request.promise;
 		} catch (err) {
-			if (err.message === 'Timeout') {
+			const timedOut = err === timeoutError;
+			let failedWorkers = request.workerFailures;
+
+			if (timedOut) {
+				failedWorkers += request.responseHandlers.size;
+			}
+
+			clusterWorkerScrapeFailures.observe(failedWorkers);
+
+			// Sending can throw before request.promise is awaited.
+			for (const response of request.responseHandlers.values()) {
+				response.reject(err);
+			}
+			request.promise.catch(() => {});
+
+			if (timedOut) {
 				throw new Error(
 					`Operation timed out. ${request.responseHandlers.size} outstanding responses.`,
 				);
@@ -118,8 +146,22 @@ class AggregatorRegistry extends Registry {
 	}
 
 	async #selfMetrics() {
+		const metrics = await Promise.all(
+			registries.map(r => r.getMetricsAsJSON()),
+		);
+		// Custom registries may not contain the coordinator's internal metric.
+		if (
+			!registries.some(
+				r =>
+					r.getSingleMetric(clusterWorkerScrapeFailures.name) ===
+					clusterWorkerScrapeFailures,
+			)
+		) {
+			metrics.push([await clusterWorkerScrapeFailures.get()]);
+		}
+
 		return {
-			metrics: await Promise.all(registries.map(r => r.getMetricsAsJSON())),
+			metrics,
 		};
 	}
 
@@ -348,6 +390,7 @@ async function primaryListener(worker, event) {
 		request.responseHandlers.delete(worker.id);
 
 		if (event.error) {
+			request.workerFailures++;
 			response.reject(new Error(event.error));
 		} else {
 			response.resolve({

lib/util.js

@@ -173,9 +173,10 @@ exports.nowTimestamp = function nowTimestamp() {
  * Async functions with a timeout.
  * @param promise {Promise}
  * @param limit {number}
+ * @param {Error} [timeoutError] - Error used when this deadline expires.
  * @returns {Promise}
  */
-exports.waitFor = async function waitFor(promise, limit = 5_000) {
+exports.waitFor = async function waitFor(promise, limit = 5_000, timeoutError) {
 	let resolve, reject;
 
 	const resultPromise = new Promise((res, rej) => {
@@ -184,7 +185,7 @@ exports.waitFor = async function waitFor(promise, limit = 5_000) {
 	});
 
 	const timeout = setTimeout(() => {
-		reject(new Error('Timeout'));
+		reject(timeoutError ?? new Error('Timeout'));
 	}, limit);
 
 	try {

lib/worker.js

@@ -24,6 +24,7 @@
 
 const { debuglog } = require('node:util');
 const worker = require('node:worker_threads');
+const Histogram = require('./histogram');
 const Registry = require('./registry');
 const { waitFor } = require('./util');
 
@@ -36,6 +37,14 @@ const GET_METRICS_REQ = '@prometheus-io/client:getMetricsReq';
 const GET_METRICS_RES = '@prometheus-io/client:getMetricsRes';
 const GOODBYE = '@prometheus-io/client:goodbye';
 
+const workerScrapeFailures = new Histogram({
+	name: 'prom_client_worker_scrape_failures',
+	help: 'Number of workers that failed to return metrics during a failed worker scrape.',
+	buckets: [0, 1, 2, 4, 8, 16, 32],
+	// Register on construction, so importing the module does not add metrics.
+	registers: [],
+});
+
 const ANNOUNCEMENT_CHANNEL = new BroadcastChannel(
 	'@prometheus-io/client:announce',
 ).unref();
@@ -60,6 +69,7 @@ class WorkerRegistry extends Registry {
 		primary = isMainThread,
 	) {
 		super(regContentType);
+		Registry.globalRegistry.registerMetric(workerScrapeFailures);
 		this.primary = primary;
 
 		addListeners(primary);
@@ -78,17 +88,12 @@ class WorkerRegistry extends Registry {
 		);
 
 		if (orderedWorkers.length === 0) {
-			if (historicMetrics.length === 0) {
-				debug('No data found for requestId', requestId);
-				return '';
-			} else {
-				debug('No workers found for requestId', requestId);
-			}
+			debug('No workers found for requestId', requestId);
 		}
 
 		const metricSnapshot = historicMetrics;
 		const responseHandlers = new Map();
-		const responsePromises = orderedWorkers.map(
+		const workerMetrics = orderedWorkers.map(
 			entry =>
 				new Promise((resolveResponse, rejectResponse) => {
 					responseHandlers.set(entry.name, {
@@ -98,11 +103,15 @@ class WorkerRegistry extends Registry {
 				}),
 		);
 
+		const responsePromises = [this.#selfMetrics(), ...workerMetrics];
+		const timeoutError = new Error('Timeout');
 		const request = {
 			responseHandlers,
+			workerFailures: 0,
 			promise: waitFor(
 				this.#gather(requestId, metricSnapshot, responsePromises),
 				5_000,
+				timeoutError,
 			),
 		};
 
@@ -117,7 +126,22 @@ class WorkerRegistry extends Registry {
 
 			return await request.promise;
 		} catch (err) {
-			if (err.message === 'Timeout') {
+			const timedOut = err === timeoutError;
+			let failedWorkers = request.workerFailures;
+
+			if (timedOut) {
+				failedWorkers += request.responseHandlers.size;
+			}
+
+			workerScrapeFailures.observe(failedWorkers);
+
+			// Sending can throw before request.promise is awaited.
+			for (const response of request.responseHandlers.values()) {
+				response.reject(err);
+			}
+			request.promise.catch(() => {});
+
+			if (timedOut) {
 				throw new Error(
 					`Operation timed out. ${request.responseHandlers.size} outstanding responses.`,
 				);
@@ -129,6 +153,23 @@ class WorkerRegistry extends Registry {
 		}
 	}
 
+	async #selfMetrics() {
+		const metrics = await Promise.all(
+			registries.map(r => r.getMetricsAsJSON()),
+		);
+		// Custom registries may not contain the coordinator's internal metric.
+		if (
+			!registries.some(
+				r =>
+					r.getSingleMetric(workerScrapeFailures.name) === workerScrapeFailures,
+			)
+		) {
+			metrics.push([await workerScrapeFailures.get()]);
+		}
+
+		return { metrics };
+	}
+
 	/**
 	 * Collect the data for a metrics request.
 	 * @param requestId {number}
@@ -277,11 +318,11 @@ function addListeners(primary) {
 				announce(name, false);
 			}
 		} else if (message.type === GET_METRICS_REQ) {
-			const metrics = await Promise.all(
-				registries.map(r => r.getMetricsAsJSON()),
-			);
-
 			try {
+				const metrics = await Promise.all(
+					registries.map(r => r.getMetricsAsJSON()),
+				);
+
 				channel.postMessage({
 					type: GET_METRICS_RES,
 					requestId: message.requestId,
@@ -375,6 +416,7 @@ async function primaryListener(event) {
 				request.responseHandlers.delete(workerName);
 
 				if (workerMessage.error) {
+					request.workerFailures++;
 					response.reject(new Error(workerMessage.error));
 				} else {
 					response.resolve({