Support worker termination by collecting sum stats on exit.

CHANGELOG.md

@@ -45,16 +45,17 @@ This release marks our first release under the Prometheus umbrella.
 - perf: Use faster `process.memoryUsage.rss()` API for resident memory collection (30-40% more ops/sec)
 - fix: Browser compatibility for Gauge.startTimer()
 - ci: Run benchmarks for pull requests
-- ci: switch out deprecated benchmark-regression library for replacement
 - AggregatorRegistry renamed to ClusterRegistry, old name deprecated
 - chore: replace benchmark-regression dependency with faceoff
 - 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
+- Support for worker threads added
+- Cluster and Worker improvements:
+  - Workers now opt in to collection
+  - Make cluster and worker-thread metric aggregation order deterministic
+  - Graceful shutdown support, and data retention past worker termination
 - Export `MetricObject`, `MetricObjectWithValues`, `MetricValue` and `MetricValueWithName` from the TypeScript definitions
 - chore: Old label processing code marked as deprecated
-- Improve cluster support to allow workers to opt out
-- Abort cluster metric responses during process termination
 
 ### Added
 

README.md

@@ -18,7 +18,10 @@ See example folder for a sample usage. The library does not bundle any web
 framework. To expose the metrics, respond to Prometheus's scrape requests with
 the result of `await registry.metrics()`.
 
-### Usage with Node.js's `cluster` module
+### Usage with Node.js's `cluster` or Worker module
+
+Note: the Prometheus client now also supports worker threads, with much the same
+constraints as cluster workers. See `example/worker.js`.
 
 Node.js's `cluster` module spawns multiple processes and hands off socket
 connections to those workers. Returning metrics from a worker's local registry
@@ -41,13 +44,23 @@ for an example.)
 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
-`worker_${cluster.worker.id}` as a label value.)
+`worker_${cluster.worker.id}` as a label value.) But this will result in a high
+cardinality situation, which the Aggregator is generally meant to avoid.
 
 Metrics are aggregated from the global registry by default. To use a different
 registry, call
 `client.AggregatorRegistry.setRegistries(registryOrArrayOfRegistries)` from the
 worker processes.
 
+#### Process Lifecycle
+
+The `shutdown()` method is provided to help cleanly shut down the application while
+metrics calls are pending. In the future this will also help with workers that are
+short-lived or need to be restarted.
+
+Please also see [The Workers Readme](Workers.md) for special notes on handling workers
+that do not survive for the entire run time of the application.
+
 ## API
 
 ### Default metrics
@@ -636,3 +649,33 @@ To avoid native dependencies in this module, GC statistics for bytes reclaimed
 in each GC sweep are kept in a separate module:
 https://github.com/SimenB/node-prometheus-gc-stats. (Note that that metric may
 no longer be accurate now that v8 uses parallel garbage collection.)
+
+## Notes
+
+### Hot Reloading
+
+While it is unusual for NodeJS applications to reload modules at runtime, some more esoteric
+codebases may still do so. You should be aware that there is quite a bit of undefined behavior
+potential when combining this strategy with @prometheus-io/client_js, and one should proceed with
+caution if they do so.
+
+@prometheus-io/client_js has some primitive support for resetting itself between calls, but these
+features were intended almost exclusively for compatibility with test harnesses in order to
+support unit and integration tests. It has not been hardened to support workloads where a live
+application forcibly reloads modules to for instance pick up changes from the filesystem without
+restarting the server.
+
+Using this in production will result in data loss. Lost telemetry can trigger anything from alert to
+incorrect triage of production issues when 'sum' metrics show glitches or sawtooth patterns in their
+charts.
+
+While reloading your own code will often not also reload your dependencies, some extra care may be
+needed to avoid the common error message:
+
+> "A metric with the name #### has already been registered."
+
+Because while your code may have forgotten about that Gauge you already initialized, Prometheus
+remembers. One option is to utilize `Registry.getSingleMetric(name)` to do collision checks on
+initialization. Note that you will not be able to handle live code changes where the labels
+associated with a Metric without first removing the metric, which of course will also result in
+artifacts in your prometheus mackend due to existing counts and histograms being zeroed out.

Workers.md

@@ -0,0 +1,81 @@
+# Notes on collection and short-lived processes
+
+## Introduction
+
+Worker threads present a large surface area for potential loss of telemetry data. In order to avoid
+strange artifacts in your statistics, it's recommended that you hook the process lifecycle events
+in order to guarantee that the aggregator does not lose access to historical data when the worker
+exits.
+
+While the examples here are specific to worker threads, the same general advice also applies to
+cluster workers as well, as the implementations are nearly identical.
+
+## Background
+
+Statsd uses a fire and forget method for dumping stats to an external handler that is responsible
+for the persistence of that data. OpenTelemetry and Prometheus, in contrast, assume that the
+services are stable enough that we can ask them every so often for the data, and rely on them still
+being there when we ask again later. This creates some challenges for gathering telemetry from
+short running programs, child processes, and isolates.
+
+Telemetry data is a mixture of values, such as Gauges, and sums, such as Counts and
+Histograms. If a process crashes or becomes unresponsive, then it is no longer available to report
+those sum values, causing them to be deducted from the aggregated data. This causes artifacts
+in the telemetry data - places where numbers go down or remain flat when they should be
+monotonically increasing. These data artifacts can and do result in judgment errors by human
+operators performing triage, capacity planning, and a number of other tasks. As the person gathering
+the telemetry, it is incumbent upon you to do your level best to avoid bad data being recorded.
+
+We generally let processes crash on unhandled exceptions and rejections, but that complicates
+telemetry collection. If you're thinking of adding Prometheus telemetry to your application, or to
+a worker thread, one of your first concerns should be in reducing the number of unrecoverable errors
+your code contains.
+
+## Strategies
+
+### Delegation
+
+In the case of worker threads, sometimes short-lived is a feature, and in others it's an
+inevitability. For extremely short-lived processes, it may be best for you to summarize the work
+that was done in the worker and let the parent convert this information into the parent's own
+statistics. This reduces the amount of aggregation that needs to be done by limiting the number of
+processes that are being directly tracked. This is especially attractive in situations where the
+worker thread is running computationally intensive tasks - these workers may not even respond in a
+timely manner to messages sent to them because they are saturating the event loop with long,
+synchronous tasks.
+
+### Graceful shutdown
+
+However, if your workers are loading modules that are in common with the rest of your stack, then
+it may be that some of these modules expect telemetry to be running wherever they are running, in
+which case you will want to do graceful shutdowns in the case of errors or orderly shutdown to
+ensure that the aggregator sees this data in between scrape intervals. For this we have the
+`shutdown()` function.
+
+When a worker or cluster worker knows it is terminating, it can flush its latest telemetry to the
+aggregator. This orderly shutdown is the most memory efficient option, as the prometheus client
+can aggregate the data from all dead workers into a single data structure.
+
+```javascript
+// In worker bootstrapping code:
+
+['SIGHUP', 'SIGINT', 'SIGQUIT', 'SIGTERM'].forEach(sig => {
+  process.on(sig, async () => {
+    await registry.shutdown();
+    process.exit(0);
+  });
+});
+```
+
+See [the example](examples/workerTest.js) for a complete rundown, including the parent process
+signalling workers to shut themselves down.
+
+#### Space Complexity
+
+The `shutdown()` function causes the main or the 'primary' process to aggregate all the 'sum'
+metrics from all defunct workers that ran the shutdown to completion. The space needed in the
+aggregator thread is proportional to the union of the cardinality of the stats from all of the
+defunct workers. Therefore, so long as the cardinality of statistics is relatively common across
+all workers (eg, workers do not label their own stats with threadId), then the space needed in the
+aggregator thread is less than what the most prolific worker required - since the data is stored
+as a snapshot instead of bringing forward the storage structure that underlies active Metrics.

example/cluster.js

@@ -30,6 +30,34 @@ if (cluster.isPrimary) {
 		cluster.fork({ ...process.env, PORT: 3000 + i });
 	}
 
+	async function gracefulShutdown(worker) {
+		return new Promise((resolve, reject) => {
+			worker.send('shutdown');
+			worker.once('exit', event => {
+				resolve(event);
+			});
+			worker.once('error', event => {
+				reject(event);
+			});
+		});
+	}
+
+	async function shutdown() {
+		console.log('Shutting down...');
+
+		await Promise.all(Object.values(cluster.workers).map(gracefulShutdown));
+
+		console.log('Workers terminated');
+	}
+
+	['SIGHUP', 'SIGINT', 'SIGQUIT', 'SIGTERM'].forEach(sig => {
+		process.on(sig, async () => {
+			await shutdown(sig);
+			// eslint-disable-next-line n/no-process-exit
+			process.exit(0);
+		});
+	});
+
 	metricsServer.get('/cluster_metrics', async (req, res) => {
 		try {
 			const metrics = await clusterRegistry.clusterMetrics();
@@ -48,5 +76,20 @@ if (cluster.isPrimary) {
 		);
 	});
 } else {
+	process.on('message', async message => {
+		if (message === 'shutdown') {
+			console.log('worker shutting down');
+			try {
+				await clusterRegistry.shutdown();
+				// eslint-disable-next-line n/no-process-exit
+				process.exit(0);
+			} catch (error) {
+				console.error(error);
+				// eslint-disable-next-line n/no-process-exit
+				process.exit(1);
+			}
+		}
+	});
+
 	require('./server.js');
 }

example/worker.js

@@ -15,7 +15,13 @@
 'use strict';
 
 const Path = require('path');
-const { Worker, isMainThread, workerData } = require('node:worker_threads');
+const {
+	Worker,
+	isMainThread,
+	parentPort,
+	workerData,
+	threadId,
+} = require('node:worker_threads');
 const express = require('express');
 const WorkerRegistry = require('../').WorkerRegistry;
 
@@ -28,17 +34,63 @@ const workerRegistry = new WorkerRegistry(
 
 if (isMainThread) {
 	// By default the main thread is the collector. Demonstrating off-loading.
-	new Worker(Path.join(__filename), {
+	const collectorWorker = new Worker(Path.join(__filename), {
 		env: { ...process.env, PORT: 3333 },
 		workerData: {
 			'@prometheus-io/client': { collector: true },
 		},
 	});
 
+	const workers = [];
+
 	for (let i = 1; i <= 10; i++) {
 		const opts = { env: { ...process.env, PORT: 3000 + i } };
-		new Worker(Path.join(__filename), opts);
+		workers.push(new Worker(Path.join(__filename), opts));
+	}
+
+	async function gracefulShutdown(worker) {
+		return new Promise((resolve, reject) => {
+			worker.postMessage('shutdown');
+			worker.once('exit', event => {
+				resolve(event);
+			});
+			worker.once('error', event => {
+				reject(event);
+			});
+		});
+	}
+
+	async function shutdown() {
+		console.log('Shutting down...');
+
+		await Promise.all(workers.map(gracefulShutdown));
+
+		await gracefulShutdown(collectorWorker);
+		console.log('Workers terminated');
 	}
+
+	['SIGHUP', 'SIGINT', 'SIGQUIT', 'SIGTERM'].forEach(sig => {
+		process.on(sig, async () => {
+			await shutdown(sig);
+			// eslint-disable-next-line n/no-process-exit
+			process.exit(0);
+		});
+	});
+} else {
+	parentPort.on('message', async message => {
+		if (message === 'shutdown') {
+			console.log('worker shutting down', threadId);
+			try {
+				await workerRegistry.shutdown();
+				// eslint-disable-next-line n/no-process-exit
+				process.exit(0);
+			} catch (error) {
+				console.error(error);
+				// eslint-disable-next-line n/no-process-exit
+				process.exit(1);
+			}
+		}
+	});
 }
 
 if (collector) {

index.d.ts

@@ -69,6 +69,14 @@ export class Registry<
 	 */
 	getMetricsAsJSON(): Promise<MetricObjectWithValues<MetricValue<string>>[]>;
 
+	/**
+	 * Get all metrics as objects
+	 * @param aggregator Filter by aggregator type
+	 */
+	getMetricsAsJSON(
+		aggregator: string,
+	): Promise<MetricObjectWithValues<MetricValue<string>>[]>;
+
 	/**
 	 * Get string representation for a metric
 	 * @param metric Metric to convert to a string
@@ -80,6 +88,12 @@ export class Registry<
 	 */
 	getMetricsAsArray(): MetricObject[];
 
+	/**
+	 * Get all metrics as objects
+	 * @param aggregator Filter by aggregator type
+	 */
+	getMetricsAsArray(aggregator: string): MetricObject[];
+
 	/**
 	 * Remove a single metric
 	 * @param name The name of the metric to remove
@@ -206,6 +220,17 @@ export class WorkerRegistry<T extends RegistryContentType> extends Registry<T> {
 	 */
 	workerMetrics(): Promise<string>;
 
+	/**
+	 * Orderly shutdown of the registry.
+	 *
+	 * This is meant to be called prior to`process.exit()` to facilitate accurate metrics.
+	 *
+	 * If this instance is the primary, then it will wait for any in-flight
+	 * metrics to finish collecting or time out prior to returning.
+	 * @returns {Promise<void>}
+	 */
+	shutdown(): Promise<void>;
+
 	/**
 	 * Sets the registry or registries to be aggregated. Call from workers to
 	 * use a registry/registries other than the default global registry.
@@ -236,6 +261,17 @@ export class AggregatorRegistry<
 	 */
 	clusterMetrics(): Promise<string>;
 
+	/**
+	 * Orderly shutdown of the registry.
+	 *
+	 * This is meant to be called prior to`process.exit()` to facilitate accurate metrics.
+	 *
+	 * If this instance is the primary, then it will wait for any in-flight
+	 * metrics to finish collecting or time out prior to returning.
+	 * @returns {Promise<void>}
+	 */
+	shutdown(): Promise<void>;
+
 	/**
 	 * Sets the registry or registries to be aggregated. Call from workers to
 	 * use a registry/registries other than the default global registry.