SigV4 Authentication Support for OTLP HTTP Logs Exporter

aws-distro-opentelemetry-node-autoinstrumentation/package.json

@@ -41,7 +41,7 @@
     ],
     "exclude": [
       "src/third-party/**/*.ts",
-      "src/otlp-aws-span-exporter.ts"
+      "src/exporter/otlp/aws/common/aws-authenticator.ts" 
     ]
   },
   "bugs": {
@@ -108,6 +108,9 @@
     "@opentelemetry/exporter-metrics-otlp-grpc": "0.57.1",
     "@opentelemetry/exporter-metrics-otlp-http": "0.57.1",
     "@opentelemetry/exporter-trace-otlp-proto": "0.57.1",
+    "@opentelemetry/exporter-logs-otlp-grpc": "0.57.1",
+    "@opentelemetry/exporter-logs-otlp-http": "0.57.1",
+    "@opentelemetry/exporter-logs-otlp-proto": "0.57.1",
     "@opentelemetry/exporter-zipkin": "1.30.1",
     "@opentelemetry/id-generator-aws-xray": "1.2.3",
     "@opentelemetry/instrumentation": "0.57.1",
@@ -119,6 +122,7 @@
     "@opentelemetry/sdk-metrics": "1.30.1",
     "@opentelemetry/sdk-node": "0.57.1",
     "@opentelemetry/sdk-trace-base": "1.30.1",
+    "@opentelemetry/sdk-logs": "0.57.1",
     "@opentelemetry/semantic-conventions": "1.28.0"
   },
   "files": [

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

@@ -0,0 +1,100 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+import { diag } from '@opentelemetry/api';
+import { getNodeVersion } from '../../../../utils';
+let SignatureV4: any;
+let HttpRequest: any;
+let defaultProvider: any;
+let Sha256: any;
+
+let dependenciesLoaded = false;
+
+if (getNodeVersion() >= 16) {
+  try {
+    defaultProvider = require('@aws-sdk/credential-provider-node').defaultProvider;
+    Sha256 = require('@aws-crypto/sha256-js').Sha256;
+    SignatureV4 = require('@smithy/signature-v4').SignatureV4;
+    HttpRequest = require('@smithy/protocol-http').HttpRequest;
+    dependenciesLoaded = true;
+  } catch (error) {
+    diag.error(`Failed to load required AWS dependency for SigV4 Signing: ${error}`);
+  }
+} else {
+  diag.error('SigV4 signing requires at least Node major version 16');
+}
+
+// See: https://docs.aws.amazon.com/AmazonS3/latest/API/sig-v4-authenticating-requests.html
+export const AUTHORIZATION_HEADER = 'authorization';
+export const X_AMZ_DATE_HEADER = 'x-amz-date';
+export const X_AMZ_SECURITY_TOKEN_HEADER = 'x-amz-security-token';
+export const X_AMZ_CONTENT_SHA256_HEADER = 'x-amz-content-sha256';
+
+export class AwsAuthenticator {
+  private endpoint: URL;
+  private region: string;
+  private service: string;
+
+  constructor(endpoint: string, service: string) {
+    // The endpoint is pre-validated by the config with isAwsOtlpEndpoint, so then endpoint is guaranteed to be well formatted and
+    //  new URL() will not throw
+    this.endpoint = new URL(endpoint);
+    this.region = endpoint.split('.')[1];
+    this.service = service;
+  }
+
+  public async authenticate(headers: Record<string, string>, serializedData: Uint8Array | undefined) {
+    // Only do SigV4 Signing if the required dependencies are installed.
+    if (dependenciesLoaded && serializedData) {
+      const cleanedHeaders = this.removeSigV4Headers(headers);
+
+      const request = new HttpRequest({
+        method: 'POST',
+        protocol: 'https',
+        hostname: this.endpoint.hostname,
+        path: this.endpoint.pathname,
+        body: serializedData,
+        headers: {
+          ...cleanedHeaders,
+          host: this.endpoint.hostname,
+        },
+      });
+
+      try {
+        const signer = new SignatureV4({
+          credentials: defaultProvider(),
+          region: this.region,
+          service: this.service,
+          sha256: Sha256,
+        });
+
+        const signedRequest = await signer.sign(request);
+
+        return signedRequest.headers;
+      } catch (exception) {
+        diag.debug(`Failed to sign/authenticate the given export request with error: ${exception}`);
+        return undefined;
+      }
+    }
+
+    diag.debug('No serialized data provided. Not authenticating.');
+    return undefined;
+  }
+
+  // Cleans up Sigv4 from headers to avoid accidentally copying them to the new headers
+  private removeSigV4Headers(headers: Record<string, string>) {
+    const newHeaders: Record<string, string> = {};
+    const sigv4Headers = [
+      AUTHORIZATION_HEADER,
+      X_AMZ_CONTENT_SHA256_HEADER,
+      X_AMZ_DATE_HEADER,
+      X_AMZ_CONTENT_SHA256_HEADER,
+    ];
+
+    for (const key in headers) {
+      if (!sigv4Headers.includes(key.toLowerCase())) {
+        newHeaders[key] = headers[key];
+      }
+    }
+    return newHeaders;
+  }
+}

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

@@ -0,0 +1,152 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+import { CompressionAlgorithm, OTLPExporterBase } from '@opentelemetry/otlp-exporter-base';
+import { gzipSync } from 'zlib';
+import { ExportResult, ExportResultCode } from '@opentelemetry/core';
+import { AwsAuthenticator } from './aws-authenticator';
+import { ISerializer } from '@opentelemetry/otlp-transformer';
+
+/**
+ * Base class for AWS OTLP exporters
+ */
+export abstract class OTLPAwsBaseExporter<Payload, Response> extends OTLPExporterBase<Payload> {
+  protected parentExporter: OTLPExporterBase<Payload>;
+  private readonly compression?: CompressionAlgorithm;
+  private endpoint: string;
+  private serializer: PassthroughSerializer<Response>;
+  private authenticator: AwsAuthenticator;
+  private parentSerializer: ISerializer<Payload, Response>;
+
+  constructor(
+    endpoint: string,
+    service: string,
+    parentExporter: OTLPExporterBase<Payload>,
+    parentSerializer: ISerializer<Payload, Response>,
+    compression?: CompressionAlgorithm
+  ) {
+    super(parentExporter['_delegate']);
+    this.compression = compression;
+    this.endpoint = endpoint;
+    this.authenticator = new AwsAuthenticator(this.endpoint, service);
+    this.parentExporter = parentExporter;
+    this.parentSerializer = parentSerializer;
+
+    // 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.
+    // This is used in order to prevent serializing and compressing the data again when calling parentExporter.export().
+    // To see why this works:
+    // 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;
+  }
+
+  /**
+   * Overrides the upstream implementation of export.
+   * All behaviors are the same except if the endpoint is an AWS OTLP endpoint, we will sign the request with SigV4
+   * in headers before sending it to the endpoint.
+   * @param items - Array of signal data to export
+   * @param resultCallback - Callback function to handle export result
+   */
+  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;
+    }
+
+    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(`Failed to compress: ${exception}`),
+        });
+        return;
+      }
+    }
+
+    this.serializer.setSerializedData(serializedData);
+    const signedHeaders = await this.authenticator.authenticate(headers, serializedData);
+
+    if (!signedHeaders) {
+      resultCallback({
+        code: ExportResultCode.FAILED,
+        error: new Error('Sigv4 Signing Failed. Not exporting'),
+      });
+      return;
+    }
+
+    this.parentExporter['_delegate']._transport._transport._parameters.headers = () => signedHeaders;
+    this.parentExporter.export(items, resultCallback);
+  }
+
+  override shutdown(): Promise<void> {
+    return this.parentExporter.shutdown();
+  }
+
+  override forceFlush(): Promise<void> {
+    return this.parentExporter.forceFlush();
+  }
+}
+
+/**
+ * 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;
+
+  /**
+   * 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

@@ -0,0 +1,32 @@
+// 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 { CompressionAlgorithm, OTLPExporterNodeConfigBase } from '@opentelemetry/otlp-exporter-base';
+import { IExportLogsServiceResponse, ProtobufLogsSerializer } from '@opentelemetry/otlp-transformer';
+import { LogRecordExporter, ReadableLogRecord } from '@opentelemetry/sdk-logs';
+import { OTLPAwsBaseExporter } from '../common/otlp-aws-base-exporter';
+
+/**
+ * This exporter extends the functionality of the OTLPProtoLogExporter to allow logs to be exported
+ * to the CloudWatch Logs OTLP endpoint https://logs.[AWSRegion].amazonaws.com/v1/logs. Utilizes the aws-sdk
+ * library to sign and directly inject SigV4 Authentication to the exported request's headers. <a
+ * href="https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-OTLPEndpoint.html">...</a>
+ *
+ * This only works with version >=16 Node.js environments.
+ * @param endpoint - The AWS CloudWatch Logs OTLP endpoint URL
+ * @param config - Optional OTLP exporter configuration
+ */
+export class OTLPAwsLogExporter
+  extends OTLPAwsBaseExporter<ReadableLogRecord[], IExportLogsServiceResponse>
+  implements LogRecordExporter
+{
+  constructor(endpoint: string, config?: OTLPExporterNodeConfigBase) {
+    const modifiedConfig: OTLPExporterNodeConfigBase = {
+      ...config,
+      url: endpoint,
+      compression: CompressionAlgorithm.NONE,
+    };
+
+    super(endpoint, 'logs', new OTLPProtoLogExporter(modifiedConfig), ProtobufLogsSerializer, config?.compression);
+  }
+}

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

@@ -0,0 +1,77 @@
+// 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 { CompressionAlgorithm, OTLPExporterNodeConfigBase } from '@opentelemetry/otlp-exporter-base';
+import { IExportTraceServiceResponse, ProtobufTraceSerializer } from '@opentelemetry/otlp-transformer';
+import { ReadableSpan, SpanExporter } from '@opentelemetry/sdk-trace-base';
+import { OTLPAwsBaseExporter } from '../common/otlp-aws-base-exporter';
+import { LLOHandler } from '../../../../llo-handler';
+import { LoggerProvider as APILoggerProvider, logs } from '@opentelemetry/api-logs';
+import { ExportResult } from '@opentelemetry/core';
+import { isAgentObservabilityEnabled } from '../../../../utils';
+import { diag } from '@opentelemetry/api';
+import { LoggerProvider } from '@opentelemetry/sdk-logs';
+
+/**
+ * This exporter extends the functionality of the OTLPProtoTraceExporter to allow spans to be exported
+ * to the XRay OTLP endpoint https://xray.[AWSRegion].amazonaws.com/v1/traces. Utilizes the aws-sdk
+ * library to sign and directly inject SigV4 Authentication to the exported request's headers. <a
+ * href="https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-OTLPEndpoint.html">...</a>
+ *
+ * This only works with version >=16 Node.js environments.
+ *
+ * @param endpoint - The AWS X-Ray OTLP endpoint URL
+ * @param config - Optional OTLP exporter configuration
+ */
+export class OTLPAwsSpanExporter
+  extends OTLPAwsBaseExporter<ReadableSpan[], IExportTraceServiceResponse>
+  implements SpanExporter
+{
+  private loggerProvider: APILoggerProvider | undefined;
+  private lloHandler: LLOHandler | undefined;
+
+  constructor(endpoint: string, config?: OTLPExporterNodeConfigBase, loggerProvider?: APILoggerProvider) {
+    const modifiedConfig: OTLPExporterNodeConfigBase = {
+      ...config,
+      url: endpoint,
+      compression: CompressionAlgorithm.NONE,
+    };
+
+    super(endpoint, 'xray', new OTLPProtoTraceExporter(modifiedConfig), ProtobufTraceSerializer, config?.compression);
+
+    this.lloHandler = undefined;
+    this.loggerProvider = loggerProvider;
+  }
+
+  // Lazily initialize LLO handler when needed to avoid initialization order issues
+  private ensureLloHandler(): boolean {
+    if (!this.lloHandler && isAgentObservabilityEnabled()) {
+      // If loggerProvider wasn't provided, try to get the current one
+      if (!this.loggerProvider) {
+        try {
+          this.loggerProvider = logs.getLoggerProvider();
+        } catch (e: unknown) {
+          diag.debug('Failed to get logger provider', e);
+          return false;
+        }
+      }
+
+      if (this.loggerProvider instanceof LoggerProvider) {
+        this.lloHandler = new LLOHandler(this.loggerProvider);
+        return true;
+      }
+    }
+
+    return !!this.lloHandler;
+  }
+
+  override async export(items: ReadableSpan[], resultCallback: (result: ExportResult) => void): Promise<void> {
+    let itemsToSerialize: ReadableSpan[] = items;
+    if (isAgentObservabilityEnabled() && this.ensureLloHandler() && this.lloHandler) {
+      // items to serialize are now the lloProcessedSpans
+      itemsToSerialize = this.lloHandler.processSpans(items);
+    }
+
+    return super.export(itemsToSerialize, resultCallback);
+  }
+}