Use http keep-alive in collector exporter

packages/opentelemetry-exporter-collector-proto/src/CollectorExporterNodeBase.ts

@@ -55,7 +55,7 @@ export abstract class CollectorExporterNodeBase<
     this._sendingPromises.push(promise);
   }
 
-  onInit(config: collectorTypes.CollectorExporterConfigBase): void {
+  onInit(config: collectorTypes.CollectorExporterNodeConfigBase): void {
     this._isShutdown = false;
     // defer to next tick and lazy load to avoid loading protobufjs too early
     // and making this impossible to be instrumented

packages/opentelemetry-exporter-collector-proto/src/CollectorMetricExporter.ts

@@ -47,15 +47,17 @@ export class CollectorMetricExporter
     );
   }
 
-  getDefaultUrl(config: collectorTypes.CollectorExporterConfigBase): string {
+  getDefaultUrl(
+    config: collectorTypes.CollectorExporterNodeConfigBase
+  ): string {
     if (!config.url) {
       return DEFAULT_COLLECTOR_URL;
     }
     return config.url;
   }
 
   getDefaultServiceName(
-    config: collectorTypes.CollectorExporterConfigBase
+    config: collectorTypes.CollectorExporterNodeConfigBase
   ): string {
     return config.serviceName || DEFAULT_SERVICE_NAME;
   }

packages/opentelemetry-exporter-collector-proto/src/CollectorTraceExporter.ts

@@ -40,15 +40,17 @@ export class CollectorTraceExporter
     return toCollectorExportTraceServiceRequest(spans, this);
   }
 
-  getDefaultUrl(config: collectorTypes.CollectorExporterConfigBase): string {
+  getDefaultUrl(
+    config: collectorTypes.CollectorExporterNodeConfigBase
+  ): string {
     if (!config.url) {
       return DEFAULT_COLLECTOR_URL;
     }
     return config.url;
   }
 
   getDefaultServiceName(
-    config: collectorTypes.CollectorExporterConfigBase
+    config: collectorTypes.CollectorExporterNodeConfigBase
   ): string {
     return config.serviceName || DEFAULT_SERVICE_NAME;
   }

packages/opentelemetry-exporter-collector-proto/src/util.ts

@@ -33,7 +33,7 @@ export function getExportRequestProto(): Type | undefined {
 
 export function onInit<ExportItem, ServiceRequest>(
   collector: CollectorExporterNodeBase<ExportItem, ServiceRequest>,
-  _config: collectorTypes.CollectorExporterConfigBase
+  _config: collectorTypes.CollectorExporterNodeConfigBase
 ): void {
   const dir = path.resolve(__dirname, '..', 'protos');
   const root = new protobufjs.Root();

packages/opentelemetry-exporter-collector-proto/test/CollectorMetricExporter.test.ts

@@ -53,7 +53,7 @@ const waitTimeMS = 20;
 
 describe('CollectorMetricExporter - node with proto over http', () => {
   let collectorExporter: CollectorMetricExporter;
-  let collectorExporterConfig: collectorTypes.CollectorExporterConfigBase;
+  let collectorExporterConfig: collectorTypes.CollectorExporterNodeConfigBase;
   let spyRequest: sinon.SinonSpy;
   let spyWrite: sinon.SinonSpy;
   let metrics: MetricRecord[];
@@ -70,6 +70,8 @@ describe('CollectorMetricExporter - node with proto over http', () => {
         serviceName: 'bar',
         attributes: {},
         url: 'http://foo.bar.com',
+        keepAlive: true,
+        httpAgentOptions: { keepAliveMsecs: 2000 },
       };
       collectorExporter = new CollectorMetricExporter(collectorExporterConfig);
       // Overwrites the start time to make tests consistent
@@ -116,6 +118,19 @@ describe('CollectorMetricExporter - node with proto over http', () => {
       }, waitTimeMS);
     });
 
+    it('should have keep alive and keepAliveMsecs option set', done => {
+      collectorExporter.export(metrics, () => {});
+
+      setTimeout(() => {
+        const args = spyRequest.args[0];
+        const options = args[0];
+        const agent = options.agent;
+        assert.strictEqual(agent.keepAlive, true);
+        assert.strictEqual(agent.options.keepAliveMsecs, 2000);
+        done();
+      });
+    });
+
     it('should successfully send metrics', done => {
       collectorExporter.export(metrics, () => {});
 

packages/opentelemetry-exporter-collector-proto/test/CollectorTraceExporter.test.ts

@@ -50,7 +50,7 @@ const waitTimeMS = 20;
 
 describe('CollectorTraceExporter - node with proto over http', () => {
   let collectorExporter: CollectorTraceExporter;
-  let collectorExporterConfig: collectorTypes.CollectorExporterConfigBase;
+  let collectorExporterConfig: collectorTypes.CollectorExporterNodeConfigBase;
   let spyRequest: sinon.SinonSpy;
   let spyWrite: sinon.SinonSpy;
   let spans: ReadableSpan[];
@@ -67,6 +67,8 @@ describe('CollectorTraceExporter - node with proto over http', () => {
         serviceName: 'bar',
         attributes: {},
         url: 'http://foo.bar.com',
+        keepAlive: true,
+        httpAgentOptions: { keepAliveMsecs: 2000 },
       };
       collectorExporter = new CollectorTraceExporter(collectorExporterConfig);
       spans = [];
@@ -102,6 +104,19 @@ describe('CollectorTraceExporter - node with proto over http', () => {
       }, waitTimeMS);
     });
 
+    it('should have keep alive and keepAliveMsecs option set', done => {
+      collectorExporter.export(spans, () => {});
+
+      setTimeout(() => {
+        const args = spyRequest.args[0];
+        const options = args[0];
+        const agent = options.agent;
+        assert.strictEqual(agent.keepAlive, true);
+        assert.strictEqual(agent.options.keepAliveMsecs, 2000);
+        done();
+      });
+    });
+
     it('should successfully send the spans', done => {
       collectorExporter.export(spans, () => {});
 

packages/opentelemetry-exporter-collector/src/platform/node/CollectorExporterNodeBase.ts

@@ -14,8 +14,11 @@
  * limitations under the License.
  */
 
+import * as http from 'http';
+import * as https from 'https';
+
 import { CollectorExporterBase } from '../../CollectorExporterBase';
-import { CollectorExporterConfigBase } from '../../types';
+import { CollectorExporterNodeConfigBase } from '../../types';
 import * as collectorTypes from '../../types';
 import { parseHeaders } from '../../util';
 import { sendWithHttp } from './util';
@@ -27,22 +30,30 @@ export abstract class CollectorExporterNodeBase<
   ExportItem,
   ServiceRequest
 > extends CollectorExporterBase<
-  CollectorExporterConfigBase,
+  CollectorExporterNodeConfigBase,
   ExportItem,
   ServiceRequest
 > {
   DEFAULT_HEADERS: Record<string, string> = {};
   headers: Record<string, string>;
-  constructor(config: CollectorExporterConfigBase = {}) {
+  keepAlive: boolean = true;
+  httpAgentOptions: http.AgentOptions | https.AgentOptions = {};
+  constructor(config: CollectorExporterNodeConfigBase = {}) {
     super(config);
     if ((config as any).metadata) {
       this.logger.warn('Metadata cannot be set when using http');
     }
     this.headers =
       parseHeaders(config.headers, this.logger) || this.DEFAULT_HEADERS;
+    if (config.keepAlive !== undefined && !config.keepAlive) {
+      this.keepAlive = config.keepAlive;
+    }
+    if (config.httpAgentOptions) {
+      this.httpAgentOptions = config.httpAgentOptions;
+    }
   }
 
-  onInit(_config: CollectorExporterConfigBase): void {
+  onInit(_config: CollectorExporterNodeConfigBase): void {
     this._isShutdown = false;
   }
 

packages/opentelemetry-exporter-collector/src/platform/node/CollectorMetricExporter.ts

@@ -15,7 +15,6 @@
  */
 
 import { MetricRecord, MetricExporter } from '@opentelemetry/metrics';
-import { CollectorExporterConfigBase } from '../../types';
 import * as collectorTypes from '../../types';
 import { CollectorExporterNodeBase } from './CollectorExporterNodeBase';
 import { toCollectorExportMetricServiceRequest } from '../../transformMetrics';
@@ -45,14 +44,18 @@ export class CollectorMetricExporter
     );
   }
 
-  getDefaultUrl(config: CollectorExporterConfigBase): string {
+  getDefaultUrl(
+    config: collectorTypes.CollectorExporterNodeConfigBase
+  ): string {
     if (!config.url) {
       return DEFAULT_COLLECTOR_URL;
     }
     return config.url;
   }
 
-  getDefaultServiceName(config: CollectorExporterConfigBase): string {
+  getDefaultServiceName(
+    config: collectorTypes.CollectorExporterNodeConfigBase
+  ): string {
     return config.serviceName || DEFAULT_SERVICE_NAME;
   }
 }

packages/opentelemetry-exporter-collector/src/platform/node/CollectorTraceExporter.ts

@@ -15,7 +15,6 @@
  */
 
 import { ReadableSpan, SpanExporter } from '@opentelemetry/tracing';
-import { CollectorExporterConfigBase } from '../../types';
 import { CollectorExporterNodeBase } from './CollectorExporterNodeBase';
 import * as collectorTypes from '../../types';
 import { toCollectorExportTraceServiceRequest } from '../../transform';
@@ -38,14 +37,18 @@ export class CollectorTraceExporter
     return toCollectorExportTraceServiceRequest(spans, this, true);
   }
 
-  getDefaultUrl(config: CollectorExporterConfigBase): string {
+  getDefaultUrl(
+    config: collectorTypes.CollectorExporterNodeConfigBase
+  ): string {
     if (!config.url) {
       return DEFAULT_COLLECTOR_URL;
     }
     return config.url;
   }
 
-  getDefaultServiceName(config: CollectorExporterConfigBase): string {
+  getDefaultServiceName(
+    config: collectorTypes.CollectorExporterNodeConfigBase
+  ): string {
     return config.serviceName || DEFAULT_SERVICE_NAME;
   }
 }

packages/opentelemetry-exporter-collector/src/platform/node/util.ts

@@ -36,7 +36,7 @@ export function sendWithHttp<ExportItem, ServiceRequest>(
 ): void {
   const parsedUrl = new url.URL(collector.url);
 
-  const options = {
+  const options: http.RequestOptions | https.RequestOptions = {
     hostname: parsedUrl.hostname,
     port: parsedUrl.port,
     path: parsedUrl.pathname,
@@ -49,6 +49,14 @@ export function sendWithHttp<ExportItem, ServiceRequest>(
   };
 
   const request = parsedUrl.protocol === 'http:' ? http.request : https.request;
+  const Agent = parsedUrl.protocol === 'http:' ? http.Agent : https.Agent;
+  if (collector.keepAlive) {
+    options.agent = new Agent({
+      ...collector.httpAgentOptions,
+      keepAlive: true,
+    });
+  }
+
   const req = request(options, (res: http.IncomingMessage) => {
     if (res.statusCode && res.statusCode < 299) {
       collector.logger.debug(`statusCode: ${res.statusCode}`);

packages/opentelemetry-exporter-collector/src/types.ts

@@ -16,6 +16,8 @@
 
 import { SpanKind, Logger, Attributes } from '@opentelemetry/api';
 import * as api from '@opentelemetry/api';
+import * as http from 'http';
+import * as https from 'https';
 
 /* eslint-disable @typescript-eslint/no-namespace */
 export namespace opentelemetryProto {
@@ -341,6 +343,15 @@ export interface CollectorExporterConfigBase {
   url?: string;
 }
 
+/**
+ * Collector Exporter node base config
+ */
+export interface CollectorExporterNodeConfigBase
+  extends CollectorExporterConfigBase {
+  keepAlive?: boolean;
+  httpAgentOptions?: http.AgentOptions | https.AgentOptions;
+}
+
 /**
  * Mapping between api SpanKind and proto SpanKind
  */

packages/opentelemetry-exporter-collector/test/node/CollectorMetricExporter.test.ts

@@ -20,7 +20,6 @@ import * as http from 'http';
 import * as assert from 'assert';
 import * as sinon from 'sinon';
 import { CollectorMetricExporter } from '../../src/platform/node';
-import { CollectorExporterConfigBase } from '../../src/types';
 import * as collectorTypes from '../../src/types';
 
 import {
@@ -48,7 +47,7 @@ const address = 'localhost:1501';
 
 describe('CollectorMetricExporter - node with json over http', () => {
   let collectorExporter: CollectorMetricExporter;
-  let collectorExporterConfig: CollectorExporterConfigBase;
+  let collectorExporterConfig: collectorTypes.CollectorExporterNodeConfigBase;
   let spyRequest: sinon.SinonSpy;
   let spyWrite: sinon.SinonSpy;
   let metrics: MetricRecord[];
@@ -81,6 +80,8 @@ describe('CollectorMetricExporter - node with json over http', () => {
         serviceName: 'bar',
         attributes: {},
         url: 'http://foo.bar.com',
+        keepAlive: true,
+        httpAgentOptions: { keepAliveMsecs: 2000 },
       };
       collectorExporter = new CollectorMetricExporter(collectorExporterConfig);
       // Overwrites the start time to make tests consistent
@@ -128,6 +129,19 @@ describe('CollectorMetricExporter - node with json over http', () => {
       });
     });
 
+    it('should have keep alive and keepAliveMsecs option set', done => {
+      collectorExporter.export(metrics, () => {});
+
+      setTimeout(() => {
+        const args = spyRequest.args[0];
+        const options = args[0];
+        const agent = options.agent;
+        assert.strictEqual(agent.keepAlive, true);
+        assert.strictEqual(agent.options.keepAliveMsecs, 2000);
+        done();
+      });
+    });
+
     it('should successfully send metrics', done => {
       collectorExporter.export(metrics, () => {});
 

packages/opentelemetry-exporter-collector/test/node/CollectorTraceExporter.test.ts

@@ -21,7 +21,6 @@ import * as http from 'http';
 import * as assert from 'assert';
 import * as sinon from 'sinon';
 import { CollectorTraceExporter } from '../../src/platform/node';
-import { CollectorExporterConfigBase } from '../../src/types';
 import * as collectorTypes from '../../src/types';
 
 import {
@@ -44,7 +43,7 @@ const address = 'localhost:1501';
 
 describe('CollectorTraceExporter - node with json over http', () => {
   let collectorExporter: CollectorTraceExporter;
-  let collectorExporterConfig: CollectorExporterConfigBase;
+  let collectorExporterConfig: collectorTypes.CollectorExporterNodeConfigBase;
   let spyRequest: sinon.SinonSpy;
   let spyWrite: sinon.SinonSpy;
   let spans: ReadableSpan[];
@@ -77,6 +76,8 @@ describe('CollectorTraceExporter - node with json over http', () => {
         serviceName: 'bar',
         attributes: {},
         url: 'http://foo.bar.com',
+        keepAlive: true,
+        httpAgentOptions: { keepAliveMsecs: 2000 },
       };
       collectorExporter = new CollectorTraceExporter(collectorExporterConfig);
       spans = [];
@@ -112,6 +113,19 @@ describe('CollectorTraceExporter - node with json over http', () => {
       });
     });
 
+    it('should have keep alive and keepAliveMsecs option set', done => {
+      collectorExporter.export(spans, () => {});
+
+      setTimeout(() => {
+        const args = spyRequest.args[0];
+        const options = args[0];
+        const agent = options.agent;
+        assert.strictEqual(agent.keepAlive, true);
+        assert.strictEqual(agent.options.keepAliveMsecs, 2000);
+        done();
+      });
+    });
+
     it('should successfully send the spans', done => {
       collectorExporter.export(spans, () => {});