Use http keep-alive in collector exporter

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

@@ -17,6 +17,7 @@
 import {
  CollectorExporterNodeBase as CollectorExporterBaseMain,
  collectorTypes,
+ CollectorExporterNodeConfigBase,
 } from '@opentelemetry/exporter-collector';
 import { ServiceClientType } from './types';
 
@@ -55,7 +56,7 @@ export abstract class CollectorExporterNodeBase<
  this._sendingPromises.push(promise);
  }
 
- onInit(config: collectorTypes.CollectorExporterConfigBase): void {
+ onInit(config: 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

@@ -17,6 +17,7 @@
 import {
  collectorTypes,
  toCollectorExportMetricServiceRequest,
+ CollectorExporterNodeConfigBase,
 } from '@opentelemetry/exporter-collector';
 import { MetricRecord, MetricExporter } from '@opentelemetry/metrics';
 import { ServiceClientType } from './types';
@@ -47,16 +48,14 @@ export class CollectorMetricExporter
  );
  }
 
- getDefaultUrl(config: collectorTypes.CollectorExporterConfigBase): string {
+ getDefaultUrl(config: CollectorExporterNodeConfigBase): string {
  if (!config.url) {
  return DEFAULT_COLLECTOR_URL;
  }
  return config.url;
  }
 
- getDefaultServiceName(
- config: collectorTypes.CollectorExporterConfigBase
- ): string {
+ getDefaultServiceName(config: CollectorExporterNodeConfigBase): string {
  return config.serviceName || DEFAULT_SERVICE_NAME;
  }
 

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

@@ -19,6 +19,7 @@ import { CollectorExporterNodeBase } from './CollectorExporterNodeBase';
 import {
  collectorTypes,
  toCollectorExportTraceServiceRequest,
+ CollectorExporterNodeConfigBase,
 } from '@opentelemetry/exporter-collector';
 import { ServiceClientType } from './types';
 
@@ -40,16 +41,14 @@ export class CollectorTraceExporter
  return toCollectorExportTraceServiceRequest(spans, this);
  }
 
- getDefaultUrl(config: collectorTypes.CollectorExporterConfigBase): string {
+ getDefaultUrl(config: CollectorExporterNodeConfigBase): string {
  if (!config.url) {
  return DEFAULT_COLLECTOR_URL;
  }
  return config.url;
  }
 
- getDefaultServiceName(
- config: collectorTypes.CollectorExporterConfigBase
- ): string {
+ getDefaultServiceName(config: CollectorExporterNodeConfigBase): string {
  return config.serviceName || DEFAULT_SERVICE_NAME;
  }
 

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

@@ -17,6 +17,7 @@
 import {
  collectorTypes,
  sendWithHttp,
+ CollectorExporterNodeConfigBase,
 } from '@opentelemetry/exporter-collector';
 import * as path from 'path';
 
@@ -33,7 +34,7 @@ export function getExportRequestProto(): Type | undefined {
 
 export function onInit<ExportItem, ServiceRequest>(
  collector: CollectorExporterNodeBase<ExportItem, ServiceRequest>,
- _config: collectorTypes.CollectorExporterConfigBase
+ _config: CollectorExporterNodeConfigBase
 ): void {
  const dir = path.resolve(__dirname, '..', 'protos');
  const root = new protobufjs.Root();

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

@@ -14,7 +14,10 @@
  * limitations under the License.
  */
 
-import { collectorTypes } from '@opentelemetry/exporter-collector';
+import {
+ collectorTypes,
+ CollectorExporterNodeConfigBase,
+} from '@opentelemetry/exporter-collector';
 import * as core from '@opentelemetry/core';
 import * as http from 'http';
 import * as assert from 'assert';
@@ -47,7 +50,7 @@ const waitTimeMS = 20;
 
 describe('CollectorMetricExporter - node with proto over http', () => {
  let collectorExporter: CollectorMetricExporter;
- let collectorExporterConfig: collectorTypes.CollectorExporterConfigBase;
+ let collectorExporterConfig: CollectorExporterNodeConfigBase;
  let spyRequest: sinon.SinonSpy;
  let spyWrite: sinon.SinonSpy;
  let metrics: MetricRecord[];
@@ -64,6 +67,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
@@ -110,6 +115,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

@@ -14,7 +14,10 @@
  * limitations under the License.
  */
 
-import { collectorTypes } from '@opentelemetry/exporter-collector';
+import {
+ collectorTypes,
+ CollectorExporterNodeConfigBase,
+} from '@opentelemetry/exporter-collector';
 
 import * as core from '@opentelemetry/core';
 import { ReadableSpan } from '@opentelemetry/tracing';
@@ -43,7 +46,7 @@ const waitTimeMS = 20;
 
 describe('CollectorTraceExporter - node with proto over http', () => {
  let collectorExporter: CollectorTraceExporter;
- let collectorExporterConfig: collectorTypes.CollectorExporterConfigBase;
+ let collectorExporterConfig: CollectorExporterNodeConfigBase;
  let spyRequest: sinon.SinonSpy;
  let spyWrite: sinon.SinonSpy;
  let spans: ReadableSpan[];
@@ -60,6 +63,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 = [];
@@ -95,6 +100,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 type * as http from 'http';
+import type * 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,35 @@ 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 (typeof config.keepAlive === 'boolean') {
+ this.keepAlive = config.keepAlive;
+ }
+ if (config.httpAgentOptions) {
+ if (!this.keepAlive) {
+ this.logger.warn(
+ 'httpAgentOptions is used only when keepAlive is true'
+ );
+ }
+ this.httpAgentOptions = Object.assign({}, config.httpAgentOptions);
+ }
  }
 
- onInit(_config: CollectorExporterConfigBase): void {
+ onInit(_config: CollectorExporterNodeConfigBase): void {
  this._isShutdown = false;
  }
 

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

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

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

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

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

@@ -18,3 +18,4 @@ export * from './CollectorTraceExporter';
 export * from './CollectorMetricExporter';
 export * from './CollectorExporterNodeBase';
 export * from './util';
+export * from './types';

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

@@ -0,0 +1,28 @@
+/*
+ * 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 type * as http from 'http';
+import type * as https from 'https';
+
+import { CollectorExporterConfigBase } from '../../types';
+
+/**
+ * Collector Exporter node base config
+ */
+export interface CollectorExporterNodeConfigBase
+ extends CollectorExporterConfigBase {
+ keepAlive?: boolean;
+ httpAgentOptions?: http.AgentOptions | https.AgentOptions;
+}

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) => {
  let data = '';
  res.on('data', chunk => (data += chunk));