fix: use socket from the request (#1939)

Co-authored-by: Daniel Dyla <dyladan@users.noreply.github.com>
Co-authored-by: Valentin Marchaud <contact@vmarchaud.fr>

packages/opentelemetry-instrumentation-http/src/utils.ts

@@ -465,7 +465,10 @@ export const getIncomingRequestAttributesOnResponse = (
   request: IncomingMessage & { __ot_middlewares?: string[] },
   response: ServerResponse & { socket: Socket }
 ): SpanAttributes => {
-  const { statusCode, statusMessage, socket } = response;
+  // take socket from the request,
+  // since it may be detached from the response object in keep-alive mode
+  const { socket } = request;
+  const { statusCode, statusMessage } = response;
   const { localAddress, localPort, remoteAddress, remotePort } = socket;
   const { __ot_middlewares } = (request as unknown) as {
     [key: string]: unknown;

packages/opentelemetry-instrumentation-http/test/functionals/utils.test.ts

@@ -291,6 +291,7 @@ describe('Utility', () => {
     it('should correctly parse the middleware stack if present', () => {
       const request = {
         __ot_middlewares: ['/test', '/toto', '/'],
+        socket: {},
       } as IncomingMessage & { __ot_middlewares?: string[] };
 
       const attributes = utils.getIncomingRequestAttributesOnResponse(request, {
@@ -300,7 +301,9 @@ describe('Utility', () => {
     });
 
     it('should succesfully process without middleware stack', () => {
-      const request = {} as IncomingMessage;
+      const request = {
+        socket: {},
+      } as IncomingMessage;
       const attributes = utils.getIncomingRequestAttributesOnResponse(request, {
         socket: {},
       } as ServerResponse & { socket: Socket });

packages/opentelemetry-instrumentation-http/test/integrations/http-enable.test.ts

@@ -39,6 +39,7 @@ import * as http from 'http';
 import { httpRequest } from '../utils/httpRequest';
 import { DummyPropagation } from '../utils/DummyPropagation';
 import { Socket } from 'net';
+import { sendRequestTwice } from '../utils/rawRequest';
 
 const protocol = 'http';
 const serverPort = 32345;
@@ -345,5 +346,14 @@ describe('HttpInstrumentation Integration tests', () => {
         });
       });
     }
+
+    it('should work for multiple active requests in keep-alive mode', async () => {
+      await sendRequestTwice(hostname, mockServerPort);
+      const spans = memoryExporter.getFinishedSpans();
+      const span = spans.find((s: any) => s.kind === SpanKind.SERVER);
+      assert.ok(span);
+      assert.strictEqual(spans.length, 2);
+      assert.strictEqual(span.name, 'HTTP GET');
+    });
   });
 });

packages/opentelemetry-instrumentation-http/test/utils/rawRequest.ts

@@ -0,0 +1,38 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import * as net from 'net';
+
+// used to reproduce this issue:
+// https://github.com/open-telemetry/opentelemetry-js/pull/1939
+export async function sendRequestTwice(
+  host: string,
+  port: number
+): Promise<Buffer> {
+  return new Promise((resolve, reject) => {
+    const request = 'GET /raw HTTP/1.1\n\n';
+    const socket = net.createConnection({ host, port }, () => {
+      socket.write(`${request}${request}`, err => {
+        if (err) reject(err);
+      });
+    });
+    socket.on('data', data => {
+      // since it's <1kb size, we expect both responses to come in a single chunk
+      socket.destroy();
+      resolve(data);
+    });
+    socket.on('error', err => reject(err));
+  });
+}

packages/opentelemetry-plugin-http/src/utils.ts

@@ -31,7 +31,6 @@ import {
   RequestOptions,
   ServerResponse,
 } from 'http';
-import { Socket } from 'net';
 import * as url from 'url';
 import { Err, IgnoreMatcher, ParsedRequestOptions } from './types';
 
@@ -463,9 +462,12 @@ export const getIncomingRequestAttributes = (
  */
 export const getIncomingRequestAttributesOnResponse = (
   request: IncomingMessage & { __ot_middlewares?: string[] },
-  response: ServerResponse & { socket: Socket }
+  response: ServerResponse
 ): SpanAttributes => {
-  const { statusCode, statusMessage, socket } = response;
+  // use socket from the request,
+  // since it may be detached from the response object in keep-alive mode
+  const { socket } = request;
+  const { statusCode, statusMessage } = response;
   const { localAddress, localPort, remoteAddress, remotePort } = socket;
   const { __ot_middlewares } = (request as unknown) as {
     [key: string]: unknown;

packages/opentelemetry-plugin-http/test/functionals/utils.test.ts

@@ -25,7 +25,6 @@ import { HttpAttribute } from '@opentelemetry/semantic-conventions';
 import * as assert from 'assert';
 import * as http from 'http';
 import { IncomingMessage, ServerResponse } from 'http';
-import { Socket } from 'net';
 import * as sinon from 'sinon';
 import * as url from 'url';
 import { IgnoreMatcher } from '../../src/types';
@@ -291,19 +290,23 @@ describe('Utility', () => {
     it('should correctly parse the middleware stack if present', () => {
       const request = {
         __ot_middlewares: ['/test', '/toto', '/'],
-      } as IncomingMessage & { __ot_middlewares?: string[] };
-
-      const attributes = utils.getIncomingRequestAttributesOnResponse(request, {
         socket: {},
-      } as ServerResponse & { socket: Socket });
+      } as IncomingMessage & { __ot_middlewares?: string[] };
+      const response = {} as ServerResponse;
+      const attributes = utils.getIncomingRequestAttributesOnResponse(
+        request,
+        response
+      );
       assert.deepEqual(attributes[HttpAttribute.HTTP_ROUTE], '/test/toto');
     });
 
     it('should succesfully process without middleware stack', () => {
-      const request = {} as IncomingMessage;
-      const attributes = utils.getIncomingRequestAttributesOnResponse(request, {
-        socket: {},
-      } as ServerResponse & { socket: Socket });
+      const request = { socket: {} } as IncomingMessage;
+      const response = {} as ServerResponse;
+      const attributes = utils.getIncomingRequestAttributesOnResponse(
+        request,
+        response
+      );
       assert.deepEqual(attributes[HttpAttribute.HTTP_ROUTE], undefined);
     });
   });

packages/opentelemetry-plugin-http/test/integrations/http-enable.test.ts

@@ -35,6 +35,7 @@ import {
 import { HttpPluginConfig } from '../../src/types';
 import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks';
 import { Socket } from 'net';
+import { sendRequestTwice } from '../utils/rawRequest';
 const protocol = 'http';
 const serverPort = 32345;
 const hostname = 'localhost';
@@ -337,5 +338,14 @@ describe('HttpPlugin Integration tests', () => {
         });
       });
     }
+
+    it('should work for multiple active requests in keep-alive mode', async () => {
+      await sendRequestTwice(hostname, mockServerPort);
+      const spans = memoryExporter.getFinishedSpans();
+      const span = spans.find((s: any) => s.kind === SpanKind.SERVER);
+      assert.ok(span);
+      assert.strictEqual(spans.length, 2);
+      assert.strictEqual(span.name, 'HTTP GET');
+    });
   });
 });

packages/opentelemetry-plugin-http/test/utils/rawRequest.ts

@@ -0,0 +1,38 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import * as net from 'net';
+
+// used to reproduce this issue:
+// https://github.com/open-telemetry/opentelemetry-js/pull/1939
+export async function sendRequestTwice(
+  host: string,
+  port: number
+): Promise<Buffer> {
+  return new Promise((resolve, reject) => {
+    const request = 'GET /raw HTTP/1.1\n\n';
+    const socket = net.createConnection({ host, port }, () => {
+      socket.write(`${request}${request}`, err => {
+        if (err) reject(err);
+      });
+    });
+    socket.on('data', data => {
+      // since it's <1kb size, we expect both responses to come in a single chunk
+      socket.destroy();
+      resolve(data);
+    });
+    socket.on('error', err => reject(err));
+  });
+}