Add PropertiesFactory support to EmfMetricLoggingPublisher

.changes/next-release/feature-AmazonCloudWatchEMFMetricPublisher-33792c9.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "Amazon CloudWatch EMF Metric Publisher",
+    "contributor": "humanzz",
+    "description": "Add `PropertiesFactory` and `propertiesFactory` to `EmfMetricLoggingPublisher.Builder`, enabling users to enrich EMF records with custom key-value properties derived from the metric collection or ambient context, searchable in CloudWatch Logs Insights. See [#6595](https://github.com/aws/aws-sdk-java-v2/issues/6595)."
+}

metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/EmfMetricLoggingPublisher.java

@@ -82,6 +82,7 @@ private EmfMetricLoggingPublisher(Builder builder) {
             .dimensions(builder.dimensions)
             .metricLevel(builder.metricLevel)
             .metricCategories(builder.metricCategories)
+            .propertiesFactory(builder.propertiesFactory)
             .build();
 
         this.metricConverter = new MetricEmfConverter(config);
@@ -123,6 +124,7 @@ public static final class Builder {
         private Collection<SdkMetric<String>> dimensions;
         private Collection<MetricCategory> metricCategories;
         private MetricLevel metricLevel;
+        private PropertiesFactory propertiesFactory;
 
         private Builder() {
         }
@@ -217,6 +219,28 @@ public Builder metricLevel(MetricLevel metricLevel) {
         }
 
 
+        /**
+         * Configure a factory for custom properties to include in each EMF record.
+         * The factory is invoked on each {@link #publish(MetricCollection)} call with the
+         * {@link MetricCollection} being published, and the returned map entries are written
+         * as top-level key-value pairs in the EMF JSON output. These appear as searchable
+         * fields in CloudWatch Logs Insights.
+         *
+         * <p>Keys that collide with reserved EMF fields ({@code _aws}), configured
+         * dimension names, or reported metric names are silently skipped.
+         *
+         * <p>If this is not specified, no custom properties are added.
+         *
+         * @param propertiesFactory a factory returning a map of property names to values,
+         *                          or {@code null} to disable custom properties
+         * @return this builder
+         * @see PropertiesFactory
+         */
+        public Builder propertiesFactory(PropertiesFactory propertiesFactory) {
+            this.propertiesFactory = propertiesFactory;
+            return this;
+        }
+
         /**
          * Build a {@link EmfMetricLoggingPublisher} using the configuration currently configured on this publisher.
          */

metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/PropertiesFactory.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * 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.awssdk.metrics.publishers.emf;
+
+import java.util.Map;
+import software.amazon.awssdk.annotations.SdkPublicApi;
+import software.amazon.awssdk.metrics.MetricCollection;
+
+/**
+ * A factory for producing custom properties to include in each EMF record.
+ *
+ * <p>Implementations receive the {@link MetricCollection} being published, allowing properties
+ * to be derived from the SDK metrics (e.g. service endpoint, request ID) or from ambient
+ * context (e.g. Lambda request ID, trace ID).
+ *
+ * <p>The returned map entries are written as top-level key-value pairs in the EMF JSON output,
+ * making them searchable in CloudWatch Logs Insights. Keys that collide with reserved EMF
+ * fields ({@code _aws}), dimension names, or metric names are silently skipped.
+ *
+ * <p>If the factory returns {@code null} or throws an exception, no custom properties are added
+ * and a warning is logged.
+ *
+ * <p>Example using ambient context:
+ * <pre>{@code
+ * EmfMetricLoggingPublisher.builder()
+ *     .propertiesFactory(metrics -> Collections.singletonMap("RequestId", requestId))
+ *     .build();
+ * }</pre>
+ *
+ * <p>Example using metric collection values:
+ * <pre>{@code
+ * EmfMetricLoggingPublisher.builder()
+ *     .propertiesFactory(metrics -> {
+ *         Map<String, String> props = new HashMap<>();
+ *         metrics.metricValues(CoreMetric.SERVICE_ENDPOINT)
+ *                .stream().findFirst()
+ *                .ifPresent(uri -> props.put("ServiceEndpoint", uri.toString()));
+ *         return props;
+ *     })
+ *     .build();
+ * }</pre>
+ *
+ * @see EmfMetricLoggingPublisher.Builder#propertiesFactory(PropertiesFactory)
+ */
+@FunctionalInterface
+@SdkPublicApi
+public interface PropertiesFactory {
+
+    /**
+     * Create a map of custom properties to include in the EMF record for the given metric collection.
+     *
+     * @param metricCollection the SDK metric collection being published
+     * @return a map of property names to string values, or {@code null} for no custom properties
+     */
+    Map<String, String> create(MetricCollection metricCollection);
+}

metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/internal/EmfMetricConfiguration.java

@@ -26,6 +26,7 @@
 import software.amazon.awssdk.metrics.MetricCategory;
 import software.amazon.awssdk.metrics.MetricLevel;
 import software.amazon.awssdk.metrics.SdkMetric;
+import software.amazon.awssdk.metrics.publishers.emf.PropertiesFactory;
 import software.amazon.awssdk.utils.Validate;
 import software.amazon.awssdk.utils.internal.SystemSettingUtils;
 
@@ -43,13 +44,17 @@ public final class EmfMetricConfiguration {
     private final Set<SdkMetric<String>> dimensions;
     private final Collection<MetricCategory> metricCategories;
     private final MetricLevel metricLevel;
+    private final PropertiesFactory propertiesFactory;
 
     private EmfMetricConfiguration(Builder builder) {
         this.namespace = builder.namespace == null ? DEFAULT_NAMESPACE : builder.namespace;
         this.logGroupName = Validate.paramNotNull(resolveLogGroupName(builder), "logGroupName");
         this.dimensions = builder.dimensions == null ? DEFAULT_DIMENSIONS : new HashSet<>(builder.dimensions);
         this.metricCategories = builder.metricCategories == null ? DEFAULT_CATEGORIES : new HashSet<>(builder.metricCategories);
         this.metricLevel = builder.metricLevel == null ? DEFAULT_METRIC_LEVEL : builder.metricLevel;
+        this.propertiesFactory = builder.propertiesFactory == null
+            ? mc -> Collections.emptyMap()
+            : builder.propertiesFactory;
     }
 
 
@@ -59,6 +64,7 @@ public static class Builder {
         private Collection<SdkMetric<String>> dimensions;
         private Collection<MetricCategory> metricCategories;
         private MetricLevel metricLevel;
+        private PropertiesFactory propertiesFactory;
 
         public Builder namespace(String namespace) {
             this.namespace = namespace;
@@ -85,6 +91,11 @@ public Builder metricLevel(MetricLevel metricLevel) {
             return this;
         }
 
+        public Builder propertiesFactory(PropertiesFactory propertiesFactory) {
+            this.propertiesFactory = propertiesFactory;
+            return this;
+        }
+
         public EmfMetricConfiguration build() {
             return new EmfMetricConfiguration(this);
         }
@@ -110,6 +121,10 @@ public MetricLevel metricLevel() {
         return metricLevel;
     }
 
+    public PropertiesFactory propertiesFactory() {
+        return propertiesFactory;
+    }
+
     private String resolveLogGroupName(Builder builder) {
         return builder.logGroupName != null ? builder.logGroupName :
                SystemSettingUtils.resolveEnvironmentVariable("AWS_LAMBDA_LOG_GROUP_NAME").orElse(null);

metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/internal/MetricEmfConverter.java

@@ -19,7 +19,9 @@
 import java.time.Clock;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -31,6 +33,7 @@
 import software.amazon.awssdk.metrics.MetricCollection;
 import software.amazon.awssdk.metrics.MetricRecord;
 import software.amazon.awssdk.metrics.SdkMetric;
+import software.amazon.awssdk.metrics.publishers.emf.PropertiesFactory;
 import software.amazon.awssdk.protocols.jsoncore.JsonWriter;
 import software.amazon.awssdk.utils.Logger;
 import software.amazon.awssdk.utils.MetricValueNormalizer;
@@ -61,17 +64,21 @@ public class MetricEmfConverter {
      */
     private static final int MAX_METRIC_NUM = 100;
 
+    private static final String AWS_METADATA_KEY = "_aws";
+
     private static final Logger logger = Logger.loggerFor(MetricEmfConverter.class);
     private final List<String> dimensions = new ArrayList<>();
     private final EmfMetricConfiguration config;
     private final boolean metricCategoriesContainsAll;
     private final Clock clock;
+    private final PropertiesFactory propertiesFactory;
 
     @SdkTestInternalApi
     public MetricEmfConverter(EmfMetricConfiguration config, Clock clock) {
         this.config = config;
         this.clock = clock;
         this.metricCategoriesContainsAll = config.metricCategories().contains(MetricCategory.ALL);
+        this.propertiesFactory = config.propertiesFactory();
     }
 
     public MetricEmfConverter(EmfMetricConfiguration config) {
@@ -136,7 +143,18 @@ public List<String> convertMetricCollectionToEmf(MetricCollection metricCollecti
             }
         }
 
-        return createEmfStrings(aggregatedMetrics);
+        Map<String, String> properties = resolveProperties(metricCollection);
+        return createEmfStrings(aggregatedMetrics, properties);
+    }
+
+    private Map<String, String> resolveProperties(MetricCollection metricCollection) {
+        try {
+            Map<String, String> result = propertiesFactory.create(metricCollection);
+            return result == null ? Collections.emptyMap() : result;
+        } catch (Exception e) {
+            logger.warn(() -> "Properties factory threw an exception, publishing without custom properties", e);
+            return Collections.emptyMap();
+        }
     }
 
     /**
@@ -188,7 +206,8 @@ private void processAndWriteValue(JsonWriter jsonWriter, MetricRecord<?> mRecord
         }
     }
 
-    private List<String> createEmfStrings(Map<SdkMetric<?>, List<MetricRecord<?>>> aggregatedMetrics) {
+    private List<String> createEmfStrings(Map<SdkMetric<?>, List<MetricRecord<?>>> aggregatedMetrics,
+                                          Map<String, String> properties) {
         List<String> emfStrings = new ArrayList<>();
         Map<SdkMetric<?>, List<MetricRecord<?>>> currentMetricBatch = new HashMap<>();
 
@@ -204,35 +223,55 @@ private List<String> createEmfStrings(Map<SdkMetric<?>, List<MetricRecord<?>>> a
             }
 
             if (currentMetricBatch.size() == MAX_METRIC_NUM) {
-                emfStrings.add(createEmfString(currentMetricBatch));
+                emfStrings.add(createEmfString(currentMetricBatch, properties));
                 currentMetricBatch = new HashMap<>();
             }
 
             currentMetricBatch.put(metric, records);
         }
 
-        emfStrings.add(createEmfString(currentMetricBatch));
+        emfStrings.add(createEmfString(currentMetricBatch, properties));
 
         return emfStrings;
     }
 
 
-    private String createEmfString(Map<SdkMetric<?>, List<MetricRecord<?>>> metrics) {
+    private String createEmfString(Map<SdkMetric<?>, List<MetricRecord<?>>> metrics,
+                                    Map<String, String> properties) {
 
         JsonWriter jsonWriter = JsonWriter.create();
         jsonWriter.writeStartObject();
 
         writeAwsObject(jsonWriter, metrics.keySet());
         writeMetricValues(jsonWriter, metrics);
+        writeCustomProperties(jsonWriter, properties, metrics.keySet());
 
         jsonWriter.writeEndObject();
 
         return new String(jsonWriter.getBytes(), StandardCharsets.UTF_8);
 
     }
 
+    private void writeCustomProperties(JsonWriter jsonWriter, Map<String, String> properties,
+                                        Set<SdkMetric<?>> metrics) {
+        if (properties.isEmpty()) {
+            return;
+        }
+        Set<String> reservedKeys = new HashSet<>();
+        reservedKeys.add(AWS_METADATA_KEY);
+        for (SdkMetric<?> metric : metrics) {
+            reservedKeys.add(metric.name());
+        }
+        for (Map.Entry<String, String> entry : properties.entrySet()) {
+            if (!reservedKeys.contains(entry.getKey())) {
+                jsonWriter.writeFieldName(entry.getKey());
+                jsonWriter.writeValue(entry.getValue());
+            }
+        }
+    }
+
     private void writeAwsObject(JsonWriter jsonWriter, Set<SdkMetric<?>> metricNames) {
-        jsonWriter.writeFieldName("_aws");
+        jsonWriter.writeFieldName(AWS_METADATA_KEY);
         jsonWriter.writeStartObject();
 
         jsonWriter.writeFieldName("Timestamp");