Add support for batching in PeriodicMetricReader

CHANGELOG.md

@@ -20,6 +20,8 @@
 
 #### Incubating
 
+* Add support for configuring `setMaxExportBatchSize` in `PeriodicMetricReader` ([#8296](https://github.com/open-telemetry/opentelemetry-java/pull/8296))
+
 * **BREAKING** Update `EnvironmentGetter` and `EnvironmentSetter` key normalization to reflect spec
   changes
   ([#8233](https://github.com/open-telemetry/opentelemetry-java/pull/8233))

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

@@ -0,0 +1,187 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.metrics.export;
+
+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 MetricExportBatcher() {}
+
+  private static void validateMaxExportBatchSize(int maxExportBatchSize) {
+    if (maxExportBatchSize <= 0) {
+      throw new IllegalArgumentException("maxExportBatchSize must be positive");
+    }
+  }
+
+  /**
+   * 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.
+   */
+  static Collection<Collection<MetricData>> batchMetrics(
+      Collection<MetricData> metrics, int maxExportBatchSize) {
+    validateMaxExportBatchSize(maxExportBatchSize);
+    if (metrics.isEmpty()) {
+      return Collections.emptyList();
+    }
+    Collection<Collection<MetricData>> preparedBatchesForExport = new ArrayList<>();
+    List<MetricData> currentBatch = new ArrayList<>(maxExportBatchSize);
+    int currentPointsInBatch = 0;
+    for (MetricData metricData : metrics) {
+      int totalPointsInMetric = metricData.getData().getPoints().size();
+      if (currentPointsInBatch + totalPointsInMetric <= maxExportBatchSize) {
+        currentBatch.add(metricData);
+        currentPointsInBatch += totalPointsInMetric;
+        continue;
+      }
+      int currentIndex = 0;
+      List<PointData> originalPointsList = new ArrayList<>(metricData.getData().getPoints());
+      while (currentIndex < totalPointsInMetric) {
+        if (currentPointsInBatch == maxExportBatchSize) {
+          preparedBatchesForExport.add(currentBatch);
+          currentBatch = new ArrayList<>(maxExportBatchSize);
+          currentPointsInBatch = 0;
+        }
+        int pointsToTake =
+            Math.min(maxExportBatchSize - currentPointsInBatch, totalPointsInMetric - currentIndex);
+        currentBatch.add(
+            copyMetricData(metricData, originalPointsList, currentIndex, pointsToTake));
+        currentPointsInBatch += pointsToTake;
+        currentIndex += pointsToTake;
+      }
+    }
+    if (!currentBatch.isEmpty()) {
+      preparedBatchesForExport.add(currentBatch);
+    }
+    return Collections.unmodifiableCollection(preparedBatchesForExport);
+  }
+
+  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());
+  }
+}

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;
+  private final int maxExportBatchSize;
 
   /**
    * 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,
+      int maxExportBatchSize) {
     this.exporter = exporter;
     this.intervalNanos = intervalNanos;
     this.scheduler = scheduler;
+    this.maxExportBatchSize = maxExportBatchSize;
     this.scheduled = new Scheduled();
   }
 
@@ -163,6 +169,8 @@ public String toString() {
         + exporter
         + ", intervalNanos="
         + intervalNanos
+        + ", maxExportBatchSize="
+        + maxExportBatchSize
         + '}';
   }
 
@@ -187,13 +195,56 @@ private final class Scheduled implements Runnable {
 
     private Scheduled() {}
 
+    private CompletableResultCode exportMetrics(Collection<MetricData> metricData) {
+      if (maxExportBatchSize == 0) {
+        return exporter.export(metricData);
+      }
+      Collection<Collection<MetricData>> batches =
+          MetricExportBatcher.batchMetrics(metricData, maxExportBatchSize);
+      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();
+      return sequentialResult;
+    }
+
     void setMeterProvider(MeterProvider meterProvider) {
       instrumentation = new MetricReaderInstrumentation(COMPONENT_ID, meterProvider);
     }
 
     @Override
     public void run() {
-      // Ignore the CompletableResultCode from doRun() in order to keep run() asynchronous
+      // Ignore the CompletableResultCode from doRun() in order to keep run()
+      // asynchronous
       doRun();
     }
 
@@ -217,11 +268,11 @@ CompletableResultCode doRun() {
             exportAvailable.set(true);
             flushResult.succeed();
           } else {
-            CompletableResultCode result = exporter.export(metricData);
+            CompletableResultCode result = exportMetrics(metricData);
             result.whenComplete(
                 () -> {
                   if (!result.isSuccess()) {
-                    logger.log(Level.FINE, "Exporter failed");
+                    logger.log(Level.WARNING, "Exporter failed");
                   }
                   exportAvailable.set(true);
                   flushResult.succeed();

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

@@ -30,6 +30,8 @@ public final class PeriodicMetricReaderBuilder {
 
   @Nullable private ScheduledExecutorService executor;
 
+  private int maxExportBatchSize;
+
   PeriodicMetricReaderBuilder(MetricExporter metricExporter) {
     this.metricExporter = metricExporter;
   }
@@ -59,13 +61,27 @@ public PeriodicMetricReaderBuilder setExecutor(ScheduledExecutorService executor
     return this;
   }
 
+  /**
+   * Sets the maximum number of data points to include in a single export batch. If unset, no
+   * batching will be performed. The maximum number of data points is considered across MetricData
+   * objects scheduled for export.
+   *
+   * @param maxExportBatchSize The maximum number of data points to include in a single export
+   *     batch.
+   */
+  PeriodicMetricReaderBuilder setMaxExportBatchSize(int maxExportBatchSize) {
+    checkArgument(maxExportBatchSize > 0, "maxExportBatchSize must be positive");
+    this.maxExportBatchSize = maxExportBatchSize;
+    return this;
+  }
+
   /** Build a {@link PeriodicMetricReader} with the configuration of this builder. */
   public PeriodicMetricReader build() {
     ScheduledExecutorService executor = this.executor;
     if (executor == null) {
       executor =
           Executors.newScheduledThreadPool(1, new DaemonThreadFactory("PeriodicMetricReader"));
     }
-    return new PeriodicMetricReader(metricExporter, intervalNanos, executor);
+    return new PeriodicMetricReader(metricExporter, intervalNanos, executor, maxExportBatchSize);
   }
 }

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/SdkMeterProviderUtil.java

@@ -76,6 +76,21 @@ public static SdkMeterProviderBuilder addMeterConfiguratorCondition(
     return sdkMeterProviderBuilder;
   }
 
+  /** Reflectively set the max export batch size for the {@link SdkMeterProviderBuilder}. */
+  public static SdkMeterProviderBuilder setMaxExportBatchSize(
+      SdkMeterProviderBuilder sdkMeterProviderBuilder, int maxExportBatchSize) {
+    try {
+      Method method =
+          SdkMeterProviderBuilder.class.getDeclaredMethod("setMaxExportBatchSize", int.class);
+      method.setAccessible(true);
+      method.invoke(sdkMeterProviderBuilder, maxExportBatchSize);
+    } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
+      throw new IllegalStateException(
+          "Error calling setMaxExportBatchSize on SdkMeterProviderBuilder", e);
+    }
+    return sdkMeterProviderBuilder;
+  }
+
   /**
    * Reflectively add an {@link AttributesProcessor} to the {@link ViewBuilder} which appends
    * key-values from baggage to all measurements.