Fix metric sdk when multiple readers are present

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

@@ -14,10 +14,11 @@
 import io.opentelemetry.api.metrics.MeterProvider;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.metrics.data.MetricData;
-import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo;
+import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader;
 import io.opentelemetry.sdk.metrics.internal.state.MeterProviderSharedState;
 import io.opentelemetry.sdk.metrics.internal.state.MeterSharedState;
 import java.util.Collection;
+import java.util.List;
 
 /** {@link SdkMeter} is SDK implementation of {@link Meter}. */
 final class SdkMeter implements Meter {
@@ -37,10 +38,11 @@ final class SdkMeter implements Meter {
 
   SdkMeter(
       MeterProviderSharedState meterProviderSharedState,
-      InstrumentationScopeInfo instrumentationScopeInfo) {
+      InstrumentationScopeInfo instrumentationScopeInfo,
+      List<RegisteredReader> registeredReaders) {
     this.instrumentationScopeInfo = instrumentationScopeInfo;
     this.meterProviderSharedState = meterProviderSharedState;
-    this.meterSharedState = MeterSharedState.create(instrumentationScopeInfo);
+    this.meterSharedState = MeterSharedState.create(instrumentationScopeInfo, registeredReaders);
   }
 
   // Visible for testing
@@ -49,10 +51,8 @@ InstrumentationScopeInfo getInstrumentationScopeInfo() {
   }
 
   /** Collects all the metric recordings that changed since the previous call. */
-  Collection<MetricData> collectAll(
-      CollectionInfo collectionInfo, long epochNanos, boolean suppressSynchronousCollection) {
-    return meterSharedState.collectAll(
-        collectionInfo, meterProviderSharedState, epochNanos, suppressSynchronousCollection);
+  Collection<MetricData> collectAll(RegisteredReader registeredReader, long epochNanos) {
+    return meterSharedState.collectAll(registeredReader, meterProviderSharedState, epochNanos);
   }
 
   /** Reset the meter, clearing all registered instruments. */

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

@@ -5,6 +5,8 @@
 
 package io.opentelemetry.sdk.metrics;
 
+import static java.util.stream.Collectors.toList;
+
 import io.opentelemetry.api.metrics.MeterBuilder;
 import io.opentelemetry.api.metrics.MeterProvider;
 import io.opentelemetry.sdk.common.Clock;
@@ -13,24 +15,18 @@
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricReader;
 import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarFilter;
-import io.opentelemetry.sdk.metrics.internal.export.CollectionHandle;
-import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo;
 import io.opentelemetry.sdk.metrics.internal.export.MetricProducer;
+import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader;
 import io.opentelemetry.sdk.metrics.internal.state.MeterProviderSharedState;
 import io.opentelemetry.sdk.metrics.internal.view.ViewRegistry;
 import io.opentelemetry.sdk.resources.Resource;
 import java.io.Closeable;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.Supplier;
 import java.util.logging.Logger;
 
 /** SDK implementation for {@link MeterProvider}. */
@@ -39,13 +35,10 @@ public final class SdkMeterProvider implements MeterProvider, Closeable {
   private static final Logger LOGGER = Logger.getLogger(SdkMeterProvider.class.getName());
   static final String DEFAULT_METER_NAME = "unknown";
 
-  private final List<MetricReader> metricReaders;
-  private final ComponentRegistry<SdkMeter> registry;
+  private final List<RegisteredReader> registeredReaders;
   private final MeterProviderSharedState sharedState;
-  private final Map<CollectionHandle, CollectionInfo> collectionInfoMap;
+  private final ComponentRegistry<SdkMeter> registry;
   private final AtomicBoolean isClosed = new AtomicBoolean(false);
-  private final AtomicLong lastCollectionTimestamp;
-  private final long minimumCollectionIntervalNanos;
 
   /**
    * Returns a new {@link SdkMeterProviderBuilder} for {@link SdkMeterProvider}.
@@ -57,39 +50,27 @@ public static SdkMeterProviderBuilder builder() {
   }
 
   SdkMeterProvider(
-      List<MetricReader> metricReaders,
+      List<RegisteredReader> registeredReaders,
       Clock clock,
       Resource resource,
       ViewRegistry viewRegistry,
-      ExemplarFilter exemplarFilter,
-      long minimumCollectionIntervalNanos) {
-    this.metricReaders = metricReaders;
+      ExemplarFilter exemplarFilter) {
+    this.registeredReaders = registeredReaders;
     this.sharedState =
         MeterProviderSharedState.create(clock, resource, viewRegistry, exemplarFilter);
     this.registry =
         new ComponentRegistry<>(
-            instrumentationLibraryInfo -> new SdkMeter(sharedState, instrumentationLibraryInfo));
-    this.lastCollectionTimestamp =
-        new AtomicLong(clock.nanoTime() - minimumCollectionIntervalNanos);
-    this.minimumCollectionIntervalNanos = minimumCollectionIntervalNanos;
-
-    // Here we construct our own unique handle ids for this SDK.
-    // These are guaranteed to be unique per-reader for this SDK, and only this SDK.
-    // These are *only* mutated in our constructor, and safe to use concurrently after construction.
-    Set<CollectionHandle> collectors = CollectionHandle.mutableSet();
-    collectionInfoMap = new HashMap<>();
-    Supplier<CollectionHandle> handleSupplier = CollectionHandle.createSupplier();
-    for (MetricReader metricReader : metricReaders) {
-      CollectionHandle handle = handleSupplier.get();
-      collectionInfoMap.put(handle, CollectionInfo.create(handle, collectors, metricReader));
-      metricReader.register(new LeasedMetricProducer(handle));
-      collectors.add(handle);
+            instrumentationLibraryInfo ->
+                new SdkMeter(sharedState, instrumentationLibraryInfo, registeredReaders));
+    for (RegisteredReader registeredReader : registeredReaders) {
+      MetricProducer producer = new LeasedMetricProducer(registry, sharedState, registeredReader);
+      registeredReader.getReader().register(producer);
     }
   }
 
   @Override
   public MeterBuilder meterBuilder(String instrumentationScopeName) {
-    if (collectionInfoMap.isEmpty()) {
+    if (registeredReaders.isEmpty()) {
       return MeterProvider.noop().meterBuilder(instrumentationScopeName);
     }
     if (instrumentationScopeName == null || instrumentationScopeName.isEmpty()) {
@@ -109,12 +90,12 @@ void resetForTest() {
    * resulting {@link CompletableResultCode} completes when all complete.
    */
   public CompletableResultCode forceFlush() {
-    if (collectionInfoMap.isEmpty()) {
+    if (registeredReaders.isEmpty()) {
       return CompletableResultCode.ofSuccess();
     }
     List<CompletableResultCode> results = new ArrayList<>();
-    for (CollectionInfo collectionInfo : collectionInfoMap.values()) {
-      results.add(collectionInfo.getReader().forceFlush());
+    for (RegisteredReader registeredReader : registeredReaders) {
+      results.add(registeredReader.getReader().forceFlush());
     }
     return CompletableResultCode.ofAll(results);
   }
@@ -128,11 +109,11 @@ public CompletableResultCode shutdown() {
       LOGGER.info("Multiple close calls");
       return CompletableResultCode.ofSuccess();
     }
-    if (collectionInfoMap.isEmpty()) {
+    if (registeredReaders.isEmpty()) {
       return CompletableResultCode.ofSuccess();
     }
     List<CompletableResultCode> results = new ArrayList<>();
-    for (CollectionInfo info : collectionInfoMap.values()) {
+    for (RegisteredReader info : registeredReaders) {
       results.add(info.getReader().shutdown());
     }
     return CompletableResultCode.ofAll(results);
@@ -152,45 +133,34 @@ public String toString() {
         + ", resource="
         + sharedState.getResource()
         + ", metricReaders="
-        + metricReaders
+        + registeredReaders.stream().map(RegisteredReader::getReader).collect(toList())
         + ", views="
         + sharedState.getViewRegistry().getViews()
         + "}";
   }
 
   /** Helper class to expose registered metric exports. */
-  private class LeasedMetricProducer implements MetricProducer {
-    private final CollectionHandle handle;
-
-    LeasedMetricProducer(CollectionHandle handle) {
-      this.handle = handle;
+  private static class LeasedMetricProducer implements MetricProducer {
+
+    private final ComponentRegistry<SdkMeter> registry;
+    private final MeterProviderSharedState sharedState;
+    private final RegisteredReader registeredReader;
+
+    LeasedMetricProducer(
+        ComponentRegistry<SdkMeter> registry,
+        MeterProviderSharedState sharedState,
+        RegisteredReader registeredReader) {
+      this.registry = registry;
+      this.sharedState = sharedState;
+      this.registeredReader = registeredReader;
     }
 
     @Override
     public Collection<MetricData> collectAllMetrics() {
       Collection<SdkMeter> meters = registry.getComponents();
-      // Suppress too-frequent-collection.
-      long currentNanoTime = sharedState.getClock().nanoTime();
-      long pastNanoTime = lastCollectionTimestamp.get();
-      // It hasn't been long enough since the last collection.
-      boolean disableSynchronousCollection =
-          (currentNanoTime - pastNanoTime) < minimumCollectionIntervalNanos;
-      // If we're not disabling metrics, write the current collection time.
-      // We don't care if this happens in more than one thread, suppression is optimistic, and the
-      // interval is small enough some jitter isn't important.
-      if (!disableSynchronousCollection) {
-        lastCollectionTimestamp.lazySet(currentNanoTime);
-      }
-      CollectionInfo info = collectionInfoMap.get(handle);
-      if (info == null) {
-        throw new IllegalStateException(
-            "No collection info for handle, this is a bug in the OpenTelemetry SDK.");
-      }
-
       List<MetricData> result = new ArrayList<>();
       for (SdkMeter meter : meters) {
-        result.addAll(
-            meter.collectAll(info, sharedState.getClock().now(), disableSynchronousCollection));
+        result.addAll(meter.collectAll(registeredReader, sharedState.getClock().now()));
       }
       return Collections.unmodifiableCollection(result);
     }

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

@@ -5,16 +5,14 @@
 
 package io.opentelemetry.sdk.metrics;
 
-import static io.opentelemetry.api.internal.Utils.checkArgument;
-
 import io.opentelemetry.sdk.common.Clock;
 import io.opentelemetry.sdk.metrics.export.MetricReader;
 import io.opentelemetry.sdk.metrics.internal.debug.SourceInfo;
 import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarFilter;
+import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader;
 import io.opentelemetry.sdk.metrics.internal.view.ViewRegistry;
 import io.opentelemetry.sdk.metrics.internal.view.ViewRegistryBuilder;
 import io.opentelemetry.sdk.resources.Resource;
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
@@ -29,19 +27,11 @@ public final class SdkMeterProviderBuilder {
    */
   private static final ExemplarFilter DEFAULT_EXEMPLAR_FILTER = ExemplarFilter.sampleWithTraces();
 
-  /**
-   * By default, the minimum collection interval is 0ns.
-   *
-   * @see #setMinimumCollectionInterval(Duration)
-   */
-  private static final long DEFAULT_MIN_COLLECTION_INTERVAL_NANOS = 0;
-
   private Clock clock = Clock.getDefault();
   private Resource resource = Resource.getDefault();
   private final ViewRegistryBuilder viewRegistryBuilder = ViewRegistry.builder();
-  private final List<MetricReader> metricReaders = new ArrayList<>();
+  private final List<RegisteredReader> registeredReaders = new ArrayList<>();
   private ExemplarFilter exemplarFilter = DEFAULT_EXEMPLAR_FILTER;
-  private long minimumCollectionIntervalNanos = DEFAULT_MIN_COLLECTION_INTERVAL_NANOS;
 
   SdkMeterProviderBuilder() {}
 
@@ -123,21 +113,7 @@ public SdkMeterProviderBuilder registerView(InstrumentSelector selector, View vi
    * @return this
    */
   public SdkMeterProviderBuilder registerMetricReader(MetricReader reader) {
-    metricReaders.add(reader);
-    return this;
-  }
-
-  /**
-   * Configure the minimum duration between synchronous collections. If collections occur more
-   * frequently than this, synchronous collection will be suppressed.
-   *
-   * @param duration The duration.
-   * @return this
-   */
-  SdkMeterProviderBuilder setMinimumCollectionInterval(Duration duration) {
-    Objects.requireNonNull(duration, "duration");
-    checkArgument(!duration.isNegative(), "duration must not be negative");
-    minimumCollectionIntervalNanos = duration.toNanos();
+    registeredReaders.add(RegisteredReader.create(reader));
     return this;
   }
 
@@ -147,11 +123,6 @@ SdkMeterProviderBuilder setMinimumCollectionInterval(Duration duration) {
    */
   public SdkMeterProvider build() {
     return new SdkMeterProvider(
-        metricReaders,
-        clock,
-        resource,
-        viewRegistryBuilder.build(),
-        exemplarFilter,
-        minimumCollectionIntervalNanos);
+        registeredReaders, clock, resource, viewRegistryBuilder.build(), exemplarFilter);
   }
 }

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

@@ -13,7 +13,6 @@
 import io.opentelemetry.sdk.metrics.internal.view.StringPredicates;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
-import java.time.Duration;
 import java.util.function.Predicate;
 
 /**
@@ -43,27 +42,6 @@ public static void setExemplarFilter(
     }
   }
 
-  /**
-   * Reflectively set the minimum duration between synchronous collections for the {@link
-   * SdkMeterProviderBuilder}. If collections occur more frequently than this, synchronous
-   * collection will be suppressed.
-   *
-   * @param duration The duration.
-   */
-  public static void setMinimumCollectionInterval(
-      SdkMeterProviderBuilder sdkMeterProviderBuilder, Duration duration) {
-    try {
-      Method method =
-          SdkMeterProviderBuilder.class.getDeclaredMethod(
-              "setMinimumCollectionInterval", Duration.class);
-      method.setAccessible(true);
-      method.invoke(sdkMeterProviderBuilder, duration);
-    } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
-      throw new IllegalStateException(
-          "Error calling setMinimumCollectionInterval on SdkMeterProviderBuilder", e);
-    }
-  }
-
   /**
    * Reflectively add an {@link AttributesProcessor} to the {@link ViewBuilder} which appends
    * key-values from baggage to all measurements.