From 134e7ab83a94344374bec1909ce2a6adb4a7c42b Mon Sep 17 00:00:00 2001
From: Stefan Belk <belk.stefan@gmail.com>
Date: Thu, 13 Apr 2023 09:18:29 +0200
Subject: [PATCH] Reduces overall memory allocation in DynatraceExporterV2
 while exporting (#3766)

Previously, in applications with a lot of metrics the Dynatrace exporter collects all lines to be exported in one list before sending them in batches. This allocates quite a lot of unnecessary memory.
This changes this behavior so that batches are sent immediately and already sent lines are not kept in memory.

Co-authored-by: Stefan Belk <stefan.belk@dynatrace.com>
---
 .../dynatrace/v2/DynatraceExporterV2.java     | 54 +++++++++----------
 1 file changed, 25 insertions(+), 29 deletions(-)

diff --git a/implementations/micrometer-registry-dynatrace/src/main/java/io/micrometer/dynatrace/v2/DynatraceExporterV2.java b/implementations/micrometer-registry-dynatrace/src/main/java/io/micrometer/dynatrace/v2/DynatraceExporterV2.java
index 7ba3bd616d..634d54a63b 100644
--- a/implementations/micrometer-registry-dynatrace/src/main/java/io/micrometer/dynatrace/v2/DynatraceExporterV2.java
+++ b/implementations/micrometer-registry-dynatrace/src/main/java/io/micrometer/dynatrace/v2/DynatraceExporterV2.java
@@ -17,9 +17,9 @@
 
 import com.dynatrace.metric.util.*;
 import io.micrometer.core.instrument.*;
+import io.micrometer.core.instrument.Timer;
 import io.micrometer.core.instrument.distribution.HistogramSnapshot;
 import io.micrometer.core.instrument.distribution.ValueAtPercentile;
-import io.micrometer.core.instrument.util.AbstractPartition;
 import io.micrometer.core.instrument.util.StringUtils;
 import io.micrometer.core.ipc.http.HttpSender;
 import io.micrometer.core.util.internal.logging.InternalLogger;
@@ -33,10 +33,7 @@
 import java.net.MalformedURLException;
 import java.net.URI;
 import java.time.Instant;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import java.util.regex.Matcher;
@@ -136,12 +133,27 @@ private DimensionList parseDefaultDimensions(Map<String, String> defaultDimensio
      */
     @Override
     public void export(List<Meter> meters) {
-        // Lines that are too long to be ingested into Dynatrace, as well as lines that
-        // contain NaN or Inf values are not returned from "toMetricLines", and are
-        // therefore dropped.
-        List<String> metricLines = meters.stream().flatMap(this::toMetricLines).collect(Collectors.toList());
+        int partitionSize = Math.min(config.batchSize(), DynatraceMetricApiConstants.getPayloadLinesLimit());
+        List<String> batch = new ArrayList<>(partitionSize);
+
+        for (Meter meter : meters) {
+            // Lines that are too long to be ingested into Dynatrace, as well as lines
+            // that contain NaN or Inf values are not returned from "toMetricLines",
+            // and are therefore dropped.
+            Stream<String> metricLines = toMetricLines(meter);
+
+            metricLines.forEach(line -> {
+                batch.add(line);
+                if (batch.size() == partitionSize) {
+                    send(batch);
+                    batch.clear();
+                }
+            });
+        }
 
-        sendInBatches(metricLines);
+        if (!batch.isEmpty()) {
+            send(batch);
+        }
     }
 
     private Stream<String> toMetricLines(Meter meter) {
@@ -316,7 +328,8 @@ private void send(List<String> metricLines) {
             return;
         }
         try {
-            logger.debug("Sending {} lines to {}", metricLines.size(), endpoint);
+            int lineCount = metricLines.size();
+            logger.debug("Sending {} lines to {}", lineCount, endpoint);
 
             String body = String.join("\n", metricLines);
             logger.debug("Sending lines:\n{}", body);
@@ -329,7 +342,7 @@ private void send(List<String> metricLines) {
             requestBuilder.withHeader("User-Agent", "micrometer")
                 .withPlainText(body)
                 .send()
-                .onSuccess(response -> handleSuccess(metricLines.size(), response))
+                .onSuccess(response -> handleSuccess(lineCount, response))
                 .onError(response -> logger.error("Failed metric ingestion: Error Code={}, Response Body={}",
                         response.code(), getTruncatedBody(response)));
         }
@@ -369,21 +382,4 @@ private void handleSuccess(int totalSent, HttpSender.Response response) {
         }
     }
 
-    private void sendInBatches(List<String> metricLines) {
-        int partitionSize = Math.min(config.batchSize(), DynatraceMetricApiConstants.getPayloadLinesLimit());
-        MetricLinePartition.partition(metricLines, partitionSize).forEach(this::send);
-    }
-
-    static class MetricLinePartition extends AbstractPartition<String> {
-
-        private MetricLinePartition(List<String> list, int partitionSize) {
-            super(list, partitionSize);
-        }
-
-        static List<List<String>> partition(List<String> list, int partitionSize) {
-            return new MetricLinePartition(list, partitionSize);
-        }
-
-    }
-
 }