feat(core): HTTP server diagnostics channel utility (#20779)

Add platform-portable building blocks that server SDKs use to instrument
incoming HTTP requests without depending on OTel HTTP instrumentation:

- `getHttpServerSubscriptions`: diagnostics_channel listener for
  `http.server.request.start`, set up isolation scope, request data,
  trace continuation, optional body capture and request-session tracking
- if `spans: true` option set, then also create the root server span
  around the request lifecycle, applying static-asset/status-code
  filtering, `ignoreIncomingRequests` and `onSpanCreated` hooks
- `recordRequestSession`: release-health session aggregation per request
- `patchRequestToCaptureBody`: opt-in incoming request body capture

Add `kind` field on `StartSpanOptions` so OTel-based SDKs can set
SpanKind on the underlying span, and update `headersToDict` to allow
`number`-valued headers to support Node.js types.

Replace inline `instrumentServer` Proxy/emit-wrapping implementation in
node-light's HTTP integration with core's `getHttpServerSubscriptions`,
which does the same work (isolation scope, request data, body capture,
trace continuation, best-effort transaction name).

Replace inline `instrumentServer` Proxy/emit-wrapping in node-core's
`httpServerIntegration.ts` with core's `getHttpServerSubscriptions`.

OTel-specific concerns (header propagation, double-wrap context guard,
`_startSpanCallback` dispatch) move into a `wrapServerEmitRequest`
callback that `instrumentServer` invokes inside the per-request
lifecycle.

Re-export `recordRequestSession` from core so existing test continues
to pass.

Duplicate request-isolation/session/body-capture plumbing removed, logic
now lives in `@sentry/core`'s subscription factory.

Author: isaacs

…ode-core/src/utils/captureRequestBody.ts …ns/http/patch-request-to-capture-body.tspackages/node-core/src/utils/captureRequestBody.ts renamed to packages/core/src/integrations/http/patch-request-to-capture-body.ts packages/node-core/src/utils/captureRequestBody.ts renamed to packages/core/src/integrations/http/patch-request-to-capture-body.ts

@@ -1,16 +1,18 @@
-import type { IncomingMessage } from 'node:http';
-import type { Scope } from '@sentry/core';
-import { debug, getMaxBodyByteLength, type MaxRequestBodySize } from '@sentry/core';
-import { DEBUG_BUILD } from '../debug-build';
+import type { Scope } from '../../scope';
+import { debug } from '../../utils/debug-logger';
+import { DEBUG_BUILD } from '../../debug-build';
+import type { HttpIncomingMessage } from './types';
+import { getMaxBodyByteLength, type MaxRequestBodySize } from '../../utils/request';
 
 /**
  * This method patches the request object to capture the body.
- * Instead of actually consuming the streamed body ourselves, which has potential side effects,
- * we monkey patch `req.on('data')` to intercept the body chunks.
- * This way, we only read the body if the user also consumes the body, ensuring we do not change any behavior in unexpected ways.
+ * Instead of actually consuming the streamed body ourselves, which has
+ * potential side effects, we monkey patch `req.on('data')` to intercept
+ * the body chunks. This way, we only read the body if the user also consumes
+ * the body, ensuring we do not change any behavior in unexpected ways.
  */
 export function patchRequestToCaptureBody(
-  req: IncomingMessage,
+  req: HttpIncomingMessage,
   isolationScope: Scope,
   maxIncomingRequestBodySize: Exclude<MaxRequestBodySize, 'none'>,
   integrationName: string,
@@ -20,18 +22,16 @@ export function patchRequestToCaptureBody(
 
   DEBUG_BUILD && debug.log(integrationName, 'Patching request.on');
 
-  /**
-   * 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
-   * original listeners to our wrapped ones.
-   */
+  // keep track of the original callbacks to remove listeners later
+  // `off` depends on having the exact same function reference passed in,
+  // so we need to be able to map original listeners to our wrapped ones.
   const callbackMap = new WeakMap();
 
   const maxBodySize = getMaxBodyByteLength(maxIncomingRequestBodySize);
 
   try {
     // eslint-disable-next-line @typescript-eslint/unbound-method
-    req.on = new Proxy(req.on, {
+    req.on = req.addListener = new Proxy(req.on, {
       apply: (target, thisArg, args: Parameters<typeof req.on>) => {
         const [event, listener, ...restArgs] = args;
 
@@ -73,7 +73,7 @@ export function patchRequestToCaptureBody(
 
     // Ensure we also remove callbacks correctly
     // eslint-disable-next-line @typescript-eslint/unbound-method
-    req.off = new Proxy(req.off, {
+    req.off = req.removeListener = new Proxy(req.off, {
       apply: (target, thisArg, args: Parameters<typeof req.off>) => {
         const [, listener] = args;
 

packages/core/src/integrations/http/record-request-session.ts

@@ -0,0 +1,88 @@
+import type { Client } from '../../client';
+import { debug } from '../../utils/debug-logger';
+import { DEBUG_BUILD } from '../../debug-build';
+import type { Scope } from '../../scope';
+import type { HttpServerResponse } from './types';
+import type { AggregationCounts } from '../../types-hoist/session';
+import { safeUnref } from '../../utils/timer';
+
+const clientToRequestSessionAggregatesMap = new WeakMap<
+  Client,
+  { [timestampRoundedToSeconds: string]: { exited: number; crashed: number; errored: number } }
+>();
+
+/**
+ * Starts a session and tracks it in the context of a given isolation scope.
+ * When the passed response is finished, the session is put into a task and
+ * is aggregated with other sessions that may happen in a certain time window
+ * (sessionFlushingDelayMs).
+ *
+ * The sessions are always aggregated by the client that is on the current
+ * scope at the time of ending the response (if there is one).
+ */
+export function recordRequestSession(
+  client: Client,
+  {
+    requestIsolationScope,
+    response,
+    sessionFlushingDelayMS,
+  }: {
+    requestIsolationScope: Scope;
+    response: HttpServerResponse;
+    sessionFlushingDelayMS?: number;
+  },
+): void {
+  requestIsolationScope.setSDKProcessingMetadata({
+    requestSession: { status: 'ok' },
+  });
+
+  response.once('close', () => {
+    const requestSession = requestIsolationScope.getScopeData().sdkProcessingMetadata.requestSession;
+
+    if (client && requestSession) {
+      DEBUG_BUILD && debug.log(`Recorded request session with status: ${requestSession.status}`);
+
+      const roundedDate = new Date();
+      roundedDate.setSeconds(0, 0);
+      const dateBucketKey = roundedDate.toISOString();
+
+      const existingClientAggregate = clientToRequestSessionAggregatesMap.get(client);
+      const bucket = existingClientAggregate?.[dateBucketKey] || { exited: 0, crashed: 0, errored: 0 };
+      bucket[({ ok: 'exited', crashed: 'crashed', errored: 'errored' } as const)[requestSession.status]]++;
+
+      if (existingClientAggregate) {
+        existingClientAggregate[dateBucketKey] = bucket;
+      } else {
+        DEBUG_BUILD && debug.log('Opened new request session aggregate.');
+        const newClientAggregate = { [dateBucketKey]: bucket };
+        clientToRequestSessionAggregatesMap.set(client, newClientAggregate);
+
+        const flushPendingClientAggregates = (): void => {
+          clearTimeout(timeout);
+          unregisterClientFlushHook();
+          clientToRequestSessionAggregatesMap.delete(client);
+
+          const aggregatePayload: AggregationCounts[] = Object.entries(newClientAggregate).map(
+            ([timestamp, value]) => ({
+              started: timestamp,
+              exited: value.exited,
+              errored: value.errored,
+              crashed: value.crashed,
+            }),
+          );
+          client.sendSession({ aggregates: aggregatePayload });
+        };
+
+        const unregisterClientFlushHook = client.on('flush', () => {
+          DEBUG_BUILD && debug.log('Sending request session aggregate due to client flush');
+          flushPendingClientAggregates();
+        });
+        const timeout = setTimeout(() => {
+          DEBUG_BUILD && debug.log('Sending request session aggregate due to flushing schedule');
+          flushPendingClientAggregates();
+        }, sessionFlushingDelayMS);
+        safeUnref(timeout);
+      }
+    }
+  });
+}