add preprocessing for compression

Author: liustve

aws-distro-opentelemetry-node-autoinstrumentation/src/aws-opentelemetry-configurator.ts

@@ -7,6 +7,7 @@ import { getPropagator } from '@opentelemetry/auto-configuration-propagators';
 import { getResourceDetectors as getResourceDetectorsFromEnv } from '@opentelemetry/auto-instrumentations-node';
 import { ENVIRONMENT, TracesSamplerValues, getEnv, getEnvWithoutDefaults } from '@opentelemetry/core';
 import { OTLPMetricExporter as OTLPGrpcOTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-grpc';
+import { CompressionAlgorithm } from '@opentelemetry/otlp-exporter-base';
 import {
   AggregationTemporalityPreference,
   OTLPMetricExporter as OTLPHttpOTLPMetricExporter,
@@ -474,7 +475,9 @@ export class AwsLoggerProcessorProvider {
               validateLogsHeaders()
             ) {
               diag.debug('Detected CloudWatch Logs OTLP endpoint. Switching exporter to OTLPAwsLogExporter');
-              exporters.push(new OTLPAwsLogExporter(otlpExporterLogsEndpoint));
+              exporters.push(
+                new OTLPAwsLogExporter(otlpExporterLogsEndpoint, { compression: CompressionAlgorithm.GZIP })
+              );
             } else {
               exporters.push(new OTLPProtoLogExporter());
             }
@@ -491,7 +494,9 @@ export class AwsLoggerProcessorProvider {
               validateLogsHeaders()
             ) {
               diag.debug('Detected CloudWatch Logs OTLP endpoint. Switching exporter to OTLPAwsLogExporter');
-              exporters.push(new OTLPAwsLogExporter(otlpExporterLogsEndpoint));
+              exporters.push(
+                new OTLPAwsLogExporter(otlpExporterLogsEndpoint, { compression: CompressionAlgorithm.GZIP })
+              );
             } else {
               exporters.push(new OTLPProtoLogExporter());
             }

aws-distro-opentelemetry-node-autoinstrumentation/src/exporter/otlp/aws/common/passthrough-serializer.ts

@@ -0,0 +1,46 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+import { ISerializer } from '@opentelemetry/otlp-transformer';
+
+/**
+ * A serializer that bypasses request serialization by returning pre-serialized data.
+ * @template Response The type of the deserialized response
+ */
+export class PassthroughSerializer<Response> implements ISerializer<Uint8Array, Response> {
+  private serializedData: Uint8Array = new Uint8Array();
+  private deserializer: (data: Uint8Array) => Response;
+
+  /**
+   * Creates a new PassthroughSerializer instance.
+   * @param deserializer Function to deserialize response data
+   */
+  constructor(deserializer: (data: Uint8Array) => Response) {
+    this.deserializer = deserializer;
+  }
+
+  /**
+   * Sets the pre-serialized data to be returned when serializeRequest is called.
+   * @param data The serialized data to use
+   */
+  setSerializedData(data: Uint8Array): void {
+    this.serializedData = data;
+  }
+
+  /**
+   * Returns the pre-serialized data, ignoring the request parameter.
+   * @param request Ignored parameter.
+   * @returns The pre-serialized data
+   */
+  serializeRequest(request: Uint8Array): Uint8Array {
+    return this.serializedData;
+  }
+
+  /**
+   * Deserializes response data using the provided deserializer function.
+   * @param data The response data to deserialize
+   * @returns The deserialized response
+   */
+  deserializeResponse(data: Uint8Array): Response {
+    return this.deserializer(data);
+  }
+}

aws-distro-opentelemetry-node-autoinstrumentation/src/exporter/otlp/aws/logs/otlp-aws-log-exporter.ts

@@ -1,12 +1,13 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 import { OTLPLogExporter as OTLPProtoLogExporter } from '@opentelemetry/exporter-logs-otlp-proto';
-import { OTLPExporterNodeConfigBase } from '@opentelemetry/otlp-exporter-base';
-import { ProtobufLogsSerializer } from '@opentelemetry/otlp-transformer';
+import { CompressionAlgorithm, OTLPExporterNodeConfigBase } from '@opentelemetry/otlp-exporter-base';
+import { gzipSync } from 'zlib';
+import { IExportLogsServiceResponse, ProtobufLogsSerializer } from '@opentelemetry/otlp-transformer';
 import { ReadableLogRecord } from '@opentelemetry/sdk-logs';
 import { AwsAuthenticator } from '../common/aws-authenticator';
-import { changeUrlConfig } from '../common/utils';
-import { ExportResult } from '@opentelemetry/core';
+import { ExportResult, ExportResultCode } from '@opentelemetry/core';
+import { PassthroughSerializer } from '../common/passthrough-serializer';
 
 /**
  * This exporter extends the functionality of the OTLPProtoLogExporter to allow spans to be exported
@@ -16,34 +17,78 @@ import { ExportResult } from '@opentelemetry/core';
  *
  * This only works with version >=16 Node.js environments.
  */
+
 export class OTLPAwsLogExporter extends OTLPProtoLogExporter {
+  private compression: CompressionAlgorithm | undefined;
   private endpoint: string;
   private region: string;
+  private serializer: PassthroughSerializer<IExportLogsServiceResponse>;
   private authenticator: AwsAuthenticator;
 
   constructor(endpoint: string, config?: OTLPExporterNodeConfigBase) {
-    super(changeUrlConfig(endpoint, config));
+    const modifiedConfig: OTLPExporterNodeConfigBase = {
+      ...config,
+      url: endpoint,
+      compression: CompressionAlgorithm.NONE,
+    };
+
+    super(modifiedConfig);
+    this.compression = config?.compression;
     this.region = endpoint.split('.')[1];
     this.endpoint = endpoint;
     this.authenticator = new AwsAuthenticator(this.region, 'logs');
+
+    // This is used in order to prevent serializing and compressing the data twice. Once for signing Sigv4 and
+    // once when we pass the data to super.export() which will serialize and compress the data again.
+    this.serializer = new PassthroughSerializer(ProtobufLogsSerializer.deserializeResponse);
+    this['_delegate']._serializer = this.serializer;
   }
 
   /**
-   * Overrides the upstream implementation of export. All behaviors are the same except if the
-   * endpoint is the CloudWatch Logs OTLP endpoint, we will sign the request with SigV4 in headers before
-   * sending it to the endpoint. Otherwise, we will skip signing.
+   * Overrides the upstream implementation of export. If the
+   * endpoint is the CloudWatch Logs OTLP endpoint, we sign the request with SigV4 in headers.
+   * To prevent performance degradation from serializing and compressing data twice, we handle serialization and compression
+   * locally in this exporter and pass the pre-processed data to the upstream export functionality.
    */
+
+  // Upstream already implements a retry mechanism:
+  // https://github.com/open-telemetry/opentelemetry-js/blob/main/experimental/packages/otlp-exporter-base/src/retrying-transport.ts
+
   public override async export(
     items: ReadableLogRecord[],
     resultCallback: (result: ExportResult) => void
   ): Promise<void> {
-    const serializedLogs: Uint8Array | undefined = ProtobufLogsSerializer.serializeRequest(items);
+    let serializedLogs: Uint8Array | undefined = ProtobufLogsSerializer.serializeRequest(items);
+
+    if (serializedLogs === undefined) {
+      resultCallback({
+        code: ExportResultCode.FAILED,
+        error: new Error('Nothing to send'),
+      });
+      return;
+    }
+
+    const shouldCompress = this.compression && this.compression !== CompressionAlgorithm.NONE;
+    if (shouldCompress) {
+      serializedLogs = gzipSync(serializedLogs);
+    }
+
+    // Pass pre-processed data to passthrough serializer. When super.export() is called, the Passthrough Serializer will
+    // use the pre-processed data instead of serializing and compressing the data again.
+    this.serializer.setSerializedData(serializedLogs);
+
+    // See type: https://github.com/open-telemetry/opentelemetry-js/blob/experimental/v0.57.1/experimental/packages/otlp-exporter-base/src/transport/http-transport-types.ts#L31
     const headers = this['_delegate']._transport?._transport?._parameters?.headers();
 
     if (headers) {
+      if (shouldCompress) {
+        headers['Content-Encoding'] = 'gzip';
+      } else {
+        delete headers['Content-Encoding'];
+      }
+
       const signedRequest = await this.authenticator.authenticate(this.endpoint, headers, serializedLogs);
 
-      // See type: https://github.com/open-telemetry/opentelemetry-js/blob/experimental/v0.57.1/experimental/packages/otlp-exporter-base/src/transport/http-transport-types.ts#L31
       const newHeaders: () => Record<string, string> = () => signedRequest;
       this['_delegate']._transport._transport._parameters.headers = newHeaders;
     }

aws-distro-opentelemetry-node-autoinstrumentation/src/exporter/otlp/aws/traces/otlp-aws-span-exporter.ts

@@ -1,12 +1,13 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 import { OTLPTraceExporter as OTLPProtoTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto';
-import { OTLPExporterNodeConfigBase } from '@opentelemetry/otlp-exporter-base';
-import { ProtobufTraceSerializer } from '@opentelemetry/otlp-transformer';
+import { CompressionAlgorithm, OTLPExporterNodeConfigBase } from '@opentelemetry/otlp-exporter-base';
+import { IExportTraceServiceResponse, ProtobufTraceSerializer } from '@opentelemetry/otlp-transformer';
 import { ReadableSpan } from '@opentelemetry/sdk-trace-base';
-import { ExportResult } from '@opentelemetry/core';
+import { ExportResult, ExportResultCode } from '@opentelemetry/core';
 import { AwsAuthenticator } from '../common/aws-authenticator';
-import { changeUrlConfig } from '../common/utils';
+import { PassthroughSerializer } from '../common/passthrough-serializer';
+import { gzipSync } from 'zlib';
 
 /**
  * This exporter extends the functionality of the OTLPProtoTraceExporter to allow spans to be exported
@@ -17,27 +18,63 @@ import { changeUrlConfig } from '../common/utils';
  * This only works with version >=16 Node.js environments.
  */
 export class OTLPAwsSpanExporter extends OTLPProtoTraceExporter {
+  private compression: CompressionAlgorithm | undefined;
   private endpoint: string;
   private region: string;
+  private serializer: PassthroughSerializer<IExportTraceServiceResponse>;
   private authenticator: AwsAuthenticator;
 
   constructor(endpoint: string, config?: OTLPExporterNodeConfigBase) {
-    super(changeUrlConfig(endpoint, config));
+    const modifiedConfig: OTLPExporterNodeConfigBase = {
+      ...config,
+      url: endpoint,
+      compression: CompressionAlgorithm.NONE,
+    };
+
+    super(modifiedConfig);
     this.region = endpoint.split('.')[1];
     this.endpoint = endpoint;
     this.authenticator = new AwsAuthenticator(this.region, 'xray');
+    this.serializer = new PassthroughSerializer(ProtobufTraceSerializer.deserializeResponse);
+    this['_delegate']._serializer = this.serializer;
   }
 
   /**
    * Overrides the upstream implementation of export. All behaviors are the same except if the
    * endpoint is an XRay OTLP endpoint, we will sign the request with SigV4 in headers before
    * sending it to the endpoint. Otherwise, we will skip signing.
+   * To prevent performance degradation from serializing and compressing data twice, we handle serialization and compression
+   * locally in this exporter and pass the pre-processed data to the upstream export functionality.
    */
   public override async export(items: ReadableSpan[], resultCallback: (result: ExportResult) => void): Promise<void> {
-    const serializedSpans: Uint8Array | undefined = ProtobufTraceSerializer.serializeRequest(items);
+    let serializedSpans: Uint8Array | undefined = ProtobufTraceSerializer.serializeRequest(items);
+
+    if (serializedSpans === undefined) {
+      resultCallback({
+        code: ExportResultCode.FAILED,
+        error: new Error('Nothing to send'),
+      });
+      return;
+    }
+
+    // Pass pre-processed data to passthrough serializer. When super.export() is called, the Passthrough Serializer will
+    // use the pre-processed data instead of serializing and compressing the data again.
+    const shouldCompress = this.compression && this.compression !== CompressionAlgorithm.NONE;
+    if (shouldCompress) {
+      serializedSpans = gzipSync(serializedSpans);
+    }
+
+    this.serializer.setSerializedData(serializedSpans);
+
     const headers = this['_delegate']._transport?._transport?._parameters?.headers();
 
     if (headers) {
+      if (shouldCompress) {
+        headers['Content-Encoding'] = 'gzip';
+      } else {
+        delete headers['Content-Encoding'];
+      }
+
       const signedRequest = await this.authenticator.authenticate(this.endpoint, headers, serializedSpans);
 
       // See type: https://github.com/open-telemetry/opentelemetry-js/blob/experimental/v0.57.1/experimental/packages/otlp-exporter-base/src/transport/http-transport-types.ts#L31

aws-distro-opentelemetry-node-autoinstrumentation/src/register.ts

@@ -84,9 +84,6 @@ try {
   diag.info('Setting TraceProvider for instrumentations at the end of initialization');
   for (const instrumentation of instrumentations) {
     instrumentation.setTracerProvider(trace.getTracerProvider());
-    if (instrumentation.setLoggerProvider) {
-      instrumentation.setLoggerProvider(logs.getLoggerProvider());
-    }
   }
 
   diag.debug(`Environment variable OTEL_PROPAGATORS is set to '${process.env.OTEL_PROPAGATORS}'`);