Add setMeterConfigurator support to MeterProvider

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

@@ -88,7 +88,8 @@ final class SdkMeter implements Meter {
   private final MeterProviderSharedState meterProviderSharedState;
   private final InstrumentationScopeInfo instrumentationScopeInfo;
   private final Map<RegisteredReader, MetricStorageRegistry> readerStorageRegistries;
-  private final boolean meterEnabled;
+
+  private boolean meterEnabled;
 
   SdkMeter(
       MeterProviderSharedState meterProviderSharedState,
@@ -103,6 +104,18 @@ final class SdkMeter implements Meter {
     this.meterEnabled = meterConfig.isEnabled();
   }
 
+  void updateMeterConfig(MeterConfig meterConfig) {
+    meterEnabled = meterConfig.isEnabled();
+
+    for (RegisteredReader registeredReader : readerStorageRegistries.keySet()) {
+      Collection<MetricStorage> storages =
+          Objects.requireNonNull(readerStorageRegistries.get(registeredReader)).getStorages();
+      for (MetricStorage storage : storages) {
+        storage.setEnabled(meterEnabled);
+      }
+    }
+  }
+
   // Visible for testing
   InstrumentationScopeInfo getInstrumentationScopeInfo() {
     return instrumentationScopeInfo;

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

@@ -52,9 +52,10 @@ public final class SdkMeterProvider implements MeterProvider, Closeable {
   private final List<MetricProducer> metricProducers;
   private final MeterProviderSharedState sharedState;
   private final ComponentRegistry<SdkMeter> registry;
-  private final ScopeConfigurator<MeterConfig> meterConfigurator;
   private final AtomicBoolean isClosed = new AtomicBoolean(false);
 
+  private ScopeConfigurator<MeterConfig> meterConfigurator;
+
   /** Returns a new {@link SdkMeterProviderBuilder} for {@link SdkMeterProvider}. */
   public static SdkMeterProviderBuilder builder() {
     return new SdkMeterProviderBuilder();
@@ -105,6 +106,15 @@ private MeterConfig getMeterConfig(InstrumentationScopeInfo instrumentationScope
     return meterConfig == null ? MeterConfig.defaultConfig() : meterConfig;
   }
 
+  void setMeterConfigurator(ScopeConfigurator<MeterConfig> meterConfigurator) {
+    this.meterConfigurator = meterConfigurator;
+    this.registry
+        .getComponents()
+        .forEach(
+            sdkMeter ->
+                sdkMeter.updateMeterConfig(getMeterConfig(sdkMeter.getInstrumentationScopeInfo())));
+  }
+
   @Override
   public MeterBuilder meterBuilder(String instrumentationScopeName) {
     if (registeredReaders.isEmpty()) {

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/SdkMeterProviderUtil.java

@@ -49,6 +49,19 @@
     return sdkMeterProviderBuilder;
   }
 
+  /** Reflectively set the {@link ScopeConfigurator} to the {@link SdkMeterProvider}. */
+  public static void setMeterConfigurator(
+      SdkMeterProvider sdkMeterProvider, ScopeConfigurator<MeterConfig> scopeConfigurator) {
+    try {
+      Method method =
+          SdkMeterProvider.class.getDeclaredMethod("setMeterConfigurator", ScopeConfigurator.class);
+      method.setAccessible(true);
+      method.invoke(sdkMeterProvider, scopeConfigurator);
+    } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
+      throw new IllegalStateException("Error calling setMeterConfigurator on SdkMeterProvider", e);
+    }
+  }
+
   /** Reflectively set the {@link ScopeConfigurator} to the {@link SdkMeterProviderBuilder}. */
   public static SdkMeterProviderBuilder setMeterConfigurator(
       SdkMeterProviderBuilder sdkMeterProviderBuilder,

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

@@ -7,7 +7,6 @@
 
 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;
@@ -27,8 +26,9 @@
 @Immutable
 public interface Aggregator<T extends PointData, U extends ExemplarData> {
   /** Returns the drop aggregator, an aggregator that drops measurements. */
-  static Aggregator<?, DoubleExemplarData> drop() {
-    return DropAggregator.INSTANCE;
+  @SuppressWarnings("unchecked")
+  static <T extends PointData, U extends ExemplarData> Aggregator<T, U> drop() {
+    return (Aggregator<T, U>) DropAggregator.INSTANCE;
   }
 
   /**

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

@@ -8,7 +8,6 @@
 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;
@@ -25,7 +24,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, DoubleExemplarData> {
+public final class DropAggregator implements Aggregator<PointData, ExemplarData> {
 
   private static final PointData POINT_DATA =
       new PointData() {
@@ -50,16 +49,16 @@ public List<? extends ExemplarData> getExemplars() {
         }
       };
 
-  public static final Aggregator<PointData, DoubleExemplarData> INSTANCE = new DropAggregator();
+  public static final Aggregator<PointData, ExemplarData> INSTANCE = new DropAggregator();
 
-  private static final AggregatorHandle<PointData, DoubleExemplarData> HANDLE =
-      new AggregatorHandle<PointData, DoubleExemplarData>(ExemplarReservoir.doubleNoSamples()) {
+  private static final AggregatorHandle<PointData, ExemplarData> HANDLE =
+      new AggregatorHandle<PointData, ExemplarData>(ExemplarReservoir.anyNoSamples()) {
         @Override
         protected PointData doAggregateThenMaybeReset(
             long startEpochNanos,
             long epochNanos,
             Attributes attributes,
-            List<DoubleExemplarData> exemplars,
+            List<ExemplarData> exemplars,
             boolean reset) {
           return POINT_DATA;
         }
@@ -74,7 +73,7 @@ protected void doRecordDouble(double value) {}
   private DropAggregator() {}
 
   @Override
-  public AggregatorHandle<PointData, DoubleExemplarData> createHandle() {
+  public AggregatorHandle<PointData, ExemplarData> createHandle() {
     return HANDLE;
   }
 

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

@@ -55,6 +55,11 @@ 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,6 +20,7 @@ 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

@@ -53,9 +53,11 @@ 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
@@ -100,6 +102,7 @@ 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.reusablePointsPool = new ObjectPool<>(aggregator::createReusablePoint);
@@ -309,6 +312,20 @@ private Collection<T> collectWithCumulativeAggregationTemporality() {
     return currentPoints;
   }
 
+  @Override
+  public void setEnabled(boolean enabled) {
+    if (enabled) {
+      if (aggregator == Aggregator.drop()) {
+        aggregator = originalAggregator;
+      }
+    } else {
+      aggregator = Aggregator.drop();
+
+      aggregatorHandles.clear();
+      lastPoints.clear();
+    }
+  }
+
   @Override
   public boolean isEmpty() {
     return aggregator == Aggregator.drop();