feat: [Java] EMF Exporter Implementation

CHANGELOG.md

@@ -15,6 +15,8 @@ If your change does not need a CHANGELOG entry, add the "skip changelog" label t
 
 ### Enhancements
 
+- Add CloudWatch EMF metrics exporter with auto instrumentation configuration 
+  ([#1209](https://github.com/aws-observability/aws-otel-java-instrumentation/pull/1209))
 - Support X-Ray Trace Id extraction from Lambda Context object, and respect user-configured OTEL_PROPAGATORS in AWS Lamdba instrumentation
   ([#1191](https://github.com/aws-observability/aws-otel-java-instrumentation/pull/1191)) ([#1218](https://github.com/aws-observability/aws-otel-java-instrumentation/pull/1218))
 - Adaptive Sampling improvements: Ensure propagation of sampling rule across services and AWS accounts. Remove unnecessary B3 propagator.

awsagentprovider/build.gradle.kts

@@ -53,6 +53,8 @@ dependencies {
   runtimeOnly("software.amazon.awssdk:sts")
   implementation("software.amazon.awssdk:auth")
   implementation("software.amazon.awssdk:http-auth-aws")
+  // For EMF exporter
+  implementation("software.amazon.awssdk:cloudwatchlogs")
 
   testImplementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure")
   testImplementation("io.opentelemetry:opentelemetry-sdk-testing")

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsApplicationSignalsConfigValidator.java → awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsApplicationSignalsConfigUtils.java

@@ -18,16 +18,45 @@
 import static software.amazon.opentelemetry.javaagent.providers.AwsApplicationSignalsCustomizerProvider.*;
 
 import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
-import java.util.Arrays;
+import java.util.*;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.regex.Pattern;
 
 /** Utilities class to validate ADOT environment variable configuration. */
-public final class AwsApplicationSignalsConfigValidator {
+public final class AwsApplicationSignalsConfigUtils {
   private static final Logger logger =
       Logger.getLogger(AwsApplicationSignalsCustomizerProvider.class.getName());
 
+  /**
+   * Removes "awsemf" from OTEL_METRICS_EXPORTER if present to prevent validation errors from OTel
+   * dependencies which would try to load metric exporters. We will contribute emf exporter to
+   * upstream for supporting OTel metrics in SDK
+   *
+   * @param configProps the configuration properties
+   * @return Optional containing string with "awsemf" removed if the original OTEL_METRICS_EXPORTER
+   *     contains "awsemf", otherwise empty Optional if "awsemf" is not found
+   */
+  static Optional<String> removeEmfExporterIfEnabled(ConfigProperties configProps) {
+    String metricExporters = configProps.getString(OTEL_METRICS_EXPORTER);
+
+    if (metricExporters == null || !metricExporters.contains("awsemf")) {
+      return Optional.empty();
+    }
+
+    String[] exporters = metricExporters.split(",");
+    List<String> filtered =
+        Arrays.stream(exporters)
+            .map(String::trim)
+            .filter(exp -> !exp.equals("awsemf"))
+            .collect(java.util.stream.Collectors.toList());
+
+    // Return empty string instead of "none" because upstream will not call
+    // customizeMetricExporter if OTEL_METRICS_EXPORTER is set to "none" as it assumes
+    // no metrics exporter is configured
+    return Optional.of(filtered.isEmpty() ? "" : String.join(",", filtered));
+  }
+
   /**
    * Is the given configuration correct to enable SigV4 for Logs?
    *
@@ -61,27 +90,21 @@ static boolean isSigV4EnabledLogs(ConfigProperties config) {
 
     if (logsHeaders == null || logsHeaders.isEmpty()) {
       logger.warning(
-          "Improper configuration: Please configure the environment variable OTEL_EXPORTER_OTLP_LOGS_HEADERS to include x-aws-log-group and x-aws-log-stream");
+          String.format(
+              "Improper configuration: Please configure the environment variable OTEL_EXPORTER_OTLP_LOGS_HEADERS to include %s and %s",
+              AWS_OTLP_LOGS_GROUP_HEADER, AWS_OTLP_LOGS_STREAM_HEADER));
 
       return false;
     }
+    Map<String, String> parsedHeaders =
+        AwsApplicationSignalsConfigUtils.parseOtlpHeaders(logsHeaders);
 
-    long filteredLogHeaders =
-        Arrays.stream(logsHeaders.split(","))
-            .filter(
-                pair -> {
-                  if (pair.contains("=")) {
-                    String key = pair.split("=", 2)[0];
-                    return key.equals(AWS_OTLP_LOGS_GROUP_HEADER)
-                        || key.equals(AWS_OTLP_LOGS_STREAM_HEADER);
-                  }
-                  return false;
-                })
-            .count();
-
-    if (filteredLogHeaders != 2) {
+    if (!(parsedHeaders.containsKey(AWS_OTLP_LOGS_GROUP_HEADER)
+        && parsedHeaders.containsKey(AWS_OTLP_LOGS_STREAM_HEADER))) {
       logger.warning(
-          "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");
+          String.format(
+              "Improper configuration: Please configure the environment variable OTEL_EXPORTER_OTLP_LOGS_HEADERS to have values for %s and %s",
+              AWS_OTLP_LOGS_GROUP_HEADER, AWS_OTLP_LOGS_STREAM_HEADER));
       return false;
     }
 
@@ -168,4 +191,26 @@ private static boolean isSigv4ValidConfig(
 
     return false;
   }
+
+  /**
+   * Parse OTLP headers and return a map of header key to value. See: <a
+   * href="https://opentelemetry.io/docs/languages/sdk-configuration/otlp-exporter/#otel_exporter_otlp_headers">...</a>
+   *
+   * @param headersString the headers string in format "key1=value1,key2=value2"
+   * @return map of header keys to values
+   */
+  static Map<String, String> parseOtlpHeaders(String headersString) {
+    Map<String, String> headers = new HashMap<>();
+    if (headersString == null || headersString.isEmpty()) {
+      return headers;
+    }
+
+    for (String pair : headersString.split(",")) {
+      if (pair.contains("=")) {
+        String[] keyValue = pair.split("=", 2);
+        headers.put(keyValue[0].trim(), keyValue[1].trim());
+      }
+    }
+    return headers;
+  }
 }

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsApplicationSignalsCustomizerProvider.java

@@ -65,6 +65,8 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.concurrent.Immutable;
+import software.amazon.opentelemetry.javaagent.providers.exporter.aws.metrics.AwsCloudWatchEmfExporter;
+import software.amazon.opentelemetry.javaagent.providers.exporter.aws.metrics.ConsoleEmfExporter;
 import software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.logs.OtlpAwsLogsExporterBuilder;
 import software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.traces.OtlpAwsSpanExporterBuilder;
 
@@ -86,6 +88,9 @@
 @Immutable
 public final class AwsApplicationSignalsCustomizerProvider
     implements AutoConfigurationCustomizerProvider {
+  // https://docs.aws.amazon.com/cli/v1/userguide/cli-configure-envvars.html
+  static final String AWS_REGION = "aws.region";
+  static final String AWS_DEFAULT_REGION = "aws.default.region";
   static final String AWS_LAMBDA_FUNCTION_NAME_CONFIG = "AWS_LAMBDA_FUNCTION_NAME";
   static final String LAMBDA_APPLICATION_SIGNALS_REMOTE_ENVIRONMENT =
       "LAMBDA_APPLICATION_SIGNALS_REMOTE_ENVIRONMENT";
@@ -103,6 +108,7 @@ public final class AwsApplicationSignalsCustomizerProvider
   // https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-OTLPEndpoint.html#CloudWatch-LogsEndpoint
   static final String AWS_OTLP_LOGS_GROUP_HEADER = "x-aws-log-group";
   static final String AWS_OTLP_LOGS_STREAM_HEADER = "x-aws-log-stream";
+  static final String AWS_EMF_METRICS_NAMESPACE = "x-aws-metric-namespace";
 
   private static final String DEPRECATED_SMP_ENABLED_CONFIG = "otel.smp.enabled";
   private static final String DEPRECATED_APP_SIGNALS_ENABLED_CONFIG =
@@ -132,7 +138,7 @@ public final class AwsApplicationSignalsCustomizerProvider
   private static final String OTEL_BSP_MAX_EXPORT_BATCH_SIZE_CONFIG =
       "otel.bsp.max.export.batch.size";
 
-  private static final String OTEL_METRICS_EXPORTER = "otel.metrics.exporter";
+  static final String OTEL_METRICS_EXPORTER = "otel.metrics.exporter";
   static final String OTEL_LOGS_EXPORTER = "otel.logs.exporter";
   static final String OTEL_TRACES_EXPORTER = "otel.traces.exporter";
   static final String OTEL_EXPORTER_OTLP_TRACES_PROTOCOL = "otel.exporter.otlp.traces.protocol";
@@ -161,6 +167,7 @@ public final class AwsApplicationSignalsCustomizerProvider
   private static final int LAMBDA_SPAN_EXPORT_BATCH_SIZE = 10;
 
   private Sampler sampler;
+  private boolean isEmfExporterEnabled = false;
 
   public void customize(AutoConfigurationCustomizer autoConfiguration) {
     autoConfiguration.addPropertiesCustomizer(this::customizeProperties);
@@ -171,6 +178,15 @@ public void customize(AutoConfigurationCustomizer autoConfiguration) {
     autoConfiguration.addMeterProviderCustomizer(this::customizeMeterProvider);
     autoConfiguration.addSpanExporterCustomizer(this::customizeSpanExporter);
     autoConfiguration.addLogRecordExporterCustomizer(this::customizeLogsExporter);
+    autoConfiguration.addMetricExporterCustomizer(this::customizeMetricExporter);
+  }
+
+  private static Optional<String> getAwsRegionFromConfig(ConfigProperties configProps) {
+    String region = configProps.getString(AWS_REGION);
+    if (region != null) {
+      return Optional.of(region);
+    }
+    return Optional.ofNullable(configProps.getString(AWS_DEFAULT_REGION));
   }
 
   static boolean isLambdaEnvironment() {
@@ -190,10 +206,18 @@ private boolean isApplicationSignalsRuntimeEnabled(ConfigProperties configProps)
         && configProps.getBoolean(APPLICATION_SIGNALS_RUNTIME_ENABLED_CONFIG, true);
   }
 
-  private Map<String, String> customizeProperties(ConfigProperties configProps) {
+  Map<String, String> customizeProperties(ConfigProperties configProps) {
     Map<String, String> propsOverride = new HashMap<>();
     boolean isLambdaEnvironment = isLambdaEnvironment();
 
+    // Check if awsemf was specified and remove it from OTEL_METRICS_EXPORTER
+    Optional<String> filteredExporters =
+        AwsApplicationSignalsConfigUtils.removeEmfExporterIfEnabled(configProps);
+    if (filteredExporters.isPresent()) {
+      this.isEmfExporterEnabled = true;
+      propsOverride.put(OTEL_METRICS_EXPORTER, filteredExporters.get());
+    }
+
     // Enable AWS Resource Providers
     propsOverride.put(OTEL_RESOURCE_PROVIDERS_AWS_ENABLED, "true");
 
@@ -394,7 +418,6 @@ private SdkTracerProviderBuilder customizeTracerProviderBuilder(
 
   private SdkMeterProviderBuilder customizeMeterProvider(
       SdkMeterProviderBuilder sdkMeterProviderBuilder, ConfigProperties configProps) {
-
     if (isApplicationSignalsRuntimeEnabled(configProps)) {
       Set<String> registeredScopeNames = new HashSet<>(1);
       String jmxRuntimeScopeName = "io.opentelemetry.jmx";
@@ -434,7 +457,7 @@ SpanExporter customizeSpanExporter(SpanExporter spanExporter, ConfigProperties c
       }
     }
 
-    if (AwsApplicationSignalsConfigValidator.isSigV4EnabledTraces(configProps)) {
+    if (AwsApplicationSignalsConfigUtils.isSigV4EnabledTraces(configProps)) {
       // can cast here since we've checked that the configuration for OTEL_TRACES_EXPORTER is otlp
       // and OTEL_EXPORTER_OTLP_TRACES_PROTOCOL is http/protobuf
       // so the given spanExporter will be an instance of OtlpHttpSpanExporter
@@ -480,7 +503,7 @@ private boolean isOtlpSpanExporter(SpanExporter spanExporter) {
 
   LogRecordExporter customizeLogsExporter(
       LogRecordExporter logsExporter, ConfigProperties configProps) {
-    if (AwsApplicationSignalsConfigValidator.isSigV4EnabledLogs(configProps)) {
+    if (AwsApplicationSignalsConfigUtils.isSigV4EnabledLogs(configProps)) {
       // can cast here since we've checked that the configuration for OTEL_LOGS_EXPORTER is otlp and
       // OTEL_EXPORTER_OTLP_LOGS_PROTOCOL is http/protobuf
       // so the given logsExporter will be an instance of OtlpHttpLogRecorderExporter
@@ -509,6 +532,45 @@ LogRecordExporter customizeLogsExporter(
     return logsExporter;
   }
 
+  MetricExporter customizeMetricExporter(
+      MetricExporter metricExporter, ConfigProperties configProps) {
+    if (isEmfExporterEnabled) {
+      Map<String, String> headers =
+          AwsApplicationSignalsConfigUtils.parseOtlpHeaders(
+              configProps.getString(OTEL_EXPORTER_OTLP_LOGS_HEADERS));
+      Optional<String> awsRegion = getAwsRegionFromConfig(configProps);
+
+      if (awsRegion.isPresent()) {
+        String namespace = headers.get(AWS_EMF_METRICS_NAMESPACE);
+
+        if (headers.containsKey(AWS_OTLP_LOGS_GROUP_HEADER)
+            && headers.containsKey(AWS_OTLP_LOGS_STREAM_HEADER)) {
+          String logGroup = headers.get(AWS_OTLP_LOGS_GROUP_HEADER);
+          String logStream = headers.get(AWS_OTLP_LOGS_STREAM_HEADER);
+          return new AwsCloudWatchEmfExporter(namespace, logGroup, logStream, awsRegion.get());
+        }
+        if (isLambdaEnvironment()) {
+          return new ConsoleEmfExporter(namespace);
+        }
+        logger.warning(
+            String.format(
+                "Improper EMF Exporter configuration: Please configure the environment variable %s to have values for %s, %s, and %s",
+                OTEL_EXPORTER_OTLP_LOGS_HEADERS,
+                AWS_OTLP_LOGS_GROUP_HEADER,
+                AWS_OTLP_LOGS_STREAM_HEADER,
+                AWS_EMF_METRICS_NAMESPACE));
+
+      } else {
+        logger.warning(
+            String.format(
+                "Improper EMF Exporter configuration: AWS region not found in environment variables please set %s or %s",
+                AWS_REGION, AWS_DEFAULT_REGION));
+      }
+    }
+
+    return metricExporter;
+  }
+
   static AwsXrayAdaptiveSamplingConfig parseConfigString(String config)
       throws JsonProcessingException {
     if (config == null) {

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/exporter/aws/metrics/AwsCloudWatchEmfExporter.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.opentelemetry.javaagent.providers.exporter.aws.metrics;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import java.util.logging.Logger;
+import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient;
+import software.amazon.opentelemetry.javaagent.providers.exporter.aws.metrics.common.BaseEmfExporter;
+import software.amazon.opentelemetry.javaagent.providers.exporter.aws.metrics.common.emitter.CloudWatchLogsClientEmitter;
+import software.amazon.opentelemetry.javaagent.providers.exporter.aws.metrics.common.emitter.LogEventEmitter;
+
+/**
+ * EMF metrics exporter for sending data directly to CloudWatch Logs.
+ *
+ * <p>This exporter converts OTel metrics into CloudWatch EMF logs which are then sent to CloudWatch
+ * Logs. CloudWatch Logs automatically extracts the metrics from the EMF logs.
+ *
+ * <p><a
+ * href="https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html">...</a>
+ */
+public class AwsCloudWatchEmfExporter extends BaseEmfExporter<CloudWatchLogsClient> {
+  private static final Logger logger = Logger.getLogger(AwsCloudWatchEmfExporter.class.getName());
+
+  /**
+   * Initialize the CloudWatch EMF exporter.
+   *
+   * @param namespace CloudWatch namespace for metrics (default: "default")
+   * @param logGroupName CloudWatch log group name
+   * @param logStreamName CloudWatch log stream name (auto-generated if null)
+   * @param awsRegion AWS region
+   */
+  public AwsCloudWatchEmfExporter(
+      String namespace, String logGroupName, String logStreamName, String awsRegion) {
+    super(namespace, new CloudWatchLogsClientEmitter(logGroupName, logStreamName, awsRegion));
+  }
+
+  /**
+   * Initialize the CloudWatch EMF exporter with a custom emitter.
+   *
+   * @param namespace CloudWatch namespace for metrics
+   * @param emitter Custom log emitter
+   */
+  public AwsCloudWatchEmfExporter(String namespace, LogEventEmitter<CloudWatchLogsClient> emitter) {
+    super(namespace, emitter);
+  }
+
+  @Override
+  public CompletableResultCode flush() {
+    this.emitter.flushEvents();
+    logger.fine("AwsCloudWatchEmfExporter force flushes the buffered metrics");
+    return CompletableResultCode.ofSuccess();
+  }
+
+  @Override
+  public CompletableResultCode shutdown() {
+    this.flush();
+    logger.fine("AwsCloudWatchEmfExporter shutdown called");
+    return CompletableResultCode.ofSuccess();
+  }
+}