fix tests

Author: fandreuz

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorage.java

@@ -53,6 +53,7 @@ public final class AsynchronousMetricStorage<T extends PointData, U extends Exem
   private final AggregationTemporality aggregationTemporality;
   private final Aggregator<T, U> aggregator;
   private final AttributesProcessor attributesProcessor;
+  private final MemoryMode memoryMode;
 
   /**
    * This field is set to 1 less than the actual intended cardinality limit, allowing the last slot
@@ -63,14 +64,14 @@ public final class AsynchronousMetricStorage<T extends PointData, U extends Exem
   // Only populated if aggregationTemporality == DELTA
   private Map<Attributes, T> lastPoints;
 
+  private Map<Attributes, AggregatorHandle<T, U>> aggregatorHandles = new HashMap<>();
+
   // Only populated if memoryMode == REUSABLE_DATA
   private final ObjectPool<T> reusablePointsPool;
   private final List<T> reusableDeltaPoints = new ArrayList<>();
 
-  private Map<Attributes, AggregatorHandle<T, U>> aggregatorHandles = new HashMap<>();
-  private Map<Attributes, Measurement> latestRecordedMeasurements = new HashMap<>();
-
-  private final MemoryMode memoryMode;
+  private long startEpochNanos;
+  private long epochNanos;
 
   private AsynchronousMetricStorage(
       RegisteredReader registeredReader,
@@ -91,13 +92,11 @@ private AsynchronousMetricStorage(
     this.reusablePointsPool = new ObjectPool<>(aggregator::createReusablePoint);
 
     if (memoryMode == REUSABLE_DATA) {
-      lastPoints = new PooledHashMap<>();
-      aggregatorHandles = new PooledHashMap<>();
-      latestRecordedMeasurements = new PooledHashMap<>();
+      this.lastPoints = new PooledHashMap<>();
+      this.aggregatorHandles = new PooledHashMap<>();
     } else {
-      lastPoints = new HashMap<>();
-      aggregatorHandles = new HashMap<>();
-      latestRecordedMeasurements = new HashMap<>();
+      this.lastPoints = new HashMap<>();
+      this.aggregatorHandles = new HashMap<>();
     }
   }
 
@@ -127,24 +126,34 @@ AsynchronousMetricStorage<T, U> create(
         registeredView.getCardinalityLimit());
   }
 
-  /**
-   * Record callback measurement from {@link ObservableLongMeasurement} or {@link
-   * ObservableDoubleMeasurement}.
-   */
-  void record(Measurement measurement) {
-    Context context = Context.current();
-    Attributes processedAttributes = attributesProcessor.process(measurement.attributes(), context);
-    long start =
-        aggregationTemporality == AggregationTemporality.DELTA
-            ? registeredReader.getLastCollectEpochNanos()
-            : measurement.startEpochNanos();
+  /** Record callback measurement from {@link ObservableLongMeasurement}. */
+  void record(Attributes attributes, long value) {
+    attributes = validateAndProcessAttributes(attributes);
+    AggregatorHandle<T, U> handle =
+        aggregatorHandles.computeIfAbsent(attributes, key -> aggregator.createHandle());
+    handle.recordLong(value, attributes, Context.current());
+  }
 
-    measurement = measurement.withAttributes(processedAttributes).withStartEpochNanos(start);
+  /** Record callback measurement from {@link ObservableDoubleMeasurement}. */
+  void record(Attributes attributes, double value) {
+    attributes = validateAndProcessAttributes(attributes);
+    AggregatorHandle<T, U> handle =
+        aggregatorHandles.computeIfAbsent(attributes, key -> aggregator.createHandle());
+    handle.recordDouble(value, attributes, Context.current());
+  }
 
-    recordPoint(processedAttributes, measurement);
+  void setEpochInformation(long startEpochNanos, long epochNanos) {
+    this.startEpochNanos =
+        aggregationTemporality == AggregationTemporality.DELTA
+            ? registeredReader.getLastCollectEpochNanos()
+            : startEpochNanos;
+    this.epochNanos = epochNanos;
   }
 
-  private void recordPoint(Attributes attributes, Measurement measurement) {
+  private Attributes validateAndProcessAttributes(Attributes attributes) {
+    Context context = Context.current();
+    attributes = attributesProcessor.process(attributes, context);
+
     if (aggregatorHandles.size() >= maxCardinality) {
       throttlingLogger.log(
           Level.WARNING,
@@ -153,9 +162,10 @@ private void recordPoint(Attributes attributes, Measurement measurement) {
               + " has exceeded the maximum allowed cardinality ("
               + maxCardinality
               + ").");
-      attributes = MetricStorage.CARDINALITY_OVERFLOW;
-      measurement = measurement.withAttributes(attributes);
-    } else if (aggregatorHandles.containsKey(
+      return MetricStorage.CARDINALITY_OVERFLOW;
+    }
+
+    if (aggregatorHandles.containsKey(
         attributes)) { // Check there is not already a recording for the attributes
       throttlingLogger.log(
           Level.WARNING,
@@ -165,31 +175,7 @@ private void recordPoint(Attributes attributes, Measurement measurement) {
               + attributes);
     }
 
-    Measurement recentMeasurement = latestRecordedMeasurements.get(attributes);
-    if (recentMeasurement != null
-        && (recentMeasurement.startEpochNanos() != measurement.startEpochNanos()
-            || recentMeasurement.epochNanos() != measurement.epochNanos())) {
-      String msg =
-          String.format(
-              "Instrument "
-                  + metricDescriptor.getSourceInstrument().getName()
-                  + " has recorded multiple values for the same attributes (%s) with different metadata: %s/%s",
-              attributes,
-              recentMeasurement,
-              measurement);
-      throttlingLogger.log(Level.WARNING, msg);
-    }
-    latestRecordedMeasurements.put(attributes, measurement);
-
-    AggregatorHandle<T, U> handle =
-        aggregatorHandles.computeIfAbsent(attributes, key -> aggregator.createHandle());
-    if (measurement.hasDoubleValue()) {
-      handle.recordDouble(measurement.doubleValue(), attributes, Context.current());
-    } else if (measurement.hasLongValue()) {
-      handle.recordLong(measurement.longValue(), attributes, Context.current());
-    } else {
-      throw new IllegalStateException();
-    }
+    return attributes;
   }
 
   @Override
@@ -220,31 +206,24 @@ public MetricData collect(
     Map<Attributes, T> currentPoints = new HashMap<>();
     aggregatorHandles.forEach(
         (attributes, handle) -> {
-          Measurement latestRecordedMeasurement = latestRecordedMeasurements.get(attributes);
-          if (latestRecordedMeasurement == null) {
-            throw new IllegalStateException("Unexpected");
-          }
           T value =
               handle.aggregateThenMaybeReset(
-                  latestRecordedMeasurement.startEpochNanos(),
-                  latestRecordedMeasurement.epochNanos(),
+                  AsynchronousMetricStorage.this.startEpochNanos,
+                  AsynchronousMetricStorage.this.epochNanos,
                   attributes,
                   reset);
           currentPoints.put(attributes, value);
         });
 
     if (memoryMode == REUSABLE_DATA) {
       aggregatorHandles.clear();
-      latestRecordedMeasurements.clear();
     } else {
       aggregatorHandles = new HashMap<>();
-      latestRecordedMeasurements = new HashMap<>();
     }
 
     Collection<T> result;
     if (aggregationTemporality == AggregationTemporality.DELTA) {
-      Collection<T> deltaPoints =
-          memoryMode == REUSABLE_DATA ? reusableDeltaPoints : new ArrayList<>();
+      result = memoryMode == REUSABLE_DATA ? reusableDeltaPoints : new ArrayList<>();
       currentPoints.forEach(
           (attributes, currentPoint) -> {
             T lastPoint = lastPoints.get(attributes);
@@ -265,15 +244,13 @@ public MetricData collect(
               }
             }
 
-            deltaPoints.add(deltaPoint);
+            result.add(deltaPoint);
           });
 
       if (memoryMode == REUSABLE_DATA) {
         lastPoints.forEach((attributes, value) -> reusablePointsPool.returnObject(value));
         lastPoints.clear();
       }
-
-      result = deltaPoints;
     } else {
       result = currentPoints.values();
     }

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/SdkObservableMeasurement.java

@@ -5,19 +5,14 @@
 
 package io.opentelemetry.sdk.metrics.internal.state;
 
-import static io.opentelemetry.sdk.metrics.internal.state.ImmutableMeasurement.createDouble;
-import static io.opentelemetry.sdk.metrics.internal.state.ImmutableMeasurement.createLong;
-
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.metrics.ObservableDoubleMeasurement;
 import io.opentelemetry.api.metrics.ObservableLongMeasurement;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
-import io.opentelemetry.sdk.common.export.MemoryMode;
 import io.opentelemetry.sdk.internal.ThrottlingLogger;
 import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor;
 import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader;
 import java.util.List;
-import java.util.Objects;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
@@ -38,9 +33,6 @@ public final class SdkObservableMeasurement
   private final InstrumentDescriptor instrumentDescriptor;
   private final List<AsynchronousMetricStorage<?, ?>> storages;
 
-  /** Only used when {@code activeReader}'s memoryMode is {@link MemoryMode#REUSABLE_DATA}. */
-  private final MutableMeasurement mutableMeasurement = new MutableMeasurement();
-
   // These fields are set before invoking callbacks. They allow measurements to be recorded to the
   // storages for correct reader, and with the correct time.
   @Nullable private volatile RegisteredReader activeReader;
@@ -114,18 +106,12 @@ public void record(long value, Attributes attributes) {
       return;
     }
 
-    Measurement measurement;
-
-    MemoryMode memoryMode = activeReader.getReader().getMemoryMode();
-    if (Objects.requireNonNull(memoryMode) == MemoryMode.IMMUTABLE_DATA) {
-      measurement = createLong(startEpochNanos, epochNanos, value, attributes);
-    } else {
-      MutableMeasurement.setLongMeasurement(
-          mutableMeasurement, startEpochNanos, epochNanos, value, attributes);
-      measurement = mutableMeasurement;
+    for (AsynchronousMetricStorage<?, ?> storage : storages) {
+      if (storage.getRegisteredReader().equals(activeReader)) {
+        storage.setEpochInformation(startEpochNanos, epochNanos);
+        storage.record(attributes, value);
+      }
     }
-
-    doRecord(measurement);
   }
 
   @Override
@@ -135,10 +121,12 @@ public void record(double value) {
 
   @Override
   public void record(double value, Attributes attributes) {
+    RegisteredReader activeReader = this.activeReader;
     if (activeReader == null) {
       logNoActiveReader();
       return;
     }
+
     if (Double.isNaN(value)) {
       logger.log(
           Level.FINE,
@@ -150,24 +138,10 @@ public void record(double value, Attributes attributes) {
       return;
     }
 
-    Measurement measurement;
-    MemoryMode memoryMode = activeReader.getReader().getMemoryMode();
-    if (Objects.requireNonNull(memoryMode) == MemoryMode.IMMUTABLE_DATA) {
-      measurement = createDouble(startEpochNanos, epochNanos, value, attributes);
-    } else {
-      MutableMeasurement.setDoubleMeasurement(
-          mutableMeasurement, startEpochNanos, epochNanos, value, attributes);
-      measurement = mutableMeasurement;
-    }
-
-    doRecord(measurement);
-  }
-
-  private void doRecord(Measurement measurement) {
-    RegisteredReader activeReader = this.activeReader;
     for (AsynchronousMetricStorage<?, ?> storage : storages) {
       if (storage.getRegisteredReader().equals(activeReader)) {
-        storage.record(measurement);
+        storage.setEpochInformation(startEpochNanos, epochNanos);
+        storage.record(attributes, value);
       }
     }
   }

sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/CardinalityTest.java

@@ -421,7 +421,7 @@ void readerAndViewCardinalityConfiguration() {
                                             0,
                                             asyncCounterLimit,
                                             LongPointData::getValue,
-                                            1))));
+                                            DEFAULT_MAX_CARDINALITY - (asyncCounterLimit - 1)))));
     assertThat(cumulativeReader.collectAllMetrics())
         .as("cumulative collection")
         .satisfiesExactlyInAnyOrder(
@@ -485,7 +485,7 @@ void readerAndViewCardinalityConfiguration() {
                                             0,
                                             asyncCounterLimit,
                                             LongPointData::getValue,
-                                            1))));
+                                            DEFAULT_MAX_CARDINALITY - (asyncCounterLimit - 1)))));
 
     // Record another round of measurements, again exceeding cardinality limits
     for (int i = DEFAULT_MAX_CARDINALITY; i < DEFAULT_MAX_CARDINALITY * 2; i++) {
@@ -625,7 +625,7 @@ void readerAndViewCardinalityConfiguration() {
                                             0,
                                             asyncCounterLimit,
                                             LongPointData::getValue,
-                                            1))));
+                                            DEFAULT_MAX_CARDINALITY - (asyncCounterLimit - 1)))));
   }
 
   /**