From 62de49f3c17d4ff2c45590b2cde977ba9b120eda Mon Sep 17 00:00:00 2001 From: Jason Marshall Date: Thu, 13 Aug 2026 10:53:08 -0700 Subject: [PATCH 01/10] Documentation on the care and feeding of flaky or short-lived workers. Signed-off-by: Jason Marshall --- README.md | 9 ++++++- Workers.md | 70 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 1 deletion(-) create mode 100644 Workers.md diff --git a/README.md b/README.md index 79f1455e..722afa96 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,9 @@ the result of `await registry.metrics()`. ### Usage with Node.js's `cluster` 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 will only reveal that individual worker's metrics, which is generally @@ -41,13 +44,17 @@ 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. +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 diff --git a/Workers.md b/Workers.md new file mode 100644 index 00000000..2549662d --- /dev/null +++ b/Workers.md @@ -0,0 +1,70 @@ +# Notes on collection and short-lived processes + +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. + +In the case of worker threads, sometimes short-lived is a feature, and in others it's an +inevitability. In these cases, the prometheus client will need a little help from you on tracking +the lifecycle of those workers. + +The biggest challenge is that if a worker is unresponsive, then the prometheus client will time out +while trying to collect the aggregated metrics, resulting in NO telemetry being reported at all. +Avoiding this problem would come at a substantial memory premium, as the sum values from every +worker would need to be retained. + +Additionally, the Prometheus client retains metadata for every worker it knows about. If you cycle +workers frequently, then that table will grow without bounds. If we knew for certain that a worker +was gone, then some of that metadata can be consolidated across all defunct workers, and as long as +the cardinality of the metrics does not include process-unique data, such as the threadId, then +twenty dead worker is no more expensive than one. + +Because of the nature of workers, it is expected that they may saturate the event loop. That means +that if we 'ping' them to see if they are still responsive, then they might not reply until after +we decided they are dead. If they intermittently respond to requests, then the bookkeeping gets +quite elaborate (expensive). + +As the application author, you have more control and visibility over the lifecycle of your workers, +especially for worker threads. + +## Graceful shutdown + +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. + +TBD: The final values for gauges may or may not be lost when the process exits. + +``` +// Code example goes here +``` + +## Lifecycle events + +The main thread can also listen for lifecycle events for its workers and inform us when +any of them exit prematurely. This solution will still result in data loss, and telemetry artifacts, +but will also reduce the number of collection errors and can help the Prometheus client to clean up +metadata related to the lost worker. + +We could fix the data loss by retaining data from the previous collection interval, that would +require a good deal of extra storage to facilitate, and therefore would be a substantial tax on +well-behaved workers. Alternatively, we could flag some workers as problematic (example: you have +three pools of workers, and only one tends to crash), but that is currently not supported. + +For now, it is recommended that you hook the unhandled exceptions in the Worker itself, then flush +the telemetry data prior to calling `process.exit()`. From 67679c388f48009b66b856d72c160c295407985e Mon Sep 17 00:00:00 2001 From: Jason Marshall Date: Thu, 13 Aug 2026 21:35:47 -0700 Subject: [PATCH 02/10] Add a function to handle timeouts. This removes a little bit of duplicate code from cluster.js and worker.js, and reduces some complexity in those functions to support new lifecycle functionality for process.exit() Signed-off-by: Jason Marshall --- lib/util.js | 27 +++++++++++++++++++++++++++ test/utilTest.js | 19 +++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/lib/util.js b/lib/util.js index 8f8dd8d9..96dc0eec 100644 --- a/lib/util.js +++ b/lib/util.js @@ -169,6 +169,33 @@ exports.nowTimestamp = function nowTimestamp() { return Date.now() / 1000; }; +/** + * Async functions with a timeout. + * @param promise {Promise} + * @param limit {number} + * @returns {Promise} + */ +exports.waitFor = async function waitFor(promise, limit = 5_000) { + let resolve, reject; + + const resultPromise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + + const timeout = setTimeout(() => { + reject(new Error('Timeout')); + }, limit); + + try { + promise.then(resolve, reject); + + return await resultPromise; + } finally { + clearTimeout(timeout); + } +}; + /** * @typedef StatsEntry {} * @property value {*} diff --git a/test/utilTest.js b/test/utilTest.js index c1fe8ebc..11f5ceef 100644 --- a/test/utilTest.js +++ b/test/utilTest.js @@ -43,6 +43,25 @@ describe('utils', () => { }); }); + describe('waitFor', () => { + const waitFor = require('../lib/util').waitFor; + + it('times out if the promise exceeds the limit', async () => { + await expect(waitFor(new Promise(() => {}), 1)).rejects.toThrow( + 'Timeout', + ); + }); + + it('Resolves on a success', async () => { + await expect(waitFor(Promise.resolve('foo'))).resolves.toEqual('foo'); + }); + + it('Rejects on a promise rejection', async () => { + const promise = waitFor(Promise.reject(new Error('nope'))); + await expect(promise).rejects.toThrow('nope'); + }); + }); + describe('getLabels', () => { const getLabels = require('../lib/util').getLabels; From 7dd391b64e1dfeb88827d228c761a88358aa2c5e Mon Sep 17 00:00:00 2001 From: Jason Marshall Date: Thu, 13 Aug 2026 21:41:50 -0700 Subject: [PATCH 03/10] rework clusterMetrics() and workerMetrics() to use simpler async logic and the new util.waitFor function. This implementation stores the promise in the requests table, which will be needed to wait for in-flight metrics collections during an orderly shutdown. Signed-off-by: Jason Marshall --- lib/cluster.js | 118 +++++++++++++++++++++++++------------------------ lib/worker.js | 107 ++++++++++++++++++++++---------------------- 2 files changed, 113 insertions(+), 112 deletions(-) diff --git a/lib/cluster.js b/lib/cluster.js index 62e3b99b..7407908d 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -24,6 +24,8 @@ const { debuglog } = require('node:util'); const Registry = require('./registry'); +const { waitFor } = require('./util'); + // We need to lazy-load the 'cluster' module as some application servers - // namely Passenger - crash when it is imported. let cluster = () => { @@ -31,15 +33,15 @@ let cluster = () => { cluster = () => data; 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; +let requestCtr = 0; // Concurrency control const requests = new Map(); // Pending requests for workers' local metrics. const workers = new Map(); @@ -60,72 +62,70 @@ class AggregatorRegistry extends Registry { * @returns {Promise} Promise that resolves with the aggregated * metrics. */ - clusterMetrics() { + async clusterMetrics() { const requestId = requestCtr++; const orderedWorkers = [...workers.values()] .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 (orderedWorkers.length === 0) { + debug('No workers found for requestId', requestId); + } - if (err !== undefined) { - reject(err); - } else { - resolve(result); - } - } + const responseHandlers = new Map(); + const workerMetrics = orderedWorkers.map( + worker => + new Promise((resolveResponse, rejectResponse) => { + responseHandlers.set(worker.id, { + resolve: resolveResponse, + reject: rejectResponse, + }); + }), + ); + + const promises = [this.#selfMetrics(), ...workerMetrics]; + const request = { + responseHandlers, + promise: waitFor(this.#gather(requestId, promises), 5_000), + }; + + requests.set(requestId, request); - const responseHandlers = new Map(); - const request = { - responseHandlers, - done, - errorTimeout: setTimeout(() => { - const err = new Error( - `Operation timed out. ${request.responseHandlers.size} outstanding responses.`, - ); - request.done(err); - }, 5_000), - }; - requests.set(requestId, request); - const workerMetrics = orderedWorkers.map( - worker => - new Promise((resolveResponse, rejectResponse) => { - responseHandlers.set(worker.id, { - resolve: resolveResponse, - reject: rejectResponse, - }); - - worker.send({ - type: GET_METRICS_REQ, - requestId, - }); - }), + try { + orderedWorkers.forEach(worker => + worker.send({ type: GET_METRICS_REQ, requestId }), ); - const myMetrics = Promise.all( - registries.map(r => r.getMetricsAsJSON()), - ).then(metrics => { - return { metrics }; - }); - - if (workerMetrics.length === 0) { - debug('No workers found for requestId', requestId); + return await request.promise; + } catch (err) { + if (err.message === 'Timeout') { + throw new Error( + `Operation timed out. ${request.responseHandlers.size} outstanding responses.`, + ); } - const allMetrics = [myMetrics, ...workerMetrics]; + throw err; + } finally { + requests.delete(requestId); + } + } - Promise.all(allMetrics) - .then(responses => responses.flatMap(response => response.metrics)) - .then(metrics => Registry.aggregate(metrics).metrics()) - .then(result => done(undefined, result), done); - }); + async #selfMetrics() { + return { + metrics: await Promise.all(registries.map(r => r.getMetricsAsJSON())), + }; + } + + /** + * Collect the data for a metrics request. + * @param requestId + * @param promises + * @returns {Promise} + */ + async #gather(requestId, promises) { + const responses = await Promise.all(promises); + const metrics = responses.flatMap(response => response.metrics); + return Registry.aggregate(metrics).metrics(); } get contentType() { @@ -255,15 +255,17 @@ async function primaryListener(worker, event) { workers.set(worker.id, worker); } else if (event.type === GET_METRICS_RES) { - const request = requests.get(event.requestId); + const requestId = event.requestId; + const request = requests.get(requestId); if (request === undefined) { - debug('unexpected results from worker', worker.id); + debug('unexpected results for', requestId, 'from worker', worker.id); return; } const response = request.responseHandlers.get(worker.id); if (response === undefined) { + debug('unexpected results from worker', worker.id); return; } request.responseHandlers.delete(worker.id); diff --git a/lib/worker.js b/lib/worker.js index 6b524d3a..058fd6f6 100644 --- a/lib/worker.js +++ b/lib/worker.js @@ -23,11 +23,13 @@ */ const { debuglog } = require('node:util'); -const Registry = require('./registry'); const worker = require('node:worker_threads'); -const { isMainThread, threadId, BroadcastChannel } = worker; +const Registry = require('./registry'); +const { waitFor } = require('./util'); +const { isMainThread, threadId, BroadcastChannel } = worker; const debug = debuglog('prom:metrics:worker'); + const ANNOUNCEMENT = '@prometheus-io/client:announcement'; const GET_METRICS_REQ = '@prometheus-io/client:getMetricsReq'; const GET_METRICS_RES = '@prometheus-io/client:getMetricsRes'; @@ -44,8 +46,8 @@ const workers = new Map(); class WorkerRegistry extends Registry { /** * Create a Registry. - * If set to primary, this thread will handle coordination of all of the - * other workers. + * If set to primary, this thread will handle coordination of all the other + * workers. * @param regContentType * @param primary {boolean} whether this is the coordinating process */ @@ -65,69 +67,66 @@ class WorkerRegistry extends Registry { * @returns {Promise} Promise that resolves with the aggregated * metrics. */ - workerMetrics() { - //TODO: We should be able to collect metrics for the collector thread. + async workerMetrics() { const requestId = requestCtr++; + const orderedWorkers = [...workers.values()].sort( + (left, right) => left.threadId - right.threadId, + ); - return new Promise((resolve, reject) => { - let settled = false; - function done(err, result) { - if (settled) return; - settled = true; - - clearTimeout(request.errorTimeout); - requests.delete(requestId); + if (orderedWorkers.length === 0) { + debug('No workers found for requestId', requestId); + return ''; + } - if (err !== undefined) { - reject(err); - } else { - resolve(result); - } - } + const responseHandlers = new Map(); + const responsePromises = orderedWorkers.map( + entry => + new Promise((resolveResponse, rejectResponse) => { + responseHandlers.set(entry.name, { + resolve: resolveResponse, + reject: rejectResponse, + }); + }), + ); - const responseHandlers = new Map(); - const request = { - responseHandlers, - done, - errorTimeout: setTimeout(() => { - const err = new Error( - `Operation timed out. ${request.responseHandlers.size} outstanding responses.`, - ); - request.done(err); - }, 5_000), - }; - requests.set(requestId, request); - - const orderedWorkers = [...workers.values()].sort( - (left, right) => left.threadId - right.threadId, - ); + const request = { + responseHandlers, + promise: waitFor(this.#gather(requestId, responsePromises), 5_000), + }; - const responsePromises = orderedWorkers.map( - entry => - new Promise((resolveResponse, rejectResponse) => { - responseHandlers.set(entry.name, { - resolve: resolveResponse, - reject: rejectResponse, - }); - }), - ); + requests.set(requestId, request); + try { ANNOUNCEMENT_CHANNEL.postMessage({ type: GET_METRICS_REQ, threadId, requestId, }); - if (responsePromises.length === 0) { - debug('No workers found for requestId', requestId); - process.nextTick(() => done(undefined, '')); - } else { - Promise.all(responsePromises) - .then(responses => responses.flatMap(response => response.metrics)) - .then(metrics => Registry.aggregate(metrics).metrics()) - .then(result => done(undefined, result), done); + return await request.promise; + } catch (err) { + if (err.message === 'Timeout') { + throw new Error( + `Operation timed out. ${request.responseHandlers.size} outstanding responses.`, + ); } - }); + + throw err; + } finally { + requests.delete(requestId); + } + } + + /** + * Collect the data for a metrics request. + * @param requestId {number} + * @param promises {Promise[]} + * @returns {Promise} + */ + async #gather(requestId, promises) { + const responses = await Promise.all(promises); + const metrics = responses.flatMap(response => response.metrics); + return Registry.aggregate(metrics).metrics(); } get contentType() { From 6bcd9f0f5988d8814ba9db5f997a1f74f92d3bdf Mon Sep 17 00:00:00 2001 From: Jason Marshall Date: Thu, 13 Aug 2026 23:34:44 -0700 Subject: [PATCH 04/10] Orderly shutdown of the aggregator. Waits for metrics() to finish processing. Signed-off-by: Jason Marshall --- README.md | 8 ++- index.d.ts | 22 +++++++ lib/cluster.js | 17 ++++++ lib/worker.js | 17 ++++++ test/clusterTest.js | 49 ++++++++++++++- test/workerTest.js | 143 ++++++++++++++++++++++++++++++-------------- 6 files changed, 209 insertions(+), 47 deletions(-) diff --git a/README.md b/README.md index 722afa96..9325f755 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ 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`. @@ -52,6 +52,12 @@ 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. diff --git a/index.d.ts b/index.d.ts index 7712e2c0..fbd402c6 100644 --- a/index.d.ts +++ b/index.d.ts @@ -206,6 +206,17 @@ export class WorkerRegistry extends Registry { */ workerMetrics(): Promise; + /** + * 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} + */ + shutdown(): Promise; + /** * Sets the registry or registries to be aggregated. Call from workers to * use a registry/registries other than the default global registry. @@ -236,6 +247,17 @@ export class AggregatorRegistry< */ clusterMetrics(): Promise; + /** + * 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} + */ + shutdown(): Promise; + /** * Sets the registry or registries to be aggregated. Call from workers to * use a registry/registries other than the default global registry. diff --git a/lib/cluster.js b/lib/cluster.js index 7407908d..14e2012b 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -132,6 +132,23 @@ class AggregatorRegistry extends Registry { return super.contentType; } + /** + * 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} + */ + async shutdown() { + if (cluster().isPrimary) { + const outstanding = requests.values().map(entry => entry.promise); + + await Promise.all(outstanding); + } + } + /** * Creates a new Registry instance from an array of metrics that were * created by `registry.getMetricsAsJSON()`. Metrics are aggregated using diff --git a/lib/worker.js b/lib/worker.js index 058fd6f6..859051d7 100644 --- a/lib/worker.js +++ b/lib/worker.js @@ -133,6 +133,23 @@ class WorkerRegistry extends Registry { return super.contentType; } + /** + * 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} + */ + async shutdown() { + if (this.primary) { + const outstanding = [...requests.values().map(entry => entry.promise)]; + + await Promise.allSettled(outstanding); + } + } + /** * Creates a new Registry instance from an array of metrics that were * created by `registry.getMetricsAsJSON()`. Metrics are aggregated using diff --git a/test/clusterTest.js b/test/clusterTest.js index 2b41c9a3..49826fec 100644 --- a/test/clusterTest.js +++ b/test/clusterTest.js @@ -17,6 +17,9 @@ const cluster = require('cluster'); const process = require('process'); const Registry = require('../lib/cluster'); +const AggregatorRegistry = require('../lib/cluster'); +const { BroadcastChannel } = require('worker_threads'); +const { setTimeout: delay } = require('timers/promises'); const ANNOUNCEMENT = '@prometheus-io/client:announcement'; const GET_METRICS_REQ = '@prometheus-io/client:getMetricsReq'; @@ -128,7 +131,7 @@ describe.each([ }); cluster.workers = originalWorkers; } - }, 6_000); + }); it('aggregates telemetry from primary thread', async () => { jest.resetModules(); @@ -152,6 +155,50 @@ describe.each([ }); }); + describe('shutdown()', () => { + it('returns immediately on no outstanding requests', async () => { + const AggregatorRegistry = require('../lib/cluster'); + const ar = new AggregatorRegistry(); + + await expect(ar.shutdown()).resolves.not.toThrow(); + }); + + it('waits for pending requests', async () => { + const originalWorkers = cluster.workers; + jest.resetModules(); + const AggregatorRegistry = require('../lib/cluster'); + const registry = new AggregatorRegistry(regType); + const worker = { + id: 53, + isConnected: () => true, + send: jest.fn(), + }; + + cluster.workers = [worker]; + + cluster.emit('message', worker, { type: ANNOUNCEMENT }); + + try { + const results = []; + const promise = registry.clusterMetrics().then(() => results.push(1)); + const shutdown = registry.shutdown().then(() => results.push(2)); + + cluster.emit('message', worker, { + type: GET_METRICS_RES, + requestId: 0, + metrics: [[metric(7)]], + }); + + await Promise.all([promise, shutdown]); + + expect(results).toEqual([1, 2]); + } finally { + cluster.emit('disconnect', worker); + cluster.workers = originalWorkers; + } + }); + }); + describe('message handling', () => { it('does not error out on unexpected (or late) responses', () => { jest.resetModules(); diff --git a/test/workerTest.js b/test/workerTest.js index 17fd0fa8..4993bf27 100644 --- a/test/workerTest.js +++ b/test/workerTest.js @@ -48,59 +48,112 @@ describe.each([ 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-io/client:announce', - ).unref(); - const responders = [1, 2, 3].map(threadId => { - const name = `@prometheus-io/client:test-worker:${threadId}`; - const channel = new BroadcastChannel(name).unref(); - - announcementChannel.postMessage({ - type: ANNOUNCEMENT, - name, + it('aggregates worker responses in thread id order', async () => { + jest.resetModules(); + const AggregatorRegistry = require('../lib/worker'); + const registry = new AggregatorRegistry(regType); + const announcementChannel = new BroadcastChannel( + '@prometheus-io/client:announce', + ).unref(); + const responders = [1, 2, 3].map(threadId => { + const name = `@prometheus-io/client:test-worker:${threadId}`; + const channel = new BroadcastChannel(name).unref(); + + announcementChannel.postMessage({ + type: ANNOUNCEMENT, + name, + threadId, + }); + + return { threadId, channel }; + }); + + await delay(5); // Let announcements arrive + + 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(); + }); - return { threadId, channel }; - }); + 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(); + } + }); + }); - await delay(5); // Let announcements arrive + describe('shutdown()', () => { + let AggregatorRegistry; + beforeEach(() => { + jest.resetModules(); + AggregatorRegistry = require('../lib/worker'); + }); - 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(); - }); + it('returns immediately on no outstanding requests', async () => { + const registry = new AggregatorRegistry(); - 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(); + await expect(registry.shutdown()).resolves.not.toThrow(); + }); + + it('waits for pending requests', async () => { + const registry = new AggregatorRegistry(); + + const announcementChannel = new BroadcastChannel( + '@prometheus-io/client:announce', + ).unref(); + + const threadId = 22; + const name = `@prometheus-io/client:test-worker:${threadId}`; + const channel = new BroadcastChannel(name).unref(); + + announcementChannel.postMessage({ + type: ANNOUNCEMENT, + name, + threadId, + }); + + await delay(5); // Let announcements arrive + + announcementChannel.addEventListener('message', async event => { + if (event.data.type === GET_METRICS_REQ) { + channel.postMessage({ + type: GET_METRICS_RES, + requestId: event.data.requestId, + threadId, + metrics: [[metric(2)]], + }); } }); + + const results = []; + const promise = registry.workerMetrics().then(() => results.push(1)); + const shutdown = registry.shutdown().then(() => results.push(2)); + await Promise.all([promise, shutdown]); + + expect(results).toEqual([1, 2]); + }); }); describe('message handling', () => { From b20c7b54b427c34122e5ecdc5516ead4f427a00d Mon Sep 17 00:00:00 2001 From: Jason Marshall Date: Fri, 14 Aug 2026 10:32:36 -0700 Subject: [PATCH 05/10] Add filtering support to Registry. This is prep work for tracking defunct workers. Signed-off-by: Jason Marshall --- index.d.ts | 14 ++++++++++++++ lib/registry.js | 29 +++++++++++++++++++++-------- test/registerTest.js | 40 +++++++++++++++++++++++++++++++++++++++- 3 files changed, 74 insertions(+), 9 deletions(-) diff --git a/index.d.ts b/index.d.ts index fbd402c6..ed3e6eb3 100644 --- a/index.d.ts +++ b/index.d.ts @@ -69,6 +69,14 @@ export class Registry< */ getMetricsAsJSON(): Promise>[]>; + /** + * Get all metrics as objects + * @param aggregator Filter by aggregator type + */ + getMetricsAsJSON( + aggregator: string, + ): Promise>[]>; + /** * 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 diff --git a/lib/registry.js b/lib/registry.js index 9c9aeaf7..482750e4 100644 --- a/lib/registry.js +++ b/lib/registry.js @@ -42,11 +42,20 @@ class Registry { /** * Return all metrics * + * @see lib/metricAggregators.js + * @param {string} [aggregator] - filter by aggregator type, typically used for sum. * @returns {object[]} */ - getMetricsAsArray() { - return Array.from(this._metrics.values()); + getMetricsAsArray(aggregator) { + const values = this._metrics.values(); + if (aggregator === undefined) { + return Array.from(values); + } else { + return Array.from( + values.filter(metric => aggregator === (metric.aggregator ?? 'sum')), + ); + } } async getMetricsAsString(metrics) { @@ -142,18 +151,22 @@ class Registry { this._defaultLabels = {}; } - async getMetricsAsJSON() { + /** + * Retrieve metrics as Objects fit for JSON formatting. + * + * @param {string} [aggregator] - filter by aggregator type, typically used for sum. + * @returns {Promise<*[]>} + */ + async getMetricsAsJSON(aggregator) { const metrics = []; let defaultLabelNames = Object.keys(this._defaultLabels); if (defaultLabelNames.length === 0) { defaultLabelNames = undefined; } - const promises = []; - - for (const metric of this.getMetricsAsArray()) { - promises.push(metric.get()); - } + const promises = this.getMetricsAsArray(aggregator).map(metric => + metric.get(), + ); const resolves = await Promise.all(promises); diff --git a/test/registerTest.js b/test/registerTest.js index ce4a8757..c7133e72 100644 --- a/test/registerTest.js +++ b/test/registerTest.js @@ -340,7 +340,45 @@ describe('Register', () => { expect(escapedResult).toMatch(/\\"/); }); - describe('should output metrics as JSON', () => { + describe('getMetricsAsArray()', () => { + it('should return metrics', async () => { + register.registerMetric(getMetric()); + const output = await register.getMetricsAsArray(); + + expect(output.length).toEqual(1); + expect(output[0].name).toEqual('test_metric'); + expect(output[0].get).toBeInstanceOf(Function); + }); + + describe('with aggregator argument', () => { + it('should filter out other aggregators', async () => { + const max = getMetric('max_metric'); + max.aggregator = 'max'; + const min = getMetric('min_metric'); + min.aggregator = 'min'; + + register.registerMetric(max); + register.registerMetric(min); + + const output = await register.getMetricsAsArray('min'); + + expect(output.length).toEqual(1); + expect(output[0].name).toEqual('min_metric'); + expect(output[0].get).toBeInstanceOf(Function); + }); + + it('should treat "sum" as the default', async () => { + register.registerMetric(getMetric()); + const output = await register.getMetricsAsArray('sum'); + + expect(output.length).toEqual(1); + expect(output[0].name).toEqual('test_metric'); + expect(output[0].get).toBeInstanceOf(Function); + }); + }); + }); + + describe('getMetricsAsJSON()', () => { it('should output metrics as JSON', async () => { register.registerMetric(getMetric()); const output = await register.getMetricsAsJSON(); From ef1cea25d2fdba39eb1f00ec6991894620c6fdc0 Mon Sep 17 00:00:00 2001 From: Jason Marshall Date: Fri, 14 Aug 2026 19:03:01 -0700 Subject: [PATCH 06/10] Add shutdown lifecycle for workers. Signed-off-by: Jason Marshall --- example/worker.js | 58 +++++++++++++- lib/registry.js | 2 +- lib/worker.js | 139 ++++++++++++++++++++++++++++++--- test/clusterTest.js | 2 - test/registerTest.js | 9 --- test/workerTest.js | 181 +++++++++++++++++++++++++++++++++++-------- 6 files changed, 334 insertions(+), 57 deletions(-) diff --git a/example/worker.js b/example/worker.js index ba24662d..6a87fcdc 100644 --- a/example/worker.js +++ b/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) { diff --git a/lib/registry.js b/lib/registry.js index 482750e4..9f0fc67d 100644 --- a/lib/registry.js +++ b/lib/registry.js @@ -53,7 +53,7 @@ class Registry { return Array.from(values); } else { return Array.from( - values.filter(metric => aggregator === (metric.aggregator ?? 'sum')), + values.filter(metric => aggregator === metric.aggregator), ); } } diff --git a/lib/worker.js b/lib/worker.js index 859051d7..866f7d1b 100644 --- a/lib/worker.js +++ b/lib/worker.js @@ -30,9 +30,12 @@ const { waitFor } = require('./util'); const { isMainThread, threadId, BroadcastChannel } = worker; const debug = debuglog('prom:metrics:worker'); +const ACK = '@prometheus-io/client:ack'; const ANNOUNCEMENT = '@prometheus-io/client:announcement'; const GET_METRICS_REQ = '@prometheus-io/client:getMetricsReq'; const GET_METRICS_RES = '@prometheus-io/client:getMetricsRes'; +const GOODBYE = '@prometheus-io/client:goodbye'; + const ANNOUNCEMENT_CHANNEL = new BroadcastChannel( '@prometheus-io/client:announce', ).unref(); @@ -42,6 +45,7 @@ let listenersAdded = false; let requestCtr = 0; // Concurrency control const requests = new Map(); // Pending requests for workers' local metrics. const workers = new Map(); +let historicMetrics = []; // Metrics from dead workers class WorkerRegistry extends Registry { /** @@ -74,10 +78,15 @@ class WorkerRegistry extends Registry { ); if (orderedWorkers.length === 0) { - debug('No workers found for requestId', requestId); - return ''; + if (historicMetrics.length === 0) { + debug('No data found for requestId', requestId); + return ''; + } else { + debug('No workers found for requestId', requestId); + } } + const metricSnapshot = historicMetrics; const responseHandlers = new Map(); const responsePromises = orderedWorkers.map( entry => @@ -91,7 +100,10 @@ class WorkerRegistry extends Registry { const request = { responseHandlers, - promise: waitFor(this.#gather(requestId, responsePromises), 5_000), + promise: waitFor( + this.#gather(requestId, metricSnapshot, responsePromises), + 5_000, + ), }; requests.set(requestId, request); @@ -120,13 +132,16 @@ class WorkerRegistry extends Registry { /** * Collect the data for a metrics request. * @param requestId {number} + * @param {any[]} historical - Previously collected values * @param promises {Promise[]} * @returns {Promise} */ - async #gather(requestId, promises) { + async #gather(requestId, historical, promises) { + debug('Gathering data...', requestId); const responses = await Promise.all(promises); + debug('Aggregating data...', requestId); const metrics = responses.flatMap(response => response.metrics); - return Registry.aggregate(metrics).metrics(); + return Registry.aggregate([historical, ...metrics]).metrics(); } get contentType() { @@ -136,17 +151,61 @@ class WorkerRegistry extends Registry { /** * Orderly shutdown of the registry. * - * This is meant to be called prior to`process.exit()` to facilitate accurate metrics. + * 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. + * If this instance is a worker, then it will flush all sum stats to the + * primary to prevent data loss on subsequent scrapes. + * If this function is called twice, it will only wait for the outstanding + * shutdown request to complete or timeout. + * + * @param {number} [timeout] - how long to wait for an orderly shutdown * @returns {Promise} */ - async shutdown() { - if (this.primary) { - const outstanding = [...requests.values().map(entry => entry.promise)]; + async shutdown(timeout = 5_000) { + const outstanding = [...requests.values().map(entry => entry.promise)]; + if (outstanding.length > 0) { await Promise.allSettled(outstanding); + } else if (!this.primary) { + const name = `@prometheus-io/client:worker:${threadId}`; + const channel = new BroadcastChannel(name).unref(); + + const responseHandlers = new Map(); + const acknowledgement = new Promise((resolveResponse, rejectResponse) => { + responseHandlers.set('ack', { + resolve: resolveResponse, + reject: rejectResponse, + }); + }); + + const requestId = requestCtr++; + const request = { + responseHandlers, + promise: acknowledgement, + }; + + requests.set(requestId, request); + + try { + const metrics = await Promise.all( + registries.map(r => r.getMetricsAsJSON('sum')), + ); + + debug('sending goodbye message from', threadId); + + channel.postMessage({ + type: GOODBYE, + requestId, + threadId, + metrics, + }); + + return await waitFor(acknowledgement, timeout); + } finally { + channel.close(); + } } } @@ -168,6 +227,10 @@ class WorkerRegistry extends Registry { return Registry.aggregate(metricsArr, registryType); } + static workerCount() { + return workers.size; + } + /** * Sets the registry or registries to be aggregated. Call from workers to * use a registry/registries other than the default global registry. @@ -223,12 +286,36 @@ function addListeners(primary) { metrics, }); } catch (error) { + debug(error); channel.postMessage({ type: GET_METRICS_RES, requestId: message.requestId, error: error.message, }); } + } else { + debug('unexpected message for', threadId, message); + } + }); + + channel.addEventListener('message', async event => { + const message = event.data; + + if (message.type === ACK) { + const request = requests.get(message.requestId); + + if (request === undefined) { + debug('unexpected ACK message for', threadId); + } else { + const resolve = request.responseHandlers.get('ack').resolve; + + debug( + 'received acknowledgement for goodbye message from', + message.threadId, + ); + + resolve(message); + } } }); @@ -254,6 +341,8 @@ async function primaryListener(event) { return; } + debug('Registering worker', message.threadId); + const workerChannel = new BroadcastChannel(workerName, {}).unref(); workers.set(workerName, { name: workerName, @@ -262,10 +351,10 @@ async function primaryListener(event) { }); workerChannel.addEventListener('close', () => { - workers.delete(workerName); + debug('channel closed for worker', workerName); }); - workerChannel.addEventListener('message', workerEvent => { + workerChannel.addEventListener('message', async workerEvent => { const workerMessage = workerEvent.data; if (workerMessage.type === GET_METRICS_RES) { @@ -290,6 +379,34 @@ async function primaryListener(event) { metrics: workerMessage.metrics, }); } + } else if (workerMessage.type === GOODBYE) { + if (!workers.has(workerName)) { + debug('goodbye message from unknown worker', workerMessage.threadId); + } else { + debug('received goodbye message from', workerMessage.threadId); + + workerChannel.postMessage({ + type: ACK, + requestId: workerMessage.requestId, + threadId, + }); + + try { + const metrics = [historicMetrics, ...workerMessage.metrics]; + + historicMetrics = Registry.aggregate(metrics).getMetricsAsArray(); + debug('collected metrics data from', workerMessage.threadId); + } catch (error) { + console.error( + 'error collecting shutdown metrics', + workerMessage.threadId, + error, + ); + } finally { + workers.delete(workerName); + workerChannel.close(); + } + } } }); } diff --git a/test/clusterTest.js b/test/clusterTest.js index 49826fec..de89a4d1 100644 --- a/test/clusterTest.js +++ b/test/clusterTest.js @@ -17,8 +17,6 @@ const cluster = require('cluster'); const process = require('process'); const Registry = require('../lib/cluster'); -const AggregatorRegistry = require('../lib/cluster'); -const { BroadcastChannel } = require('worker_threads'); const { setTimeout: delay } = require('timers/promises'); const ANNOUNCEMENT = '@prometheus-io/client:announcement'; diff --git a/test/registerTest.js b/test/registerTest.js index c7133e72..4ac6a346 100644 --- a/test/registerTest.js +++ b/test/registerTest.js @@ -366,15 +366,6 @@ describe('Register', () => { expect(output[0].name).toEqual('min_metric'); expect(output[0].get).toBeInstanceOf(Function); }); - - it('should treat "sum" as the default', async () => { - register.registerMetric(getMetric()); - const output = await register.getMetricsAsArray('sum'); - - expect(output.length).toEqual(1); - expect(output[0].name).toEqual('test_metric'); - expect(output[0].get).toBeInstanceOf(Function); - }); }); }); diff --git a/test/workerTest.js b/test/workerTest.js index 4993bf27..68419c42 100644 --- a/test/workerTest.js +++ b/test/workerTest.js @@ -18,9 +18,11 @@ const { setTimeout: delay } = require('timers/promises'); const { BroadcastChannel } = require('worker_threads'); const Registry = require('../lib/worker'); +const ACK = '@prometheus-io/client:ack'; const ANNOUNCEMENT = '@prometheus-io/client:announcement'; const GET_METRICS_REQ = '@prometheus-io/client:getMetricsReq'; const GET_METRICS_RES = '@prometheus-io/client:getMetricsRes'; +const GOODBYE = '@prometheus-io/client:goodbye'; function metric(value) { return { @@ -56,7 +58,7 @@ describe.each([ '@prometheus-io/client:announce', ).unref(); const responders = [1, 2, 3].map(threadId => { - const name = `@prometheus-io/client:test-worker:${threadId}`; + const name = `@prometheus-io/client:worker:${threadId}`; const channel = new BroadcastChannel(name).unref(); announcementChannel.postMessage({ @@ -102,57 +104,174 @@ describe.each([ for (const responder of responders) responder.channel.close(); } }); + + it('accumulate stats from terminated workers', async () => { + jest.resetModules(); + const AggregatorRegistry = require('../lib/worker'); + const registry = new AggregatorRegistry(regType); + const announcementChannel = new BroadcastChannel( + '@prometheus-io/client:announce', + ).unref(); + + const threadId = 134; + const name = `@prometheus-io/client:worker:${threadId}`; + const channel = new BroadcastChannel(name).unref(); + + announcementChannel.postMessage({ + type: ANNOUNCEMENT, + name, + threadId, + }); + + await delay(5); // Let announcements arrive + + const ack = new Promise(resolve => { + channel.addEventListener('message', async event => { + if (event.data.type === ACK) { + resolve(event); + } + }); + }); + + channel.postMessage({ + type: GOODBYE, + threadId, + metrics: [[metric(0.123456)]], + }); + + await ack; + + try { + const result = await registry.workerMetrics(); + expect(result).toContain('test_metric 0.123456'); + } finally { + announcementChannel.close(); + channel.close(); + } + }); }); describe('shutdown()', () => { let AggregatorRegistry; - beforeEach(() => { + let announcementChannel; + let registry; + let discovery; + + beforeEach(async () => { jest.resetModules(); AggregatorRegistry = require('../lib/worker'); + + announcementChannel = new BroadcastChannel( + '@prometheus-io/client:announce', + ).unref(); + + registry = new AggregatorRegistry(regType); + + discovery = new Promise(resolve => { + announcementChannel.addEventListener('message', async event => { + if (event.data.type === ANNOUNCEMENT && !event.data.primary) { + resolve(event); + } + }); + }); }); - it('returns immediately on no outstanding requests', async () => { - const registry = new AggregatorRegistry(); + afterEach(() => { + announcementChannel.close(); + }); + it('returns immediately on no outstanding requests', async () => { await expect(registry.shutdown()).resolves.not.toThrow(); }); - it('waits for pending requests', async () => { - const registry = new AggregatorRegistry(); - - const announcementChannel = new BroadcastChannel( - '@prometheus-io/client:announce', - ).unref(); + it('sends data back to the primary', async () => { + jest.resetModules(); + AggregatorRegistry = require('../lib/worker'); - const threadId = 22; - const name = `@prometheus-io/client:test-worker:${threadId}`; + const workerRegistry = new AggregatorRegistry(regType, false); + const name = `@prometheus-io/client:worker:0`; const channel = new BroadcastChannel(name).unref(); - announcementChannel.postMessage({ - type: ANNOUNCEMENT, - name, - threadId, + const { Gauge } = require('../index'); + const gauge = new Gauge({ name: 'primary_gauge_test', help: 'test' }); + + gauge.set(0.8675309); + + let metrics; + + // wait until worker has processed the ACk before continuing + const acknowledged = new Promise(resolve => { + channel.addEventListener('message', async event => { + if (event.data.type === GOODBYE) { + metrics = event.data.metrics; + channel.postMessage({ type: ACK, requestId: 0, threadId: 0 }); + } else if (event.data.type === ACK) { + resolve(metrics); + } + }); }); - await delay(5); // Let announcements arrive + try { + await workerRegistry.shutdown(); + const expected = { + aggregator: 'sum', + help: 'test', + name: 'primary_gauge_test', + type: 'gauge', + values: [ + { + labels: {}, + value: 0.8675309, + }, + ], + }; + + await expect(acknowledged).resolves.toEqual([[expected]]); + } finally { + channel.close(); + } + }); - announcementChannel.addEventListener('message', async event => { - if (event.data.type === GET_METRICS_REQ) { - channel.postMessage({ - type: GET_METRICS_RES, - requestId: event.data.requestId, - threadId, - metrics: [[metric(2)]], - }); - } + describe('with workers', () => { + let channel; + + beforeEach(async () => { + const threadId = 22; + const name = `@prometheus-io/client:test-worker:${threadId}`; + channel = new BroadcastChannel(name).unref(); + + announcementChannel.postMessage({ + type: ANNOUNCEMENT, + name, + threadId, + }); + + announcementChannel.addEventListener('message', async event => { + if (event.data.type === GET_METRICS_REQ) { + channel.postMessage({ + type: GET_METRICS_RES, + requestId: event.data.requestId, + threadId: 22, + metrics: [[metric(2)]], + }); + } + }); + + await discovery; }); - const results = []; - const promise = registry.workerMetrics().then(() => results.push(1)); - const shutdown = registry.shutdown().then(() => results.push(2)); - await Promise.all([promise, shutdown]); + afterEach(() => { + channel.close(); + }); - expect(results).toEqual([1, 2]); + it('waits for pending requests', async () => { + const results = []; + const promise = registry.workerMetrics().then(() => results.push(1)); + const shutdown = registry.shutdown().then(() => results.push(2)); + await Promise.all([promise, shutdown]); + + expect(results).toEqual([1, 2]); + }); }); }); From 8d446a57699024cbd12909f14a3ad295df453754 Mon Sep 17 00:00:00 2001 From: Jason Marshall Date: Sat, 15 Aug 2026 13:17:43 -0700 Subject: [PATCH 07/10] Copy new shutdown logic from worker.js to cluster.js TODO: Needs example updated Signed-off-by: Jason Marshall --- example/cluster.js | 43 ++++++++++++ lib/cluster.js | 129 ++++++++++++++++++++++++++++------ test/clusterTest.js | 168 ++++++++++++++++++++++++++++++++++++++++---- test/workerTest.js | 10 +-- 4 files changed, 310 insertions(+), 40 deletions(-) diff --git a/example/cluster.js b/example/cluster.js index 40b77421..e0719707 100644 --- a/example/cluster.js +++ b/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'); } diff --git a/lib/cluster.js b/lib/cluster.js index 14e2012b..bc5dc49e 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -35,15 +35,18 @@ let cluster = () => { }; const debug = debuglog('prom:metrics:cluster'); +const ACK = '@prometheus-io/client:ack'; const ANNOUNCEMENT = '@prometheus-io/client:announcement'; const GET_METRICS_REQ = '@prometheus-io/client:getMetricsReq'; const GET_METRICS_RES = '@prometheus-io/client:getMetricsRes'; +const GOODBYE = '@prometheus-io/client:goodbye'; let registries = [Registry.globalRegistry]; let listenersAdded = false; let requestCtr = 0; // Concurrency control const requests = new Map(); // Pending requests for workers' local metrics. const workers = new Map(); +let historicMetrics = []; // Metrics from dead workers class AggregatorRegistry extends Registry { /** @@ -72,6 +75,7 @@ class AggregatorRegistry extends Registry { debug('No workers found for requestId', requestId); } + const metricSnapshot = historicMetrics; const responseHandlers = new Map(); const workerMetrics = orderedWorkers.map( worker => @@ -83,10 +87,13 @@ class AggregatorRegistry extends Registry { }), ); - const promises = [this.#selfMetrics(), ...workerMetrics]; + const responsePromises = [this.#selfMetrics(), ...workerMetrics]; const request = { responseHandlers, - promise: waitFor(this.#gather(requestId, promises), 5_000), + promise: waitFor( + this.#gather(requestId, metricSnapshot, responsePromises), + 5_000, + ), }; requests.set(requestId, request); @@ -118,14 +125,15 @@ class AggregatorRegistry extends Registry { /** * Collect the data for a metrics request. - * @param requestId - * @param promises - * @returns {Promise} + * @param requestId {number} + * @param {any[]} historical - Previously collected values + * @param promises {Promise[]} + * @returns {Promise} */ - async #gather(requestId, promises) { + async #gather(requestId, historical, promises) { const responses = await Promise.all(promises); const metrics = responses.flatMap(response => response.metrics); - return Registry.aggregate(metrics).metrics(); + return Registry.aggregate([historical, ...metrics]).metrics(); } get contentType() { @@ -135,17 +143,57 @@ class AggregatorRegistry extends Registry { /** * Orderly shutdown of the registry. * - * This is meant to be called prior to`process.exit()` to facilitate accurate metrics. + * 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. + * If this instance is a worker, then it will flush all sum stats to the + * primary to prevent data loss on subsequent scrapes. + * If this function is called twice, it will only wait for the outstanding + * shutdown request to complete or timeout. + * + * @param {number} [timeout] - how long to wait for an orderly shutdown * @returns {Promise} */ - async shutdown() { - if (cluster().isPrimary) { - const outstanding = requests.values().map(entry => entry.promise); + async shutdown(timeout = 5_000) { + const outstanding = [...requests.values().map(entry => entry.promise)]; + + if (outstanding.length > 0) { + await Promise.allSettled(outstanding); + } else if (!cluster().isPrimary) { + const responseHandlers = new Map(); + const acknowledgement = new Promise((resolveResponse, rejectResponse) => { + responseHandlers.set('ack', { + resolve: resolveResponse, + reject: rejectResponse, + }); + }); + + const requestId = requestCtr++; + const request = { + responseHandlers, + promise: acknowledgement, + }; + + requests.set(requestId, request); - await Promise.all(outstanding); + try { + const metrics = await Promise.all( + registries.map(r => r.getMetricsAsJSON('sum')), + ); + + debug('sending goodbye message from', process.pid); + + processSend({ + type: GOODBYE, + requestId, + metrics, + }); + + return await waitFor(acknowledgement, timeout); + } finally { + /* empty */ + } } } @@ -204,14 +252,7 @@ function addListeners() { announce(); } else { replaceListener('message', process, workerListener); - - 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 }); - } + processSend({ type: ANNOUNCEMENT }); } } @@ -252,6 +293,18 @@ async function workerListener(message) { }); } } + } else if (message.type === ACK) { + const request = requests.get(message.requestId); + + if (request === undefined) { + debug('unexpected goodbye message for', process.pid); + } else { + const resolve = request.responseHandlers.get('ack').resolve; + + debug('received acknowledgement for goodbye message from parent'); + + resolve(message); + } } } @@ -295,6 +348,28 @@ async function primaryListener(worker, event) { metrics: event.metrics, }); } + } else if (event.type === GOODBYE) { + if (!workers.has(worker.id)) { + debug('goodbye message from unknown worker', worker.id); + } else { + debug('received goodbye message from', worker.id); + + worker.send({ + type: ACK, + requestId: event.requestId, + }); + + try { + const metrics = [historicMetrics, ...event.metrics]; + + historicMetrics = Registry.aggregate(metrics).getMetricsAsArray(); + debug('collected metrics data from', worker.id); + } catch (error) { + console.error('error collecting shutdown metrics', worker.id, error); + } finally { + workers.delete(worker.id); + } + } } } @@ -334,4 +409,18 @@ function replaceListener(messageType, emitter, fn) { emitter.on(messageType, fn); } +/** + * Wrap process.send() to deal with odd runtimes and lifecycle teardown issues. + * @param {object} message - message payload + */ +function processSend(message) { + 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(message); + } +} + module.exports = AggregatorRegistry; diff --git a/test/clusterTest.js b/test/clusterTest.js index de89a4d1..4b812b90 100644 --- a/test/clusterTest.js +++ b/test/clusterTest.js @@ -19,9 +19,11 @@ const process = require('process'); const Registry = require('../lib/cluster'); const { setTimeout: delay } = require('timers/promises'); +const ACK = '@prometheus-io/client:ack'; const ANNOUNCEMENT = '@prometheus-io/client:announcement'; const GET_METRICS_REQ = '@prometheus-io/client:getMetricsReq'; const GET_METRICS_RES = '@prometheus-io/client:getMetricsRes'; +const GOODBYE = '@prometheus-io/client:goodbye'; function metric(value) { return { @@ -70,8 +72,29 @@ describe.each([ }); describe('aggregatorRegistry.clusterMetrics()', () => { + let AggregatorRegistry; + let listener; + let discovery; + + beforeEach(() => { + jest.resetModules(); + AggregatorRegistry = require('../lib/cluster'); + + discovery = new Promise(resolve => { + listener = message => { + resolve(message); + }; + + cluster.on('message', listener); + }); + }); + + afterEach(() => { + cluster.off('message', listener); + jest.restoreAllMocks(); + }); + it('works properly if there are no cluster workers', async () => { - const AggregatorRegistry = require('../lib/cluster'); const ar = new AggregatorRegistry(regType); const metrics = await ar.clusterMetrics(); expect(metrics.trim()).toEqual(''); @@ -88,8 +111,6 @@ describe.each([ it('aggregates worker responses in worker id order', async () => { const originalWorkers = cluster.workers; - jest.resetModules(); - const AggregatorRegistry = require('../lib/cluster'); const registry = new AggregatorRegistry(regType); const workers = Object.fromEntries( [1, 2, 3].map(id => [ @@ -108,8 +129,9 @@ describe.each([ }); try { - const result = registry.clusterMetrics(); + await discovery; + const result = registry.clusterMetrics(); for (const [id, value] of [ [3, 0.3437699], [1, 0.5848208], @@ -132,8 +154,6 @@ describe.each([ }); it('aggregates telemetry from primary thread', async () => { - jest.resetModules(); - require('../lib/cluster'); const { Gauge } = require('../index'); @@ -151,12 +171,70 @@ describe.each([ gauge.remove(); } }); + + it('accumulate stats from terminated workers', async () => { + const originalWorkers = cluster.workers; + const registry = new AggregatorRegistry(regType); + const worker = { + id: 37, + isConnected: () => true, + send: jest.fn(), + }; + + cluster.workers = [worker]; + + const metrics = new Promise(resolve => { + worker.send.mockImplementationOnce(message => { + resolve(message.metrics); + }); + }); + + try { + cluster.emit('message', worker, { type: ANNOUNCEMENT }); + + await discovery; + + cluster.emit('message', worker, { + type: GOODBYE, + requestId: 0, + metrics: [[metric(0.654321)]], + }); + + await metrics; + + const result = registry.clusterMetrics(); + await expect(result).resolves.toContain('test_metric 0.654321'); + } finally { + cluster.workers = originalWorkers; + } + }); }); describe('shutdown()', () => { + let AggregatorRegistry; + let listener; + let discovery; + + beforeEach(() => { + jest.resetModules(); + AggregatorRegistry = require('../lib/cluster'); + + discovery = new Promise(resolve => { + listener = message => { + resolve(message); + }; + + cluster.on('message', listener); + }); + }); + + afterEach(() => { + cluster.off('message', listener); + jest.restoreAllMocks(); + }); + it('returns immediately on no outstanding requests', async () => { - const AggregatorRegistry = require('../lib/cluster'); - const ar = new AggregatorRegistry(); + const ar = new AggregatorRegistry(regType); await expect(ar.shutdown()).resolves.not.toThrow(); }); @@ -177,6 +255,8 @@ describe.each([ cluster.emit('message', worker, { type: ANNOUNCEMENT }); try { + await discovery; + const results = []; const promise = registry.clusterMetrics().then(() => results.push(1)); const shutdown = registry.shutdown().then(() => results.push(2)); @@ -195,6 +275,64 @@ describe.each([ cluster.workers = originalWorkers; } }); + + it('sends data back to main', async () => { + jest.resetModules(); + + // Fake a worker thread + jest.doMock('cluster', () => { + return { isPrimary: false }; + }); + + const connectedDescriptor = Object.getOwnPropertyDescriptor( + process, + 'connected', + ); + + process.connected = true; + + const send = jest.spyOn(process, 'send'); + const AggregatorRegistry = require('../lib/cluster'); + const workerRegistry = new AggregatorRegistry(regType); + + const { Gauge } = require('../index'); + const gauge = new Gauge({ name: 'primary_gauge_test', help: 'test' }); + + gauge.set(0.8675309); + + try { + const metrics = new Promise(resolve => { + send.mockImplementationOnce(message => { + process.emit('message', { type: ACK, requestId: 0 }); + resolve(message.metrics); + }); + }); + + await workerRegistry.shutdown(); + const expected = { + aggregator: 'sum', + help: 'test', + name: 'primary_gauge_test', + type: 'gauge', + values: [ + { + labels: {}, + value: 0.8675309, + }, + ], + }; + + await expect(metrics).resolves.toEqual([[expected]]); + } finally { + jest.dontMock('cluster'); + gauge.remove(); + if (connectedDescriptor) { + Object.defineProperty(process, 'connected', connectedDescriptor); + } else { + delete process.connected; + } + } + }); }); describe('message handling', () => { @@ -230,10 +368,13 @@ describe('worker message handling', () => { 'connected', ); + const send = jest.spyOn(process, 'send'); + const AggregatorRegistry = require('../lib/cluster'); new AggregatorRegistry(); - const send = jest.spyOn(process, 'send'); + send.mockReset(); + let listener; try { @@ -253,18 +394,19 @@ describe('worker message handling', () => { process.connected = false; listener({ type: GET_METRICS_REQ, requestId: 1 }); - await new Promise(resolve => setImmediate(resolve)); - expect(send).not.toHaveBeenCalled(); + await delay(0); + + expect(send).toHaveBeenCalledTimes(0); // Announcement } finally { + jest.resetModules(); + jest.dontMock('cluster'); process.removeListener('message', listener); if (connectedDescriptor) { Object.defineProperty(process, 'connected', connectedDescriptor); } else { delete process.connected; } - jest.resetModules(); - jest.clearAllMocks(); } }); }); diff --git a/test/workerTest.js b/test/workerTest.js index 68419c42..0e7eb0c3 100644 --- a/test/workerTest.js +++ b/test/workerTest.js @@ -197,16 +197,12 @@ describe.each([ gauge.set(0.8675309); - let metrics; - // wait until worker has processed the ACk before continuing - const acknowledged = new Promise(resolve => { + const metrics = new Promise(resolve => { channel.addEventListener('message', async event => { if (event.data.type === GOODBYE) { - metrics = event.data.metrics; channel.postMessage({ type: ACK, requestId: 0, threadId: 0 }); - } else if (event.data.type === ACK) { - resolve(metrics); + resolve(event.data.metrics); } }); }); @@ -226,7 +222,7 @@ describe.each([ ], }; - await expect(acknowledged).resolves.toEqual([[expected]]); + await expect(metrics).resolves.toEqual([[expected]]); } finally { channel.close(); } From aea554af69a8bf0584833a79addf26413b8ac3fa Mon Sep 17 00:00:00 2001 From: Jason Marshall Date: Sun, 16 Aug 2026 16:48:21 -0700 Subject: [PATCH 08/10] Update the advice in Workers.md to include options and a conversation about space complexity. Signed-off-by: Jason Marshall --- Workers.md | 85 ++++++++++++++++++++++++++++++------------------------ 1 file changed, 48 insertions(+), 37 deletions(-) diff --git a/Workers.md b/Workers.md index 2549662d..1c41ed81 100644 --- a/Workers.md +++ b/Workers.md @@ -1,5 +1,17 @@ # 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 @@ -19,52 +31,51 @@ telemetry collection. If you're thinking of adding Prometheus telemetry to your a worker thread, one of your first concerns should be in reducing the number of unrecoverable errors your code contains. -In the case of worker threads, sometimes short-lived is a feature, and in others it's an -inevitability. In these cases, the prometheus client will need a little help from you on tracking -the lifecycle of those workers. - -The biggest challenge is that if a worker is unresponsive, then the prometheus client will time out -while trying to collect the aggregated metrics, resulting in NO telemetry being reported at all. -Avoiding this problem would come at a substantial memory premium, as the sum values from every -worker would need to be retained. +## Strategies -Additionally, the Prometheus client retains metadata for every worker it knows about. If you cycle -workers frequently, then that table will grow without bounds. If we knew for certain that a worker -was gone, then some of that metadata can be consolidated across all defunct workers, and as long as -the cardinality of the metrics does not include process-unique data, such as the threadId, then -twenty dead worker is no more expensive than one. +### Delegation -Because of the nature of workers, it is expected that they may saturate the event loop. That means -that if we 'ping' them to see if they are still responsive, then they might not reply until after -we decided they are dead. If they intermittently respond to requests, then the bookkeeping gets -quite elaborate (expensive). - -As the application author, you have more control and visibility over the lifecycle of your workers, -especially for worker threads. - -## Graceful shutdown +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. -TBD: The final values for gauges may or may not be lost when the process exits. +```javascript +// In worker bootstrapping code: +['SIGHUP', 'SIGINT', 'SIGQUIT', 'SIGTERM'].forEach(sig => { + process.on(sig, async () => { + await registry.shutdown(); + process.exit(0); + }); +}); ``` -// Code example goes here -``` - -## Lifecycle events -The main thread can also listen for lifecycle events for its workers and inform us when -any of them exit prematurely. This solution will still result in data loss, and telemetry artifacts, -but will also reduce the number of collection errors and can help the Prometheus client to clean up -metadata related to the lost worker. +See [the example](examples/workerTest.js) for a complete rundown, including the parent process +signalling workers to shut themselves down. -We could fix the data loss by retaining data from the previous collection interval, that would -require a good deal of extra storage to facilitate, and therefore would be a substantial tax on -well-behaved workers. Alternatively, we could flag some workers as problematic (example: you have -three pools of workers, and only one tends to crash), but that is currently not supported. +#### Space Complexity -For now, it is recommended that you hook the unhandled exceptions in the Worker itself, then flush -the telemetry data prior to calling `process.exit()`. +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. From a4cb92af7595192ce7a9f7f4145136f35af52f13 Mon Sep 17 00:00:00 2001 From: Jason Marshall Date: Sun, 16 Aug 2026 17:09:51 -0700 Subject: [PATCH 09/10] Rework replaceListener to only do cleanup in test and dev mode ('watch') and warn of undefined behavior in production mode. This reduces the surface area of potential undefined behavior and puts the user on notice to review their application design if they are unintentionally triggering this in pre-prod or production modes. Add a big warning to the bottom of the README defining the bounds of expected behavior from this feature. Signed-off-by: Jason Marshall --- CHANGELOG.md | 9 +++++---- README.md | 30 ++++++++++++++++++++++++++++++ lib/cluster.js | 34 ++++++++++++++++++++++++---------- 3 files changed, 59 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 919dc723..f2abd9cd 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 9325f755..13f78213 100644 --- a/README.md +++ b/README.md @@ -649,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. diff --git a/lib/cluster.js b/lib/cluster.js index bc5dc49e..06d1553d 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -246,12 +246,15 @@ function addListeners() { listenersAdded = true; if (cluster().isPrimary) { - replaceListener('message', cluster(), primaryListener); - replaceListener('disconnect', cluster(), disconnect); + scanListeners('message', cluster(), primaryListener); + cluster().on('message', primaryListener); + scanListeners('disconnect', cluster(), disconnect); + cluster().on('disconnect', disconnect); announce(); } else { - replaceListener('message', process, workerListener); + scanListeners('message', process, workerListener); + process.on('message', workerListener); processSend({ type: ANNOUNCEMENT }); } } @@ -313,6 +316,7 @@ async function workerListener(message) { * * Whereas clusters are a top-level activity, multiple modules may start their * own workers and require telemetry collection. + * @param worker {Worker} * @param event {MessageEvent} */ @@ -387,26 +391,36 @@ function announce() { } /** - * Replace any listeners with new ones. + * Look for and complain about duplicate listeners. + * + * In test and development mode, this function will remove duplicate listeners. + * In any other mode (eg, production) it issues a warning about undefined behavior. * * @param messageType * @param emitter {EventEmitter} * @param fn */ -function replaceListener(messageType, emitter, fn) { +function scanListeners(messageType, emitter, fn) { // Reloading a module creates a unique instance of each function, so the - // identity checks is cluster.off() will fail. + // identity checks in 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); + if (['test', 'development'].includes(process.env.NODE_ENV)) { + debug('removing duplicate listener', messageType); + emitter.off(messageType, listener); + } else { + console.warn( + 'Loading multiple instances of @prometheus-io/client_js will result in data loss.', + ); + console.warn( + 'Please review your architecture to ensure that a single copy is loaded at startup and retained throughout the application lifecycle.', + ); + } } } - - emitter.on(messageType, fn); } /** From d8be5a10c76584bd143c6b67c1b1b760a680203e Mon Sep 17 00:00:00 2001 From: Jason Marshall Date: Tue, 18 Aug 2026 14:37:04 -0700 Subject: [PATCH 10/10] Apply same timing fix for ANNOUNCEMENT that I applied to other tests to avoid timing issues with tests. Signed-off-by: Jason Marshall --- test/workerTest.js | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/test/workerTest.js b/test/workerTest.js index 0e7eb0c3..ae3be168 100644 --- a/test/workerTest.js +++ b/test/workerTest.js @@ -57,6 +57,15 @@ describe.each([ const announcementChannel = new BroadcastChannel( '@prometheus-io/client:announce', ).unref(); + + const discovery = new Promise(resolve => { + announcementChannel.addEventListener('message', async event => { + if (event.data.type === ANNOUNCEMENT && !event.data.primary) { + resolve(event); + } + }); + }); + const responders = [1, 2, 3].map(threadId => { const name = `@prometheus-io/client:worker:${threadId}`; const channel = new BroadcastChannel(name).unref(); @@ -70,7 +79,7 @@ describe.each([ return { threadId, channel }; }); - await delay(5); // Let announcements arrive + await discovery; // Let announcements arrive let finishSendingResponses; const responsesSent = new Promise(resolve => {