fix(node): Make body capturing more robust (#16105)

I have a hunch that not all frameworks call `req.on('end')` but may only
do `req.on('close')` or whatever else and this is why we are not
reliably capturing bodies.

It should be safe to attach a `req.on('end')` handler, as that doesn't
change any semantics AFAIK.

Fixes #16090

---------

Co-authored-by: Francesco Gringl-Novy <[email protected]>

Author: lforst

dev-packages/e2e-tests/test-applications/node-fastify/src/app.ts

@@ -119,6 +119,10 @@ app.get('/test-outgoing-http-external-disallowed', async function (req, res) {
   res.send(data);
 });
 
+app.post('/test-post', function (req, res) {
+  res.send({ status: 'ok', body: req.body });
+});
+
 app.listen({ port: port });
 
 // A second app so we can test header propagation between external URLs

dev-packages/e2e-tests/test-applications/node-fastify/tests/transactions.test.ts

@@ -124,3 +124,40 @@ test('Sends an API route transaction', async ({ baseURL }) => {
     origin: 'manual',
   });
 });
+
+test('Captures request metadata', async ({ baseURL }) => {
+  const transactionEventPromise = waitForTransaction('node-fastify', transactionEvent => {
+    return (
+      transactionEvent?.contexts?.trace?.op === 'http.server' && transactionEvent?.transaction === 'POST /test-post'
+    );
+  });
+
+  const res = await fetch(`${baseURL}/test-post`, {
+    method: 'POST',
+    body: JSON.stringify({ foo: 'bar', other: 1 }),
+    headers: {
+      'Content-Type': 'application/json',
+    },
+  });
+  const resBody = await res.json();
+
+  expect(resBody).toEqual({ status: 'ok', body: { foo: 'bar', other: 1 } });
+
+  const transactionEvent = await transactionEventPromise;
+
+  expect(transactionEvent.request).toEqual({
+    cookies: {},
+    url: expect.stringMatching(/^http:\/\/localhost:(\d+)\/test-post$/),
+    method: 'POST',
+    headers: expect.objectContaining({
+      'user-agent': expect.stringContaining(''),
+      'content-type': 'application/json',
+    }),
+    data: JSON.stringify({
+      foo: 'bar',
+      other: 1,
+    }),
+  });
+
+  expect(transactionEvent.user).toEqual(undefined);
+});

packages/node/src/integrations/http/SentryHttpInstrumentation.ts

@@ -3,7 +3,7 @@ import { context, propagation } from '@opentelemetry/api';
 import { VERSION } from '@opentelemetry/core';
 import type { InstrumentationConfig } from '@opentelemetry/instrumentation';
 import { InstrumentationBase, InstrumentationNodeModuleDefinition } from '@opentelemetry/instrumentation';
-import type { AggregationCounts, Client, RequestEventData, SanitizedRequestData, Scope } from '@sentry/core';
+import type { AggregationCounts, Client, SanitizedRequestData, Scope } from '@sentry/core';
 import {
   addBreadcrumb,
   generateSpanId,
@@ -360,12 +360,9 @@ function getBreadcrumbData(request: http.ClientRequest): Partial<SanitizedReques
  * This way, we only read the body if the user also consumes the body, ensuring we do not change any behavior in unexpected ways.
  */
 function patchRequestToCaptureBody(req: IncomingMessage, isolationScope: Scope): void {
+  let bodyByteLength = 0;
   const chunks: Buffer[] = [];
 
-  function getChunksSize(): number {
-    return chunks.reduce((acc, chunk) => acc + chunk.byteLength, 0);
-  }
-
   /**
    * We need to keep track of the original callbacks, in order to be able to remove listeners again.
    * Since `off` depends on having the exact same function reference passed in, we need to be able to map
@@ -386,41 +383,21 @@ function patchRequestToCaptureBody(req: IncomingMessage, isolationScope: Scope):
         if (event === 'data') {
           const callback = new Proxy(listener, {
             apply: (target, thisArg, args: Parameters<typeof listener>) => {
-              // If we have already read more than the max body length, we stop adding chunks
-              // To avoid growing the memory indefinitely if a response is e.g. streamed
-              if (getChunksSize() < MAX_BODY_BYTE_LENGTH) {
-                const chunk = args[0] as Buffer;
-                chunks.push(chunk);
-              } else if (DEBUG_BUILD) {
-                logger.log(
-                  INSTRUMENTATION_NAME,
-                  `Dropping request body chunk because maximum body length of ${MAX_BODY_BYTE_LENGTH}b is exceeded.`,
-                );
-              }
-
-              return Reflect.apply(target, thisArg, args);
-            },
-          });
-
-          callbackMap.set(listener, callback);
-
-          return Reflect.apply(target, thisArg, [event, callback, ...restArgs]);
-        }
-
-        if (event === 'end') {
-          const callback = new Proxy(listener, {
-            apply: (target, thisArg, args) => {
               try {
-                const body = Buffer.concat(chunks).toString('utf-8');
-
-                if (body) {
-                  const normalizedRequest = { data: body } satisfies RequestEventData;
-                  isolationScope.setSDKProcessingMetadata({ normalizedRequest });
-                }
-              } catch (error) {
-                if (DEBUG_BUILD) {
-                  logger.error(INSTRUMENTATION_NAME, 'Error building captured request body', error);
+                const chunk = args[0] as Buffer | string;
+                const bufferifiedChunk = Buffer.from(chunk);
+
+                if (bodyByteLength < MAX_BODY_BYTE_LENGTH) {
+                  chunks.push(bufferifiedChunk);
+                  bodyByteLength += bufferifiedChunk.byteLength;
+                } else if (DEBUG_BUILD) {
+                  logger.log(
+                    INSTRUMENTATION_NAME,
+                    `Dropping request body chunk because maximum body length of ${MAX_BODY_BYTE_LENGTH}b is exceeded.`,
+                  );
                 }
+              } catch (err) {
+                DEBUG_BUILD && logger.error(INSTRUMENTATION_NAME, 'Encountered error while storing body chunk.');
               }
 
               return Reflect.apply(target, thisArg, args);
@@ -454,6 +431,19 @@ function patchRequestToCaptureBody(req: IncomingMessage, isolationScope: Scope):
         return Reflect.apply(target, thisArg, args);
       },
     });
+
+    req.on('end', () => {
+      try {
+        const body = Buffer.concat(chunks).toString('utf-8');
+        if (body) {
+          isolationScope.setSDKProcessingMetadata({ normalizedRequest: { data: body } });
+        }
+      } catch (error) {
+        if (DEBUG_BUILD) {
+          logger.error(INSTRUMENTATION_NAME, 'Error building captured request body', error);
+        }
+      }
+    });
   } catch (error) {
     if (DEBUG_BUILD) {
       logger.error(INSTRUMENTATION_NAME, 'Error patching request to capture body', error);