refactor more changes to base exporter

Author: liustve

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

@@ -608,7 +608,7 @@ export class AwsSpanProcessorProvider {
       case 'http/protobuf':
         if (otlpExporterTracesEndpoint && isAwsOtlpEndpoint(otlpExporterTracesEndpoint, 'xray')) {
           diag.debug('Detected XRay OTLP Traces endpoint. Switching exporter to OtlpAwsSpanExporter');
-          return new OTLPAwsSpanExporter(otlpExporterTracesEndpoint, { compression: CompressionAlgorithm.GZIP });
+          return new OTLPAwsSpanExporter(otlpExporterTracesEndpoint);
         }
         return new OTLPProtoTraceExporter();
       case 'udp':
@@ -618,7 +618,7 @@ export class AwsSpanProcessorProvider {
         diag.warn(`Unsupported OTLP traces protocol: ${protocol}. Using http/protobuf.`);
         if (otlpExporterTracesEndpoint && isAwsOtlpEndpoint(otlpExporterTracesEndpoint, 'xray')) {
           diag.debug('Detected XRay OTLP Traces endpoint. Switching exporter to OtlpAwsSpanExporter');
-          return new OTLPAwsSpanExporter(otlpExporterTracesEndpoint, { compression: CompressionAlgorithm.GZIP });
+          return new OTLPAwsSpanExporter(otlpExporterTracesEndpoint);
         }
         return new OTLPProtoTraceExporter();
     }

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

@@ -4,18 +4,14 @@ import { CompressionAlgorithm, OTLPExporterBase } from '@opentelemetry/otlp-expo
 import { gzipSync } from 'zlib';
 import { ExportResult, ExportResultCode } from '@opentelemetry/core';
 import { AwsAuthenticator } from './aws-authenticator';
-import { PassthroughSerializer } from './passthrough-serializer';
 import { ISerializer } from '@opentelemetry/otlp-transformer';
-import { diag } from '@opentelemetry/api';
 
 /**
  * Base class for AWS OTLP exporters
  */
-export abstract class OTLPAwsBaseExporter<Payload, Response> {
+export abstract class OTLPAwsBaseExporter<Payload, Response> extends OTLPExporterBase<Payload> {
   protected parentExporter: OTLPExporterBase<Payload>;
-  private readonly originalHeaders?: Record<string, string>;
   private readonly compression?: CompressionAlgorithm;
-  private newHeaders: Record<string, string> = {};
   private endpoint: string;
   private serializer: PassthroughSerializer<Response>;
   private authenticator: AwsAuthenticator;
@@ -28,6 +24,7 @@ export abstract class OTLPAwsBaseExporter<Payload, Response> {
     parentSerializer: ISerializer<Payload, Response>,
     compression?: CompressionAlgorithm
   ) {
+    super(parentExporter['_delegate']);
     this.compression = compression;
     this.endpoint = endpoint;
     this.authenticator = new AwsAuthenticator(this.endpoint, service);
@@ -41,7 +38,6 @@ export abstract class OTLPAwsBaseExporter<Payload, Response> {
     // https://github.com/open-telemetry/opentelemetry-js/blob/ec17ce48d0e5a99a122da5add612a20e2dd84ed5/experimental/packages/otlp-exporter-base/src/otlp-export-delegate.ts#L69
     this.serializer = new PassthroughSerializer<Response>(this.parentSerializer.deserializeResponse);
     this.parentExporter['_delegate']._serializer = this.serializer;
-    this.originalHeaders = this.getHeaders();
   }
 
   /**
@@ -51,83 +47,101 @@ export abstract class OTLPAwsBaseExporter<Payload, Response> {
    * @param items - Array of signal data to export
    * @param resultCallback - Callback function to handle export result
    */
-  public async export(items: Payload, resultCallback: (result: ExportResult) => void): Promise<void> {
-    if (this.originalHeaders) {
-      let serializedData: Uint8Array | undefined = this.parentSerializer.serializeRequest(items);
+  override async export(items: Payload, resultCallback: (result: ExportResult) => void): Promise<void> {
+    const headers = this.parentExporter['_delegate']._transport?._transport?._parameters?.headers();
+
+    if (!headers) {
+      resultCallback({
+        code: ExportResultCode.FAILED,
+        error: new Error(`Request headers are unset - unable to export to ${this.endpoint}`),
+      });
+      return;
+    }
+
+    let serializedData: Uint8Array | undefined = this.parentSerializer.serializeRequest(items);
+
+    if (!serializedData) {
+      resultCallback({
+        code: ExportResultCode.FAILED,
+        error: new Error('Nothing to send'),
+      });
+      return;
+    }
 
-      if (!serializedData) {
+    delete headers['Content-Encoding'];
+    const shouldCompress = this.compression && this.compression !== CompressionAlgorithm.NONE;
+
+    if (shouldCompress) {
+      try {
+        serializedData = gzipSync(serializedData);
+        headers['Content-Encoding'] = 'gzip';
+      } catch (exception) {
         resultCallback({
           code: ExportResultCode.FAILED,
-          error: new Error('Nothing to send'),
+          error: new Error(`Failed to compress: ${exception}`),
         });
         return;
       }
+    }
 
-      const shouldCompress = this.compression && this.compression !== CompressionAlgorithm.NONE;
-
-      if (shouldCompress) {
-        try {
-          serializedData = gzipSync(serializedData);
-          this.addHeader('Content-Encoding', 'gzip');
-        } catch (exception) {
-          resultCallback({
-            code: ExportResultCode.FAILED,
-            error: new Error(`Failed to compress: ${exception}`),
-          });
-          return;
-        }
-      }
+    this.serializer.setSerializedData(serializedData);
+    const signedHeaders = await this.authenticator.authenticate(headers, serializedData);
 
-      this.serializer.setSerializedData(serializedData);
+    if (signedHeaders) {
+      this.parentExporter['_delegate']._transport._transport._parameters.headers = () => signedHeaders;
+    }
 
-      const mergedHeaders = { ...this.newHeaders, ...this.originalHeaders };
-      const signedHeaders = await this.authenticator.authenticate(mergedHeaders, serializedData);
+    this.parentExporter.export(items, resultCallback);
+  }
 
-      if (signedHeaders) {
-        this.setTransportHeaders(signedHeaders);
-      }
+  override shutdown(): Promise<void> {
+    return this.parentExporter.shutdown();
+  }
 
-      this.parentExporter.export(items, resultCallback);
+  override forceFlush(): Promise<void> {
+    return this.parentExporter.forceFlush();
+  }
+}
 
-      this.setTransportHeaders(this.originalHeaders);
-      this.newHeaders = {};
-      return;
-    }
+/**
+ * A serializer that bypasses request serialization by returning pre-serialized data.
+ * @template Response The type of the deserialized response
+ */
+class PassthroughSerializer<Response> implements ISerializer<Uint8Array, Response> {
+  private serializedData: Uint8Array = new Uint8Array();
+  private deserializer: (data: Uint8Array) => Response;
 
-    resultCallback({
-      code: ExportResultCode.FAILED,
-      error: new Error('No headers found, cannot sign request. Not exporting.'),
-    });
+  /**
+   * Creates a new PassthroughSerializer instance.
+   * @param deserializer Function to deserialize response data
+   */
+  constructor(deserializer: (data: Uint8Array) => Response) {
+    this.deserializer = deserializer;
   }
 
-  // This is a bit ugly but need it in order safely set any new headers
-
   /**
-   * Adds a header to the exporter's transport parameters
+   * Sets the pre-serialized data to be returned when serializeRequest is called.
+   * @param data The serialized data to use
    */
-  protected addHeader(key: string, value: string): void {
-    this.newHeaders[key] = value;
+  setSerializedData(data: Uint8Array): void {
+    this.serializedData = data;
   }
 
   /**
-   * Gets headers in the transport parameters
+   * Returns the pre-serialized data, ignoring the request parameter.
+   * @param request Ignored parameter.
+   * @returns The pre-serialized data
    */
-  private getHeaders(): Record<string, string> | undefined {
-    const headersFunc = this.parentExporter['_delegate']._transport?._transport?._parameters?.headers;
-    if (!headersFunc) {
-      diag.debug('No existing headers found, using empty headers.');
-      return undefined;
-    }
-    return headersFunc();
+  serializeRequest(request: Uint8Array): Uint8Array {
+    return this.serializedData;
   }
 
   /**
-   * Sets headers in the transport parameters
+   * Deserializes response data using the provided deserializer function.
+   * @param data The response data to deserialize
+   * @returns The deserialized response
    */
-  private setTransportHeaders(headers: Record<string, string>): void {
-    const parameters = this.parentExporter['_delegate']._transport?._transport?._parameters;
-    if (parameters) {
-      parameters.headers = () => headers;
-    }
+  deserializeResponse(data: Uint8Array): Response {
+    return this.deserializer(data);
   }
 }

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

@@ -27,10 +27,6 @@ export class OTLPAwsLogExporter
       compression: CompressionAlgorithm.NONE,
     };
 
-    const parentExporter = new OTLPProtoLogExporter(modifiedConfig);
-    super(endpoint, 'logs', parentExporter, ProtobufLogsSerializer, config?.compression);
-  }
-  shutdown(): Promise<void> {
-    return this.parentExporter.shutdown();
+    super(endpoint, 'logs', new OTLPProtoLogExporter(modifiedConfig), ProtobufLogsSerializer, config?.compression);
   }
 }

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

@@ -28,11 +28,6 @@ export class OTLPAwsSpanExporter
       compression: CompressionAlgorithm.NONE,
     };
 
-    const parentExporter = new OTLPProtoTraceExporter(modifiedConfig);
-    super(endpoint, 'xray', parentExporter, ProtobufTraceSerializer, config?.compression);
-  }
-
-  shutdown(): Promise<void> {
-    return this.parentExporter.shutdown();
+    super(endpoint, 'xray', new OTLPProtoTraceExporter(modifiedConfig), ProtobufTraceSerializer, config?.compression);
   }
 }