Fix gzip compression for OTLP HTTP exporter (#3046)

Co-authored-by: Daniel Dyla <[email protected]>

Author: mattolson

experimental/CHANGELOG.md

@@ -24,6 +24,7 @@ All notable changes to experimental packages in this project will be documented
 
 * fix(otlp-transformer): remove type dependency on Long #3022 @legendecas
 * fix(grpc-exporter): use non-normalized URL to determine channel security #3019 @pichlermarc
+* fix(otlp-exporter-base): fix gzip output stream in http otlp export #3046 @mattolson
 
 ### :books: (Refine Doc)
 

experimental/packages/otlp-exporter-base/src/platform/node/OTLPExporterNodeBase.ts

@@ -88,7 +88,5 @@ export abstract class OTLPExporterNodeBase<
     promise.then(popPromise, popPromise);
   }
 
-  // TODO: end gzip stream from util.ts if not undefined
-  // It should perhaps be a class member here instead of a variable in util.ts
   onShutdown(): void {}
 }

experimental/packages/otlp-exporter-base/src/platform/node/util.ts

@@ -25,8 +25,6 @@ import { CompressionAlgorithm } from './types';
 import { getEnv } from '@opentelemetry/core';
 import { OTLPExporterError } from '../../types';
 
-let gzip: zlib.Gzip | undefined;
-
 /**
  * Sends data using http
  * @param collector
@@ -116,13 +114,10 @@ export function sendWithHttp<ExportItem, ServiceRequest>(
 
   switch (collector.compression) {
     case CompressionAlgorithm.GZIP: {
-      if (!gzip) {
-        gzip = zlib.createGzip();
-      }
       req.setHeader('Content-Encoding', 'gzip');
       const dataStream = readableFromBuffer(data);
       dataStream.on('error', onError)
-        .pipe(gzip).on('error', onError)
+        .pipe(zlib.createGzip()).on('error', onError)
         .pipe(req);
 
       break;

experimental/packages/otlp-exporter-base/test/node/util.test.ts

@@ -16,11 +16,50 @@
 
 import * as assert from 'assert';
 import { configureExporterTimeout, invalidTimeout } from '../../src/util';
+import { sendWithHttp } from '../../src/platform/node/util';
 import { CompressionAlgorithm} from '../../src/platform/node/types';
 import { configureCompression} from '../../src/platform/node/util';
 import { diag } from '@opentelemetry/api';
 import * as sinon from 'sinon';
 
+import { OTLPExporterNodeBase } from '../../src/platform/node/OTLPExporterNodeBase';
+import { OTLPExporterNodeConfigBase } from '../../src/platform/node/types';
+import { OTLPExporterError } from '../../src/types';
+import { PassThrough } from 'stream';
+import * as http from 'http';
+import * as zlib from 'zlib';
+
+// Meant to simulate http.IncomingMessage, at least the parts that sendWithHttp cares about
+// but make it a PassThrough so we can inspect it for the test
+class HttpResponse extends PassThrough {
+  statusCode: number;
+  statusMessage: string;
+
+  constructor(statusCode = 200, statusMessage = 'OK') {
+    super();
+    this.statusCode = statusCode;
+    this.statusMessage = statusMessage;
+  }
+}
+
+// Meant to simulate http.ClientRequest, at least the parts that sendWithHttp cares about
+// but make it a PassThrough so we can inspect it for the test
+class HttpRequest extends PassThrough {
+  setHeader(name: string, value: string) {}
+}
+
+// Barebones exporter for use by sendWithHttp
+type ExporterConfig = OTLPExporterNodeConfigBase;
+class Exporter extends OTLPExporterNodeBase<object,object> {
+  getDefaultUrl(config: ExporterConfig): string {
+    return config.url || '';
+  }
+
+  convert(spans: object[]): object {
+    return {};
+  }
+}
+
 describe('configureExporterTimeout', () => {
   const envSource = process.env;
   it('should use timeoutMillis parameter as export timeout value', () => {
@@ -124,3 +163,123 @@ describe('configureCompression', () => {
     assert.strictEqual(configureCompression(undefined),CompressionAlgorithm.NONE);
   });
 });
+
+describe('sendWithHttp', () => {
+  let exporter: Exporter;
+  let httpRequestStub: sinon.SinonStub;
+  let mockRequest: HttpRequest;
+  let setHeaderSpy: sinon.SinonSpy;
+
+  const spanData: object = {
+    'foo': 'bar',
+    'bar': 'baz',
+  };
+
+  beforeEach(() => {
+
+    // Create stub of http.request (used by sendWithHttp)
+    httpRequestStub = sinon.stub(http, 'request');
+
+    // Mock out a request
+    mockRequest = new HttpRequest();
+    setHeaderSpy = sinon.spy(mockRequest, 'setHeader');
+
+    // Mock out response
+    const response = new HttpResponse();
+    response.end('OK');
+
+    // Stub out http.request so it calls our callback with the mocked response
+    // and also so it returns our mocked request stream so we can observe. We don't
+    // really care about the response for the purpose of this test, but we do want
+    // to observe the request compression behavior.
+    httpRequestStub.returns(mockRequest).callsArgWith(1, response);
+  });
+
+  afterEach(function() {
+    httpRequestStub.restore();
+    setHeaderSpy.restore();
+  });
+
+  it('should send with no compression if configured to do so', () => {
+    exporter = new Exporter({
+      url: 'http://foobar.com',
+      compression: CompressionAlgorithm.NONE,
+    });
+    const data = JSON.stringify(spanData);
+
+    // Show that data is written to the request stream
+    let requestData = '';
+    mockRequest.on('data', chunk => requestData += chunk);
+    mockRequest.on('end', () => {
+      assert.strictEqual(requestData, data);
+    });
+
+    sendWithHttp(exporter, data, 'application/json', () => {
+      // Show that we aren't setting the gzip encoding header
+      assert(setHeaderSpy.withArgs('Content-Encoding', 'gzip').notCalled);
+    }, (err: OTLPExporterError) => {
+      assert.fail(err);
+    });
+  });
+
+  it('should send with gzip compression if configured to do so', () => {
+    exporter = new Exporter({
+      url: 'http://foobar.com',
+      compression: CompressionAlgorithm.GZIP,
+    });
+
+    const data = JSON.stringify(spanData);
+    const compressedData = zlib.gzipSync(Buffer.from(data));
+
+    // Show that compressed data is written to the request stream
+    const buffers: Buffer[] = [];
+    mockRequest.on('data', chunk => buffers.push(Buffer.from(chunk)));
+    mockRequest.on('end', () => {
+      assert(Buffer.concat(buffers).equals(compressedData));
+    });
+
+    sendWithHttp(exporter, data, 'application/json', () => {
+      // Show that we are setting the gzip encoding header
+      assert(setHeaderSpy.withArgs('Content-Encoding', 'gzip').calledOnce);
+    }, (err: OTLPExporterError) => {
+      assert.fail(err);
+    });
+  });
+
+  it('should work with gzip compression enabled even after multiple requests', () => {
+    exporter = new Exporter({
+      url: 'http://foobar.com',
+      compression: CompressionAlgorithm.GZIP,
+    });
+
+    const data = JSON.stringify(spanData);
+    const compressedData = zlib.gzipSync(Buffer.from(data));
+
+    for (let i = 0; i < 5; i++) {
+      mockRequest = new HttpRequest();
+      setHeaderSpy.restore();
+      setHeaderSpy = sinon.spy(mockRequest, 'setHeader');
+
+      const response = new HttpResponse();
+      response.end('OK');
+
+      httpRequestStub.restore();
+      httpRequestStub = sinon.stub(http, 'request');
+      httpRequestStub.returns(mockRequest).callsArgWith(1, response);
+
+      // Show that compressed data is written to the request stream
+      const buffers: Buffer[] = [];
+      mockRequest.on('data', chunk => buffers.push(Buffer.from(chunk)));
+      mockRequest.on('end', () => {
+        assert(Buffer.concat(buffers).equals(compressedData));
+      });
+
+      sendWithHttp(exporter, data, 'application/json', () => {
+        // Show that we are setting the gzip encoding header
+        assert(setHeaderSpy.withArgs('Content-Encoding', 'gzip').calledOnce);
+      }, (err: OTLPExporterError) => {
+        assert.fail(err);
+      });
+    }
+  });
+});