feat(otlp-grpc-exporter): add compression configuration to exporter (#…

…2813)


Signed-off-by: Svetlana Brennan <svetlana.svn@gmail.com>

Co-authored-by: legendecas <legendecas@gmail.com>
Co-authored-by: Valentin Marchaud <contact@vmarchaud.fr>

packages/exporter-trace-otlp-grpc/README.md

@@ -107,6 +107,32 @@ provider.register();
 
 Note, that this will only work if TLS is also configured on the server.
 
+By default no compression will be used. To use compression, set it programmatically in `collectorOptions` or with environment variables. Supported compression options: `gzip` and `none`.
+
+```js
+const { CompressionAlgorithm } = require('@opentelemetry/exporter-trace-otlp-grpc');
+
+const collectorOptions = {
+  // url is optional and can be omitted - default is localhost:4317
+  url: '<collector-hostname>:<port>',
+  metadata, // // an optional grpc.Metadata object to be sent with each request
+  compression: CompressionAlgorithm.GZIP,
+};
+const exporter = new OTLPTraceExporter(collectorOptions);
+```
+
+ > Providing `compression` with `collectorOptions` takes precedence and overrides compression set with environment variables.
+
+## Environment Variable Configuration
+
+Set compression with environment variables.
+
+```shell
+OTEL_EXPORTER_OTLP_TRACES_COMPRESSION=gzip
+```
+
+ > Compression set programatically in `collectorOptions` takes precedence over compression set with environment variables. `OTEL_EXPORTER_OTLP_TRACES_COMPRESSION` takes precedence and overrides `OTEL_EXPORTER_OTLP_COMPRESSION`.
+
 ## Running opentelemetry-collector locally to see the traces
 
 1. Go to examples/otlp-exporter-node

packages/exporter-trace-otlp-grpc/src/OTLPExporterNodeBase.ts

@@ -25,8 +25,9 @@ import {
   GRPCQueueItem,
   ServiceClientType,
 } from './types';
-import { ServiceClient } from './types';
+import { ServiceClient, CompressionAlgorithm } from './types';
 import { getEnv, baggageUtils } from '@opentelemetry/core';
+import { configureCompression } from './util';
 
 /**
  * OTLP Metric Exporter abstract base class
@@ -43,6 +44,7 @@ export abstract class OTLPExporterNodeBase<
   metadata?: Metadata;
   serviceClient?: ServiceClient = undefined;
   private _send!: Function;
+  compression: CompressionAlgorithm;
 
   constructor(config: OTLPExporterConfigNode = {}) {
     super(config);
@@ -54,6 +56,7 @@ export abstract class OTLPExporterNodeBase<
     for (const [k, v] of Object.entries(headers)) {
       this.metadata.set(k, v);
     }
+    this.compression = configureCompression(config.compression);
   }
 
   private _sendPromise(

packages/exporter-trace-otlp-grpc/src/types.ts

@@ -45,9 +45,18 @@ export interface OTLPExporterConfigNode
   extends otlpTypes.OTLPExporterConfigBase {
   credentials?: grpc.ChannelCredentials;
   metadata?: grpc.Metadata;
+  compression?: CompressionAlgorithm;
 }
 
 export enum ServiceClientType {
   SPANS,
   METRICS,
 }
+
+/**
+ * These values are defined by grpc client
+ */
+export enum CompressionAlgorithm {
+  NONE = 0,
+  GZIP = 2
+}

packages/exporter-trace-otlp-grpc/src/util.ts

@@ -17,7 +17,7 @@
 import * as grpc from '@grpc/grpc-js';
 import * as protoLoader from '@grpc/proto-loader';
 import { diag } from '@opentelemetry/api';
-import { globalErrorHandler } from '@opentelemetry/core';
+import { globalErrorHandler, getEnv } from '@opentelemetry/core';
 import { otlpTypes } from '@opentelemetry/exporter-trace-otlp-http';
 import * as path from 'path';
 import { OTLPExporterNodeBase } from './OTLPExporterNodeBase';
@@ -26,6 +26,7 @@ import {
   OTLPExporterConfigNode,
   GRPCQueueItem,
   ServiceClientType,
+  CompressionAlgorithm
 } from './types';
 
 export function onInit<ExportItem, ServiceRequest>(
@@ -50,17 +51,21 @@ export function onInit<ExportItem, ServiceRequest>(
     .then(packageDefinition => {
       const packageObject: any = grpc.loadPackageDefinition(packageDefinition);
 
+      const options = { 'grpc.default_compression_algorithm': collector.compression };
+
       if (collector.getServiceClientType() === ServiceClientType.SPANS) {
         collector.serviceClient =
           new packageObject.opentelemetry.proto.collector.trace.v1.TraceService(
             collector.url,
             credentials,
+            options,
           );
       } else {
         collector.serviceClient =
           new packageObject.opentelemetry.proto.collector.metrics.v1.MetricsService(
             collector.url,
             credentials,
+            options,
           );
       }
 
@@ -125,3 +130,13 @@ export function validateAndNormalizeUrl(url: string): string {
   }
   return target.host;
 }
+
+export function configureCompression(compression: CompressionAlgorithm | undefined): CompressionAlgorithm {
+  if (compression) {
+    return compression;
+  } else {
+    const definedCompression = getEnv().OTEL_EXPORTER_OTLP_TRACES_COMPRESSION || getEnv().OTEL_EXPORTER_OTLP_COMPRESSION;
+
+    return definedCompression === 'gzip' ? CompressionAlgorithm.GZIP: CompressionAlgorithm.NONE;
+  }
+}

packages/exporter-trace-otlp-grpc/test/OTLPTraceExporter.test.ts

@@ -36,6 +36,8 @@ import {
   mockedReadableSpan,
 } from './traceHelper';
 
+import { CompressionAlgorithm } from '../src/types';
+
 const traceServiceProtoPath =
   'opentelemetry/proto/collector/trace/v1/trace_service.proto';
 const includeDirs = [path.resolve(__dirname, '../protos')];
@@ -198,6 +200,77 @@ const testCollectorExporter = (params: TestParams) =>
         }, 200);
       });
     });
+    describe('export - with gzip compression', () => {
+      beforeEach(() => {
+        const credentials = params.useTLS
+        ? grpc.credentials.createSsl(
+            fs.readFileSync('./test/certs/ca.crt'),
+            fs.readFileSync('./test/certs/client.key'),
+            fs.readFileSync('./test/certs/client.crt')
+          )
+        : undefined;
+        collectorExporter = new OTLPTraceExporter({
+          url: 'grpcs://' + address,
+          credentials,
+          metadata: params.metadata,
+          compression: CompressionAlgorithm.GZIP,
+        });
+
+        const provider = new BasicTracerProvider();
+        provider.addSpanProcessor(new SimpleSpanProcessor(collectorExporter));
+      });
+      it('should successfully send the spans', done => {
+        const responseSpy = sinon.spy();
+        const spans = [Object.assign({}, mockedReadableSpan)];
+        collectorExporter.export(spans, responseSpy);
+        setTimeout(() => {
+          assert.ok(
+            typeof exportedData !== 'undefined',
+            'resource' + " doesn't exist"
+          );
+          let spans;
+          let resource;
+          if (exportedData) {
+            spans = exportedData.instrumentationLibrarySpans[0].spans;
+            resource = exportedData.resource;
+            ensureExportedSpanIsCorrect(spans[0]);
+
+            assert.ok(
+              typeof resource !== 'undefined',
+              "resource doesn't exist"
+            );
+            if (resource) {
+              ensureResourceIsCorrect(resource);
+            }
+          }
+          if (params.metadata && reqMetadata) {
+            ensureMetadataIsCorrect(reqMetadata, params.metadata);
+          }
+          done();
+        }, 500);
+      });
+    });
+    describe('Trace Exporter with compression', () => {
+      const envSource = process.env;
+      it('should return gzip compression algorithm on exporter', () => {
+        const credentials = params.useTLS
+        ? grpc.credentials.createSsl(
+            fs.readFileSync('./test/certs/ca.crt'),
+            fs.readFileSync('./test/certs/client.key'),
+            fs.readFileSync('./test/certs/client.crt')
+          )
+        : undefined;
+
+        envSource.OTEL_EXPORTER_OTLP_COMPRESSION='gzip';
+        collectorExporter = new OTLPTraceExporter({
+          url: 'grpcs://' + address,
+          credentials,
+          metadata: params.metadata,
+        });
+        assert.strictEqual(collectorExporter.compression, CompressionAlgorithm.GZIP);
+        delete envSource.OTEL_EXPORTER_OTLP_COMPRESSION;
+      });
+    });
   });
 
 describe('OTLPTraceExporter - node (getDefaultUrl)', () => {

packages/exporter-trace-otlp-grpc/test/util.test.ts

@@ -18,7 +18,8 @@ import * as sinon from 'sinon';
 import * as assert from 'assert';
 
 import { diag } from '@opentelemetry/api';
-import { validateAndNormalizeUrl } from '../src/util';
+import { validateAndNormalizeUrl, configureCompression } from '../src/util';
+import { CompressionAlgorithm} from '../src/types';
 
 // Tests added to detect breakage released in #2130
 describe('validateAndNormalizeUrl()', () => {
@@ -79,3 +80,24 @@ describe('validateAndNormalizeUrl()', () => {
     });
   });
 });
+
+describe('configureCompression', () => {
+  const envSource = process.env;
+  it('should return none for compression', () => {
+    const compression = CompressionAlgorithm.NONE;
+    assert.strictEqual(configureCompression(compression), CompressionAlgorithm.NONE);
+  });
+  it('should return gzip compression defined via env', () => {
+    envSource.OTEL_EXPORTER_OTLP_TRACES_COMPRESSION = 'gzip';
+    assert.strictEqual(configureCompression(undefined),CompressionAlgorithm.GZIP);
+    delete envSource.OTEL_EXPORTER_OTLP_TRACES_COMPRESSION;
+  });
+  it('should return none for compression defined via env', () => {
+    envSource.OTEL_EXPORTER_OTLP_TRACES_COMPRESSION = 'none';
+    assert.strictEqual(configureCompression(undefined),CompressionAlgorithm.NONE);
+    delete envSource.OTEL_EXPORTER_OTLP_TRACES_COMPRESSION;
+  });
+  it('should return none for compression when no compression is set', () => {
+    assert.strictEqual(configureCompression(undefined),CompressionAlgorithm.NONE);
+  });
+});