From c340e6ca08ae8560ed58fffcd49a145bd2570378 Mon Sep 17 00:00:00 2001 From: Bryan English Date: Mon, 2 Nov 2026 16:16:31 -0500 Subject: [PATCH] fix: keep handler errors from crashing the app Node rethrows an error from a diagnostics_channel subscriber or a timer as an uncaught exception, and an error from an event listener comes out of emit() and skips the listeners after it. Previously, any bug in a microtel handler could take down the app it was tracing, or keep the app's own listeners from running. Now every function microtel hands to Node is wrapped once, when it is subscribed or attached, and reports its errors as an OTelInstrumentationWarning. The warnings are throttled the way export failures are, one per 30 seconds with a running count, and each carries the original stack. The request it happened on carries on untraced or partly traced. Each export handler settles its export before anything that can throw, and the export deadline settles it too, so a failure a guard catches never leaves an awaited flush() waiting. Signed-off-by: Bryan English --- README.md | 5 + lib/flush.js | 100 ++++++----- lib/guard.js | 46 +++++ lib/instrumentations.js | 46 +++-- test/fixtures/export-failure-child.js | 111 ++++++++++++ .../fixtures/instrumentation-failure-child.js | 161 ++++++++++++++++++ test/test-instrumentation-failure.js | 74 ++++++++ 7 files changed, 489 insertions(+), 54 deletions(-) create mode 100644 lib/guard.js create mode 100644 test/fixtures/export-failure-child.js create mode 100644 test/fixtures/instrumentation-failure-child.js create mode 100644 test/test-instrumentation-failure.js diff --git a/README.md b/README.md index e8c5e63..cff7a94 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,11 @@ Tracing is activated at most once per process. Calling `init()` a second time is a no-op, even with different options. There is no way to stop tracing once started. Invalid options throw. +microtel catches errors in its own handlers, so a bug in microtel can't crash +the app it traces. Each error is reported as an `OTelInstrumentationWarning` +process warning, at most once every 30 seconds with a running count. The +request it happened on carries on untraced or partly traced. + ### Options | Option | Type | Default | Description | diff --git a/lib/flush.js b/lib/flush.js index 670c375..a6ce18f 100644 --- a/lib/flush.js +++ b/lib/flush.js @@ -5,6 +5,8 @@ const http = require('node:http'); const https = require('node:https'); const { clearTimeout, setImmediate, setTimeout } = require('node:timers'); +const { guard } = require('./guard'); + const { getEndpoint, getExportHeaders, @@ -163,13 +165,10 @@ function addSpan(span) { // filled the buffer. if (!flushScheduled) { flushScheduled = true; - setImmediate(() => { - flushScheduled = false; - flush(); - }); + setImmediate(guardedDeferredFlush); } } else if (flushTimer == null) { - flushTimer = setTimeout(flush, flushIntervalMs); + flushTimer = setTimeout(guardedTimerFlush, flushIntervalMs); flushTimer.unref(); } } @@ -239,6 +238,15 @@ function flush() { return settleAll(pending); } +function deferredFlush() { + flushScheduled = false; + flush(); +} + +const guardedDeferredFlush = guard('deferred flush', deferredFlush); +const guardedTimerFlush = guard('flush timer', flush); +const guardedBeforeExitFlush = guard('beforeExit flush', flush); + // Counts failures and emits an OTelExportWarning, throttled to at most one // per kWarningThrottleMs. function warnExportFailure(message) { @@ -258,7 +266,9 @@ function sendToCollector(body) { // Settled by done() from whichever comes first: the response ending, a // request or response error, or the export deadline. Settling twice is - // harmless, since a promise keeps its first resolution. + // harmless, since a promise keeps its first resolution. Each handler + // settles before anything that can throw, so a failure its guard catches + // cannot leave a flush() waiting. let settle; const exportDone = new Promise((resolve) => { settle = resolve; @@ -295,48 +305,62 @@ function sendToCollector(body) { 'content-length': encoded.length, }, }, - (res) => { + guard('export response', (res) => { // TODO(bengl): Once retry logic is added, parse the response body for // ExportTraceServiceResponse.partial_success.rejected_spans. res.resume(); - res.on('end', () => { - clearDeadline(); - // Anything outside 2xx is a failed export: redirects are not - // followed, so a 3xx would silently drop every span. - if (res.statusCode < 200 || res.statusCode >= 300) { + res.on( + 'end', + guard('export response end', () => { + clearDeadline(); + done(); + // Anything outside 2xx is a failed export: redirects are not + // followed, so a 3xx would silently drop every span. + if (res.statusCode < 200 || res.statusCode >= 300) { + warnExportFailure( + `OTLP collector responded with HTTP ${res.statusCode}`, + ); + } + }), + ); + res.on( + 'error', + guard('export response error', (err) => { + clearDeadline(); + done(); warnExportFailure( - `OTLP collector responded with HTTP ${res.statusCode}`, + `OTLP export response stream error: ${err.message}`, ); - } - done(); - }); - res.on('error', (err) => { - clearDeadline(); - warnExportFailure( - `OTLP export response stream error: ${err.message}`, - ); - done(); - }); - }, + }), + ); + }), ); // exportTimeout is an absolute deadline for the whole export, not a // socket idle timeout: a collector that trickles response bytes would - // reset an idle timer forever. - const deadline = setTimeout(() => { - req.destroy( - new Error(`exceeded the ${exportTimeoutMs}ms export timeout`), - ); - }, exportTimeoutMs); + // reset an idle timer forever. The deadline settles the export itself, + // so a flush() waiting on it resolves by then, whatever else fails. + const deadline = setTimeout( + guard('export timeout', () => { + done(); + req.destroy( + new Error(`exceeded the ${exportTimeoutMs}ms export timeout`), + ); + }), + exportTimeoutMs, + ); const clearDeadline = () => clearTimeout(deadline); - req.on('error', (err) => { - clearDeadline(); - warnExportFailure( - `Failed to export spans to ${endpoint}: ${err.message}`, - ); - done(); - }); + req.on( + 'error', + guard('export request error', (err) => { + clearDeadline(); + done(); + warnExportFailure( + `Failed to export spans to ${endpoint}: ${err.message}`, + ); + }), + ); req.end(encoded); } catch (err) { @@ -360,7 +384,7 @@ function startFlusher(options) { flushIntervalMs = options?.flushInterval ?? kDefaultFlushIntervalMs; exportTimeoutMs = options?.exportTimeout ?? kDefaultExportTimeoutMs; - process.on('beforeExit', flush); + process.on('beforeExit', guardedBeforeExitFlush); } // Resets per-run state for the test suite; tracing itself never resets diff --git a/lib/guard.js b/lib/guard.js new file mode 100644 index 0000000..a636854 --- /dev/null +++ b/lib/guard.js @@ -0,0 +1,46 @@ +'use strict'; + +// Node rethrows an error from a diagnostics_channel subscriber or a timer +// as an uncaught exception, and one from an event listener skips the +// listeners after it. Every function microtel hands to Node goes through +// guard(), so a bug in microtel can't do either to the app it traces. + +const kWarningThrottleMs = 30_000; + +let failureCount = 0; +let lastWarningTime = 0; + +// The warnings are throttled like export failures, so a bug that fires +// on every request can't flood the logs. +function warnInstrumentationFailure(name, err) { + try { + failureCount++; + const now = Date.now(); + if (now - lastWarningTime < kWarningThrottleMs) return; + lastWarningTime = now; + const suffix = failureCount > 1 ? ` (${failureCount} total failures)` : ''; + process.emitWarning( + `microtel ${name} handler failed: ${err?.message ?? err}${suffix}`, + { type: 'OTelInstrumentationWarning', detail: err?.stack }, + ); + } catch { + // Reporting a failure must not become a new failure. + } +} + +// Each wrapper is made once per handler, so a guarded call allocates +// nothing. A plain function keeps `this` for the response listeners. +function guard(name, handler) { + return function () { + try { + handler.apply(this, arguments); + } catch (err) { + warnInstrumentationFailure(name, err); + } + }; +} + +module.exports = { + guard, + warnInstrumentationFailure, +}; diff --git a/lib/instrumentations.js b/lib/instrumentations.js index 5cb80ff..ff11187 100644 --- a/lib/instrumentations.js +++ b/lib/instrumentations.js @@ -2,6 +2,8 @@ const dc = require('node:diagnostics_channel'); +const { guard } = require('./guard'); + const { Span, SPAN_KIND_SERVER, @@ -104,7 +106,7 @@ function onHttpServerRequestStart({ request, response, socket }) { // before the response completed. Request 'close' cannot be used for // this: it also fires when the request stream is merely consumed, which // can happen long before a delayed response is sent. - response.on('close', onServerResponseClose); + response.on('close', guardedServerResponseClose); } // Shared by every server response, so no closure is allocated per @@ -118,6 +120,11 @@ function onServerResponseClose() { } } +const guardedServerResponseClose = guard( + 'server response close', + onServerResponseClose, +); + function onHttpServerResponseFinish({ response }) { const span = response[kSpan]; if (span == null) return; @@ -216,8 +223,8 @@ function onHttpClientResponseFinish({ request, response }) { // covers aborted responses. Ending a span twice is a no-op, so plain // listeners do what once() would without its wrappers. response[kSpan] = span; - response.on('end', endClientResponseSpan); - response.on('close', endClientResponseSpan); + response.on('end', guardedClientResponseSpanEnd); + response.on('close', guardedClientResponseSpanEnd); } // Shared by every client response, so no closure is allocated per @@ -227,6 +234,11 @@ function endClientResponseSpan() { this[kSpan].end(); } +const guardedClientResponseSpanEnd = guard( + 'client response end', + endClientResponseSpan, +); + function onClientRequestError({ request, error }) { const span = request[kSpan]; if (span == null) return; @@ -424,37 +436,39 @@ function onUndiciRequestTrailers({ request }) { span.end(); } +const subscribe = (name, handler) => dc.subscribe(name, guard(name, handler)); + function enableInstrumentations() { spanStorage = getSpanStorage(); if (isModuleEnabled('node:http/server')) { - dc.subscribe('http.server.request.start', onHttpServerRequestStart); - dc.subscribe('http.server.response.finish', onHttpServerResponseFinish); + subscribe('http.server.request.start', onHttpServerRequestStart); + subscribe('http.server.response.finish', onHttpServerResponseFinish); } if (isModuleEnabled('node:http/client')) { - dc.subscribe('http.client.request.created', onHttpClientRequestCreated); - dc.subscribe('http.client.response.finish', onHttpClientResponseFinish); - dc.subscribe('http.client.request.error', onClientRequestError); + subscribe('http.client.request.created', onHttpClientRequestCreated); + subscribe('http.client.response.finish', onHttpClientResponseFinish); + subscribe('http.client.request.error', onClientRequestError); } // A filter of node:http2 covers node:http2/server through the prefix // match in isModuleEnabled, the same way node:http covers its client and // server subscopes. if (isModuleEnabled('node:http2/server')) { - dc.subscribe('http2.server.stream.start', onHttp2ServerStreamStart); - dc.subscribe('http2.server.stream.finish', onHttp2ServerStreamFinish); - dc.subscribe('http2.server.stream.close', onHttp2ServerStreamClose); - dc.subscribe('http2.server.stream.error', onHttp2ServerStreamError); + subscribe('http2.server.stream.start', onHttp2ServerStreamStart); + subscribe('http2.server.stream.finish', onHttp2ServerStreamFinish); + subscribe('http2.server.stream.close', onHttp2ServerStreamClose); + subscribe('http2.server.stream.error', onHttp2ServerStreamError); } // node:fetch is an alias for node:undici: both enable the same // channels, which cover every undici request, fetch() included. if (isModuleEnabled('node:undici') || isModuleEnabled('node:fetch')) { - dc.subscribe('undici:request:create', onUndiciRequestCreate); - dc.subscribe('undici:request:headers', onUndiciRequestHeaders); - dc.subscribe('undici:request:trailers', onUndiciRequestTrailers); - dc.subscribe('undici:request:error', onClientRequestError); + subscribe('undici:request:create', onUndiciRequestCreate); + subscribe('undici:request:headers', onUndiciRequestHeaders); + subscribe('undici:request:trailers', onUndiciRequestTrailers); + subscribe('undici:request:error', onClientRequestError); } } diff --git a/test/fixtures/export-failure-child.js b/test/fixtures/export-failure-child.js new file mode 100644 index 0000000..c646f1b --- /dev/null +++ b/test/fixtures/export-failure-child.js @@ -0,0 +1,111 @@ +'use strict'; + +// Spawned with JSON init options as argv[2]: fakes http.request so each +// export callback can be made to throw, then prints the sites that +// reported a failure and any failure that got past its guard. + +const { EventEmitter } = require('node:events'); +const http = require('node:http'); + +const { init, flush } = require(process.env.MICROTEL_PATH); +const { Span, SPAN_KIND_INTERNAL } = require( + `${process.env.MICROTEL_PATH}/lib/span`, +); + +init(JSON.parse(process.argv[2])); + +const warnings = []; +const escaped = []; +const uncaught = []; + +process.on('warning', (w) => { + if (w.name === 'OTelInstrumentationWarning') warnings.push(w.message); +}); +process.on('uncaughtException', (err) => uncaught.push(`${err}`)); + +const realEmitWarning = process.emitWarning; +process.emitWarning = function (warning, type, ...rest) { + if (type === 'OTelExportWarning') throw new Error('export warning boom'); + return realEmitWarning.call(this, warning, type, ...rest); +}; + +// Both throttles read Date.now(), so moving it gives each site its own +// warning. +const realDateNow = Date.now; +let clockOffset = 0; +Date.now = () => realDateNow() + clockOffset; + +const requests = []; +http.request = (options, onResponse) => { + const req = new EventEmitter(); + req.end = () => {}; + req.destroy = () => { + if (req === requests[0].req) throw new Error('destroy boom'); + }; + requests.push({ req, onResponse }); + return req; +}; + +// The sites whose flush() resolved. A guard that caught a failure before +// the export settled would leave its flush() waiting. +const settled = []; +const startExport = (site) => { + new Span('export', SPAN_KIND_INTERNAL).end(); + flush().then(() => settled.push(site)); + return requests.at(-1); +}; + +const response = (statusCode, resume = () => {}) => + Object.assign(new EventEmitter(), { statusCode, resume }); + +// A throw that reaches here got past its guard. +const trigger = (site, fn) => { + clockOffset += 31_000; + try { + fn(); + } catch { + escaped.push(site); + } +}; + +const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); +const waitForWarnings = async (count) => { + for (let i = 0; i < 50 && warnings.length < count; i++) await delay(100); +}; + +async function main() { + // Nothing answers, so the deadline fires and destroy() throws. + startExport('export timeout'); + await waitForWarnings(1); + + trigger('export response', () => { + const { onResponse } = startExport('export response'); + onResponse( + response(200, () => { + throw new Error('resume boom'); + }), + ); + }); + trigger('export response end', () => { + const { onResponse } = startExport('export response end'); + const res = response(500); + onResponse(res); + res.emit('end'); + }); + trigger('export response error', () => { + const { onResponse } = startExport('export response error'); + const res = response(200); + onResponse(res); + res.emit('error', new Error('reset')); + }); + trigger('export request error', () => { + const { req } = startExport('export request error'); + req.emit('error', new Error('refused')); + }); + + await waitForWarnings(5); + for (let i = 0; i < 50 && settled.length < 5; i++) await delay(100); + console.log(JSON.stringify({ warnings, escaped, uncaught, settled })); +} + +main(); diff --git a/test/fixtures/instrumentation-failure-child.js b/test/fixtures/instrumentation-failure-child.js new file mode 100644 index 0000000..e9e5f5a --- /dev/null +++ b/test/fixtures/instrumentation-failure-child.js @@ -0,0 +1,161 @@ +'use strict'; + +// Spawned with JSON init options as argv[2]: makes microtel's span +// methods and serialization throw, then writes its results at exit, after +// the beforeExit flush has run. + +const fs = require('node:fs'); +const http = require('node:http'); +const http2 = require('node:http2'); + +const { init } = require(process.env.MICROTEL_PATH); +const { Span } = require(`${process.env.MICROTEL_PATH}/lib/span`); +const { addSpan } = require(`${process.env.MICROTEL_PATH}/lib/flush`); + +init(JSON.parse(process.argv[2])); + +const warnings = []; +const details = []; +const uncaught = []; +let completed = 0; +let injected = 0; + +process.on('warning', (w) => { + if (w.name !== 'OTelInstrumentationWarning') return; + warnings.push(w.message); + details.push(w.detail); +}); +process.on('uncaughtException', (err) => uncaught.push(`${err}`)); +process.on('exit', () => { + const result = { completed, injected, warnings, details, uncaught }; + fs.writeSync(1, `${JSON.stringify(result)}\n`); +}); + +// Counts each failure a guard should catch. +const boom = (what) => { + injected++; + throw new Error(`${what} boom`); +}; + +// Moving the clock past the warning window gives the next failure a +// warning of its own. +const realDateNow = Date.now; +let clockOffset = 0; +Date.now = () => realDateNow() + clockOffset; +const nextWindow = () => (clockOffset += 31_000); + +// flush() catches a span that fails to serialize and names it in an +// export warning, so a throwing name getter escapes flush() itself. +const poisonedSpan = () => ({ + get name() { + return boom('flush'); + }, + getAttributes() { + throw new Error('serialize boom'); + }, +}); + +const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); +const waitForWarnings = async (count) => { + for (let i = 0; i < 50 && warnings.length < count; i++) await delay(100); +}; + +const httpGet = (url) => + new Promise((resolve) => { + http + .get(url, (res) => { + res.resume(); + res.on('end', resolve); + }) + .on('error', resolve); + }).then(() => completed++); + +// The body must be read, so the response 'end' and 'close' listeners +// run. +const fetchRequest = (url) => + fetch(url) + .then((res) => res.text()) + .catch(() => {}) + .then(() => completed++); + +const http2Get = (port, path) => + new Promise((resolve) => { + const session = http2.connect(`http://127.0.0.1:${port}`); + session.on('error', () => {}); + const stream = session.request({ ':path': path }); + stream.on('error', () => {}); + stream.on('response', () => stream.resume()); + stream.on('close', () => { + session.close(); + resolve(); + }); + }).then(() => completed++); + +async function main() { + const server = http.createServer((req, res) => { + if (req.url === '/abort') return res.destroy(); + res.end('ok'); + }); + const h2server = http2.createServer(); + h2server.on('stream', (stream, headers) => { + stream.on('error', () => {}); + if (headers[':path'] === '/destroy') + return stream.destroy(new Error('stream boom')); + stream.respond({ ':status': 200 }); + stream.end('ok'); + }); + + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + await new Promise((resolve) => h2server.listen(0, '127.0.0.1', resolve)); + const url = `http://127.0.0.1:${server.address().port}`; + const h2port = h2server.address().port; + + // Attribute writes throw, so each protocol's start handlers fail. + const realSetAttribute = Span.prototype.setAttribute; + Span.prototype.setAttribute = () => boom('setAttribute'); + Span.prototype.end = () => boom('end'); + await httpGet(`${url}/one`); + await http2Get(h2port, '/two'); + await fetchRequest(`${url}/three`); + + // Spans now start, and the handlers that end them fail. The aborted + // requests reach the error channels. + Span.prototype.setAttribute = realSetAttribute; + await httpGet(`${url}/four`); + await fetchRequest(`${url}/five`); + await httpGet(`${url}/abort`); + await fetchRequest(`${url}/abort`); + await http2Get(h2port, '/six'); + await http2Get(h2port, '/destroy'); + + // The response handlers that only set attributes fail on the status + // code they record. + Span.prototype.setAttribute = function (key, value) { + if (key === 'http.response.status_code') return boom('setAttribute'); + return realSetAttribute.call(this, key, value); + }; + await http2Get(h2port, '/seven'); + await httpGet(`${url}/eight`); + await fetchRequest(`${url}/nine`); + + server.close(); + h2server.close(); + http.globalAgent.destroy(); + + // The flush timer's flush() throws. + nextWindow(); + addSpan(poisonedSpan()); + await waitForWarnings(2); + + // A full buffer's deferred flush() throws. maxBufferSize is 2. + nextWindow(); + addSpan(poisonedSpan()); + addSpan(poisonedSpan()); + await waitForWarnings(3); + + // The beforeExit flush() throws once nothing else is left to run. + nextWindow(); + addSpan(poisonedSpan()); +} + +main(); diff --git a/test/test-instrumentation-failure.js b/test/test-instrumentation-failure.js new file mode 100644 index 0000000..21207da --- /dev/null +++ b/test/test-instrumentation-failure.js @@ -0,0 +1,74 @@ +'use strict'; + +const assert = require('node:assert'); +const { describe, it } = require('node:test'); +const { fixture, spawnMicrotel } = require('./common/otel'); + +// The fixtures patch microtel's internals, so they run in child processes. + +const siteOf = (message) => /^microtel (.+) handler failed/.exec(message)?.[1]; + +describe('otel handler failures', () => { + it('warn instead of breaking requests, timers, or exit', async () => { + const { code, stdout } = await spawnMicrotel( + [ + fixture('instrumentation-failure-child.js'), + JSON.stringify({ + endpoint: 'http://127.0.0.1:1', + flushInterval: 100, + maxBufferSize: 2, + }), + ], + {}, + // A broken guard can leave the child's requests hanging. + { timeout: 15_000 }, + ); + + assert.strictEqual(code, 0); + const { completed, injected, warnings, details, uncaught } = + JSON.parse(stdout); + + assert.deepStrictEqual(uncaught, []); + assert.strictEqual(completed, 12); + + // One warning per window, naming the first failure in it. + assert.deepStrictEqual(warnings.map(siteOf), [ + 'http.client.request.created', + 'flush timer', + 'deferred flush', + 'beforeExit flush', + ]); + assert.ok( + warnings.at(-1).endsWith(` (${injected} total failures)`), + `Expected ${injected} failures in "${warnings.at(-1)}"`, + ); + assert.match(details[0], /^Error: setAttribute boom\n\s+at /); + }); + + it('warn instead of breaking exports', async () => { + const { code, stdout } = await spawnMicrotel( + [ + fixture('export-failure-child.js'), + JSON.stringify({ endpoint: 'http://127.0.0.1:1', exportTimeout: 50 }), + ], + {}, + { timeout: 15_000 }, + ); + + assert.strictEqual(code, 0); + const { warnings, escaped, uncaught, settled } = JSON.parse(stdout); + + const sites = [ + 'export timeout', + 'export response', + 'export response end', + 'export response error', + 'export request error', + ]; + assert.deepStrictEqual(uncaught, []); + assert.deepStrictEqual(escaped, []); + assert.deepStrictEqual(warnings.map(siteOf), sites); + // Every export still settled, so no flush() was left waiting. + assert.deepStrictEqual(settled.toSorted(), sites.toSorted()); + }); +});