Break out DoubleExemplarReservoir, LongExemplarReservoir, add ExemplarReservoirFactory

Author: jack-berg

docs/apidiffs/current_vs_latest/opentelemetry-common.txt

@@ -1,2 +1,2 @@
-Comparing source compatibility of opentelemetry-common-1.55.0-SNAPSHOT.jar against opentelemetry-common-1.54.1.jar
+Comparing source compatibility of opentelemetry-common-1.55.0-SNAPSHOT.jar against opentelemetry-common-1.55.0.jar
 No changes.

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

@@ -1,2 +1,2 @@
-Comparing source compatibility of opentelemetry-sdk-metrics-1.55.0-SNAPSHOT.jar against opentelemetry-sdk-metrics-1.54.1.jar
+Comparing source compatibility of opentelemetry-sdk-metrics-1.55.0-SNAPSHOT.jar against opentelemetry-sdk-metrics-1.55.0.jar
 No changes.

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/internal/aggregator/HistogramAggregationParam.java

@@ -7,7 +7,7 @@
 
 import static io.opentelemetry.sdk.common.export.MemoryMode.IMMUTABLE_DATA;
 
-import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoir;
+import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoirFactory;
 import java.util.Collections;
 
 /** The types of histogram aggregation to benchmark. */
@@ -17,19 +17,19 @@ public enum HistogramAggregationParam {
       new DoubleExplicitBucketHistogramAggregator(
           ExplicitBucketHistogramUtils.createBoundaryArray(
               ExplicitBucketHistogramUtils.DEFAULT_HISTOGRAM_BUCKET_BOUNDARIES),
-          ExemplarReservoir::noSamples,
+          ExemplarReservoirFactory.noSamples(),
           IMMUTABLE_DATA)),
   EXPLICIT_SINGLE_BUCKET(
       new DoubleExplicitBucketHistogramAggregator(
           ExplicitBucketHistogramUtils.createBoundaryArray(Collections.emptyList()),
-          ExemplarReservoir::noSamples,
+          ExemplarReservoirFactory.noSamples(),
           IMMUTABLE_DATA)),
   EXPONENTIAL_SMALL_CIRCULAR_BUFFER(
       new DoubleBase2ExponentialHistogramAggregator(
-          ExemplarReservoir::noSamples, 20, 0, IMMUTABLE_DATA)),
+          ExemplarReservoirFactory.noSamples(), 20, 0, IMMUTABLE_DATA)),
   EXPONENTIAL_CIRCULAR_BUFFER(
       new DoubleBase2ExponentialHistogramAggregator(
-          ExemplarReservoir::noSamples, 160, 0, IMMUTABLE_DATA));
+          ExemplarReservoirFactory.noSamples(), 160, 0, IMMUTABLE_DATA));
 
   private final Aggregator<?> aggregator;
 

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/internal/aggregator/HistogramBenchmark.java

@@ -5,6 +5,8 @@
 
 package io.opentelemetry.sdk.metrics.internal.aggregator;
 
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.context.Context;
 import java.util.concurrent.TimeUnit;
 import java.util.function.DoubleSupplier;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -45,7 +47,8 @@ public final void setup() {
     public void record() {
       // Record a number of samples.
       for (int i = 0; i < 2000; i++) {
-        this.aggregatorHandle.recordDouble(valueSupplier.getAsDouble());
+        this.aggregatorHandle.recordDouble(
+            valueSupplier.getAsDouble(), Attributes.empty(), Context.current());
       }
     }
   }

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/internal/aggregator/HistogramScaleBenchmark.java

@@ -5,6 +5,8 @@
 
 package io.opentelemetry.sdk.metrics.internal.aggregator;
 
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.context.Context;
 import java.util.concurrent.TimeUnit;
 import java.util.function.DoubleSupplier;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -51,7 +53,8 @@ public final void setup() {
     public void record() {
       // Record a number of samples.
       for (int i = 0; i < 20000; i++) {
-        this.aggregatorHandle.recordDouble(valueSupplier.getAsDouble());
+        this.aggregatorHandle.recordDouble(
+            valueSupplier.getAsDouble(), Attributes.empty(), Context.current());
       }
     }
   }

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/aggregator/AggregatorHandle.java

@@ -10,8 +10,11 @@
 import io.opentelemetry.sdk.metrics.data.DoubleExemplarData;
 import io.opentelemetry.sdk.metrics.data.LongExemplarData;
 import io.opentelemetry.sdk.metrics.data.PointData;
-import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoir;
+import io.opentelemetry.sdk.metrics.internal.exemplar.DoubleExemplarReservoir;
+import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoirFactory;
+import io.opentelemetry.sdk.metrics.internal.exemplar.LongExemplarReservoir;
 import java.util.List;
+import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
@@ -27,14 +30,26 @@
 @ThreadSafe
 public abstract class AggregatorHandle<T extends PointData> {
 
+  private static final String UNSUPPORTED_LONG_MESSAGE =
+      "This aggregator does not support long values.";
+  private static final String UNSUPPORTED_DOUBLE_MESSAGE =
+      "This aggregator does not support double values.";
+
   // A reservoir of sampled exemplars for this time period.
-  private final ExemplarReservoir exemplarReservoir;
-  private volatile boolean valuesRecorded = false;
+  @Nullable private final DoubleExemplarReservoir doubleReservoirFactory;
+  @Nullable private final LongExemplarReservoir longReservoirFactory;
   private final boolean isDoubleType;
+  private volatile boolean valuesRecorded = false;
 
-  protected AggregatorHandle(ExemplarReservoir exemplarReservoir) {
-    this.exemplarReservoir = exemplarReservoir;
+  protected AggregatorHandle(ExemplarReservoirFactory reservoirFactory) {
     this.isDoubleType = isDoubleType();
+    if (isDoubleType) {
+      this.doubleReservoirFactory = reservoirFactory.createDoubleExemplarReservoir();
+      this.longReservoirFactory = null;
+    } else {
+      this.doubleReservoirFactory = null;
+      this.longReservoirFactory = reservoirFactory.createLongExemplarReservoir();
+    }
   }
 
   /**
@@ -52,14 +67,16 @@ public final T aggregateThenMaybeReset(
           startEpochNanos,
           epochNanos,
           attributes,
-          exemplarReservoir.collectAndResetDoubles(attributes),
+          throwUnsupportedIfNull(this.doubleReservoirFactory, UNSUPPORTED_DOUBLE_MESSAGE)
+              .collectAndResetDoubles(attributes),
           reset);
     }
     return doAggregateThenMaybeResetLongs(
         startEpochNanos,
         epochNanos,
         attributes,
-        exemplarReservoir.collectAndResetLongs(attributes),
+        throwUnsupportedIfNull(this.longReservoirFactory, UNSUPPORTED_LONG_MESSAGE)
+            .collectAndResetLongs(attributes),
         reset);
   }
 
@@ -83,7 +100,7 @@ protected T doAggregateThenMaybeResetDoubles(
       Attributes attributes,
       List<DoubleExemplarData> exemplars,
       boolean reset) {
-    throw new UnsupportedOperationException("This aggregator does not support double values.");
+    throw new UnsupportedOperationException(UNSUPPORTED_DOUBLE_MESSAGE);
   }
 
   /** Implementation of the {@link #aggregateThenMaybeReset(long, long, Attributes, boolean)} . */
@@ -93,22 +110,12 @@ protected T doAggregateThenMaybeResetLongs(
       Attributes attributes,
       List<LongExemplarData> exemplars,
       boolean reset) {
-    throw new UnsupportedOperationException("This aggregator does not support long values.");
+    throw new UnsupportedOperationException(UNSUPPORTED_LONG_MESSAGE);
   }
 
-  public final void recordLong(long value, Attributes attributes, Context context) {
-    exemplarReservoir.offerLongMeasurement(value, attributes, context);
-    recordLong(value);
-  }
-
-  /**
-   * Updates the current aggregator with a newly recorded {@code long} value.
-   *
-   * <p>Visible for Testing
-   *
-   * @param value the new {@code long} value to be added.
-   */
-  public final void recordLong(long value) {
+  public void recordLong(long value, Attributes attributes, Context context) {
+    throwUnsupportedIfNull(this.longReservoirFactory, UNSUPPORTED_LONG_MESSAGE)
+        .offerLongMeasurement(value, attributes, context);
     doRecordLong(value);
     valuesRecorded = true;
   }
@@ -122,18 +129,8 @@ protected void doRecordLong(long value) {
   }
 
   public final void recordDouble(double value, Attributes attributes, Context context) {
-    exemplarReservoir.offerDoubleMeasurement(value, attributes, context);
-    recordDouble(value);
-  }
-
-  /**
-   * Updates the current aggregator with a newly recorded {@code double} value.
-   *
-   * <p>Visible for Testing
-   *
-   * @param value the new {@code double} value to be added.
-   */
-  public final void recordDouble(double value) {
+    throwUnsupportedIfNull(this.doubleReservoirFactory, UNSUPPORTED_DOUBLE_MESSAGE)
+        .offerDoubleMeasurement(value, attributes, context);
     doRecordDouble(value);
     valuesRecorded = true;
   }
@@ -143,7 +140,7 @@ public final void recordDouble(double value) {
    * double values.
    */
   protected void doRecordDouble(double value) {
-    throw new UnsupportedOperationException("This aggregator does not support double values.");
+    throw new UnsupportedOperationException(UNSUPPORTED_DOUBLE_MESSAGE);
   }
 
   /**
@@ -154,4 +151,11 @@ protected void doRecordDouble(double value) {
   public boolean hasRecordedValues() {
     return valuesRecorded;
   }
+
+  private static <S> S throwUnsupportedIfNull(@Nullable S value, String message) {
+    if (value == null) {
+      throw new UnsupportedOperationException(message);
+    }
+    return value;
+  }
 }

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/aggregator/DoubleBase2ExponentialHistogramAggregator.java

@@ -6,6 +6,7 @@
 package io.opentelemetry.sdk.metrics.internal.aggregator;
 
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.context.Context;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.common.export.MemoryMode;
 import io.opentelemetry.sdk.internal.DynamicPrimitiveLongList;
@@ -21,11 +22,10 @@
 import io.opentelemetry.sdk.metrics.internal.data.MutableExponentialHistogramBuckets;
 import io.opentelemetry.sdk.metrics.internal.data.MutableExponentialHistogramPointData;
 import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
-import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoir;
+import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoirFactory;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.Collection;
 import java.util.List;
-import java.util.function.Supplier;
 import javax.annotation.Nullable;
 
 /**
@@ -37,30 +37,30 @@
 public final class DoubleBase2ExponentialHistogramAggregator
     implements Aggregator<ExponentialHistogramPointData> {
 
-  private final Supplier<ExemplarReservoir> reservoirSupplier;
+  private final ExemplarReservoirFactory reservoirFactory;
   private final int maxBuckets;
   private final int maxScale;
   private final MemoryMode memoryMode;
 
   /**
    * Constructs an exponential histogram aggregator.
    *
-   * @param reservoirSupplier Supplier of exemplar reservoirs per-stream.
+   * @param reservoirFactory Supplier of exemplar reservoirs per-stream.
    */
   public DoubleBase2ExponentialHistogramAggregator(
-      Supplier<ExemplarReservoir> reservoirSupplier,
+      ExemplarReservoirFactory reservoirFactory,
       int maxBuckets,
       int maxScale,
       MemoryMode memoryMode) {
-    this.reservoirSupplier = reservoirSupplier;
+    this.reservoirFactory = reservoirFactory;
     this.maxBuckets = maxBuckets;
     this.maxScale = maxScale;
     this.memoryMode = memoryMode;
   }
 
   @Override
   public AggregatorHandle<ExponentialHistogramPointData> createHandle() {
-    return new Handle(reservoirSupplier.get(), maxBuckets, maxScale, memoryMode);
+    return new Handle(reservoirFactory, maxBuckets, maxScale, memoryMode);
   }
 
   @Override
@@ -95,8 +95,12 @@ static final class Handle extends AggregatorHandle<ExponentialHistogramPointData
     // Used only when MemoryMode = REUSABLE_DATA
     @Nullable private final MutableExponentialHistogramPointData reusablePoint;
 
-    Handle(ExemplarReservoir reservoir, int maxBuckets, int maxScale, MemoryMode memoryMode) {
-      super(reservoir);
+    Handle(
+        ExemplarReservoirFactory reservoirFactory,
+        int maxBuckets,
+        int maxScale,
+        MemoryMode memoryMode) {
+      super(reservoirFactory);
       this.maxBuckets = maxBuckets;
       this.maxScale = maxScale;
       this.sum = 0;
@@ -262,8 +266,8 @@ protected boolean isDoubleType() {
     }
 
     @Override
-    protected void doRecordLong(long value) {
-      doRecordDouble((double) value);
+    public void recordLong(long value, Attributes attributes, Context context) {
+      super.recordDouble((double) value, attributes, context);
     }
 
     void downScale(int by) {

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/aggregator/DoubleExplicitBucketHistogramAggregator.java

@@ -7,6 +7,7 @@
 
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.internal.GuardedBy;
+import io.opentelemetry.context.Context;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.common.export.MemoryMode;
 import io.opentelemetry.sdk.internal.PrimitiveLongList;
@@ -19,14 +20,13 @@
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableMetricData;
 import io.opentelemetry.sdk.metrics.internal.data.MutableHistogramPointData;
 import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
-import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoir;
+import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoirFactory;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.function.Supplier;
 import javax.annotation.Nullable;
 
 /**
@@ -43,17 +43,17 @@ public final class DoubleExplicitBucketHistogramAggregator
   // a cache for converting to MetricData
   private final List<Double> boundaryList;
 
-  private final Supplier<ExemplarReservoir> reservoirSupplier;
+  private final ExemplarReservoirFactory reservoirFactory;
 
   /**
    * Constructs an explicit bucket histogram aggregator.
    *
    * @param boundaries Bucket boundaries, in-order.
-   * @param reservoirSupplier Supplier of exemplar reservoirs per-stream.
+   * @param reservoirFactory Supplier of exemplar reservoirs per-stream.
    * @param memoryMode The {@link MemoryMode} to use in this aggregator.
    */
   public DoubleExplicitBucketHistogramAggregator(
-      double[] boundaries, Supplier<ExemplarReservoir> reservoirSupplier, MemoryMode memoryMode) {
+      double[] boundaries, ExemplarReservoirFactory reservoirFactory, MemoryMode memoryMode) {
     this.boundaries = boundaries;
     this.memoryMode = memoryMode;
 
@@ -62,12 +62,12 @@ public DoubleExplicitBucketHistogramAggregator(
       boundaryList.add(v);
     }
     this.boundaryList = Collections.unmodifiableList(boundaryList);
-    this.reservoirSupplier = reservoirSupplier;
+    this.reservoirFactory = reservoirFactory;
   }
 
   @Override
   public AggregatorHandle<HistogramPointData> createHandle() {
-    return new Handle(this.boundaryList, this.boundaries, reservoirSupplier.get(), memoryMode);
+    return new Handle(boundaryList, boundaries, reservoirFactory, memoryMode);
   }
 
   @Override
@@ -115,9 +115,9 @@ static final class Handle extends AggregatorHandle<HistogramPointData> {
     Handle(
         List<Double> boundaryList,
         double[] boundaries,
-        ExemplarReservoir reservoir,
+        ExemplarReservoirFactory reservoirFactory,
         MemoryMode memoryMode) {
-      super(reservoir);
+      super(reservoirFactory);
       this.boundaryList = boundaryList;
       this.boundaries = boundaries;
       this.counts = new long[this.boundaries.length + 1];
@@ -130,6 +130,11 @@ static final class Handle extends AggregatorHandle<HistogramPointData> {
       }
     }
 
+    @Override
+    public void recordLong(long value, Attributes attributes, Context context) {
+      super.recordDouble((double) value, attributes, context);
+    }
+
     @Override
     protected boolean isDoubleType() {
       return true;
@@ -196,10 +201,5 @@ protected void doRecordDouble(double value) {
         this.counts[bucketIndex]++;
       }
     }
-
-    @Override
-    protected void doRecordLong(long value) {
-      doRecordDouble((double) value);
-    }
   }
 }