Memory Mode: Adding 2nd part support for synchronous instruments - exponential histogram

exporters/common/src/main/java/io/opentelemetry/exporter/internal/marshal/MarshalerUtil.java

@@ -8,6 +8,7 @@
 import io.opentelemetry.api.trace.SpanId;
 import io.opentelemetry.api.trace.TraceId;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
+import io.opentelemetry.sdk.internal.DynamicPrimitiveLongList;
 import io.opentelemetry.sdk.resources.Resource;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -128,6 +129,27 @@ public static int sizeRepeatedUInt64(ProtoFieldInfo field, long[] values) {
     return field.getTagSize() + CodedOutputStream.computeUInt32SizeNoTag(payloadSize) + payloadSize;
   }
 
+  /**
+   * Returns the size of a repeated uint64 field.
+   *
+   * <p>Packed repeated fields contain the tag, an integer representing the incoming payload size,
+   * and an actual payload of repeated varints.
+   */
+  public static int sizeRepeatedUInt64(ProtoFieldInfo field, DynamicPrimitiveLongList values) {
+    if (values.isEmpty()) {
+      return 0;
+    }
+
+    int payloadSize = 0;
+    for (int i = 0; i < values.size(); i++) {
+      long v = values.getLong(i);
+      payloadSize += CodedOutputStream.computeUInt64SizeNoTag(v);
+    }
+
+    // tag size + payload indicator size + actual payload size
+    return field.getTagSize() + CodedOutputStream.computeUInt32SizeNoTag(payloadSize) + payloadSize;
+  }
+
   /** Returns the size of a repeated double field. */
   public static int sizeRepeatedDouble(ProtoFieldInfo field, List<Double> values) {
     // Same as fixed64.

exporters/common/src/main/java/io/opentelemetry/exporter/internal/marshal/Serializer.java

@@ -5,6 +5,7 @@
 
 package io.opentelemetry.exporter.internal.marshal;
 
+import io.opentelemetry.sdk.internal.DynamicPrimitiveLongList;
 import java.io.IOException;
 import java.util.List;
 import javax.annotation.Nullable;
@@ -243,6 +244,27 @@ public void serializeRepeatedUInt64(ProtoFieldInfo field, long[] values) throws
     writeEndRepeatedVarint();
   }
 
+  /** Serializes a {@code repeated uint64} field. */
+  public void serializeRepeatedUInt64(ProtoFieldInfo field, DynamicPrimitiveLongList values)
+      throws IOException {
+    if (values.isEmpty()) {
+      return;
+    }
+
+    int payloadSize = 0;
+    for (int i = 0; i < values.size(); i++) {
+      long v = values.getLong(i);
+      payloadSize += CodedOutputStream.computeUInt64SizeNoTag(v);
+    }
+
+    writeStartRepeatedVarint(field, payloadSize);
+    for (int i = 0; i < values.size(); i++) {
+      long value = values.getLong(i);
+      writeUInt64Value(value);
+    }
+    writeEndRepeatedVarint();
+  }
+
   /** Serializes a {@code repeated double} field. */
   public void serializeRepeatedDouble(ProtoFieldInfo field, List<Double> values)
       throws IOException {

exporters/otlp/common/src/main/java/io/opentelemetry/exporter/internal/otlp/metrics/ExponentialHistogramBucketsMarshaler.java

@@ -9,6 +9,7 @@
 import io.opentelemetry.exporter.internal.marshal.MarshalerWithSize;
 import io.opentelemetry.exporter.internal.marshal.Serializer;
 import io.opentelemetry.proto.metrics.v1.internal.ExponentialHistogramDataPoint;
+import io.opentelemetry.sdk.internal.DynamicPrimitiveLongList;
 import io.opentelemetry.sdk.internal.PrimitiveLongList;
 import io.opentelemetry.sdk.metrics.data.ExponentialHistogramBuckets;
 import java.io.IOException;
@@ -35,16 +36,29 @@ private ExponentialHistogramBucketsMarshaler(int offset, List<Long> counts) {
   @Override
   protected void writeTo(Serializer output) throws IOException {
     output.serializeSInt32(ExponentialHistogramDataPoint.Buckets.OFFSET, offset);
-    output.serializeRepeatedUInt64(
-        ExponentialHistogramDataPoint.Buckets.BUCKET_COUNTS, PrimitiveLongList.toArray(counts));
+    if (counts instanceof DynamicPrimitiveLongList) {
+      output.serializeRepeatedUInt64(
+          ExponentialHistogramDataPoint.Buckets.BUCKET_COUNTS, (DynamicPrimitiveLongList) counts);
+    } else {
+      output.serializeRepeatedUInt64(
+          ExponentialHistogramDataPoint.Buckets.BUCKET_COUNTS, PrimitiveLongList.toArray(counts));
+    }
   }
 
   private static int calculateSize(int offset, List<Long> counts) {
     int size = 0;
     size += MarshalerUtil.sizeSInt32(ExponentialHistogramDataPoint.Buckets.OFFSET, offset);
-    size +=
-        MarshalerUtil.sizeRepeatedUInt64(
-            ExponentialHistogramDataPoint.Buckets.BUCKET_COUNTS, PrimitiveLongList.toArray(counts));
+    if (counts instanceof DynamicPrimitiveLongList) {
+      size +=
+          MarshalerUtil.sizeRepeatedUInt64(
+              ExponentialHistogramDataPoint.Buckets.BUCKET_COUNTS,
+              (DynamicPrimitiveLongList) counts);
+    } else {
+      size +=
+          MarshalerUtil.sizeRepeatedUInt64(
+              ExponentialHistogramDataPoint.Buckets.BUCKET_COUNTS,
+              PrimitiveLongList.toArray(counts));
+    }
     return size;
   }
 }

exporters/otlp/common/src/test/java/io/opentelemetry/exporter/internal/otlp/metrics/MetricsRequestMarshalerTest.java

@@ -40,12 +40,14 @@
 import io.opentelemetry.proto.metrics.v1.Summary;
 import io.opentelemetry.proto.metrics.v1.SummaryDataPoint;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
+import io.opentelemetry.sdk.internal.DynamicPrimitiveLongList;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.ExponentialHistogramPointData;
 import io.opentelemetry.sdk.metrics.data.HistogramPointData;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.data.PointData;
 import io.opentelemetry.sdk.metrics.data.SummaryPointData;
+import io.opentelemetry.sdk.metrics.internal.aggregator.MutableExponentialHistogramBuckets;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableDoubleExemplarData;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableDoublePointData;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableExponentialHistogramBuckets;
@@ -61,6 +63,7 @@
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableSummaryData;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableSummaryPointData;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableValueAtQuantile;
+import io.opentelemetry.sdk.metrics.internal.data.MutableExponentialHistogramPointData;
 import io.opentelemetry.sdk.resources.Resource;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -508,6 +511,105 @@ void exponentialHistogramDataPoints() {
                 .build());
   }
 
+  @SuppressWarnings("PointlessArithmeticExpression")
+  @Test
+  void exponentialHistogramReusableDataPoints() {
+    assertThat(
+            toExponentialHistogramDataPoints(
+                ImmutableList.of(
+                    new MutableExponentialHistogramPointData()
+                        .set(
+                            0,
+                            123.4,
+                            1,
+                            /* hasMin= */ false,
+                            0,
+                            /* hasMax= */ false,
+                            0,
+                            new MutableExponentialHistogramBuckets()
+                                .set(0, 0, 0, new DynamicPrimitiveLongList()),
+                            new MutableExponentialHistogramBuckets()
+                                .set(0, 0, 0, new DynamicPrimitiveLongList()),
+                            123,
+                            456,
+                            Attributes.empty(),
+                            Collections.emptyList()),
+                    new MutableExponentialHistogramPointData()
+                        .set(
+                            0,
+                            123.4,
+                            1,
+                            /* hasMin= */ true,
+                            3.3,
+                            /* hasMax= */ true,
+                            80.1,
+                            new MutableExponentialHistogramBuckets()
+                                .set(0, 1, 1 + 0 + 2, DynamicPrimitiveLongList.of(1L, 0L, 2L)),
+                            new MutableExponentialHistogramBuckets()
+                                .set(0, 0, 0, new DynamicPrimitiveLongList()),
+                            123,
+                            456,
+                            Attributes.of(stringKey("key"), "value"),
+                            ImmutableList.of(
+                                ImmutableDoubleExemplarData.create(
+                                    Attributes.of(stringKey("test"), "value"),
+                                    2,
+                                    SpanContext.create(
+                                        "00000000000000000000000000000001",
+                                        "0000000000000002",
+                                        TraceFlags.getDefault(),
+                                        TraceState.getDefault()),
+                                    1.5))))))
+        .containsExactly(
+            ExponentialHistogramDataPoint.newBuilder()
+                .setStartTimeUnixNano(123)
+                .setTimeUnixNano(456)
+                .setCount(1)
+                .setScale(0)
+                .setSum(123.4)
+                .setZeroCount(1)
+                .setPositive(
+                    ExponentialHistogramDataPoint.Buckets.newBuilder().setOffset(0)) // no buckets
+                .setNegative(
+                    ExponentialHistogramDataPoint.Buckets.newBuilder().setOffset(0)) // no buckets
+                .build(),
+            ExponentialHistogramDataPoint.newBuilder()
+                .setStartTimeUnixNano(123)
+                .setTimeUnixNano(456)
+                .setCount(4) // Counts in positive, negative, and zero count.
+                .addAllAttributes(
+                    singletonList(
+                        KeyValue.newBuilder().setKey("key").setValue(stringValue("value")).build()))
+                .setScale(0)
+                .setSum(123.4)
+                .setMin(3.3)
+                .setMax(80.1)
+                .setZeroCount(1)
+                .setPositive(
+                    ExponentialHistogramDataPoint.Buckets.newBuilder()
+                        .setOffset(1)
+                        .addBucketCounts(1)
+                        .addBucketCounts(0)
+                        .addBucketCounts(2))
+                .setNegative(
+                    ExponentialHistogramDataPoint.Buckets.newBuilder().setOffset(0)) // no buckets
+                .addExemplars(
+                    Exemplar.newBuilder()
+                        .setTimeUnixNano(2)
+                        .addFilteredAttributes(
+                            KeyValue.newBuilder()
+                                .setKey("test")
+                                .setValue(stringValue("value"))
+                                .build())
+                        .setSpanId(ByteString.copyFrom(new byte[] {0, 0, 0, 0, 0, 0, 0, 2}))
+                        .setTraceId(
+                            ByteString.copyFrom(
+                                new byte[] {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1}))
+                        .setAsDouble(1.5)
+                        .build())
+                .build());
+  }
+
   @Test
   void toProtoMetric_monotonic() {
     assertThat(

sdk/common/src/main/java/io/opentelemetry/sdk/internal/DynamicPrimitiveLongList.java

@@ -0,0 +1,142 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.internal;
+
+import java.util.AbstractList;
+
+/**
+ * A resizable list for storing primitive `long` values.
+ *
+ * <p>This class implements a dynamically resizable list specifically for primitive long values. The
+ * values are stored in a chain of arrays (named sub-array), so it can grow efficiently, by adding
+ * more sub-arrays per its defined size. The backing array also helps avoid auto-boxing and helps
+ * provide access to values as primitives without boxing.
+ *
+ * <p>The list is designed to minimize memory allocations, by:
+ *
+ * <ol>
+ *   <li>Adding sub-arrays and not creating new arrays and copying.
+ *   <li>When the size is changing to a smaller size, arrays are not removed.
+ * </ol>
+ *
+ * <p><b>Supported {@code List<Long>} methods:</b>
+ *
+ * <ul>
+ *   <li>{@link #get(int)} - Retrieves the element at the specified position in this list as a
+ *       {@code Long} object.
+ *   <li>{@link #set(int, Long)} - Replaces the element at the specified position in this list with
+ *       the specified {@code Long} object.
+ *   <li>{@link #size()} - Returns the number of elements in this list.
+ * </ul>
+ *
+ * <p><b>Additional utility methods:</b>
+ *
+ * <ul>
+ *   <li>{@link #getLong(int)} - Retrieves the element at the specified position in this list as a
+ *       primitive long.
+ *   <li>{@link #setLong(int, long)} - Replaces the element at the specified position in this list
+ *       with the specified primitive long element.
+ *   <li>{@link #resize(int)} - Resizes the list to the specified size, resetting all elements to
+ *       zero.
+ * </ul>
+ *
+ * <p>This class is an internal part of the OpenTelemetry SDK and is not intended for public use.
+ * Its API is unstable and subject to change.
+ *
+ * <p>This class is not thread-safe.
+ */
+public class DynamicPrimitiveLongList extends AbstractList<Long> {
+
+  private static final int DEFAULT_SUBARRAY_CAPACITY = 10;
+  private final int subarrayCapacity;
+  private long[][] arrays;
+  private int size;
+  private int arrayCount;
+
+  public static DynamicPrimitiveLongList of(long... values) {
+    DynamicPrimitiveLongList list = new DynamicPrimitiveLongList();
+    list.resize(values.length);
+    for (int i = 0; i < values.length; i++) {
+      list.setLong(i, values[i]);
+    }
+    return list;
+  }
+
+  public DynamicPrimitiveLongList() {
+    this(DEFAULT_SUBARRAY_CAPACITY);
+  }
+
+  public DynamicPrimitiveLongList(int subarrayCapacity) {
+    if (subarrayCapacity <= 0) {
+      throw new IllegalArgumentException("Subarray capacity must be positive");
+    }
+    this.subarrayCapacity = subarrayCapacity;
+    arrays = new long[0][subarrayCapacity];
+    arrayCount = 0;
+    size = 0;
+  }
+
+  @Override
+  public Long get(int index) {
+    return getLong(index);
+  }
+
+  public long getLong(int index) {
+    rangeCheck(index);
+    return arrays[index / subarrayCapacity][index % subarrayCapacity];
+  }
+
+  @Override
+  public Long set(int index, Long element) {
+    return setLong(index, element);
+  }
+
+  public long setLong(int index, long element) {
+    rangeCheck(index);
+    long oldValue = arrays[index / subarrayCapacity][index % subarrayCapacity];
+    arrays[index / subarrayCapacity][index % subarrayCapacity] = element;
+    return oldValue;
+  }
+
+  @Override
+  public int size() {
+    return size;
+  }
+
+  public void resize(int newSize) {
+    if (newSize < 0) {
+      throw new IllegalArgumentException("New size must be non-negative");
+    }
+    ensureCapacity(newSize);
+    size = newSize;
+    for (int i = 0; i < newSize; i++) {
+      setLong(i, 0);
+    }
+  }
+
+  private void ensureCapacity(int minCapacity) {
+    // A faster way to do ceil(minCapacity/subArrayCapacity)
+    int requiredArrays = (minCapacity + subarrayCapacity - 1) / subarrayCapacity;
+
+    if (requiredArrays > arrayCount) {
+      arrays = java.util.Arrays.copyOf(arrays, /* newLength= */ requiredArrays);
+      for (int i = arrayCount; i < requiredArrays; i++) {
+        arrays[i] = new long[subarrayCapacity];
+      }
+      arrayCount = requiredArrays;
+    }
+  }
+
+  private void rangeCheck(int index) {
+    if (index < 0 || index >= size) {
+      throw new IndexOutOfBoundsException(outOfBoundsMsg(index));
+    }
+  }
+
+  private String outOfBoundsMsg(int index) {
+    return "Index: " + index + ", Size: " + size;
+  }
+}