feat(exporters)!: use transport interface in node.js exporters (#4743)

* feat(exporters)!: use transport interface in node.js exporters

* feat(exporters): hide compression property

* feat(otlp-exporter-base)!: remove header property

* feat(otlp-exporter-base): add retrying transport

* fix: lint

* chore: add changelog entry

* fix: use queueMicrotask over nextTick

* chore: move changelog entry to unreleased

* chore: note that user-agent cannot be overwritten by users anymore

* fix: export missing ExportResponseRetryable

* fix: retry jitter

Author: pichlermarc

experimental/CHANGELOG.md

@@ -8,6 +8,12 @@ All notable changes to experimental packages in this project will be documented
 ### :boom: Breaking Change
 
 * fix(instrumentation)!:remove unused description property from interface [#4847](https://github.com/open-telemetry/opentelemetry-js/pull/4847) @blumamir
+* feat(exporter-*-otlp-*)!: use transport interface in node.js exporters [#4743](https://github.com/open-telemetry/opentelemetry-js/pull/4743) @pichlermarc
+  * (user-facing) `headers` was intended for internal use has been removed from all exporters
+  * (user-facing) `compression` was intended for internal use and has been removed from all exporters
+  * (user-facing) `hostname` was intended for use in tests and is not used by any exporters, it will be removed in a future release
+* fix(exporter-*-otlp-*)!: ensure `User-Agent` header cannot be overwritten by the user [#4743](https://github.com/open-telemetry/opentelemetry-js/pull/4743) @pichlermarc
+  * allowing overrides of the `User-Agent` header was not specification compliant.
 
 ### :rocket: (Enhancement)
 
@@ -79,7 +85,7 @@ All notable changes to experimental packages in this project will be documented
 
 ### :bug: (Bug Fix)
 
-* fix(instrumentation): Update `import-in-the-middle` to fix [numerous bugs](https://github.com/DataDog/import-in-the-middle/releases/tag/v1.8.0) [#4745](https://github.com/open-telemetry/opentelemetry-js/pull/4745) @timfish
+* fix(instrumentation): Update `import-in-the-middle` to fix [numerous bugs](https://github.com/DataDog/import-in-the-middle/pull/91) [#4745](https://github.com/open-telemetry/opentelemetry-js/pull/4745) @timfish
 
 ### :books: (Refine Doc)
 

experimental/packages/exporter-logs-otlp-http/src/platform/node/OTLPLogExporter.ts

@@ -49,16 +49,15 @@ export class OTLPLogExporter
         ...config,
       },
       JsonLogsSerializer,
-      'application/json'
+      {
+        ...baggageUtils.parseKeyPairsIntoRecord(
+          getEnv().OTEL_EXPORTER_OTLP_LOGS_HEADERS
+        ),
+        ...parseHeaders(config?.headers),
+        ...USER_AGENT,
+        'Content-Type': 'application/json',
+      }
     );
-    this.headers = {
-      ...this.headers,
-      ...USER_AGENT,
-      ...baggageUtils.parseKeyPairsIntoRecord(
-        getEnv().OTEL_EXPORTER_OTLP_LOGS_HEADERS
-      ),
-      ...parseHeaders(config?.headers),
-    };
   }
 
   getDefaultUrl(config: OTLPExporterNodeConfigBase): string {

experimental/packages/exporter-logs-otlp-http/test/node/OTLPLogExporter.test.ts

@@ -43,7 +43,7 @@ class MockedResponse extends Stream {
     super();
   }
 
-  send(data: string) {
+  send(data: Uint8Array) {
     this.emit('data', data);
     this.emit('end');
   }
@@ -65,6 +65,9 @@ describe('OTLPLogExporter', () => {
 
   afterEach(() => {
     fakeRequest = new Stream.PassThrough();
+    Object.defineProperty(fakeRequest, 'setTimeout', {
+      value: function (_timeout: number) {},
+    });
     sinon.restore();
   });
 
@@ -83,15 +86,20 @@ describe('OTLPLogExporter', () => {
     it('should include user-agent header by default', () => {
       const exporter = new OTLPLogExporter();
       assert.strictEqual(
-        exporter.headers['User-Agent'],
+        exporter['_transport']['_transport']['_parameters']['headers'][
+          'User-Agent'
+        ],
         `OTel-OTLP-Exporter-JavaScript/${VERSION}`
       );
     });
 
     it('should use headers defined via env', () => {
       envSource.OTEL_EXPORTER_OTLP_LOGS_HEADERS = 'foo=bar';
       const exporter = new OTLPLogExporter();
-      assert.strictEqual(exporter.headers.foo, 'bar');
+      assert.strictEqual(
+        exporter['_transport']['_transport']['_parameters']['headers']['foo'],
+        'bar'
+      );
       delete envSource.OTEL_EXPORTER_OTLP_LOGS_HEADERS;
     });
 
@@ -106,13 +114,19 @@ describe('OTLPLogExporter', () => {
 
     it('should override headers defined via env with headers defined in constructor', () => {
       envSource.OTEL_EXPORTER_OTLP_HEADERS = 'foo=bar,bar=foo';
-      const collectorExporter = new OTLPLogExporter({
+      const exporter = new OTLPLogExporter({
         headers: {
           foo: 'constructor',
         },
       });
-      assert.strictEqual(collectorExporter.headers.foo, 'constructor');
-      assert.strictEqual(collectorExporter.headers.bar, 'foo');
+      assert.strictEqual(
+        exporter['_transport']['_transport']['_parameters']['headers']['foo'],
+        'constructor'
+      );
+      assert.strictEqual(
+        exporter['_transport']['_transport']['_parameters']['headers']['bar'],
+        'foo'
+      );
       envSource.OTEL_EXPORTER_OTLP_HEADERS = '';
     });
   });
@@ -152,10 +166,12 @@ describe('OTLPLogExporter', () => {
         assert.strictEqual(options.method, 'POST');
         assert.strictEqual(options.path, '/');
 
-        const mockRes = new MockedResponse(200);
-        cb(mockRes);
-        mockRes.send('success');
-        done();
+        queueMicrotask(() => {
+          const mockRes = new MockedResponse(200);
+          cb(mockRes);
+          mockRes.send(Buffer.from('success'));
+          done();
+        });
         return fakeRequest as any;
       });
       collectorExporter.export(logs, () => {});
@@ -165,10 +181,12 @@ describe('OTLPLogExporter', () => {
       sinon.stub(http, 'request').callsFake((options: any, cb: any) => {
         assert.strictEqual(options.headers['foo'], 'bar');
 
-        const mockRes = new MockedResponse(200);
-        cb(mockRes);
-        mockRes.send('success');
-        done();
+        queueMicrotask(() => {
+          const mockRes = new MockedResponse(200);
+          cb(mockRes);
+          mockRes.send(Buffer.from('success'));
+          done();
+        });
         return fakeRequest as any;
       });
 
@@ -180,10 +198,12 @@ describe('OTLPLogExporter', () => {
         assert.strictEqual(options.agent.keepAlive, true);
         assert.strictEqual(options.agent.options.keepAliveMsecs, 2000);
 
-        const mockRes = new MockedResponse(200);
-        cb(mockRes);
-        mockRes.send('success');
-        done();
+        queueMicrotask(() => {
+          const mockRes = new MockedResponse(200);
+          cb(mockRes);
+          mockRes.send(Buffer.from('success'));
+          done();
+        });
         return fakeRequest as any;
       });
 
@@ -192,10 +212,13 @@ describe('OTLPLogExporter', () => {
 
     it('should successfully send the logs', done => {
       const fakeRequest = new Stream.PassThrough();
-      sinon.stub(http, 'request').returns(fakeRequest as any);
+      Object.defineProperty(fakeRequest, 'setTimeout', {
+        value: function (_timeout: number) {},
+      });
 
+      sinon.stub(http, 'request').returns(fakeRequest as any);
       let buff = Buffer.from('');
-      fakeRequest.on('end', () => {
+      fakeRequest.on('finish', () => {
         const responseBody = buff.toString();
         const json = JSON.parse(responseBody) as IExportLogsServiceRequest;
         const log1 = json.resourceLogs?.[0].scopeLogs?.[0].logRecords?.[0];
@@ -222,9 +245,11 @@ describe('OTLPLogExporter', () => {
       const spyLoggerError = sinon.stub(diag, 'error');
 
       sinon.stub(http, 'request').callsFake((options: any, cb: any) => {
-        const mockRes = new MockedResponse(200);
-        cb(mockRes);
-        mockRes.send('success');
+        queueMicrotask(() => {
+          const mockRes = new MockedResponse(200);
+          cb(mockRes);
+          mockRes.send(Buffer.from('success'));
+        });
         return fakeRequest as any;
       });
 
@@ -237,9 +262,11 @@ describe('OTLPLogExporter', () => {
 
     it('should log the error message', done => {
       sinon.stub(http, 'request').callsFake((options: any, cb: any) => {
-        const mockResError = new MockedResponse(400);
-        cb(mockResError);
-        mockResError.send('failed');
+        queueMicrotask(() => {
+          const mockRes = new MockedResponse(400);
+          cb(mockRes);
+          mockRes.send(Buffer.from('failure'));
+        });
 
         return fakeRequest as any;
       });

experimental/packages/exporter-logs-otlp-proto/src/platform/node/OTLPLogExporter.ts

@@ -45,16 +45,14 @@ export class OTLPLogExporter
   implements LogRecordExporter
 {
   constructor(config: OTLPExporterConfigBase = {}) {
-    super(config, ProtobufLogsSerializer, 'application/x-protobuf');
-    const env = getEnv();
-    this.headers = {
-      ...this.headers,
-      ...USER_AGENT,
+    super(config, ProtobufLogsSerializer, {
       ...baggageUtils.parseKeyPairsIntoRecord(
-        env.OTEL_EXPORTER_OTLP_LOGS_HEADERS
+        getEnv().OTEL_EXPORTER_OTLP_LOGS_HEADERS
       ),
       ...parseHeaders(config?.headers),
-    };
+      ...USER_AGENT,
+      'Content-Type': 'application/x-protobuf',
+    });
   }
 
   getDefaultUrl(config: OTLPExporterConfigBase): string {

experimental/packages/exporter-logs-otlp-proto/test/logHelper.ts

@@ -175,7 +175,7 @@ export class MockedResponse extends Stream {
     super();
   }
 
-  send(data: string) {
+  send(data: Uint8Array) {
     this.emit('data', data);
     this.emit('end');
   }