Add support for batching in PeriodicMetricReader

docs/apidiffs/current_vs_latest/opentelemetry-sdk-metrics.txt

@@ -1,2 +1,4 @@
 Comparing source compatibility of opentelemetry-sdk-metrics-1.62.0-SNAPSHOT.jar against opentelemetry-sdk-metrics-1.61.0.jar
-No changes.
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.sdk.metrics.export.PeriodicMetricReaderBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.metrics.export.PeriodicMetricReaderBuilder setMaxExportBatchSize(int)

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/MetricExportBatcher.java

@@ -0,0 +1,280 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.metrics.export;
+
+import io.opentelemetry.sdk.metrics.data.Data;
+import io.opentelemetry.sdk.metrics.data.DoublePointData;
+import io.opentelemetry.sdk.metrics.data.ExponentialHistogramData;
+import io.opentelemetry.sdk.metrics.data.ExponentialHistogramPointData;
+import io.opentelemetry.sdk.metrics.data.HistogramData;
+import io.opentelemetry.sdk.metrics.data.HistogramPointData;
+import io.opentelemetry.sdk.metrics.data.LongPointData;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.data.PointData;
+import io.opentelemetry.sdk.metrics.data.SumData;
+import io.opentelemetry.sdk.metrics.data.SummaryPointData;
+import io.opentelemetry.sdk.metrics.internal.data.ImmutableExponentialHistogramData;
+import io.opentelemetry.sdk.metrics.internal.data.ImmutableGaugeData;
+import io.opentelemetry.sdk.metrics.internal.data.ImmutableHistogramData;
+import io.opentelemetry.sdk.metrics.internal.data.ImmutableMetricData;
+import io.opentelemetry.sdk.metrics.internal.data.ImmutableSumData;
+import io.opentelemetry.sdk.metrics.internal.data.ImmutableSummaryData;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Batches metric data into multiple batches based on the maximum export batch size. This is used by
+ * the {@link PeriodicMetricReader} to batch metric data before exporting it.
+ *
+ * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
+ * at any time.
+ */
+class MetricExportBatcher {
+  private final int maxExportBatchSize;
+
+  /**
+   * Creates a new {@link MetricExportBatcher} with the given maximum export batch size.
+   *
+   * @param maxExportBatchSize The maximum number of {@link Data#getPoints()} in each export.
+   */
+  MetricExportBatcher(int maxExportBatchSize) {
+    if (maxExportBatchSize <= 0) {
+      throw new IllegalArgumentException("maxExportBatchSize must be positive");
+    }
+    this.maxExportBatchSize = maxExportBatchSize;
+  }
+
+  @Override
+  public String toString() {
+    return "MetricExportBatcher{maxExportBatchSize=" + maxExportBatchSize + "}";
+  }
+
+  /**
+   * Batches the given metric data into multiple batches based on the maximum export batch size.
+   *
+   * @param metrics The collection of metric data objects to batch based on the number of data
+   *     points they contain.
+   * @return A collection of batches of metric data.
+   */
+  Collection<Collection<MetricData>> batchMetrics(Collection<MetricData> metrics) {
+    if (metrics.isEmpty()) {
+      return Collections.emptyList();
+    }
+    Collection<Collection<MetricData>> preparedBatchesForExport = new ArrayList<>();
+    BatchState currentBatch = new BatchState(new ArrayList<>(maxExportBatchSize), 0);
+
+    // Fill active batch and split overlapping metric points if needed
+    for (MetricData metricData : metrics) {
+      MetricDataSplitOperationResult splitResult = prepareExportBatches(metricData, currentBatch);
+      preparedBatchesForExport.addAll(splitResult.getPreparedBatches());
+      currentBatch = splitResult.getLastInProgressBatch();
+    }
+
+    // Push trailing capacity block
+    if (!currentBatch.metrics.isEmpty()) {
+      preparedBatchesForExport.add(currentBatch.metrics);
+    }
+    return Collections.unmodifiableCollection(preparedBatchesForExport);
+  }
+
+  /**
+   * Prepares export batches from a single metric data object. This function only operates on a
+   * single metric data object, fills up the current batch with as many points as possible from the
+   * metric data object, and then creates new metric data objects for the remaining points.
+   *
+   * @param metricData The metric data object to split.
+   * @param currentBatch The current batch of metric data objects.
+   * @return A result containing the prepared batches and the last in-progress batch.
+   */
+  private MetricDataSplitOperationResult prepareExportBatches(
+      MetricData metricData, BatchState currentBatch) {
+    int remainingCapacityInCurrentBatch = maxExportBatchSize - currentBatch.points;
+    int totalPointsInMetricData = metricData.getData().getPoints().size();
+
+    if (remainingCapacityInCurrentBatch >= totalPointsInMetricData) {
+      currentBatch.metrics.add(metricData);
+      currentBatch.points += totalPointsInMetricData;
+      return new MetricDataSplitOperationResult(Collections.emptyList(), currentBatch);
+    } else {
+      // Remaining capacity can't hold all points, partition existing metric data object
+      List<PointData> originalPointsList = new ArrayList<>(metricData.getData().getPoints());
+      Collection<Collection<MetricData>> preparedBatches = new ArrayList<>();
+      int currentIndex = 0;
+
+      while (currentIndex < totalPointsInMetricData) {
+        int pointsToTake =
+            Math.min(totalPointsInMetricData - currentIndex, remainingCapacityInCurrentBatch);
+
+        if (pointsToTake > 0) {
+          currentBatch.metrics.add(
+              copyMetricData(metricData, originalPointsList, currentIndex, pointsToTake));
+          currentBatch.points += pointsToTake;
+          currentIndex += pointsToTake;
+          remainingCapacityInCurrentBatch -= pointsToTake;
+        }
+
+        if (remainingCapacityInCurrentBatch == 0) {
+          preparedBatches.add(currentBatch.metrics);
+          currentBatch = new BatchState(new ArrayList<>(maxExportBatchSize), 0);
+          remainingCapacityInCurrentBatch = maxExportBatchSize;
+        }
+      }
+      return new MetricDataSplitOperationResult(preparedBatches, currentBatch);
+    }
+  }
+
+  private static MetricData copyMetricData(
+      MetricData original,
+      List<PointData> originalPointsList,
+      int dataPointsOffset,
+      int dataPointsToTake) {
+    List<PointData> points =
+        Collections.unmodifiableList(
+            new ArrayList<>(
+                originalPointsList.subList(dataPointsOffset, dataPointsOffset + dataPointsToTake)));
+    return createMetricDataWithPoints(original, points);
+  }
+
+  /**
+   * Creates a new MetricData with the given points.
+   *
+   * @param original The original MetricData.
+   * @param points The points to use for the new MetricData.
+   * @return A new MetricData with the given points.
+   */
+  @SuppressWarnings("unchecked")
+  private static MetricData createMetricDataWithPoints(
+      MetricData original, Collection<PointData> points) {
+    switch (original.getType()) {
+      case DOUBLE_GAUGE:
+        return ImmutableMetricData.createDoubleGauge(
+            original.getResource(),
+            original.getInstrumentationScopeInfo(),
+            original.getName(),
+            original.getDescription(),
+            original.getUnit(),
+            ImmutableGaugeData.create((Collection<DoublePointData>) (Collection<?>) points));
+      case LONG_GAUGE:
+        return ImmutableMetricData.createLongGauge(
+            original.getResource(),
+            original.getInstrumentationScopeInfo(),
+            original.getName(),
+            original.getDescription(),
+            original.getUnit(),
+            ImmutableGaugeData.create((Collection<LongPointData>) (Collection<?>) points));
+      case DOUBLE_SUM:
+        SumData<DoublePointData> doubleSumData = original.getDoubleSumData();
+        return ImmutableMetricData.createDoubleSum(
+            original.getResource(),
+            original.getInstrumentationScopeInfo(),
+            original.getName(),
+            original.getDescription(),
+            original.getUnit(),
+            ImmutableSumData.create(
+                doubleSumData.isMonotonic(),
+                doubleSumData.getAggregationTemporality(),
+                (Collection<DoublePointData>) (Collection<?>) points));
+      case LONG_SUM:
+        SumData<LongPointData> longSumData = original.getLongSumData();
+        return ImmutableMetricData.createLongSum(
+            original.getResource(),
+            original.getInstrumentationScopeInfo(),
+            original.getName(),
+            original.getDescription(),
+            original.getUnit(),
+            ImmutableSumData.create(
+                longSumData.isMonotonic(),
+                longSumData.getAggregationTemporality(),
+                (Collection<LongPointData>) (Collection<?>) points));
+      case HISTOGRAM:
+        HistogramData histogramData = original.getHistogramData();
+        return ImmutableMetricData.createDoubleHistogram(
+            original.getResource(),
+            original.getInstrumentationScopeInfo(),
+            original.getName(),
+            original.getDescription(),
+            original.getUnit(),
+            ImmutableHistogramData.create(
+                histogramData.getAggregationTemporality(),
+                (Collection<HistogramPointData>) (Collection<?>) points));
+      case EXPONENTIAL_HISTOGRAM:
+        ExponentialHistogramData expHistogramData = original.getExponentialHistogramData();
+        return ImmutableMetricData.createExponentialHistogram(
+            original.getResource(),
+            original.getInstrumentationScopeInfo(),
+            original.getName(),
+            original.getDescription(),
+            original.getUnit(),
+            ImmutableExponentialHistogramData.create(
+                expHistogramData.getAggregationTemporality(),
+                (Collection<ExponentialHistogramPointData>) (Collection<?>) points));
+      case SUMMARY:
+        return ImmutableMetricData.createDoubleSummary(
+            original.getResource(),
+            original.getInstrumentationScopeInfo(),
+            original.getName(),
+            original.getDescription(),
+            original.getUnit(),
+            ImmutableSummaryData.create((Collection<SummaryPointData>) (Collection<?>) points));
+    }
+    throw new UnsupportedOperationException("Unsupported metric type: " + original.getType());
+  }
+
+  /**
+   * A data class to store the result of a split operation performed on a single {@link MetricData}
+   * object.
+   */
+  private static class MetricDataSplitOperationResult {
+    private final Collection<Collection<MetricData>> preparedBatches;
+    private final BatchState lastInProgressBatch;
+
+    /**
+     * Creates a new MetricDataSplitOperationResult.
+     *
+     * @param preparedBatches The collection of prepared batches of metric data for export. Each
+     *     batch of {@link MetricData} objects is guaranteed to have at most {@link
+     *     #maxExportBatchSize} points.
+     * @param lastInProgressBatch The last batch that is still in progress. This batch may have less
+     *     than {@link #maxExportBatchSize} points.
+     */
+    MetricDataSplitOperationResult(
+        Collection<Collection<MetricData>> preparedBatches, BatchState lastInProgressBatch) {
+      this.preparedBatches = preparedBatches;
+      this.lastInProgressBatch = lastInProgressBatch;
+    }
+
+    Collection<Collection<MetricData>> getPreparedBatches() {
+      return preparedBatches;
+    }
+
+    BatchState getLastInProgressBatch() {
+      return lastInProgressBatch;
+    }
+  }
+
+  /**
+   * Tracks the active batch while batching stays linear: {@code metrics} is the current export
+   * payload being assembled and {@code points} is its running point count, so callers do not need
+   * to rescan the batch on every append.
+   */
+  private static final class BatchState {
+    private final Collection<MetricData> metrics;
+    private int points;
+
+    /**
+     * Creates the mutable state for the current in-progress batch.
+     *
+     * @param metrics metric entries collected into the current export batch
+     * @param points running total of data points across {@code metrics}
+     */
+    private BatchState(Collection<MetricData> metrics, int points) {
+      this.metrics = metrics;
+      this.points = points;
+    }
+  }
+}

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReader.java

@@ -17,6 +17,7 @@
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import java.util.Collection;
+import java.util.Iterator;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -51,6 +52,7 @@ public final class PeriodicMetricReader implements MetricReader {
   private volatile CollectionRegistration collectionRegistration = CollectionRegistration.noop();
 
   @Nullable private volatile ScheduledFuture<?> scheduledFuture;
+  @Nullable private final MetricExportBatcher metricsBatcher;
 
   /**
    * Returns a new {@link PeriodicMetricReader} which exports to the {@code exporter} once every
@@ -66,10 +68,14 @@ public static PeriodicMetricReaderBuilder builder(MetricExporter exporter) {
   }
 
   PeriodicMetricReader(
-      MetricExporter exporter, long intervalNanos, ScheduledExecutorService scheduler) {
+      MetricExporter exporter,
+      long intervalNanos,
+      ScheduledExecutorService scheduler,
+      @Nullable MetricExportBatcher metricsBatcher) {
     this.exporter = exporter;
     this.intervalNanos = intervalNanos;
     this.scheduler = scheduler;
+    this.metricsBatcher = metricsBatcher;
     this.scheduled = new Scheduled();
   }
 
@@ -163,6 +169,8 @@ public String toString() {
         + exporter
         + ", intervalNanos="
         + intervalNanos
+        + ", metricsBatcher="
+        + metricsBatcher
         + '}';
   }
 
@@ -217,7 +225,47 @@ CompletableResultCode doRun() {
             exportAvailable.set(true);
             flushResult.succeed();
           } else {
-            CompletableResultCode result = exporter.export(metricData);
+            CompletableResultCode result;
+            if (metricsBatcher != null) {
+              Collection<Collection<MetricData>> batches = metricsBatcher.batchMetrics(metricData);
+              CompletableResultCode sequentialResult = new CompletableResultCode();
+              AtomicBoolean anyFailed = new AtomicBoolean(false);
+              Iterator<Collection<MetricData>> batchIterator = batches.iterator();
+
+              Runnable exportNext =
+                  new Runnable() {
+                    @Override
+                    public void run() {
+                      while (batchIterator.hasNext()) {
+                        Collection<MetricData> currentBatch = batchIterator.next();
+                        CompletableResultCode currentResult = exporter.export(currentBatch);
+                        if (currentResult.isDone()) {
+                          if (!currentResult.isSuccess()) {
+                            anyFailed.set(true);
+                          }
+                        } else {
+                          currentResult.whenComplete(
+                              () -> {
+                                if (!currentResult.isSuccess()) {
+                                  anyFailed.set(true);
+                                }
+                                this.run();
+                              });
+                          return;
+                        }
+                      }
+                      if (anyFailed.get()) {
+                        sequentialResult.fail();
+                      } else {
+                        sequentialResult.succeed();
+                      }
+                    }
+                  };
+              exportNext.run();
+              result = sequentialResult;
+            } else {
+              result = exporter.export(metricData);
+            }
             result.whenComplete(
                 () -> {
                   if (!result.isSuccess()) {