fix(plugin-http): ensure no leaks (#398)

* fix(plugin-http): ensure no leaks

closes #397

Signed-off-by: Olivier Albertini <[email protected]>

* fix: add @Flarna recommandations

Signed-off-by: Olivier Albertini <[email protected]>

Author: OlivierAlbertini

packages/opentelemetry-plugin-http/package.json

@@ -38,12 +38,12 @@
     "access": "public"
   },
   "devDependencies": {
-    "@types/got": "^9.6.6",
+    "@types/got": "^9.6.7",
     "@types/mocha": "^5.2.7",
-    "@types/nock": "^10.0.3",
-    "@types/node": "^12.6.9",
-    "@types/request-promise-native": "^1.0.16",
-    "@types/semver": "^6.0.1",
+    "@types/nock": "^11.1.0",
+    "@types/node": "^12.7.9",
+    "@types/request-promise-native": "^1.0.17",
+    "@types/semver": "^6.0.2",
     "@types/shimmer": "^1.0.1",
     "@types/sinon": "^7.0.13",
     "@types/superagent": "^4.1.3",
@@ -55,17 +55,17 @@
     "request": "^2.88.0",
     "request-promise-native": "^1.0.7",
     "superagent": "5.1.0",
-    "codecov": "^3.5.0",
-    "gts": "^1.0.0",
-    "mocha": "^6.2.0",
-    "nock": "^11.0.0",
+    "codecov": "^3.6.1",
+    "gts": "^1.1.0",
+    "mocha": "^6.2.1",
+    "nock": "^11.3.5",
     "nyc": "^14.1.1",
     "rimraf": "^3.0.0",
-    "sinon": "^7.3.2",
+    "sinon": "^7.5.0",
     "tslint-microsoft-contrib": "^6.2.0",
-    "tslint-consistent-codestyle": "^1.15.1",
+    "tslint-consistent-codestyle": "^1.16.0",
     "ts-mocha": "^6.0.0",
-    "ts-node": "^8.3.0",
+    "ts-node": "^8.4.1",
     "typescript": "^3.6.3"
   },
   "dependencies": {

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

@@ -21,6 +21,7 @@ import {
   SpanOptions,
   Attributes,
   CanonicalCode,
+  Status,
 } from '@opentelemetry/types';
 import {
   ClientRequest,
@@ -39,6 +40,7 @@ import {
   ResponseEndArgs,
   ParsedRequestOptions,
   HttpRequestArgs,
+  Err,
 } from './types';
 import { Format } from './enums/Format';
 import { AttributeNames } from './enums/AttributeNames';
@@ -50,11 +52,14 @@ import * as utils from './utils';
 export class HttpPlugin extends BasePlugin<Http> {
   readonly component: string;
   protected _config!: HttpPluginConfig;
+  /** keep track on spans not ended */
+  private readonly _spanNotEnded: WeakSet<Span>;
 
   constructor(readonly moduleName: string, readonly version: string) {
     super();
     // For now component is equal to moduleName but it can change in the future.
     this.component = this.moduleName;
+    this._spanNotEnded = new WeakSet<Span>();
     this._config = {};
   }
 
@@ -192,29 +197,38 @@ export class HttpPlugin extends BasePlugin<Http> {
         [AttributeNames.HTTP_HOSTNAME]: host,
         [AttributeNames.HTTP_METHOD]: method,
         [AttributeNames.HTTP_PATH]: options.path || '/',
-        [AttributeNames.HTTP_USER_AGENT]: userAgent || '',
       });
 
+      if (userAgent !== undefined) {
+        span.setAttribute(AttributeNames.HTTP_USER_AGENT, userAgent);
+      }
+
       request.on(
         'response',
-        (response: IncomingMessage & { aborted?: boolean }) => {
+        (
+          response: IncomingMessage & { aborted?: boolean; req: ClientRequest }
+        ) => {
+          if (response.statusCode) {
+            span.setAttributes({
+              [AttributeNames.HTTP_STATUS_CODE]: response.statusCode,
+              [AttributeNames.HTTP_STATUS_TEXT]: response.statusMessage,
+            });
+          }
+
           this._tracer.bind(response);
           this._logger.debug('outgoingRequest on response()');
           response.on('end', () => {
             this._logger.debug('outgoingRequest on end()');
-            if (response.statusCode) {
-              span.setAttributes({
-                [AttributeNames.HTTP_STATUS_CODE]: response.statusCode,
-                [AttributeNames.HTTP_STATUS_TEXT]: response.statusMessage,
-              });
-            }
+            let status: Status;
 
             if (response.aborted && !response.complete) {
-              span.setStatus({ code: CanonicalCode.ABORTED });
+              status = { code: CanonicalCode.ABORTED };
             } else {
-              span.setStatus(utils.parseResponseStatus(response.statusCode!));
+              status = utils.parseResponseStatus(response.statusCode!);
             }
 
+            span.setStatus(status);
+
             if (this._config.applyCustomAttributesOnSpan) {
               this._safeExecute(
                 span,
@@ -228,13 +242,23 @@ export class HttpPlugin extends BasePlugin<Http> {
               );
             }
 
-            span.end();
+            this._closeHttpSpan(span);
+          });
+          response.on('error', (error: Err) => {
+            utils.setSpanWithError(span, error, response);
+            this._closeHttpSpan(span);
           });
-          utils.setSpanOnError(span, response);
         }
       );
-
-      utils.setSpanOnError(span, request);
+      request.on('close', () => {
+        if (!request.aborted) {
+          this._closeHttpSpan(span);
+        }
+      });
+      request.on('error', (error: Err) => {
+        utils.setSpanWithError(span, error, request);
+        this._closeHttpSpan(span);
+      });
 
       this._logger.debug('makeRequestTrace return request');
       return request;
@@ -331,7 +355,7 @@ export class HttpPlugin extends BasePlugin<Http> {
             attributes[AttributeNames.HTTP_ROUTE] = requestUrl.pathname || '/';
           }
 
-          if (userAgent) {
+          if (userAgent !== undefined) {
             attributes[AttributeNames.HTTP_USER_AGENT] = userAgent;
           }
 
@@ -352,7 +376,7 @@ export class HttpPlugin extends BasePlugin<Http> {
             );
           }
 
-          span.end();
+          plugin._closeHttpSpan(span);
           return returned;
         };
 
@@ -438,10 +462,22 @@ export class HttpPlugin extends BasePlugin<Http> {
   }
 
   private _startHttpSpan(name: string, options: SpanOptions) {
-    return this._tracer
+    const span = this._tracer
       .startSpan(name, options)
       .setAttribute(AttributeNames.COMPONENT, this.component);
+    this._spanNotEnded.add(span);
+    return span;
+  }
+
+  private _closeHttpSpan(span: Span) {
+    if (!this._spanNotEnded.has(span)) {
+      return;
+    }
+
+    span.end();
+    this._spanNotEnded.delete(span);
   }
+
   private _safeExecute<
     T extends (...args: unknown[]) => ReturnType<T>,
     K extends boolean
@@ -460,7 +496,7 @@ export class HttpPlugin extends BasePlugin<Http> {
     } catch (error) {
       if (rethrow) {
         utils.setSpanWithError(span, error);
-        span.end();
+        this._closeHttpSpan(span);
         throw error;
       }
       this._logger.error('caught error ', error);

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

@@ -154,21 +154,6 @@ export const isIgnored = (
   return false;
 };
 
-/**
- * Will subscribe obj on error event and will set attributes when emitting event
- * @param span to set
- * @param obj to subscribe on 'error' event (must extend `EventEmitter`)
- */
-export const setSpanOnError = (
-  span: Span,
-  obj: IncomingMessage | ClientRequest
-) => {
-  obj.on('error', (error: Err) => {
-    setSpanWithError(span, error, obj);
-    span.end();
-  });
-};
-
 /**
  * Sets the span with the error passed in params
  * @param {Span} span the span that need to be set
@@ -189,7 +174,6 @@ export const setSpanWithError = (
 
   if (!obj) {
     span.setStatus({ code: CanonicalCode.UNKNOWN, message });
-    span.end();
     return;
   }
 

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

@@ -30,6 +30,7 @@ import { DummyPropagation } from '../utils/DummyPropagation';
 import { httpRequest } from '../utils/httpRequest';
 import * as utils from '../../src/utils';
 import { HttpPluginConfig, Http } from '../../src/types';
+import { AttributeNames } from '../../src/enums/AttributeNames';
 
 const applyCustomAttributesOnSpanErrorMessage =
   'bad applyCustomAttributesOnSpan function';
@@ -541,12 +542,12 @@ describe('HttpPlugin', () => {
           const [span] = spans;
           assert.strictEqual(spans.length, 1);
           assert.strictEqual(span.status.code, CanonicalCode.ABORTED);
-          assert.ok(Object.keys(span.attributes).length > 7);
+          assert.ok(Object.keys(span.attributes).length > 6);
         }
       });
 
       it('should have 1 ended span when request is aborted after receiving response', async () => {
-        nock('http://my.server.com')
+        nock(`${protocol}://my.server.com`)
           .get('/')
           .delay({
             body: 50,
@@ -555,7 +556,7 @@ describe('HttpPlugin', () => {
 
         const promiseRequest = new Promise((resolve, reject) => {
           const req = http.request(
-            'http://my.server.com',
+            `${protocol}://my.server.com`,
             (resp: http.IncomingMessage) => {
               let data = '';
               resp.on('data', chunk => {
@@ -582,6 +583,44 @@ describe('HttpPlugin', () => {
           assert.ok(Object.keys(span.attributes).length > 7);
         }
       });
+
+      it("should have 1 ended span when request doesn't listening response", done => {
+        nock.cleanAll();
+        nock.enableNetConnect();
+        const req = http.request(`${protocol}://${hostname}/`);
+        req.on('close', () => {
+          const spans = memoryExporter.getFinishedSpans();
+          const [span] = spans;
+          assert.strictEqual(spans.length, 1);
+          assert.ok(Object.keys(span.attributes).length > 6);
+          done();
+        });
+        req.end();
+      });
+
+      it("should have 1 ended span when response is listened by using req.on('response')", done => {
+        const host = `${protocol}://${hostname}`;
+        nock(host)
+          .get('/')
+          .reply(404);
+        const req = http.request(`${host}/`);
+        req.on('response', response => {
+          response.on('data', () => {});
+          response.on('end', () => {
+            const spans = memoryExporter.getFinishedSpans();
+            const [span] = spans;
+            assert.strictEqual(spans.length, 1);
+            assert.ok(Object.keys(span.attributes).length > 6);
+            assert.strictEqual(
+              span.attributes[AttributeNames.HTTP_STATUS_CODE],
+              404
+            );
+            assert.strictEqual(span.status.code, CanonicalCode.NOT_FOUND);
+            done();
+          });
+        });
+        req.end();
+      });
     });
   });
 });

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

@@ -15,7 +15,7 @@
  */
 
 import { NoopLogger } from '@opentelemetry/core';
-import { SpanKind, Span } from '@opentelemetry/types';
+import { SpanKind } from '@opentelemetry/types';
 import * as assert from 'assert';
 import * as http from 'http';
 import * as nock from 'nock';
@@ -34,11 +34,10 @@ import {
   SimpleSpanProcessor,
 } from '@opentelemetry/tracing';
 
-const memoryExporter = new InMemorySpanExporter();
+import { HttpPluginConfig } from '../../src/types';
+import { customAttributeFunction } from './http-enable.test';
 
-export const customAttributeFunction = (span: Span): void => {
-  span.setAttribute('span kind', SpanKind.CLIENT);
-};
+const memoryExporter = new InMemorySpanExporter();
 
 describe('Packages', () => {
   describe('get', () => {
@@ -55,7 +54,10 @@ describe('Packages', () => {
     });
 
     before(() => {
-      plugin.enable(http, tracer, tracer.logger);
+      const config: HttpPluginConfig = {
+        applyCustomAttributesOnSpan: customAttributeFunction,
+      };
+      plugin.enable(http, tracer, tracer.logger, config);
     });
 
     after(() => {

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

@@ -17,7 +17,7 @@
 import * as assert from 'assert';
 import * as sinon from 'sinon';
 import * as url from 'url';
-import { CanonicalCode, Attributes, SpanKind } from '@opentelemetry/types';
+import { CanonicalCode, SpanKind } from '@opentelemetry/types';
 import { NoopScopeManager } from '@opentelemetry/scope-base';
 import { IgnoreMatcher } from '../../src/types';
 import * as utils from '../../src/utils';
@@ -235,27 +235,6 @@ describe('Utility', () => {
       assert.strictEqual(result, 'http://localhost:8080/helloworld');
     });
   });
-  describe('setSpanOnError()', () => {
-    it('should call span methods when we get an error event', done => {
-      /* tslint:disable-next-line:no-any */
-      const span: any = {
-        setAttributes: (obj: Attributes) => {},
-        setStatus: (status: unknown) => {},
-        end: () => {},
-      };
-      sinon.spy(span, 'setAttributes');
-      sinon.spy(span, 'setStatus');
-      sinon.spy(span, 'end');
-      const req = http.get('http://noop');
-      utils.setSpanOnError(span, req);
-      req.on('error', () => {
-        assert.strictEqual(span.setAttributes.callCount, 1);
-        assert.strictEqual(span.setStatus.callCount, 1);
-        assert.strictEqual(span.end.callCount, 1);
-        done();
-      });
-    });
-  });
 
   describe('setSpanWithError()', () => {
     it('should have error attributes', () => {

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

@@ -436,7 +436,7 @@ describe('HttpsPlugin', () => {
         const [span] = spans;
         assert.strictEqual(spans.length, 1);
         assert.strictEqual(span.status.code, CanonicalCode.ABORTED);
-        assert.ok(Object.keys(span.attributes).length > 7);
+        assert.ok(Object.keys(span.attributes).length > 6);
       }
     });