EMF Exporter - Initial implementation with Gauge Support

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

@@ -69,16 +69,16 @@
     "instrumentation"
   ],
   "devDependencies": {
-    "@aws-sdk/client-bedrock": "3.632.0",
-    "@aws-sdk/client-bedrock-agent": "3.632.0",
-    "@aws-sdk/client-bedrock-agent-runtime": "3.632.0",
-    "@aws-sdk/client-bedrock-runtime": "3.632.0",
-    "@aws-sdk/client-kinesis": "3.632.0",
-    "@aws-sdk/client-lambda": "3.632.0",
-    "@aws-sdk/client-s3": "3.632.0",
-    "@aws-sdk/client-secrets-manager": "3.632.0",
-    "@aws-sdk/client-sfn": "3.632.0",
-    "@aws-sdk/client-sns": "3.632.0",
+    "@aws-sdk/client-bedrock": "3.631.0",
+    "@aws-sdk/client-bedrock-agent": "3.631.0",
+    "@aws-sdk/client-bedrock-agent-runtime": "3.631.0",
+    "@aws-sdk/client-bedrock-runtime": "3.631.0",
+    "@aws-sdk/client-kinesis": "3.631.0",
+    "@aws-sdk/client-lambda": "3.631.0",
+    "@aws-sdk/client-s3": "3.631.0",
+    "@aws-sdk/client-secrets-manager": "3.631.0",
+    "@aws-sdk/client-sfn": "3.631.0",
+    "@aws-sdk/client-sns": "3.631.0",
     "@opentelemetry/contrib-test-utils": "^0.45.0",
     "@smithy/protocol-http": "^5.0.1",
     "@smithy/signature-v4": "^5.0.1",
@@ -94,9 +94,10 @@
     "rimraf": "5.0.5",
     "sinon": "15.2.0",
     "ts-mocha": "10.0.0",
-    "typescript": "4.4.4"
+    "typescript": "4.9.5"
   },
   "dependencies": {
+    "@aws-sdk/client-cloudwatch-logs": "3.621.0",
     "@opentelemetry/api": "1.9.0",
     "@opentelemetry/auto-configuration-propagators": "0.3.2",
     "@opentelemetry/auto-instrumentations-node": "0.56.0",

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

@@ -31,6 +31,7 @@ import {
 import {
   Aggregation,
   AggregationSelector,
+  AggregationTemporality,
   InstrumentType,
   MeterProvider,
   PeriodicExportingMetricReader,
@@ -61,6 +62,8 @@ import { OTLPUdpSpanExporter } from './otlp-udp-exporter';
 import { AwsXRayRemoteSampler } from './sampler/aws-xray-remote-sampler';
 // This file is generated via `npm run compile`
 import { LIB_VERSION } from './version';
+import { AWSCloudWatchEMFExporter } from './exporter/otlp/aws/metrics/otlp-aws-emf-exporter';
+import { CloudWatchLogsClientConfig } from '@aws-sdk/client-cloudwatch-logs';
 
 const XRAY_OTLP_ENDPOINT_PATTERN = '^https://xray\\.([a-z0-9-]+)\\.amazonaws\\.com/v1/traces$';
 
@@ -77,6 +80,17 @@ const FORMAT_OTEL_UNSAMPLED_TRACES_BINARY_PREFIX = 'T1U';
 const LAMBDA_SPAN_EXPORT_BATCH_SIZE = 10;
 export const LAMBDA_APPLICATION_SIGNALS_REMOTE_ENVIRONMENT: string = 'LAMBDA_APPLICATION_SIGNALS_REMOTE_ENVIRONMENT';
 
+const AWS_OTLP_LOGS_GROUP_HEADER = 'x-aws-log-group';
+const AWS_OTLP_LOGS_STREAM_HEADER = 'x-aws-log-stream';
+const AWS_EMF_METRICS_NAMESPACE = 'x-aws-metric-namespace';
+
+interface OtlpLogHeaderSetting {
+  logGroup?: string;
+  logStream?: string;
+  namespace?: string;
+  isValid: boolean;
+}
+
 /**
  * Aws Application Signals Config Provider creates a configuration object that can be provided to
  * the OTel NodeJS SDK for Auto Instrumentation with Application Signals Functionality.
@@ -97,6 +111,7 @@ export class AwsOpentelemetryConfigurator {
   private sampler: Sampler;
   private spanProcessors: SpanProcessor[];
   private propagator: TextMapPropagator;
+  private metricReader: PeriodicExportingMetricReader | undefined;
 
   /**
    * The constructor will setup the AwsOpentelemetryConfigurator object to be able to provide a
@@ -180,6 +195,10 @@ export class AwsOpentelemetryConfigurator {
     const awsSpanProcessorProvider: AwsSpanProcessorProvider = new AwsSpanProcessorProvider(this.resource);
     this.spanProcessors = awsSpanProcessorProvider.getSpanProcessors();
     AwsOpentelemetryConfigurator.customizeSpanProcessors(this.spanProcessors, this.resource);
+
+    if (checkEmfExporterEnabled()) {
+      this.metricReader = AwsOpentelemetryConfigurator.createEmfExporterMetricReader();
+    }
   }
 
   private customizeVersions(autoResource: Resource): Resource {
@@ -211,6 +230,10 @@ export class AwsOpentelemetryConfigurator {
       textMapPropagator: this.propagator,
     };
 
+    if (this.metricReader) {
+      config.metricReader = this.metricReader;
+    }
+
     return config;
   }
 
@@ -223,13 +246,7 @@ export class AwsOpentelemetryConfigurator {
     return isApplicationSignalsEnabled.toLowerCase() === 'true';
   }
 
-  static customizeSpanProcessors(spanProcessors: SpanProcessor[], resource: Resource): void {
-    if (!AwsOpentelemetryConfigurator.isApplicationSignalsEnabled()) {
-      return;
-    }
-
-    diag.info('AWS Application Signals enabled.');
-
+  static geMetricExportInterval(): number {
     let exportIntervalMillis: number = Number(process.env[METRIC_EXPORT_INTERVAL_CONFIG]);
     diag.debug(`AWS Application Signals Metrics export interval: ${exportIntervalMillis}`);
 
@@ -239,13 +256,23 @@ export class AwsOpentelemetryConfigurator {
       diag.info(`AWS Application Signals metrics export interval capped to ${exportIntervalMillis}`);
     }
 
+    return exportIntervalMillis;
+  }
+
+  static customizeSpanProcessors(spanProcessors: SpanProcessor[], resource: Resource): void {
+    if (!AwsOpentelemetryConfigurator.isApplicationSignalsEnabled()) {
+      return;
+    }
+
+    diag.info('AWS Application Signals enabled.');
+
     spanProcessors.push(AttributePropagatingSpanProcessorBuilder.create().build());
 
     const applicationSignalsMetricExporter: PushMetricExporter =
       ApplicationSignalsExporterProvider.Instance.createExporter();
     const periodicExportingMetricReader: PeriodicExportingMetricReader = new PeriodicExportingMetricReader({
       exporter: applicationSignalsMetricExporter,
-      exportIntervalMillis: exportIntervalMillis,
+      exportIntervalMillis: AwsOpentelemetryConfigurator.geMetricExportInterval(),
     });
 
     // Register BatchUnsampledSpanProcessor to export unsampled traces in Lambda
@@ -281,6 +308,22 @@ export class AwsOpentelemetryConfigurator {
     }
   }
 
+  static createEmfExporterMetricReader() {
+    const headersResult = validateLogsHeaders();
+    if (!headersResult.logGroup) {
+      diag.warn('Log group is not set. Set Log Group with OTEL_EXPORTER_OTLP_LOGS_HEADERS=<EMF Log Group>');
+      return;
+    }
+
+    const emfExporter = createEmfExporter(headersResult.logGroup, headersResult.logStream, headersResult.namespace);
+    const periodicExportingMetricReader = new PeriodicExportingMetricReader({
+      exporter: emfExporter,
+      exportIntervalMillis: AwsOpentelemetryConfigurator.geMetricExportInterval(),
+    });
+
+    return periodicExportingMetricReader;
+  }
+
   static customizeSampler(sampler: Sampler): Sampler {
     if (AwsOpentelemetryConfigurator.isApplicationSignalsEnabled()) {
       return AlwaysRecordSampler.create(sampler);
@@ -428,7 +471,7 @@ export class AwsSpanProcessorProvider {
   private resource: Resource;
 
   static configureOtlp(): SpanExporter {
-    const otlp_exporter_traces_endpoint = process.env['OTEL_EXPORTER_OTLP_TRACES_ENDPOINT'];
+    const otlpExporterTracesEndpoint = process.env['OTEL_EXPORTER_OTLP_TRACES_ENDPOINT'];
     // eslint-disable-next-line @typescript-eslint/typedef
     let protocol = this.getOtlpProtocol();
 
@@ -445,19 +488,19 @@ export class AwsSpanProcessorProvider {
       case 'http/json':
         return new OTLPHttpTraceExporter();
       case 'http/protobuf':
-        if (otlp_exporter_traces_endpoint && isXrayOtlpEndpoint(otlp_exporter_traces_endpoint)) {
+        if (otlpExporterTracesEndpoint && isXrayOtlpEndpoint(otlpExporterTracesEndpoint)) {
           diag.debug('Detected XRay OTLP Traces endpoint. Switching exporter to OtlpAwsSpanExporter');
-          return new OTLPAwsSpanExporter(otlp_exporter_traces_endpoint);
+          return new OTLPAwsSpanExporter(otlpExporterTracesEndpoint);
         }
         return new OTLPProtoTraceExporter();
       case 'udp':
         diag.debug('Detected AWS Lambda environment and enabling UDPSpanExporter');
         return new OTLPUdpSpanExporter(getXrayDaemonEndpoint(), FORMAT_OTEL_SAMPLED_TRACES_BINARY_PREFIX);
       default:
         diag.warn(`Unsupported OTLP traces protocol: ${protocol}. Using http/protobuf.`);
-        if (otlp_exporter_traces_endpoint && isXrayOtlpEndpoint(otlp_exporter_traces_endpoint)) {
+        if (otlpExporterTracesEndpoint && isXrayOtlpEndpoint(otlpExporterTracesEndpoint)) {
           diag.debug('Detected XRay OTLP Traces endpoint. Switching exporter to OtlpAwsSpanExporter');
-          return new OTLPAwsSpanExporter(otlp_exporter_traces_endpoint);
+          return new OTLPAwsSpanExporter(otlpExporterTracesEndpoint);
         }
         return new OTLPProtoTraceExporter();
     }
@@ -647,6 +690,8 @@ function getSamplerProbabilityFromEnv(environment: Required<ENVIRONMENT>): numbe
   return probability;
 }
 
+// END The OpenTelemetry Authors code
+
 function getSpanExportBatchSize() {
   if (isLambdaEnvironment()) {
     return LAMBDA_SPAN_EXPORT_BATCH_SIZE;
@@ -667,8 +712,97 @@ function getXrayDaemonEndpoint() {
   return process.env[AWS_XRAY_DAEMON_ADDRESS_CONFIG];
 }
 
-function isXrayOtlpEndpoint(otlpEndpoint: string | undefined) {
+export function isXrayOtlpEndpoint(otlpEndpoint: string | undefined) {
   return otlpEndpoint && new RegExp(XRAY_OTLP_ENDPOINT_PATTERN).test(otlpEndpoint.toLowerCase());
 }
 
-// END The OpenTelemetry Authors code
+export function checkEmfExporterEnabled(): boolean {
+  const exporterValue = process.env.OTEL_METRICS_EXPORTER;
+  if (exporterValue === undefined) {
+    return false;
+  }
+
+  const exporters = exporterValue.split(',').map(exporter => exporter.trim());
+
+  const index = exporters.indexOf('awsemf');
+  if (index === -1) {
+    return false;
+  }
+
+  exporters.splice(index, 1);
+
+  const newValue = exporters ? exporters.join(',') : undefined;
+
+  if (typeof newValue === 'string' && newValue !== '') {
+    process.env.OTEL_METRICS_EXPORTER = newValue;
+  } else {
+    delete process.env.OTEL_METRICS_EXPORTER;
+  }
+
+  return true;
+}
+
+export function createEmfExporter(
+  logGroupName: string,
+  logStreamName?: string,
+  namespace?: string,
+  aggregationTemporality: AggregationTemporality = AggregationTemporality.DELTA,
+  cloudwatchLogsConfig: CloudWatchLogsClientConfig = {}
+): AWSCloudWatchEMFExporter {
+  return new AWSCloudWatchEMFExporter(
+    namespace,
+    logGroupName,
+    logStreamName,
+    aggregationTemporality,
+    cloudwatchLogsConfig
+  );
+}
+
+/**
+ * Checks if x-aws-log-group and x-aws-log-stream are present in the headers in order to send logs to
+ * AWS OTLP Logs endpoint.
+ */
+export function validateLogsHeaders(): OtlpLogHeaderSetting {
+  const logHeaders = process.env.OTEL_EXPORTER_OTLP_LOGS_HEADERS;
+
+  let logGroup: string | undefined = undefined;
+  let logStream: string | undefined = undefined;
+  let namespace: string | undefined = undefined;
+
+  if (!logHeaders) {
+    diag.warn(
+      'Improper configuration: Please configure the environment variable OTEL_EXPORTER_OTLP_LOGS_HEADERS to include x-aws-log-group and x-aws-log-stream'
+    );
+    return { isValid: false };
+  }
+
+  for (const pair of logHeaders.split(',')) {
+    const splitIndex = pair.indexOf('=');
+    if (splitIndex > -1) {
+      const key = pair.substring(0, splitIndex);
+      const value = pair.substring(splitIndex + 1);
+
+      if (key === AWS_OTLP_LOGS_GROUP_HEADER && value !== '') {
+        logGroup = value;
+      } else if (key === AWS_OTLP_LOGS_STREAM_HEADER && value !== '') {
+        logStream = value;
+      } else if (key === AWS_EMF_METRICS_NAMESPACE && value !== '') {
+        namespace = value;
+      }
+    }
+  }
+
+  const isValid = !!logGroup && !!logStream;
+  if (!isValid) {
+    diag.warn(
+      'Improper configuration: Please configure the environment variable OTEL_EXPORTER_OTLP_LOGS_HEADERS to have values for x-aws-log-group and x-aws-log-stream'
+    );
+  }
+
+  return {
+    logGroup: logGroup,
+    logStream: logStream,
+    namespace: namespace,
+    isValid: isValid,
+  };
+}