feat(plugin-http): handle aborted request (jaegertracing#365)

closes jaegertracing#364

Signed-off-by: Olivier Albertini <olivier.albertini@montreal.ca>

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

@@ -15,7 +15,13 @@
  */
 
 import { BasePlugin, isValid } from '@opentelemetry/core';
-import { Span, SpanKind, SpanOptions, Attributes } from '@opentelemetry/types';
+import {
+  Span,
+  SpanKind,
+  SpanOptions,
+  Attributes,
+  CanonicalCode,
+} from '@opentelemetry/types';
 import {
   ClientRequest,
   IncomingMessage,
@@ -165,46 +171,42 @@ export class HttpPlugin extends BasePlugin<Http> {
     return (): ClientRequest => {
       this._logger.debug('makeRequestTrace by injecting context into header');
 
+      const host = options.hostname || options.host || 'localhost';
+      const method = options.method ? options.method.toUpperCase() : 'GET';
+      const headers = options.headers || {};
+      const userAgent = headers['user-agent'];
+
+      span.setAttributes({
+        [AttributeNames.HTTP_URL]: Utils.getAbsoluteUrl(
+          options,
+          headers,
+          `${HttpPlugin.component}:`
+        ),
+        [AttributeNames.HTTP_HOSTNAME]: host,
+        [AttributeNames.HTTP_METHOD]: method,
+        [AttributeNames.HTTP_PATH]: options.path || '/',
+        [AttributeNames.HTTP_USER_AGENT]: userAgent || '',
+      });
+
       request.on(
         'response',
-        (response: IncomingMessage & { req?: { method?: string } }) => {
+        (response: IncomingMessage & { aborted?: boolean }) => {
           this._tracer.bind(response);
           this._logger.debug('outgoingRequest on response()');
           response.on('end', () => {
             this._logger.debug('outgoingRequest on end()');
-
-            const method =
-              response.req && response.req.method
-                ? response.req.method.toUpperCase()
-                : 'GET';
-            const headers = options.headers || {};
-            const userAgent = headers['user-agent'];
-
-            const host = options.hostname || options.host || 'localhost';
-
-            const attributes: Attributes = {
-              [AttributeNames.HTTP_URL]: Utils.getAbsoluteUrl(
-                options,
-                headers,
-                `${HttpPlugin.component}:`
-              ),
-              [AttributeNames.HTTP_HOSTNAME]: host,
-              [AttributeNames.HTTP_METHOD]: method,
-              [AttributeNames.HTTP_PATH]: options.path || '/',
-            };
-
-            if (userAgent) {
-              attributes[AttributeNames.HTTP_USER_AGENT] = userAgent;
-            }
-
             if (response.statusCode) {
-              attributes[AttributeNames.HTTP_STATUS_CODE] = response.statusCode;
-              attributes[AttributeNames.HTTP_STATUS_TEXT] =
-                response.statusMessage;
-              span.setStatus(Utils.parseResponseStatus(response.statusCode));
+              span.setAttributes({
+                [AttributeNames.HTTP_STATUS_CODE]: response.statusCode,
+                [AttributeNames.HTTP_STATUS_TEXT]: response.statusMessage,
+              });
             }
 
-            span.setAttributes(attributes);
+            if (response.aborted && !response.complete) {
+              span.setStatus({ code: CanonicalCode.ABORTED });
+            } else {
+              span.setStatus(Utils.parseResponseStatus(response.statusCode!));
+            }
 
             if (this._config.applyCustomAttributesOnSpan) {
               this._config.applyCustomAttributesOnSpan(span, request, response);

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

@@ -173,9 +173,17 @@ export class Utils {
       [AttributeNames.HTTP_ERROR_MESSAGE]: message,
     });
 
+    if (!obj) {
+      span.setStatus({ code: CanonicalCode.UNKNOWN, message });
+      span.end();
+      return;
+    }
+
     let status: Status;
-    if (obj && (obj as IncomingMessage).statusCode) {
+    if ((obj as IncomingMessage).statusCode) {
       status = Utils.parseResponseStatus((obj as IncomingMessage).statusCode!);
+    } else if ((obj as ClientRequest).aborted) {
+      status = { code: CanonicalCode.ABORTED };
     } else {
       status = { code: CanonicalCode.UNKNOWN };
     }

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

@@ -14,20 +14,20 @@
  * limitations under the License.
  */
 
+import {
+  InMemorySpanExporter,
+  SimpleSpanProcessor,
+} from '@opentelemetry/basic-tracer';
 import { NoopLogger } from '@opentelemetry/core';
-import { SpanKind, Span as ISpan } from '@opentelemetry/types';
+import { NodeTracer } from '@opentelemetry/node-sdk';
+import { CanonicalCode, Span as ISpan, SpanKind } from '@opentelemetry/types';
 import * as assert from 'assert';
 import * as http from 'http';
 import * as nock from 'nock';
 import { HttpPlugin, plugin } from '../../src/http';
 import { assertSpan } from '../utils/assertSpan';
 import { DummyPropagation } from '../utils/DummyPropagation';
 import { httpRequest } from '../utils/httpRequest';
-import { NodeTracer } from '@opentelemetry/node-sdk';
-import {
-  InMemorySpanExporter,
-  SimpleSpanProcessor,
-} from '@opentelemetry/basic-tracer';
 import { Utils } from '../../src/utils';
 import { HttpPluginConfig, Http } from '../../src/types';
 
@@ -345,8 +345,11 @@ describe('HttpPlugin', () => {
     }
 
     it('should have 1 ended span when request throw on bad "options" object', () => {
+      nock.cleanAll();
+      nock.enableNetConnect();
       try {
         http.request({ protocol: 'telnet' });
+        assert.fail();
       } catch (error) {
         const spans = memoryExporter.getFinishedSpans();
         assert.strictEqual(spans.length, 1);
@@ -381,5 +384,81 @@ describe('HttpPlugin', () => {
         assert.strictEqual(spans.length, 1);
       }
     });
+
+    it('should have 1 ended span when request is aborted', async () => {
+      nock('http://my.server.com')
+        .get('/')
+        .socketDelay(50)
+        .reply(200, '<html></html>');
+
+      const promiseRequest = new Promise((resolve, reject) => {
+        const req = http.request(
+          'http://my.server.com',
+          (resp: http.IncomingMessage) => {
+            let data = '';
+            resp.on('data', chunk => {
+              data += chunk;
+            });
+            resp.on('end', () => {
+              resolve(data);
+            });
+          }
+        );
+        req.setTimeout(10, () => {
+          req.abort();
+          reject('timeout');
+        });
+        return req.end();
+      });
+
+      try {
+        await promiseRequest;
+        assert.fail();
+      } catch (error) {
+        const spans = memoryExporter.getFinishedSpans();
+        const [span] = spans;
+        assert.strictEqual(spans.length, 1);
+        assert.strictEqual(span.status.code, CanonicalCode.ABORTED);
+        assert.ok(Object.keys(span.attributes).length > 7);
+      }
+    });
+
+    it('should have 1 ended span when request is aborted after receiving response', async () => {
+      nock('http://my.server.com')
+        .get('/')
+        .delay({
+          body: 50,
+        })
+        .replyWithFile(200, `${process.cwd()}/package.json`);
+
+      const promiseRequest = new Promise((resolve, reject) => {
+        const req = http.request(
+          'http://my.server.com',
+          (resp: http.IncomingMessage) => {
+            let data = '';
+            resp.on('data', chunk => {
+              req.abort();
+              data += chunk;
+            });
+            resp.on('end', () => {
+              resolve(data);
+            });
+          }
+        );
+
+        return req.end();
+      });
+
+      try {
+        await promiseRequest;
+        assert.fail();
+      } catch (error) {
+        const spans = memoryExporter.getFinishedSpans();
+        const [span] = spans;
+        assert.strictEqual(spans.length, 1);
+        assert.strictEqual(span.status.code, CanonicalCode.ABORTED);
+        assert.ok(Object.keys(span.attributes).length > 7);
+      }
+    });
   });
 });