feat(exporter-collector): log upstream error #1459

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

@@ -30,24 +30,18 @@ import {
  ensureExportedCounterIsCorrect,
  ensureExportedObserverIsCorrect,
  ensureExportedValueRecorderIsCorrect,
+ MockedResponse,
 } from './helper';
 import { MetricRecord } from '@opentelemetry/metrics';
 import { ExportResult, ExportResultCode } from '@opentelemetry/core';
+import { CollectorExporterError } from '@opentelemetry/exporter-collector/build/src/types';
 
 const fakeRequest = {
  end: function () {},
  on: function () {},
  write: function () {},
 };
 
-const mockRes = {
- statusCode: 200,
-};
-
-const mockResError = {
- statusCode: 400,
-};
-
 // send is lazy loading file so need to wait a bit
 const waitTimeMS = 20;
 
@@ -164,9 +158,11 @@ describe('CollectorMetricExporter - node with proto over http', () => {
  collectorExporter.export(metrics, responseSpy);
 
  setTimeout(() => {
+ const mockRes = new MockedResponse(200);
  const args = spyRequest.args[0];
  const callback = args[1];
  callback(mockRes);
+ mockRes.send('success');
  setTimeout(() => {
  const result = responseSpy.args[0][0] as ExportResult;
  assert.strictEqual(result.code, ExportResultCode.SUCCESS);
@@ -181,14 +177,17 @@ describe('CollectorMetricExporter - node with proto over http', () => {
  collectorExporter.export(metrics, responseSpy);
 
  setTimeout(() => {
+ const mockRes = new MockedResponse(400);
  const args = spyRequest.args[0];
  const callback = args[1];
- callback(mockResError);
+ callback(mockRes);
+ mockRes.send('failed');
  setTimeout(() => {
  const result = responseSpy.args[0][0] as ExportResult;
  assert.strictEqual(result.code, ExportResultCode.FAILED);
- // @ts-expect-error
- assert.strictEqual(result.error?.code, 400);
+ const error = result.error as CollectorExporterError;
+ assert.strictEqual(error.code, 400);
+ assert.strictEqual(error.data, 'failed');
  done();
  });
  }, waitTimeMS);

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

@@ -28,6 +28,7 @@ import {
  ensureExportTraceServiceRequestIsSet,
  ensureProtoSpanIsCorrect,
  mockedReadableSpan,
+ MockedResponse,
 } from './helper';
 import { ExportResult, ExportResultCode } from '@opentelemetry/core';
 
@@ -37,14 +38,6 @@ const fakeRequest = {
  write: function () {},
 };
 
-const mockRes = {
- statusCode: 200,
-};
-
-const mockResError = {
- statusCode: 400,
-};
-
 // send is lazy loading file so need to wait a bit
 const waitTimeMS = 20;
 
@@ -130,9 +123,11 @@ describe('CollectorTraceExporter - node with proto over http', () => {
  collectorExporter.export(spans, responseSpy);
 
  setTimeout(() => {
+ const mockRes = new MockedResponse(200);
  const args = spyRequest.args[0];
  const callback = args[1];
  callback(mockRes);
+ mockRes.send('success');
  setTimeout(() => {
  const result = responseSpy.args[0][0] as ExportResult;
  assert.strictEqual(result.code, ExportResultCode.SUCCESS);
@@ -147,9 +142,11 @@ describe('CollectorTraceExporter - node with proto over http', () => {
  collectorExporter.export(spans, responseSpy);
 
  setTimeout(() => {
+ const mockResError = new MockedResponse(400);
  const args = spyRequest.args[0];
  const callback = args[1];
  callback(mockResError);
+ mockResError.send('failed');
  setTimeout(() => {
  const result = responseSpy.args[0][0] as ExportResult;
  assert.strictEqual(result.code, ExportResultCode.FAILED);

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

@@ -21,6 +21,7 @@ import { Resource } from '@opentelemetry/resources';
 import { collectorTypes } from '@opentelemetry/exporter-collector';
 import * as assert from 'assert';
 import { MeterProvider, MetricRecord } from '@opentelemetry/metrics';
+import { Stream } from 'stream';
 
 const meterProvider = new MeterProvider({
  interval: 30000,
@@ -424,3 +425,22 @@ export function ensureExportMetricsServiceRequestIsSet(
  const metrics = resourceMetrics[0].instrumentationLibraryMetrics[0].metrics;
  assert.strictEqual(metrics.length, 3, 'Metrics are missing');
 }
+
+export class MockedResponse extends Stream {
+ constructor(private _code: number, private _msg?: string) {
+ super();
+ }
+
+ send(data: string) {
+ this.emit('data', data);
+ this.emit('end');
+ }
+
+ get statusCode() {
+ return this._code;
+ }
+
+ get statusMessage() {
+ return this._msg;
+ }
+}

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

@@ -50,16 +50,21 @@ export function sendWithHttp<ExportItem, ServiceRequest>(
 
  const request = parsedUrl.protocol === 'http:' ? http.request : https.request;
  const req = request(options, (res: http.IncomingMessage) => {
- if (res.statusCode && res.statusCode < 299) {
- collector.logger.debug(`statusCode: ${res.statusCode}`);
- onSuccess();
- } else {
- const error = new collectorTypes.CollectorExporterError(
- res.statusMessage,
- res.statusCode
- );
- onError(error);
- }
+ let data = '';
+ res.on('data', chunk => (data += chunk));
+ res.on('end', () => {
+ if (res.statusCode && res.statusCode < 299) {
+ collector.logger.debug(`statusCode: ${res.statusCode}`, data);
+ onSuccess();
+ } else {
+ const error = new collectorTypes.CollectorExporterError(
+ res.statusMessage,
+ res.statusCode,
+ data
+ );
+ onError(error);
+ }
+ });
  });
 
  req.on('error', (error: Error) => {

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

@@ -310,9 +310,11 @@ export namespace opentelemetryProto {
 export class CollectorExporterError extends Error {
  readonly code?: number;
  readonly name: string = 'CollectorExporterError';
+ readonly data?: string;
 
- constructor(message?: string, code?: number) {
+ constructor(message?: string, code?: number, data?: string) {
  super(message);
+ this.data = data;
  this.code = code;
  }
 }

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

@@ -14,15 +14,20 @@
  * limitations under the License.
  */
 
-import { ConsoleLogger, ExportResultCode, LogLevel } from '@opentelemetry/core';
+import {
+ ConsoleLogger,
+ ExportResult,
+ ExportResultCode,
+ LogLevel,
+} from '@opentelemetry/core';
 import * as core from '@opentelemetry/core';
 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 { MockedResponse } from './nodeHelpers';
 import {
  mockCounter,
  mockObserver,
@@ -40,10 +45,6 @@ const fakeRequest = {
  write: function () {},
 };
 
-const mockRes = {
- statusCode: 200,
-};
-
 const address = 'localhost:1501';
 
 describe('CollectorMetricExporter - node with json over http', () => {
@@ -167,19 +168,18 @@ describe('CollectorMetricExporter - node with json over http', () => {
  });
 
  it('should log the successful message', done => {
- const spyLoggerDebug = sinon.stub(collectorExporter.logger, 'debug');
  const spyLoggerError = sinon.stub(collectorExporter.logger, 'error');
 
  const responseSpy = sinon.spy();
  collectorExporter.export(metrics, responseSpy);
 
  setTimeout(() => {
+ const mockRes = new MockedResponse(200);
  const args = spyRequest.args[0];
  const callback = args[1];
  callback(mockRes);
+ mockRes.send('success');
  setTimeout(() => {
- const response: any = spyLoggerDebug.args[1][0];
- assert.strictEqual(response, 'statusCode: 200');
  assert.strictEqual(spyLoggerError.args.length, 0);
  assert.strictEqual(
  responseSpy.args[0][0].code,
@@ -189,6 +189,37 @@ describe('CollectorMetricExporter - node with json over http', () => {
  });
  });
  });
+
+ it('should log the error message', done => {
+ const spyLoggerError = sinon.spy();
+ const handler = core.loggingErrorHandler({
+ debug: sinon.fake(),
+ info: sinon.fake(),
+ warn: sinon.fake(),
+ error: spyLoggerError,
+ });
+ core.setGlobalErrorHandler(handler);
+
+ const responseSpy = sinon.spy();
+ collectorExporter.export(metrics, responseSpy);
+
+ setTimeout(() => {
+ const mockRes = new MockedResponse(400);
+ const args = spyRequest.args[0];
+ const callback = args[1];
+ callback(mockRes);
+ mockRes.send('failed');
+ setTimeout(() => {
+ const result = responseSpy.args[0][0] as ExportResult;
+ assert.strictEqual(result.code, ExportResultCode.FAILED);
+ const error = result.error as collectorTypes.CollectorExporterError;
+ assert.ok(error !== undefined);
+ assert.strictEqual(error.code, 400);
+ assert.strictEqual(error.data, 'failed');
+ done();
+ });
+ });
+ });
  });
  describe('CollectorMetricExporter - node (getDefaultUrl)', () => {
  it('should default to localhost', done => {

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

@@ -14,7 +14,12 @@
  * limitations under the License.
  */
 
-import { ConsoleLogger, ExportResultCode, LogLevel } from '@opentelemetry/core';
+import {
+ ConsoleLogger,
+ ExportResultCode,
+ ExportResult,
+ LogLevel,
+} from '@opentelemetry/core';
 import * as core from '@opentelemetry/core';
 import { ReadableSpan } from '@opentelemetry/tracing';
 import * as http from 'http';
@@ -23,6 +28,7 @@ import * as sinon from 'sinon';
 import { CollectorTraceExporter } from '../../src/platform/node';
 import { CollectorExporterConfigBase } from '../../src/types';
 import * as collectorTypes from '../../src/types';
+import { MockedResponse } from './nodeHelpers';
 
 import {
  ensureExportTraceServiceRequestIsSet,
@@ -36,10 +42,6 @@ const fakeRequest = {
  write: function () {},
 };
 
-const mockRes = {
- statusCode: 200,
-};
-
 const address = 'localhost:1501';
 
 describe('CollectorTraceExporter - node with json over http', () => {
@@ -134,19 +136,17 @@ describe('CollectorTraceExporter - node with json over http', () => {
  });
 
  it('should log the successful message', done => {
- const spyLoggerDebug = sinon.stub(collectorExporter.logger, 'debug');
  const spyLoggerError = sinon.stub(collectorExporter.logger, 'error');
-
  const responseSpy = sinon.spy();
  collectorExporter.export(spans, responseSpy);
 
  setTimeout(() => {
+ const mockRes = new MockedResponse(200);
  const args = spyRequest.args[0];
  const callback = args[1];
  callback(mockRes);
+ mockRes.send('success');
  setTimeout(() => {
- const response: any = spyLoggerDebug.args[1][0];
- assert.strictEqual(response, 'statusCode: 200');
  assert.strictEqual(spyLoggerError.args.length, 0);
  assert.strictEqual(
  responseSpy.args[0][0].code,
@@ -156,6 +156,28 @@ describe('CollectorTraceExporter - node with json over http', () => {
  });
  });
  });
+
+ it('should log the error message', done => {
+ const responseSpy = sinon.spy();
+ collectorExporter.export(spans, responseSpy);
+
+ setTimeout(() => {
+ const mockResError = new MockedResponse(400);
+ const args = spyRequest.args[0];
+ const callback = args[1];
+ callback(mockResError);
+ mockResError.send('failed');
+ setTimeout(() => {
+ const result = responseSpy.args[0][0] as ExportResult;
+ assert.strictEqual(result.code, ExportResultCode.FAILED);
+ const error = result.error as collectorTypes.CollectorExporterError;
+ assert.ok(error !== undefined);
+ assert.strictEqual(error.code, 400);
+ assert.strictEqual(error.data, 'failed');
+ done();
+ });
+ });
+ });
  });
  describe('CollectorTraceExporter - node (getDefaultUrl)', () => {
  it('should default to localhost', done => {

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

@@ -0,0 +1,36 @@
+/*
+ * 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 { Stream } from 'stream';
+
+export class MockedResponse extends Stream {
+ constructor(private _code: number, private _msg?: string) {
+ super();
+ }
+
+ send(data: string) {
+ this.emit('data', data);
+ this.emit('end');
+ }
+
+ get statusCode() {
+ return this._code;
+ }
+
+ get statusMessage() {
+ return this._msg;
+ }
+}