Skip to content

Record cluster worker scrape failures #850

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nightjard wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Record cluster worker scrape failures #850

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
801c14c
Record cluster worker scrape failures
nightjard Sep 4, 2026
1d303f6
Address scrape failure review feedback
nightjard Sep 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -630,6 +630,12 @@ aggregation method, set the `aggregator` property in the metric config to one of
'sum', 'first', 'min', 'max', 'average' or 'omit'. (See `lib/metrics/version.js`
for an example.)

Failed cluster collections are recorded in the
`prom_client_cluster_worker_scrape_failures` histogram. Each observation is the
number of workers that failed to return metrics. Since failed collections reject
without returning partial metrics, the observation is exposed by the next
successful call to `clusterMetrics()`.

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
Expand Down
27 changes: 26 additions & 1 deletion lib/cluster.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
*/

const { debuglog } = require('node:util');
const Histogram = require('./histogram');
const Registry = require('./registry');
const { waitFor } = require('./util');

Expand All @@ -41,6 +42,13 @@ const GET_METRICS_REQ = '@prometheus-io/client:getMetricsReq';
const GET_METRICS_RES = '@prometheus-io/client:getMetricsRes';
const GOODBYE = '@prometheus-io/client:goodbye';

const clusterWorkerScrapeFailures = new Histogram({
name: 'prom_client_cluster_worker_scrape_failures',
help: 'Number of workers that failed to return metrics during a failed cluster scrape.',
buckets: [1, 2, 4, 8, 16, 32],
registers: [],
});

let registries = [Registry.globalRegistry];
let listenersAdded = false;
let requestCtr = 0; // Concurrency control
Expand Down Expand Up @@ -90,6 +98,7 @@ class AggregatorRegistry extends Registry {
const responsePromises = [this.#selfMetrics(), ...workerMetrics];
const request = {
responseHandlers,
workerFailures: 0,
promise: waitFor(
this.#gather(requestId, metricSnapshot, responsePromises),
5_000,
Expand All @@ -105,6 +114,16 @@ class AggregatorRegistry extends Registry {

return await request.promise;
} catch (err) {
let failedWorkers = request.workerFailures;

if (err.message === 'Timeout') {
failedWorkers += request.responseHandlers.size;
}

if (failedWorkers > 0) {
clusterWorkerScrapeFailures.observe(failedWorkers);

@jdmarshall jdmarshall Sep 4, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think observing 0's when there are no workers is probably unlikely to happen but is also extra information about the class of failure we may be dealing with. As written this will never catch unclassified errors such as an NPE in the try block.

For instance, regressions in the Node API are not unknown. I work on a project that doesn't work on node 24.7-24.16, for instance, due to breaking changes in child_process.fork()

}

if (err.message === 'Timeout') {
throw new Error(
`Operation timed out. ${request.responseHandlers.size} outstanding responses.`,
Expand All @@ -118,8 +137,13 @@ class AggregatorRegistry extends Registry {
}

async #selfMetrics() {
const metrics = await Promise.all(
registries.map(r => r.getMetricsAsJSON()),
);
metrics.push([await clusterWorkerScrapeFailures.get()]);

@jdmarshall jdmarshall Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are you sure you're not double-reporting this? That should be in the default registry.


return {
metrics: await Promise.all(registries.map(r => r.getMetricsAsJSON())),
metrics,
};
}

Expand Down Expand Up @@ -348,6 +372,7 @@ async function primaryListener(worker, event) {
request.responseHandlers.delete(worker.id);

if (event.error) {
request.workerFailures++;
response.reject(new Error(event.error));
} else {
response.resolve({
Expand Down
117 changes: 111 additions & 6 deletions test/clusterTest.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ 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 CLUSTER_WORKER_SCRAPE_FAILURES =
'prom_client_cluster_worker_scrape_failures';

function metric(value) {
return {
Expand Down Expand Up @@ -91,18 +93,15 @@ describe.each([

afterEach(() => {
cluster.off('message', listener);
jest.useRealTimers();
jest.restoreAllMocks();
});

it('works properly if there are no cluster workers', async () => {
const ar = new AggregatorRegistry(regType);
const metrics = await ar.clusterMetrics();

if (regType === Registry.OPENMETRICS_CONTENT_TYPE) {
expect(metrics).toContain('# EOF\n');
} else {
expect(metrics.trim()).toEqual('');
}
expect(metrics).toContain(`${CLUSTER_WORKER_SCRAPE_FAILURES}_count 0`);
});

it('formats in the correct content type', async () => {
Expand All @@ -112,7 +111,7 @@ describe.each([
if (regType === Registry.OPENMETRICS_CONTENT_TYPE) {
expect(metrics).toContain('# EOF\n');
} else {
expect(metrics.trim()).toEqual('');
expect(metrics).not.toContain('# EOF\n');
}
});

Expand Down Expand Up @@ -188,6 +187,112 @@ describe.each([
}
});

it('records the number of workers that time out', async () => {
jest.useFakeTimers();

const originalWorkers = cluster.workers;
const registry = new AggregatorRegistry(regType);
const workers = Object.fromEntries(
[1, 2, 3].map(id => [
id,
{
id,
isConnected: () => true,
send: jest.fn(),
},
]),
);
cluster.workers = workers;

Object.values(workers).forEach(worker => {
cluster.emit('message', worker, { type: ANNOUNCEMENT });
});

try {
await discovery;

const failedResult = registry.clusterMetrics();
const rejection = expect(failedResult).rejects.toThrow(
'Operation timed out. 2 outstanding responses.',
);

cluster.emit('message', workers[1], {
type: GET_METRICS_RES,
requestId: 0,
metrics: [[]],
});

await jest.advanceTimersByTimeAsync(5_000);
await rejection;

const recoveredResult = registry.clusterMetrics();
Object.values(workers).forEach(worker => {
cluster.emit('message', worker, {
type: GET_METRICS_RES,
requestId: 1,
metrics: [[]],
});
});

await expect(recoveredResult).resolves.toContain(
`${CLUSTER_WORKER_SCRAPE_FAILURES}_sum 2`,
);
await expect(recoveredResult).resolves.toContain(
`${CLUSTER_WORKER_SCRAPE_FAILURES}_count 1`,
);
} finally {
Object.values(workers).forEach(worker => {
cluster.emit('disconnect', worker);
});
cluster.workers = originalWorkers;
}
});

it('records worker-reported scrape errors', async () => {
const originalWorkers = cluster.workers;
const registry = new AggregatorRegistry(regType);
const worker = {
id: 1,
isConnected: () => true,
send: jest.fn(),
};
cluster.workers = [worker];
cluster.emit('message', worker, { type: ANNOUNCEMENT });

try {
await discovery;

const failedResult = registry.clusterMetrics();
const rejection = expect(failedResult).rejects.toThrow(
'worker collection failed',
);

cluster.emit('message', worker, {
type: GET_METRICS_RES,
requestId: 0,
error: 'worker collection failed',
});
await rejection;

const recoveredResult = registry.clusterMetrics();
cluster.emit('message', worker, {
type: GET_METRICS_RES,
requestId: 1,
metrics: [[]],
});

await expect(recoveredResult).resolves.toContain(
`${CLUSTER_WORKER_SCRAPE_FAILURES}_sum 1`,
);
await expect(recoveredResult).resolves.toContain(
`${CLUSTER_WORKER_SCRAPE_FAILURES}_count 1`,
);
} finally {
cluster.emit('disconnect', worker);
cluster.workers = originalWorkers;
}
});

it('accumulate stats from terminated workers', async () => {
const originalWorkers = cluster.workers;
const registry = new AggregatorRegistry(regType);
Expand Down
Loading