Spatial aggregation for async instruments with filtering views (#7264)

Co-authored-by: jack-berg <[email protected]>
Co-authored-by: Jack Berg <[email protected]>

Author: fandreuz

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

@@ -13,7 +13,6 @@
 import io.opentelemetry.sdk.metrics.data.MetricDataType;
 import io.opentelemetry.sdk.metrics.data.PointData;
 import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
-import io.opentelemetry.sdk.metrics.internal.state.Measurement;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.Collection;
 import javax.annotation.concurrent.Immutable;
@@ -68,24 +67,6 @@ default void diffInPlace(T previousCumulativeReusable, T currentCumulative) {
     throw new UnsupportedOperationException("This aggregator does not support diffInPlace.");
   }
 
-  /**
-   * Return a new point representing the measurement.
-   *
-   * <p>Aggregators MUST implement diff if it can be used with asynchronous instruments.
-   */
-  default T toPoint(Measurement measurement) {
-    throw new UnsupportedOperationException("This aggregator does not support toPoint.");
-  }
-
-  /**
-   * Resets {@code reusablePoint} to represent the {@code measurement}.
-   *
-   * <p>Aggregators MUST implement diff if it can be used with asynchronous instruments.
-   */
-  default void toPoint(Measurement measurement, T reusablePoint) {
-    throw new UnsupportedOperationException("This aggregator does not support toPoint.");
-  }
-
   /** Creates a new reusable point. */
   default T createReusablePoint() {
     throw new UnsupportedOperationException(

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

@@ -18,11 +18,11 @@
 import io.opentelemetry.sdk.metrics.internal.data.MutableDoublePointData;
 import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
 import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoir;
-import io.opentelemetry.sdk.metrics.internal.state.Measurement;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
@@ -66,25 +66,6 @@ public void diffInPlace(DoublePointData previousReusable, DoublePointData curren
     ((MutableDoublePointData) previousReusable).set(current);
   }
 
-  @Override
-  public DoublePointData toPoint(Measurement measurement) {
-    return ImmutableDoublePointData.create(
-        measurement.startEpochNanos(),
-        measurement.epochNanos(),
-        measurement.attributes(),
-        measurement.doubleValue());
-  }
-
-  @Override
-  public void toPoint(Measurement measurement, DoublePointData reusablePoint) {
-    ((MutableDoublePointData) reusablePoint)
-        .set(
-            measurement.startEpochNanos(),
-            measurement.epochNanos(),
-            measurement.attributes(),
-            measurement.doubleValue());
-  }
-
   @Override
   public DoublePointData createReusablePoint() {
     return new MutableDoublePointData();
@@ -114,8 +95,8 @@ public MetricData toMetricData(
   }
 
   static final class Handle extends AggregatorHandle<DoublePointData, DoubleExemplarData> {
-    @Nullable private static final Double DEFAULT_VALUE = null;
-    private final AtomicReference<Double> current = new AtomicReference<>(DEFAULT_VALUE);
+    private final AtomicReference<AtomicLong> current = new AtomicReference<>(null);
+    private final AtomicLong valueBits = new AtomicLong();
 
     // Only used when memoryMode is REUSABLE_DATA
     @Nullable private final MutableDoublePointData reusablePoint;
@@ -136,20 +117,22 @@ protected DoublePointData doAggregateThenMaybeReset(
         Attributes attributes,
         List<DoubleExemplarData> exemplars,
         boolean reset) {
-      Double value = reset ? this.current.getAndSet(DEFAULT_VALUE) : this.current.get();
+      AtomicLong valueBits =
+          Objects.requireNonNull(reset ? this.current.getAndSet(null) : this.current.get());
+      double value = Double.longBitsToDouble(valueBits.get());
       if (reusablePoint != null) {
-        reusablePoint.set(
-            startEpochNanos, epochNanos, attributes, Objects.requireNonNull(value), exemplars);
+        reusablePoint.set(startEpochNanos, epochNanos, attributes, value, exemplars);
         return reusablePoint;
       } else {
         return ImmutableDoublePointData.create(
-            startEpochNanos, epochNanos, attributes, Objects.requireNonNull(value), exemplars);
+            startEpochNanos, epochNanos, attributes, value, exemplars);
       }
     }
 
     @Override
     protected void doRecordDouble(double value) {
-      current.set(value);
+      valueBits.set(Double.doubleToLongBits(value));
+      current.compareAndSet(null, valueBits);
     }
   }
 }

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

@@ -21,7 +21,6 @@
 import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor;
 import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
 import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoir;
-import io.opentelemetry.sdk.metrics.internal.state.Measurement;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.Collection;
 import java.util.List;
@@ -82,25 +81,6 @@ public void diffInPlace(DoublePointData previousReusablePoint, DoublePointData c
             currentPoint.getExemplars());
   }
 
-  @Override
-  public DoublePointData toPoint(Measurement measurement) {
-    return ImmutableDoublePointData.create(
-        measurement.startEpochNanos(),
-        measurement.epochNanos(),
-        measurement.attributes(),
-        measurement.doubleValue());
-  }
-
-  @Override
-  public void toPoint(Measurement measurement, DoublePointData reusablePoint) {
-    ((MutableDoublePointData) reusablePoint)
-        .set(
-            measurement.startEpochNanos(),
-            measurement.epochNanos(),
-            measurement.attributes(),
-            measurement.doubleValue());
-  }
-
   @Override
   public DoublePointData createReusablePoint() {
     return new MutableDoublePointData();

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

@@ -18,7 +18,6 @@
 import io.opentelemetry.sdk.metrics.internal.data.MutableLongPointData;
 import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
 import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoir;
-import io.opentelemetry.sdk.metrics.internal.state.Measurement;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.Collection;
 import java.util.List;
@@ -63,25 +62,6 @@ public void diffInPlace(LongPointData previousReusablePoint, LongPointData curre
     ((MutableLongPointData) previousReusablePoint).set(currentPoint);
   }
 
-  @Override
-  public LongPointData toPoint(Measurement measurement) {
-    return ImmutableLongPointData.create(
-        measurement.startEpochNanos(),
-        measurement.epochNanos(),
-        measurement.attributes(),
-        measurement.longValue());
-  }
-
-  @Override
-  public void toPoint(Measurement measurement, LongPointData reusablePoint) {
-    ((MutableLongPointData) reusablePoint)
-        .set(
-            measurement.startEpochNanos(),
-            measurement.epochNanos(),
-            measurement.attributes(),
-            measurement.longValue());
-  }
-
   @Override
   public LongPointData createReusablePoint() {
     return new MutableLongPointData();

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

@@ -21,7 +21,6 @@
 import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor;
 import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
 import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarReservoir;
-import io.opentelemetry.sdk.metrics.internal.state.Measurement;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.Collection;
 import java.util.List;
@@ -75,25 +74,6 @@ public void diffInPlace(LongPointData previousReusablePoint, LongPointData curre
             currentPoint.getExemplars());
   }
 
-  @Override
-  public LongPointData toPoint(Measurement measurement) {
-    return ImmutableLongPointData.create(
-        measurement.startEpochNanos(),
-        measurement.epochNanos(),
-        measurement.attributes(),
-        measurement.longValue());
-  }
-
-  @Override
-  public void toPoint(Measurement measurement, LongPointData reusablePoint) {
-    ((MutableLongPointData) reusablePoint)
-        .set(
-            measurement.startEpochNanos(),
-            measurement.epochNanos(),
-            measurement.attributes(),
-            measurement.longValue());
-  }
-
   @Override
   public LongPointData createReusablePoint() {
     return new MutableLongPointData();