diff --git a/src/instrumentation/libraries/http/Instrumentation.ts b/src/instrumentation/libraries/http/Instrumentation.ts index 2478430b..2d602845 100644 --- a/src/instrumentation/libraries/http/Instrumentation.ts +++ b/src/instrumentation/libraries/http/Instrumentation.ts @@ -413,7 +413,8 @@ export class HttpInstrumentation extends TdInstrumentationBase { const statusCode = res.statusCode; // Capture any final response data - if (chunk) { + // res.end(callback) passes the callback as the first argument; it is not body data + if (chunk && typeof chunk !== "function") { responseChunks.push(chunk); } @@ -755,7 +756,8 @@ export class HttpInstrumentation extends TdInstrumentationBase { if (originalEnd) { req.end = function (chunk?: any, encoding?: any, callback?: any): ClientRequest { // Capture final chunk if provided - if (chunk && !requestBodyCaptured) { + // req.end(callback) passes the callback as the first argument; it is not body data + if (chunk && typeof chunk !== "function" && !requestBodyCaptured) { requestBodyChunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding)); } diff --git a/src/instrumentation/libraries/http/integration-tests/end-callback.test.ts b/src/instrumentation/libraries/http/integration-tests/end-callback.test.ts new file mode 100644 index 00000000..f743e0ba --- /dev/null +++ b/src/instrumentation/libraries/http/integration-tests/end-callback.test.ts @@ -0,0 +1,176 @@ +process.env.TUSK_DRIFT_MODE = "RECORD"; + +import { TuskDrift } from "../../../../core/TuskDrift"; + +TuskDrift.initialize({ + apiKey: "test-api-key-end-callback", + env: "test", + logLevel: "silent", +}); +TuskDrift.markAppAsReady(); + +import { SpanKind } from "@opentelemetry/api"; +import { + InMemorySpanAdapter, + registerInMemoryAdapter, + clearRegisteredInMemoryAdapters, +} from "../../../../core/tracing/adapters/InMemorySpanAdapter"; +import { waitForSpans } from "./test-utils"; +import test from "ava"; +import type { Server, RequestOptions } from "http"; +import { AddressInfo } from "net"; + +const spanAdapter = new InMemorySpanAdapter(); +registerInMemoryAdapter(spanAdapter); + +// Require http after TuskDrift.initialize() so the require-in-the-middle hook patches it, +// matching how the SDK is used in an application. +const http: typeof import("http") = require("http"); + +let server: Server; +let port: number; + +function decodeBody(value: unknown): string { + return Buffer.from(String(value), "base64").toString("utf8"); +} + +function makeRequest( + options: RequestOptions, + body?: string, + endWithCallback = false, +): Promise<{ statusCode: number; data: string }> { + return new Promise((resolve, reject) => { + const req = http.request(options, (res) => { + let data = ""; + res.on("data", (chunk) => (data += chunk)); + res.on("end", () => resolve({ statusCode: res.statusCode || 0, data })); + }); + req.on("error", reject); + if (body) { + req.write(body); + } + if (endWithCallback) { + // Node's documented signature: request.end([data[, encoding]][, callback]) + req.end(() => {}); + } else { + req.end(); + } + }); +} + +test.before(async () => { + server = http.createServer((req, res) => { + if (req.url === "/upstream") { + let body = ""; + req.on("data", (chunk) => (body += chunk)); + req.on("end", () => { + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ upstreamReceived: body })); + }); + return; + } + + if (req.url === "/call-upstream") { + // Outbound request made inside the inbound span, ended with a callback-only end(). + const postData = JSON.stringify({ hello: "world" }); + try { + makeRequest( + { + hostname: "127.0.0.1", + port, + path: "/upstream", + method: "POST", + headers: { + "content-type": "application/json", + "content-length": Buffer.byteLength(postData), + }, + }, + postData, + true, + ).then( + (upstream) => { + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ upstream: JSON.parse(upstream.data) })); + }, + (error: Error) => { + res.statusCode = 500; + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ error: error.message })); + }, + ); + } catch (error) { + res.statusCode = 500; + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ error: (error as Error).message })); + } + return; + } + + // Inbound handler that writes the body, then ends with a callback-only end(). + res.setHeader("content-type", "application/json"); + res.write(JSON.stringify({ path: req.url })); + res.end(() => {}); + }); + + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + port = (server.address() as AddressInfo).port; +}); + +test.after.always(async () => { + await new Promise((resolve, reject) => + server.close((err?: Error) => (err ? reject(err) : resolve())), + ); + clearRegisteredInMemoryAdapters(); +}); + +test("inbound: response body is recorded when the handler calls res.end(callback)", async (t) => { + const response = await makeRequest({ + hostname: "127.0.0.1", + port, + path: "/inbound", + method: "GET", + }); + + t.is(response.statusCode, 200); + t.deepEqual(JSON.parse(response.data), { path: "/inbound" }); + + await waitForSpans(); + + const serverSpan = spanAdapter + .getAllSpans() + .find( + (span) => span.kind === SpanKind.SERVER && (span.inputValue as any)?.target === "/inbound", + ); + t.truthy(serverSpan, "Expected a SERVER span for /inbound"); + + const outputValue = serverSpan!.outputValue as any; + t.truthy(outputValue.body, "Expected the recorded server span to carry the response body"); + t.deepEqual(JSON.parse(decodeBody(outputValue.body)), { path: "/inbound" }); +}); + +test("outbound: req.end(callback) inside a trace does not throw and the request body is recorded", async (t) => { + const response = await makeRequest({ + hostname: "127.0.0.1", + port, + path: "/call-upstream", + method: "GET", + }); + + t.is(response.statusCode, 200, `Unexpected response: ${response.data}`); + t.deepEqual(JSON.parse(response.data), { + upstream: { upstreamReceived: JSON.stringify({ hello: "world" }) }, + }); + + await waitForSpans(); + + const clientSpan = spanAdapter + .getAllSpans() + .find( + (span) => span.kind === SpanKind.CLIENT && (span.inputValue as any)?.path === "/upstream", + ); + t.truthy(clientSpan, "Expected a CLIENT span for /upstream"); + + const inputValue = clientSpan!.inputValue as any; + t.truthy(inputValue.body, "Expected the recorded client span to carry the request body"); + t.deepEqual(JSON.parse(decodeBody(inputValue.body)), { hello: "world" }); +}); diff --git a/src/instrumentation/libraries/nextjs/Instrumentation.ts b/src/instrumentation/libraries/nextjs/Instrumentation.ts index 23202a4f..760a659c 100644 --- a/src/instrumentation/libraries/nextjs/Instrumentation.ts +++ b/src/instrumentation/libraries/nextjs/Instrumentation.ts @@ -352,7 +352,8 @@ export class NextjsInstrumentation extends TdInstrumentationBase { const originalEnd = actualRes.end?.bind(actualRes); if (originalEnd) { actualRes.end = function (chunk?: any, encoding?: any, callback?: any) { - if (chunk) { + // res.end(callback) passes the callback as the first argument; it is not body data + if (chunk && typeof chunk !== "function") { responseChunks.push(chunk); } return originalEnd.call(this, chunk, encoding, callback);