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()); + }); +});