EMF Exporter implementation with Gauge, Sum, and Histogram 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

@@ -61,6 +61,7 @@ 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/aws/metrics/aws-cloudwatch-emf-exporter';
 import { isAgentObservabilityEnabled } from './utils';
 import { BaggageSpanProcessor } from '@opentelemetry/baggage-span-processor';
 import { logs } from '@opentelemetry/api-logs';
@@ -80,6 +81,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.
@@ -100,6 +112,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
@@ -183,6 +196,9 @@ export class AwsOpentelemetryConfigurator {
     const awsSpanProcessorProvider: AwsSpanProcessorProvider = new AwsSpanProcessorProvider(this.resource);
     this.spanProcessors = awsSpanProcessorProvider.getSpanProcessors();
     AwsOpentelemetryConfigurator.customizeSpanProcessors(this.spanProcessors, this.resource);
+
+    const isEmfEnabled = checkEmfExporterEnabled();
+    this.customizeMetricReader(isEmfEnabled);
   }
 
   private customizeVersions(autoResource: Resource): Resource {
@@ -214,6 +230,10 @@ export class AwsOpentelemetryConfigurator {
       textMapPropagator: this.propagator,
     };
 
+    if (this.metricReader) {
+      config.metricReader = this.metricReader;
+    }
+
     return config;
   }
 
@@ -226,6 +246,19 @@ export class AwsOpentelemetryConfigurator {
     return isApplicationSignalsEnabled.toLowerCase() === 'true';
   }
 
+  static geMetricExportInterval(): number {
+    let exportIntervalMillis: number = Number(process.env[METRIC_EXPORT_INTERVAL_CONFIG]);
+    diag.debug(`AWS Application Signals Metrics export interval: ${exportIntervalMillis}`);
+
+    if (isNaN(exportIntervalMillis) || exportIntervalMillis.valueOf() > DEFAULT_METRIC_EXPORT_INTERVAL_MILLIS) {
+      exportIntervalMillis = DEFAULT_METRIC_EXPORT_INTERVAL_MILLIS;
+
+      diag.info(`AWS Application Signals metrics export interval capped to ${exportIntervalMillis}`);
+    }
+
+    return exportIntervalMillis;
+  }
+
   static exportUnsampledSpanForAgentObservability(spanProcessors: SpanProcessor[], resource: Resource): void {
     if (!isAgentObservabilityEnabled()) {
       return;
@@ -274,22 +307,13 @@ export class AwsOpentelemetryConfigurator {
 
     diag.info('AWS Application Signals enabled.');
 
-    let exportIntervalMillis: number = Number(process.env[METRIC_EXPORT_INTERVAL_CONFIG]);
-    diag.debug(`AWS Application Signals Metrics export interval: ${exportIntervalMillis}`);
-
-    if (isNaN(exportIntervalMillis) || exportIntervalMillis.valueOf() > DEFAULT_METRIC_EXPORT_INTERVAL_MILLIS) {
-      exportIntervalMillis = DEFAULT_METRIC_EXPORT_INTERVAL_MILLIS;
-
-      diag.info(`AWS Application Signals metrics export interval capped to ${exportIntervalMillis}`);
-    }
-
     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
@@ -325,6 +349,18 @@ export class AwsOpentelemetryConfigurator {
     }
   }
 
+  private customizeMetricReader(isEmfEnabled: boolean) {
+    if (isEmfEnabled) {
+      const emfExporter = createEmfExporter();
+      if (emfExporter) {
+        const periodicExportingMetricReader = new PeriodicExportingMetricReader({
+          exporter: emfExporter,
+        });
+        this.metricReader = periodicExportingMetricReader;
+      }
+    }
+  }
+
   static customizeSampler(sampler: Sampler): Sampler {
     if (AwsOpentelemetryConfigurator.isApplicationSignalsEnabled()) {
       return AlwaysRecordSampler.create(sampler);
@@ -472,7 +508,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();
 
@@ -489,19 +525,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();
     }
@@ -691,6 +727,8 @@ function getSamplerProbabilityFromEnv(environment: Required<ENVIRONMENT>): numbe
   return probability;
 }
 
+// END The OpenTelemetry Authors code
+
 function getSpanExportBatchSize() {
   if (isLambdaEnvironment()) {
     return LAMBDA_SPAN_EXPORT_BATCH_SIZE;
@@ -711,8 +749,98 @@ 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(): AWSCloudWatchEMFExporter | undefined {
+  const headersResult = validateAndFetchLogsHeader();
+  if (!headersResult.isValid) {
+    return undefined;
+  }
+
+  return new AWSCloudWatchEMFExporter(headersResult.namespace, headersResult.logGroup, headersResult.logStream);
+}
+
+/**
+ * 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 validateAndFetchLogsHeader(): OtlpLogHeaderSetting {
+  const logHeaders = process.env.OTEL_EXPORTER_OTLP_LOGS_HEADERS;
+
+  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 {
+      logGroup: '',
+      logStream: '',
+      namespace: '',
+      isValid: false,
+    };
+  }
+
+  let logGroup: string | undefined = undefined;
+  let logStream: string | undefined = undefined;
+  let namespace: string | undefined = undefined;
+  let filteredLogHeadersCount: number = 0;
+
+  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;
+        filteredLogHeadersCount++;
+      } else if (key === AWS_OTLP_LOGS_STREAM_HEADER && value !== '') {
+        logStream = value;
+        filteredLogHeadersCount++;
+      } else if (key === AWS_EMF_METRICS_NAMESPACE && value !== '') {
+        namespace = value;
+      }
+    }
+  }
+
+  const isValid = filteredLogHeadersCount === 2 && !!logGroup && !!logStream;
+  if (!isValid) {
+    diag.warn(
+      'Improper configuration: The environment variable OTEL_EXPORTER_OTLP_LOGS_HEADERS has invalid value(s) for x-aws-log-group or x-aws-log-stream'
+    );
+  }
+
+  return {
+    logGroup: logGroup as string,
+    logStream: logStream as string,
+    namespace: namespace,
+    isValid: isValid,
+  };
+}