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,28 +104,41 @@ 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;
   }
 
   /** 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());
@@ -275,7 +289,8 @@ WriteableMetricStorage registerSynchronousMetricStorage(InstrumentDescriptor ins
                     reader,
                     registeredView,
                     instrument,
-                    meterProviderSharedState.getExemplarFilter())));
+                    meterProviderSharedState.getExemplarFilter(),
+                    meterEnabled)));
       }
     }
 
@@ -301,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/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 @@ public static SdkMeterProviderBuilder setExemplarFilter(
     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/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;
@@ -86,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,6 +106,7 @@ private AsynchronousMetricStorage(
     this.aggregator = 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();
@@ -125,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);
@@ -140,7 +146,8 @@ AsynchronousMetricStorage<T, U> create(
         metricDescriptor,
         aggregator,
         registeredView.getViewAttributesProcessor(),
-        registeredView.getCardinalityLimit());
+        registeredView.getCardinalityLimit(),
+        enabled);
   }
 
   /** Record callback measurement from {@link ObservableLongMeasurement}. */
@@ -207,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() {
@@ -310,7 +319,7 @@ private Collection<T> collectWithCumulativeAggregationTemporality() {
   }
 
   @Override
-  public boolean isEmpty() {
-    return aggregator == Aggregator.drop();
+  public void setEnabled(boolean enabled) {
+    this.enabled = enabled;
   }
 }

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

@@ -74,12 +74,15 @@ public final class DefaultSynchronousMetricStorage<T extends PointData, U extend
   private final ConcurrentLinkedQueue<AggregatorHandle<T, U>> aggregatorHandlePool =
       new ConcurrentLinkedQueue<>();
 
+  private boolean enabled;
+
   DefaultSynchronousMetricStorage(
       RegisteredReader registeredReader,
       MetricDescriptor metricDescriptor,
       Aggregator<T, U> aggregator,
       AttributesProcessor attributesProcessor,
-      int maxCardinality) {
+      int maxCardinality,
+      boolean enabled) {
     this.registeredReader = registeredReader;
     this.metricDescriptor = metricDescriptor;
     this.aggregationTemporality =
@@ -90,6 +93,7 @@ public final class DefaultSynchronousMetricStorage<T extends PointData, U extend
     this.attributesProcessor = attributesProcessor;
     this.maxCardinality = maxCardinality - 1;
     this.memoryMode = registeredReader.getReader().getMemoryMode();
+    this.enabled = enabled;
   }
 
   // Visible for testing
@@ -99,6 +103,9 @@ Queue<AggregatorHandle<T, U>> getAggregatorHandlePool() {
 
   @Override
   public void recordLong(long value, Attributes attributes, Context context) {
+    if (!enabled) {
+      return;
+    }
     AggregatorHolder<T, U> aggregatorHolder = getHolderForRecord();
     try {
       AggregatorHandle<T, U> handle =
@@ -111,6 +118,9 @@ public void recordLong(long value, Attributes attributes, Context context) {
 
   @Override
   public void recordDouble(double value, Attributes attributes, Context context) {
+    if (!enabled) {
+      return;
+    }
     if (Double.isNaN(value)) {
       logger.log(
           Level.FINE,
@@ -131,9 +141,14 @@ public void recordDouble(double value, Attributes attributes, Context context) {
     }
   }
 
+  @Override
+  public void setEnabled(boolean enabled) {
+    this.enabled = enabled;
+  }
+
   @Override
   public boolean isEnabled() {
-    return true;
+    return enabled;
   }
 
   /**
@@ -299,7 +314,7 @@ public MetricData collect(
       previousCollectionAggregatorHandles = aggregatorHandles;
     }
 
-    if (points.isEmpty()) {
+    if (points.isEmpty() || !enabled) {
       return EmptyMetricData.getInstance();
     }
 

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

@@ -44,4 +44,9 @@ public void recordDouble(double value, Attributes attributes, Context context) {
   public boolean isEnabled() {
     return false;
   }
+
+  @Override
+  public void setEnabled(boolean enabled) {
+    // do nothing
+  }
 }

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

@@ -48,14 +48,5 @@ MetricData collect(
       long startEpochNanos,
       long epochNanos);
 
-  /**
-   * Determines whether this storage is an empty metric storage.
-   *
-   * <p>Uses the reference comparison since {@link EmptyMetricStorage} is singleton.
-   *
-   * @return true if is empty.
-   */
-  default boolean isEmpty() {
-    return this == EmptyMetricStorage.INSTANCE;
-  }
+  void setEnabled(boolean enabled);
 }

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

@@ -40,7 +40,8 @@ static <T extends PointData, U extends ExemplarData> SynchronousMetricStorage cr
       RegisteredReader registeredReader,
       RegisteredView registeredView,
       InstrumentDescriptor instrumentDescriptor,
-      ExemplarFilter exemplarFilter) {
+      ExemplarFilter exemplarFilter,
+      boolean enabled) {
     View view = registeredView.getView();
     MetricDescriptor metricDescriptor =
         MetricDescriptor.create(view, registeredView.getViewSourceInfo(), instrumentDescriptor);
@@ -57,6 +58,7 @@ static <T extends PointData, U extends ExemplarData> SynchronousMetricStorage cr
         metricDescriptor,
         aggregator,
         registeredView.getViewAttributesProcessor(),
-        registeredView.getCardinalityLimit());
+        registeredView.getCardinalityLimit(),
+        enabled);
   }
 }

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

@@ -28,8 +28,10 @@
 import io.opentelemetry.context.Scope;
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
+import io.opentelemetry.sdk.internal.ScopeConfigurator;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricReader;
+import io.opentelemetry.sdk.metrics.internal.MeterConfig;
 import io.opentelemetry.sdk.metrics.internal.SdkMeterProviderUtil;
 import io.opentelemetry.sdk.metrics.internal.view.ViewRegistry;
 import io.opentelemetry.sdk.resources.Resource;
@@ -47,6 +49,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -1030,6 +1033,34 @@ void resetForTest() {
                         sum -> sum.isCumulative().hasPointsSatisfying(point -> point.hasValue(1))));
   }
 
+  private static ScopeConfigurator<MeterConfig> flipConfigurator(boolean enabled) {
+    return scopeInfo -> enabled ? MeterConfig.disabled() : MeterConfig.enabled();
+  }
+
+  @Test
+  void propagatesEnablementToMeterDirectly() {
+    SdkMeterProvider meterProvider =
+        SdkMeterProvider.builder().registerMetricReader(InMemoryMetricReader.create()).build();
+    SdkMeter meter = (SdkMeter) meterProvider.get("test");
+    boolean isEnabled = meter.isMeterEnabled();
+
+    meterProvider.setMeterConfigurator(flipConfigurator(isEnabled));
+
+    Assertions.assertThat(meter.isMeterEnabled()).isEqualTo(!isEnabled);
+  }
+
+  @Test
+  void propagatesEnablementToMeterByUtil() {
+    SdkMeterProvider sdkMeterProvider =
+        SdkMeterProvider.builder().registerMetricReader(InMemoryMetricReader.create()).build();
+    SdkMeter sdkMeter = (SdkMeter) sdkMeterProvider.get("test");
+    boolean isEnabled = sdkMeter.isMeterEnabled();
+
+    SdkMeterProviderUtil.setMeterConfigurator(sdkMeterProvider, flipConfigurator(isEnabled));
+
+    Assertions.assertThat(sdkMeter.isMeterEnabled()).isEqualTo(!isEnabled);
+  }
+
   private static void registerViewForAllTypes(
       SdkMeterProviderBuilder meterProviderBuilder, Aggregation aggregation) {
     for (InstrumentType instrumentType : InstrumentType.values()) {

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

@@ -18,6 +18,7 @@
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.metrics.MeterProvider;
 import io.opentelemetry.internal.testing.slf4j.SuppressLogger;
+import io.opentelemetry.sdk.metrics.internal.MeterConfig;
 import io.opentelemetry.sdk.metrics.internal.state.MetricStorageRegistry;
 import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
 import java.util.Locale;
@@ -481,4 +482,16 @@ void stringRepresentation() {
                 + "attributes={}"
                 + "}}");
   }
+
+  @Test
+  void updateEnabled() {
+    SdkMeterProvider sdkMeterProvider =
+        SdkMeterProvider.builder().registerMetricReader(InMemoryMetricReader.create()).build();
+    SdkMeter meter = (SdkMeter) sdkMeterProvider.get("test");
+
+    meter.updateMeterConfig(MeterConfig.disabled());
+    assertThat(meter.isMeterEnabled()).isFalse();
+    meter.updateMeterConfig(MeterConfig.enabled());
+    assertThat(meter.isMeterEnabled()).isTrue();
+  }
 }