Handle log context with more than 100 metrics

src/main/java/software/amazon/cloudwatchlogs/emf/Constants.java

@@ -20,4 +20,6 @@ public class Constants {
     public static final int DEFAULT_AGENT_PORT = 25888;
 
     public static final String UNKNOWN = "Unknown";
+
+    public static final int MAX_METRICS_PER_EVENT = 100;
 }

src/main/java/software/amazon/cloudwatchlogs/emf/model/Metadata.java

@@ -26,12 +26,15 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.Setter;
+import lombok.With;
 import software.amazon.cloudwatchlogs.emf.serializers.InstantDeserializer;
 import software.amazon.cloudwatchlogs.emf.serializers.InstantSerializer;
 
 /** Represents the MetaData part of the EMF schema. */
+@AllArgsConstructor
 class Metadata {
 
     @Getter
@@ -44,6 +47,7 @@ class Metadata {
 
     @Getter
     @Setter
+    @With
     @JsonProperty("CloudWatchMetrics")
     private List<MetricDirective> cloudWatchMetrics;
 

src/main/java/software/amazon/cloudwatchlogs/emf/model/MetricDefinition.java

@@ -16,33 +16,48 @@
 
 package software.amazon.cloudwatchlogs.emf.model;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NonNull;
-import lombok.Setter;
 import software.amazon.cloudwatchlogs.emf.serializers.UnitDeserializer;
 import software.amazon.cloudwatchlogs.emf.serializers.UnitSerializer;
 
 /** Represents the MetricDefinition of the EMF schema. */
 @AllArgsConstructor
 class MetricDefinition {
     @NonNull
-    @Setter
     @Getter
     @JsonProperty("Name")
     private String name;
 
-    @Setter
     @Getter
     @JsonProperty("Unit")
     @JsonSerialize(using = UnitSerializer.class)
     @JsonDeserialize(using = UnitDeserializer.class)
     private Unit unit;
 
+    @JsonIgnore @NonNull @Getter private List<Double> values;
+
     MetricDefinition(String name) {
-        this(name, Unit.NONE);
+        this(name, Unit.NONE, new ArrayList<>());
+    }
+
+    MetricDefinition(String name, double value) {
+        this(name, Unit.NONE, value);
+    }
+
+    MetricDefinition(String name, Unit unit, double value) {
+        this(name, unit, new ArrayList<>(Arrays.asList(value)));
+    }
+
+    void addValue(double value) {
+        values.add(value);
     }
 }

src/main/java/software/amazon/cloudwatchlogs/emf/model/MetricDirective.java

@@ -16,43 +16,58 @@
 
 package software.amazon.cloudwatchlogs.emf.model;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
-import lombok.AccessLevel;
-import lombok.Getter;
-import lombok.Setter;
+import lombok.*;
 
 /** Represents the MetricDirective part of the EMF schema. */
+@AllArgsConstructor
 class MetricDirective {
     @Setter
     @Getter
     @JsonProperty("Namespace")
-    private String namespace = "aws-embedded-metrics";
+    private String namespace;
 
-    @Setter
-    @Getter
-    @JsonProperty("Metrics")
-    private List<MetricDefinition> metrics = new ArrayList<>();
+    @JsonIgnore @Setter @Getter @With private Map<String, MetricDefinition> metrics;
 
     @Getter(AccessLevel.PROTECTED)
-    private List<DimensionSet> dimensions = new ArrayList<>();
+    private List<DimensionSet> dimensions;
 
     @Setter
     @Getter(AccessLevel.PROTECTED)
-    private DimensionSet defaultDimensions = new DimensionSet();
+    private DimensionSet defaultDimensions;
 
-    private boolean shouldUseDefaultDimension = true;
+    private boolean shouldUseDefaultDimension;
+
+    MetricDirective() {
+        namespace = "aws-embedded-metrics";
+        metrics = new HashMap<>();
+        dimensions = new ArrayList<>();
+        defaultDimensions = new DimensionSet();
+        shouldUseDefaultDimension = true;
+    }
 
     void putDimensionSet(DimensionSet dimensionSet) {
         dimensions.add(dimensionSet);
     }
 
-    void putMetric(MetricDefinition metric) {
-        metrics.add(metric);
+    void putMetric(String key, double value) {
+        putMetric(key, value, Unit.NONE);
+    }
+
+    void putMetric(String key, double value, Unit unit) {
+        if (metrics.containsKey(key)) {
+            metrics.get(key).addValue(value);
+        } else {
+            metrics.put(key, new MetricDefinition(key, unit, value));
+        }
+    }
+
+    @JsonProperty("Metrics")
+    Collection<MetricDefinition> getAllMetrics() {
+        return metrics.values();
     }
 
     @JsonProperty("Dimensions")

src/main/java/software/amazon/cloudwatchlogs/emf/model/MetricsContext.java

@@ -17,10 +17,9 @@
 package software.amazon.cloudwatchlogs.emf.model;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import lombok.Getter;
+import software.amazon.cloudwatchlogs.emf.Constants;
 
 /** Stores metrics and their associated properties and dimensions. */
 public class MetricsContext {
@@ -101,8 +100,7 @@ public boolean hasDefaultDimensions() {
      * @param unit The unit of the metric
      */
     public void putMetric(String key, double value, Unit unit) {
-        metricDirective.putMetric(new MetricDefinition(key, unit));
-        rootNode.putMetric(key, value);
+        metricDirective.putMetric(key, value, unit);
     }
 
     /**
@@ -201,12 +199,38 @@ public MetricsContext createCopyWithContext() {
     }
 
     /**
-     * Serialize the metrics in this context to a string.
+     * Serialize the metrics in this context to strings.
      *
-     * @return the serialized string
+     * @return the serialized strings.
      * @throws JsonProcessingException if there's any object that cannot be serialized
      */
-    public String serialize() throws JsonProcessingException {
-        return this.rootNode.serialize();
+    public List<String> serialize() throws JsonProcessingException {
+        if (rootNode.metrics().size() <= Constants.MAX_METRICS_PER_EVENT) {
+            return Arrays.asList(this.rootNode.serialize());
+        } else {
+            List<RootNode> nodes = new ArrayList<>();
+            Map<String, MetricDefinition> metrics = new HashMap<>();
+            int count = 0;
+            for (MetricDefinition metric : rootNode.metrics().values()) {
+                metrics.put(metric.getName(), metric);
+                count++;
+                if (metrics.size() == Constants.MAX_METRICS_PER_EVENT
+                        || count == rootNode.metrics().size()) {
+                    Metadata metadata = rootNode.getAws();
+                    MetricDirective metricDirective = metadata.getCloudWatchMetrics().get(0);
+                    Metadata clonedMetadata =
+                            metadata.withCloudWatchMetrics(
+                                    Arrays.asList(metricDirective.withMetrics(metrics)));
+                    nodes.add(rootNode.withAws(clonedMetadata));
+                    metrics = new HashMap<>();
+                }
+            }
+
+            List<String> strings = new ArrayList<>();
+            for (RootNode node : nodes) {
+                strings.add(node.serialize());
+            }
+            return strings;
+        }
     }
 }

src/main/java/software/amazon/cloudwatchlogs/emf/model/RootNode.java

@@ -22,44 +22,39 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
+import lombok.AllArgsConstructor;
 import lombok.Getter;
+import lombok.With;
 
 /** Represents the root of the EMF schema. */
+@AllArgsConstructor
 @JsonFilter("emptyMetricFilter")
 class RootNode {
     @Getter
+    @With
     @JsonProperty("_aws")
-    private Metadata aws = new Metadata();
+    private Metadata aws;
 
-    private Map<String, Object> properties = new HashMap<>();
-    private Map<String, List<Double>> metrics = new HashMap<>();
-    private ObjectMapper objectMapper = new ObjectMapper();
+    private Map<String, Object> properties;
+    private ObjectMapper objectMapper;
 
     RootNode() {
         final SimpleFilterProvider filterProvider =
                 new SimpleFilterProvider().addFilter("emptyMetricFilter", new EmptyMetricsFilter());
+        aws = new Metadata();
+        properties = new HashMap<>();
+        objectMapper = new ObjectMapper();
         objectMapper.setFilterProvider(filterProvider);
     }
 
     public void putProperty(String key, Object value) {
         properties.put(key, value);
     }
 
-    /**
-     * Add a metric measurement. Multiple calls using the same key will be stored as an array of
-     * scalar values
-     */
-    void putMetric(String key, double value) {
-        if (!metrics.containsKey(key)) {
-            metrics.put(key, new ArrayList<>());
-        }
-        metrics.get(key).add(value);
-    }
-
     Map<String, Object> getProperties() {
         return properties;
     }
@@ -70,9 +65,13 @@ Map<String, Object> getTargetMembers() {
         Map<String, Object> targetMembers = new HashMap<>();
         targetMembers.putAll(properties);
         targetMembers.putAll(getDimensions());
-        for (Map.Entry<String, List<Double>> entry : metrics.entrySet()) {
-            List<Double> values = entry.getValue();
-            targetMembers.put(entry.getKey(), values.size() == 1 ? values.get(0) : values);
+        List<MetricDefinition> metrics =
+                aws.getCloudWatchMetrics().stream()
+                        .flatMap(metricDirective -> metricDirective.getMetrics().values().stream())
+                        .collect(Collectors.toList());
+        for (MetricDefinition metric : metrics) {
+            List<Double> values = metric.getValues();
+            targetMembers.put(metric.getName(), values.size() == 1 ? values.get(0) : values);
         }
         return targetMembers;
     }
@@ -88,6 +87,10 @@ Map<String, String> getDimensions() {
         return dimensions;
     }
 
+    Map<String, MetricDefinition> metrics() {
+        return aws.getCloudWatchMetrics().get(0).getMetrics();
+    }
+
     String serialize() throws JsonProcessingException {
         return objectMapper.writeValueAsString(this);
     }

src/main/java/software/amazon/cloudwatchlogs/emf/serializers/InstantSerializer.java

@@ -37,7 +37,6 @@ public class InstantSerializer extends StdSerializer<Instant> {
     public void serialize(Instant value, JsonGenerator jgen, SerializerProvider provider)
             throws IOException, JsonProcessingException {
 
-        // Just serialize dimensions as an array.
         jgen.writeNumber(value.toEpochMilli());
     }
 }

src/main/java/software/amazon/cloudwatchlogs/emf/sinks/AgentSink.java

@@ -48,7 +48,9 @@ public void accept(MetricsContext context) {
         }
 
         try {
-            client.sendMessage(context.serialize() + "\n");
+            for (String event : context.serialize()) {
+                client.sendMessage(event + "\n");
+            }
         } catch (JsonProcessingException e) {
             log.error("Failed to serialize the metrics with the exception: ", e);
         }

src/main/java/software/amazon/cloudwatchlogs/emf/sinks/ConsoleSink.java

@@ -33,7 +33,9 @@ public void accept(MetricsContext context) {
 
         try {
             // CHECKSTYLE OFF
-            System.out.println(context.serialize());
+            for (String event : context.serialize()) {
+                System.out.println(event);
+            }
             // CHECKSTYLE ON
         } catch (JsonProcessingException e) {
             log.error("Failed to serialize a MetricsContext: ", e);

src/test/java/software/amazon/cloudwatchlogs/emf/model/MetricDefinitionTest.java

@@ -20,6 +20,7 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Arrays;
 import org.junit.Test;
 
 public class MetricDefinitionTest {
@@ -41,9 +42,18 @@ public void testSerializeMetricDefinitionWithoutUnit() throws JsonProcessingExce
     @Test
     public void testSerializeMetricDefinition() throws JsonProcessingException {
         ObjectMapper objectMapper = new ObjectMapper();
-        MetricDefinition metricDefinition = new MetricDefinition("Time", Unit.MILLISECONDS);
+        MetricDefinition metricDefinition = new MetricDefinition("Time", Unit.MILLISECONDS, 10);
         String metricString = objectMapper.writeValueAsString(metricDefinition);
 
         assertEquals(metricString, "{\"Name\":\"Time\",\"Unit\":\"Milliseconds\"}");
     }
+
+    @Test
+    public void testAddValue() {
+        MetricDefinition md = new MetricDefinition("Time", Unit.MICROSECONDS, 10);
+        assertEquals(Arrays.asList(10d), md.getValues());
+
+        md.addValue(20);
+        assertEquals(Arrays.asList(10d, 20d), md.getValues());
+    }
 }