Simplify dynamic meter config mechanics

Author: jack-berg

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

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

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeter.java

@@ -123,21 +123,22 @@ InstrumentationScopeInfo getInstrumentationScopeInfo() {
 
   /** Collect all metrics for the meter. */
   Collection<MetricData> collectAll(RegisteredReader registeredReader, long epochNanos) {
-    // Short circuit collection process if meter is disabled
-    if (!meterEnabled) {
-      return Collections.emptyList();
-    }
     List<CallbackRegistration> currentRegisteredCallbacks;
     synchronized (callbackLock) {
       currentRegisteredCallbacks = new ArrayList<>(callbackRegistrations);
     }
     // Collections across all readers are sequential
     synchronized (collectLock) {
-      for (CallbackRegistration callbackRegistration : currentRegisteredCallbacks) {
-        callbackRegistration.invokeCallback(
-            registeredReader, meterProviderSharedState.getStartEpochNanos(), epochNanos);
+      // Only invoke callbacks if meter is enabled
+      if (meterEnabled) {
+        for (CallbackRegistration callbackRegistration : currentRegisteredCallbacks) {
+          callbackRegistration.invokeCallback(
+              registeredReader, meterProviderSharedState.getStartEpochNanos(), epochNanos);
+        }
       }
 
+      // Collect even if meter is disabled. Storage is responsible for managing state and returning
+      // empty metric if disabled.
       Collection<MetricStorage> storages =
           Objects.requireNonNull(readerStorageRegistries.get(registeredReader)).getStorages();
       List<MetricData> result = new ArrayList<>(storages.size());
@@ -288,7 +289,8 @@ WriteableMetricStorage registerSynchronousMetricStorage(InstrumentDescriptor ins
                     reader,
                     registeredView,
                     instrument,
-                    meterProviderSharedState.getExemplarFilter())));
+                    meterProviderSharedState.getExemplarFilter(),
+                    meterEnabled)));
       }
     }
 
@@ -314,7 +316,8 @@ SdkObservableMeasurement registerObservableMeasurement(
         }
         registeredStorages.add(
             registry.register(
-                AsynchronousMetricStorage.create(reader, registeredView, instrumentDescriptor)));
+                AsynchronousMetricStorage.create(
+                    reader, registeredView, instrumentDescriptor, meterEnabled)));
       }
     }
 

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

@@ -7,6 +7,7 @@
 
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
+import io.opentelemetry.sdk.metrics.data.DoubleExemplarData;
 import io.opentelemetry.sdk.metrics.data.ExemplarData;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.data.MetricDataType;
@@ -26,9 +27,8 @@
 @Immutable
 public interface Aggregator<T extends PointData, U extends ExemplarData> {
   /** Returns the drop aggregator, an aggregator that drops measurements. */
-  @SuppressWarnings("unchecked")
-  static <T extends PointData, U extends ExemplarData> Aggregator<T, U> drop() {
-    return (Aggregator<T, U>) DropAggregator.INSTANCE;
+  static Aggregator<?, DoubleExemplarData> drop() {
+    return DropAggregator.INSTANCE;
   }
 
   /**

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

@@ -8,6 +8,7 @@
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
+import io.opentelemetry.sdk.metrics.data.DoubleExemplarData;
 import io.opentelemetry.sdk.metrics.data.ExemplarData;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.data.PointData;
@@ -24,7 +25,7 @@
  * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
  * at any time.
  */
-public final class DropAggregator implements Aggregator<PointData, ExemplarData> {
+public final class DropAggregator implements Aggregator<PointData, DoubleExemplarData> {
 
   private static final PointData POINT_DATA =
       new PointData() {
@@ -49,16 +50,16 @@ public List<? extends ExemplarData> getExemplars() {
         }
       };
 
-  public static final Aggregator<PointData, ExemplarData> INSTANCE = new DropAggregator();
+  public static final Aggregator<PointData, DoubleExemplarData> INSTANCE = new DropAggregator();
 
-  private static final AggregatorHandle<PointData, ExemplarData> HANDLE =
-      new AggregatorHandle<PointData, ExemplarData>(ExemplarReservoir.anyNoSamples()) {
+  private static final AggregatorHandle<PointData, DoubleExemplarData> HANDLE =
+      new AggregatorHandle<PointData, DoubleExemplarData>(ExemplarReservoir.doubleNoSamples()) {
         @Override
         protected PointData doAggregateThenMaybeReset(
             long startEpochNanos,
             long epochNanos,
             Attributes attributes,
-            List<ExemplarData> exemplars,
+            List<DoubleExemplarData> exemplars,
             boolean reset) {
           return POINT_DATA;
         }
@@ -73,7 +74,7 @@ protected void doRecordDouble(double value) {}
   private DropAggregator() {}
 
   @Override
-  public AggregatorHandle<PointData, ExemplarData> createHandle() {
+  public AggregatorHandle<PointData, DoubleExemplarData> createHandle() {
     return HANDLE;
   }
 

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/exemplar/ExemplarReservoir.java

@@ -55,11 +55,6 @@ static ExemplarReservoir<LongExemplarData> longNoSamples() {
     return NoopExemplarReservoir.LONG_INSTANCE;
   }
 
-  /** An exemplar reservoir that stores no exemplars. */
-  static ExemplarReservoir<ExemplarData> anyNoSamples() {
-    return NoopExemplarReservoir.ANY_INSTANCE;
-  }
-
   /**
    * A double reservoir with fixed size that stores the given number of exemplars.
    *

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/exemplar/NoopExemplarReservoir.java

@@ -20,7 +20,6 @@ class NoopExemplarReservoir<T extends ExemplarData> implements ExemplarReservoir
       new NoopExemplarReservoir<>();
   static final NoopExemplarReservoir<DoubleExemplarData> DOUBLE_INSTANCE =
       new NoopExemplarReservoir<>();
-  static final NoopExemplarReservoir<ExemplarData> ANY_INSTANCE = new NoopExemplarReservoir<>();
 
   private NoopExemplarReservoir() {}
 

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

@@ -22,6 +22,7 @@
 import io.opentelemetry.sdk.metrics.internal.aggregator.Aggregator;
 import io.opentelemetry.sdk.metrics.internal.aggregator.AggregatorFactory;
 import io.opentelemetry.sdk.metrics.internal.aggregator.AggregatorHandle;
+import io.opentelemetry.sdk.metrics.internal.aggregator.EmptyMetricData;
 import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor;
 import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
 import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarFilter;
@@ -53,11 +54,9 @@ public final class AsynchronousMetricStorage<T extends PointData, U extends Exem
   private final RegisteredReader registeredReader;
   private final MetricDescriptor metricDescriptor;
   private final AggregationTemporality aggregationTemporality;
+  private final Aggregator<T, U> aggregator;
   private final AttributesProcessor attributesProcessor;
   private final MemoryMode memoryMode;
-  private final Aggregator<T, U> originalAggregator;
-
-  private Aggregator<T, U> aggregator;
 
   /**
    * This field is set to 1 less than the actual intended cardinality limit, allowing the last slot
@@ -88,12 +87,15 @@ public final class AsynchronousMetricStorage<T extends PointData, U extends Exem
   private long startEpochNanos;
   private long epochNanos;
 
+  private boolean enabled;
+
   private AsynchronousMetricStorage(
       RegisteredReader registeredReader,
       MetricDescriptor metricDescriptor,
       Aggregator<T, U> aggregator,
       AttributesProcessor attributesProcessor,
-      int maxCardinality) {
+      int maxCardinality,
+      boolean enabled) {
     this.registeredReader = registeredReader;
     this.metricDescriptor = metricDescriptor;
     this.aggregationTemporality =
@@ -102,9 +104,9 @@ private AsynchronousMetricStorage(
             .getAggregationTemporality(metricDescriptor.getSourceInstrument().getType());
     this.memoryMode = registeredReader.getReader().getMemoryMode();
     this.aggregator = aggregator;
-    this.originalAggregator = aggregator;
     this.attributesProcessor = attributesProcessor;
     this.maxCardinality = maxCardinality - 1;
+    this.enabled = enabled;
     this.reusablePointsPool = new ObjectPool<>(aggregator::createReusablePoint);
     this.reusableHandlesPool = new ObjectPool<>(aggregator::createHandle);
     this.handleBuilder = ignored -> reusableHandlesPool.borrowObject();
@@ -128,7 +130,8 @@ private AsynchronousMetricStorage(
       AsynchronousMetricStorage<T, U> create(
           RegisteredReader registeredReader,
           RegisteredView registeredView,
-          InstrumentDescriptor instrumentDescriptor) {
+          InstrumentDescriptor instrumentDescriptor,
+          boolean enabled) {
     View view = registeredView.getView();
     MetricDescriptor metricDescriptor =
         MetricDescriptor.create(view, registeredView.getViewSourceInfo(), instrumentDescriptor);
@@ -143,7 +146,8 @@ AsynchronousMetricStorage<T, U> create(
         metricDescriptor,
         aggregator,
         registeredView.getViewAttributesProcessor(),
-        registeredView.getCardinalityLimit());
+        registeredView.getCardinalityLimit(),
+        enabled);
   }
 
   /** Record callback measurement from {@link ObservableLongMeasurement}. */
@@ -210,8 +214,10 @@ public MetricData collect(
     aggregatorHandles.forEach(handleReleaser);
     aggregatorHandles.clear();
 
-    return aggregator.toMetricData(
-        resource, instrumentationScopeInfo, metricDescriptor, result, aggregationTemporality);
+    return enabled
+        ? aggregator.toMetricData(
+            resource, instrumentationScopeInfo, metricDescriptor, result, aggregationTemporality)
+        : EmptyMetricData.getInstance();
   }
 
   private Collection<T> collectWithDeltaAggregationTemporality() {
@@ -314,16 +320,7 @@ private Collection<T> collectWithCumulativeAggregationTemporality() {
 
   @Override
   public void setEnabled(boolean enabled) {
-    if (enabled) {
-      if (aggregator == Aggregator.drop()) {
-        aggregator = originalAggregator;
-      }
-    } else {
-      aggregator = Aggregator.drop();
-
-      aggregatorHandles.clear();
-      lastPoints.clear();
-    }
+    this.enabled = enabled;
   }
 
   @Override