feat(node): Compression support for http transport (#5209)

Note: This is an exact duplicate of #5139 and #5203, both originally by @timfish, which accidentally got merged into the 7.x branch rather than master.

#5139:

This is mostly stolen straight from the [Electron transport](https://github.com/getsentry/sentry-electron/blob/master/src/main/transports/electron-net.ts).

This will help with attachments!

#5203:

#5139 introduced a subtle bug where `options.headers` was modified which causes headers to leak between requests. This means requests after a compressed request will be incorrectly marked with `content-encoding: gzip`.

Author: lobsterkatie

packages/node/src/transports/http-module.ts

@@ -1,5 +1,6 @@
 import { IncomingHttpHeaders, RequestOptions as HTTPRequestOptions } from 'http';
 import { RequestOptions as HTTPSRequestOptions } from 'https';
+import { Writable } from 'stream';
 import { URL } from 'url';
 
 export type HTTPModuleRequestOptions = HTTPRequestOptions | HTTPSRequestOptions | string | URL;
@@ -15,15 +16,6 @@ export interface HTTPModuleRequestIncomingMessage {
   setEncoding(encoding: string): void;
 }
 
-/**
- * Cut version of http.ClientRequest.
- * Some transports work in a special Javascript environment where http.IncomingMessage is not available.
- */
-export interface HTTPModuleClientRequest {
-  end(chunk: string | Uint8Array): void;
-  on(event: 'error', listener: () => void): void;
-}
-
 /**
  * Internal used interface for typescript.
  * @hidden
@@ -34,10 +26,7 @@ export interface HTTPModule {
    * @param options These are {@see TransportOptions}
    * @param callback Callback when request is finished
    */
-  request(
-    options: HTTPModuleRequestOptions,
-    callback?: (res: HTTPModuleRequestIncomingMessage) => void,
-  ): HTTPModuleClientRequest;
+  request(options: HTTPModuleRequestOptions, callback?: (res: HTTPModuleRequestIncomingMessage) => void): Writable;
 
   // This is the type for nodejs versions that handle the URL argument
   // (v10.9.0+), but we do not use it just yet because we support older node

packages/node/src/transports/http.ts

@@ -8,7 +8,9 @@ import {
 } from '@sentry/types';
 import * as http from 'http';
 import * as https from 'https';
+import { Readable } from 'stream';
 import { URL } from 'url';
+import { createGzip } from 'zlib';
 
 import { HTTPModule } from './http-module';
 
@@ -23,6 +25,22 @@ export interface NodeTransportOptions extends BaseTransportOptions {
   httpModule?: HTTPModule;
 }
 
+// Estimated maximum size for reasonable standalone event
+const GZIP_THRESHOLD = 1024 * 32;
+
+/**
+ * Gets a stream from a Uint8Array or string
+ * Readable.from is ideal but was added in node.js v12.3.0 and v10.17.0
+ */
+function streamFromBody(body: Uint8Array | string): Readable {
+  return new Readable({
+    read() {
+      this.push(body);
+      this.push(null);
+    },
+  });
+}
+
 /**
  * Creates a Transport that uses native the native 'http' and 'https' modules to send events to Sentry.
  */
@@ -85,11 +103,20 @@ function createRequestExecutor(
   const { hostname, pathname, port, protocol, search } = new URL(options.url);
   return function makeRequest(request: TransportRequest): Promise<TransportMakeRequestResponse> {
     return new Promise((resolve, reject) => {
+      let body = streamFromBody(request.body);
+
+      const headers: Record<string, string> = { ...options.headers };
+
+      if (request.body.length > GZIP_THRESHOLD) {
+        headers['content-encoding'] = 'gzip';
+        body = body.pipe(createGzip());
+      }
+
       const req = httpModule.request(
         {
           method: 'POST',
           agent,
-          headers: options.headers,
+          headers,
           hostname,
           path: `${pathname}${search}`,
           port,
@@ -123,7 +150,7 @@ function createRequestExecutor(
       );
 
       req.on('error', reject);
-      req.end(request.body);
+      body.pipe(req);
     });
   };
 }

packages/node/test/transports/http.test.ts

@@ -1,8 +1,9 @@
 import { createTransport } from '@sentry/core';
 import { EventEnvelope, EventItem } from '@sentry/types';
-import { createEnvelope, serializeEnvelope } from '@sentry/utils';
+import { addItemToEnvelope, createAttachmentEnvelopeItem, createEnvelope, serializeEnvelope } from '@sentry/utils';
 import * as http from 'http';
 import { TextEncoder } from 'util';
+import { createGunzip } from 'zlib';
 
 import { makeNodeTransport } from '../../src/transports';
 
@@ -34,17 +35,19 @@ let testServer: http.Server | undefined;
 
 function setupTestServer(
   options: TestServerOptions,
-  requestInspector?: (req: http.IncomingMessage, body: string) => void,
+  requestInspector?: (req: http.IncomingMessage, body: string, raw: Uint8Array) => void,
 ) {
   testServer = http.createServer((req, res) => {
-    let body = '';
+    const chunks: Buffer[] = [];
 
-    req.on('data', data => {
-      body += data;
+    const stream = req.headers['content-encoding'] === 'gzip' ? req.pipe(createGunzip({})) : req;
+
+    stream.on('data', data => {
+      chunks.push(data);
     });
 
-    req.on('end', () => {
-      requestInspector?.(req, body);
+    stream.on('end', () => {
+      requestInspector?.(req, chunks.join(), Buffer.concat(chunks));
     });
 
     res.writeHead(options.statusCode, options.responseHeaders);
@@ -69,6 +72,16 @@ const EVENT_ENVELOPE = createEnvelope<EventEnvelope>({ event_id: 'aa3ff046696b4b
 
 const SERIALIZED_EVENT_ENVELOPE = serializeEnvelope(EVENT_ENVELOPE, new TextEncoder());
 
+const ATTACHMENT_ITEM = createAttachmentEnvelopeItem(
+  { filename: 'empty-file.bin', data: new Uint8Array(50_000) },
+  new TextEncoder(),
+);
+const EVENT_ATTACHMENT_ENVELOPE = addItemToEnvelope(EVENT_ENVELOPE, ATTACHMENT_ITEM);
+const SERIALIZED_EVENT_ATTACHMENT_ENVELOPE = serializeEnvelope(
+  EVENT_ATTACHMENT_ENVELOPE,
+  new TextEncoder(),
+) as Uint8Array;
+
 const defaultOptions = {
   url: TEST_SERVER_URL,
   recordDroppedEvent: () => undefined,
@@ -155,6 +168,40 @@ describe('makeNewHttpTransport()', () => {
     });
   });
 
+  describe('compression', () => {
+    it('small envelopes should not be compressed', async () => {
+      await setupTestServer(
+        {
+          statusCode: SUCCESS,
+          responseHeaders: {},
+        },
+        (req, body) => {
+          expect(req.headers['content-encoding']).toBeUndefined();
+          expect(body).toBe(SERIALIZED_EVENT_ENVELOPE);
+        },
+      );
+
+      const transport = makeNodeTransport(defaultOptions);
+      await transport.send(EVENT_ENVELOPE);
+    });
+
+    it('large envelopes should be compressed', async () => {
+      await setupTestServer(
+        {
+          statusCode: SUCCESS,
+          responseHeaders: {},
+        },
+        (req, _, raw) => {
+          expect(req.headers['content-encoding']).toEqual('gzip');
+          expect(raw.buffer).toStrictEqual(SERIALIZED_EVENT_ATTACHMENT_ENVELOPE.buffer);
+        },
+      );
+
+      const transport = makeNodeTransport(defaultOptions);
+      await transport.send(EVENT_ATTACHMENT_ENVELOPE);
+    });
+  });
+
   describe('proxy', () => {
     it('can be configured through option', () => {
       makeNodeTransport({
@@ -236,104 +283,106 @@ describe('makeNewHttpTransport()', () => {
     });
   });
 
-  it('should register TransportRequestExecutor that returns the correct object from server response (rate limit)', async () => {
-    await setupTestServer({
-      statusCode: RATE_LIMIT,
-      responseHeaders: {},
-    });
-
-    makeNodeTransport(defaultOptions);
-    const registeredRequestExecutor = (createTransport as jest.Mock).mock.calls[0][1];
-
-    const executorResult = registeredRequestExecutor({
-      body: serializeEnvelope(EVENT_ENVELOPE, new TextEncoder()),
-      category: 'error',
-    });
-
-    await expect(executorResult).resolves.toEqual(
-      expect.objectContaining({
+  describe('should register TransportRequestExecutor that returns the correct object from server response', () => {
+    it('rate limit', async () => {
+      await setupTestServer({
         statusCode: RATE_LIMIT,
-      }),
-    );
-  });
+        responseHeaders: {},
+      });
 
-  it('should register TransportRequestExecutor that returns the correct object from server response (OK)', async () => {
-    await setupTestServer({
-      statusCode: SUCCESS,
-    });
+      makeNodeTransport(defaultOptions);
+      const registeredRequestExecutor = (createTransport as jest.Mock).mock.calls[0][1];
 
-    makeNodeTransport(defaultOptions);
-    const registeredRequestExecutor = (createTransport as jest.Mock).mock.calls[0][1];
+      const executorResult = registeredRequestExecutor({
+        body: serializeEnvelope(EVENT_ENVELOPE, new TextEncoder()),
+        category: 'error',
+      });
 
-    const executorResult = registeredRequestExecutor({
-      body: serializeEnvelope(EVENT_ENVELOPE, new TextEncoder()),
-      category: 'error',
+      await expect(executorResult).resolves.toEqual(
+        expect.objectContaining({
+          statusCode: RATE_LIMIT,
+        }),
+      );
     });
 
-    await expect(executorResult).resolves.toEqual(
-      expect.objectContaining({
+    it('OK', async () => {
+      await setupTestServer({
         statusCode: SUCCESS,
-        headers: {
-          'retry-after': null,
-          'x-sentry-rate-limits': null,
-        },
-      }),
-    );
-  });
+      });
 
-  it('should register TransportRequestExecutor that returns the correct object from server response (OK with rate-limit headers)', async () => {
-    await setupTestServer({
-      statusCode: SUCCESS,
-      responseHeaders: {
-        'Retry-After': '2700',
-        'X-Sentry-Rate-Limits': '60::organization, 2700::organization',
-      },
-    });
+      makeNodeTransport(defaultOptions);
+      const registeredRequestExecutor = (createTransport as jest.Mock).mock.calls[0][1];
 
-    makeNodeTransport(defaultOptions);
-    const registeredRequestExecutor = (createTransport as jest.Mock).mock.calls[0][1];
+      const executorResult = registeredRequestExecutor({
+        body: serializeEnvelope(EVENT_ENVELOPE, new TextEncoder()),
+        category: 'error',
+      });
 
-    const executorResult = registeredRequestExecutor({
-      body: serializeEnvelope(EVENT_ENVELOPE, new TextEncoder()),
-      category: 'error',
+      await expect(executorResult).resolves.toEqual(
+        expect.objectContaining({
+          statusCode: SUCCESS,
+          headers: {
+            'retry-after': null,
+            'x-sentry-rate-limits': null,
+          },
+        }),
+      );
     });
 
-    await expect(executorResult).resolves.toEqual(
-      expect.objectContaining({
+    it('OK with rate-limit headers', async () => {
+      await setupTestServer({
         statusCode: SUCCESS,
-        headers: {
-          'retry-after': '2700',
-          'x-sentry-rate-limits': '60::organization, 2700::organization',
+        responseHeaders: {
+          'Retry-After': '2700',
+          'X-Sentry-Rate-Limits': '60::organization, 2700::organization',
         },
-      }),
-    );
-  });
+      });
 
-  it('should register TransportRequestExecutor that returns the correct object from server response (NOK with rate-limit headers)', async () => {
-    await setupTestServer({
-      statusCode: RATE_LIMIT,
-      responseHeaders: {
-        'Retry-After': '2700',
-        'X-Sentry-Rate-Limits': '60::organization, 2700::organization',
-      },
-    });
+      makeNodeTransport(defaultOptions);
+      const registeredRequestExecutor = (createTransport as jest.Mock).mock.calls[0][1];
 
-    makeNodeTransport(defaultOptions);
-    const registeredRequestExecutor = (createTransport as jest.Mock).mock.calls[0][1];
+      const executorResult = registeredRequestExecutor({
+        body: serializeEnvelope(EVENT_ENVELOPE, new TextEncoder()),
+        category: 'error',
+      });
 
-    const executorResult = registeredRequestExecutor({
-      body: serializeEnvelope(EVENT_ENVELOPE, new TextEncoder()),
-      category: 'error',
+      await expect(executorResult).resolves.toEqual(
+        expect.objectContaining({
+          statusCode: SUCCESS,
+          headers: {
+            'retry-after': '2700',
+            'x-sentry-rate-limits': '60::organization, 2700::organization',
+          },
+        }),
+      );
     });
 
-    await expect(executorResult).resolves.toEqual(
-      expect.objectContaining({
+    it('NOK with rate-limit headers', async () => {
+      await setupTestServer({
         statusCode: RATE_LIMIT,
-        headers: {
-          'retry-after': '2700',
-          'x-sentry-rate-limits': '60::organization, 2700::organization',
+        responseHeaders: {
+          'Retry-After': '2700',
+          'X-Sentry-Rate-Limits': '60::organization, 2700::organization',
         },
-      }),
-    );
+      });
+
+      makeNodeTransport(defaultOptions);
+      const registeredRequestExecutor = (createTransport as jest.Mock).mock.calls[0][1];
+
+      const executorResult = registeredRequestExecutor({
+        body: serializeEnvelope(EVENT_ENVELOPE, new TextEncoder()),
+        category: 'error',
+      });
+
+      await expect(executorResult).resolves.toEqual(
+        expect.objectContaining({
+          statusCode: RATE_LIMIT,
+          headers: {
+            'retry-after': '2700',
+            'x-sentry-rate-limits': '60::organization, 2700::organization',
+          },
+        }),
+      );
+    });
   });
 });