diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeter.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeter.java index 905d0851561..397f767c1c2 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeter.java +++ b/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 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 collectAll( - CollectionInfo collectionInfo, long epochNanos, boolean suppressSynchronousCollection) { - return meterSharedState.collectAll( - collectionInfo, meterProviderSharedState, epochNanos, suppressSynchronousCollection); + Collection collectAll(RegisteredReader registeredReader, long epochNanos) { + return meterSharedState.collectAll(registeredReader, meterProviderSharedState, epochNanos); } /** Reset the meter, clearing all registered instruments. */ diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeterProvider.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeterProvider.java index e0a7149b726..99b2ff46f41 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeterProvider.java +++ b/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,9 +15,8 @@ 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; @@ -23,14 +24,9 @@ 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 metricReaders; - private final ComponentRegistry registry; + private final List registeredReaders; private final MeterProviderSharedState sharedState; - private final Map collectionInfoMap; + private final ComponentRegistry 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 metricReaders, + List 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 collectors = CollectionHandle.mutableSet(); - collectionInfoMap = new HashMap<>(); - Supplier 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 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 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 registry; + private final MeterProviderSharedState sharedState; + private final RegisteredReader registeredReader; + + LeasedMetricProducer( + ComponentRegistry registry, + MeterProviderSharedState sharedState, + RegisteredReader registeredReader) { + this.registry = registry; + this.sharedState = sharedState; + this.registeredReader = registeredReader; } @Override public Collection collectAllMetrics() { Collection 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 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); } diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeterProviderBuilder.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeterProviderBuilder.java index a12ac054be4..9aecdc41316 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeterProviderBuilder.java +++ b/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 metricReaders = new ArrayList<>(); + private final List 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); } } diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/SdkMeterProviderUtil.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/SdkMeterProviderUtil.java index 965a1d6a0ee..eb89bc35435 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/SdkMeterProviderUtil.java +++ b/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. diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/export/CollectionHandle.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/export/CollectionHandle.java deleted file mode 100644 index ebc97534acf..00000000000 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/export/CollectionHandle.java +++ /dev/null @@ -1,155 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - */ - -package io.opentelemetry.sdk.metrics.internal.export; - -import java.util.AbstractSet; -import java.util.BitSet; -import java.util.Collection; -import java.util.Iterator; -import java.util.NoSuchElementException; -import java.util.Set; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Supplier; -import javax.annotation.Nullable; - -/** - * A handle for a collection-pipeline of metrics. - * - *

This class provides an efficient means of leasing and tracking exporters. - * - *

This class is internal and is hence not for public use. Its APIs are unstable and can change - * at any time. - */ -public final class CollectionHandle { - /** The index of this handle. */ - private final int index; - - private CollectionHandle(int index) { - this.index = index; - } - - /** Construct a new (efficient) mutable set for tracking collection handles. */ - public static Set mutableSet() { - return new CollectionHandleSet(); - } - - /** - * Construct a new (mutable) set consisting of the passed in collection handles. - * - *

Used by tests. - */ - static Set of(CollectionHandle... handles) { - Set result = mutableSet(); - for (CollectionHandle handle : handles) { - result.add(handle); - } - return result; - } - - /** - * Construct a new supplier of collection handles. - * - *

Handles returned by this supplier should not be used with unique handles produced by any - * other supplier. - */ - public static Supplier createSupplier() { - return new Supplier() { - private final AtomicInteger nextIndex = new AtomicInteger(1); - - @Override - public CollectionHandle get() { - return new CollectionHandle(nextIndex.getAndIncrement()); - } - }; - } - - @Override - public int hashCode() { - return index; - } - - @Override - public boolean equals(@Nullable Object other) { - if (this == other) { - return true; - } - if (other == null) { - return false; - } - if (!(other instanceof CollectionHandle)) { - return false; - } - return index == ((CollectionHandle) other).index; - } - - @Override - public String toString() { - return "CollectionHandle(" + index + ")"; - } - - /** An optimised bitset version of {@code Set}. */ - private static class CollectionHandleSet extends AbstractSet { - private final BitSet storage = new BitSet(); - - @Override - public Iterator iterator() { - return new MyIterator(); - } - - @Override - public boolean add(CollectionHandle handle) { - if (storage.get(handle.index)) { - return false; - } - storage.set(handle.index); - return true; - } - - @Override - public boolean contains(Object handle) { - if (handle instanceof CollectionHandle) { - return storage.get(((CollectionHandle) handle).index); - } - return false; - } - - @Override - public boolean containsAll(Collection other) { - if (other instanceof CollectionHandleSet) { - BitSet result = (BitSet) storage.clone(); - BitSet otherStorage = ((CollectionHandleSet) other).storage; - result.and(otherStorage); - return result.equals(otherStorage); - } - return super.containsAll(other); - } - - private class MyIterator implements Iterator { - private int currentIndex = 0; - - @Override - public boolean hasNext() { - return (currentIndex != -1) && storage.nextSetBit(currentIndex) != -1; - } - - @Override - public CollectionHandle next() { - int result = storage.nextSetBit(currentIndex); - if (result != -1) { - // Start checking next bit next time. - currentIndex = result + 1; - return new CollectionHandle(result); - } - throw new NoSuchElementException("Called `.next` on iterator with no remaining values."); - } - } - - @Override - public int size() { - return storage.cardinality(); - } - } -} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/export/CollectionInfo.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/export/CollectionInfo.java deleted file mode 100644 index 9b5b8a27984..00000000000 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/export/CollectionInfo.java +++ /dev/null @@ -1,44 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - */ - -package io.opentelemetry.sdk.metrics.internal.export; - -import com.google.auto.value.AutoValue; -import io.opentelemetry.sdk.metrics.InstrumentType; -import io.opentelemetry.sdk.metrics.data.AggregationTemporality; -import io.opentelemetry.sdk.metrics.export.MetricReader; -import java.util.Set; -import javax.annotation.concurrent.Immutable; - -/** - * Information about a {@link MetricReader} used when collecting metrics. - * - *

This class is internal and is hence not for public use. Its APIs are unstable and can change - * at any time. - */ -@AutoValue -@Immutable -public abstract class CollectionInfo { - - /** Construct a new collection info object storing information for collection against a reader. */ - public static CollectionInfo create( - CollectionHandle handle, Set allCollectors, MetricReader reader) { - return new AutoValue_CollectionInfo(handle, allCollectors, reader); - } - - CollectionInfo() {} - - /** The current collection. */ - public abstract CollectionHandle getCollector(); - /** The set of all possible collectors. */ - public abstract Set getAllCollectors(); - - public abstract MetricReader getReader(); - - /** The default aggregation temporality for the current metric collection. */ - public final AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) { - return getReader().getAggregationTemporality(instrumentType); - } -} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/export/RegisteredReader.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/export/RegisteredReader.java new file mode 100644 index 00000000000..6e8feb41b00 --- /dev/null +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/export/RegisteredReader.java @@ -0,0 +1,58 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.metrics.internal.export; + +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.export.MetricReader; +import java.util.concurrent.atomic.AtomicInteger; +import javax.annotation.Nullable; + +/** + * Represents a {@link MetricReader} registered with {@link SdkMeterProvider}. + * + *

This class is internal and is hence not for public use. Its APIs are unstable and can change + * at any time. + */ +public class RegisteredReader { + + private static final AtomicInteger ID_COUNTER = new AtomicInteger(1); + private final int id = ID_COUNTER.incrementAndGet(); + private final MetricReader metricReader; + + /** Construct a new collection info object storing information for collection against a reader. */ + public static RegisteredReader create(MetricReader reader) { + return new RegisteredReader(reader); + } + + private RegisteredReader(MetricReader metricReader) { + this.metricReader = metricReader; + } + + public MetricReader getReader() { + return metricReader; + } + + @Override + public int hashCode() { + return id; + } + + @Override + public boolean equals(@Nullable Object o) { + if (this == o) { + return true; + } + if (!(o instanceof RegisteredReader)) { + return false; + } + return id == ((RegisteredReader) o).id; + } + + @Override + public String toString() { + return "RegisteredReader{" + id + "}"; + } +} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorage.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorage.java index 39116b28c38..7e6bd67ae46 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorage.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorage.java @@ -20,7 +20,7 @@ import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor; import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor; import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarFilter; -import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; import io.opentelemetry.sdk.metrics.internal.view.AttributesProcessor; import io.opentelemetry.sdk.metrics.internal.view.RegisteredView; import io.opentelemetry.sdk.resources.Resource; @@ -39,17 +39,25 @@ final class AsynchronousMetricStorage implements Metr private static final Logger logger = Logger.getLogger(AsynchronousMetricStorage.class.getName()); private final ThrottlingLogger throttlingLogger = new ThrottlingLogger(logger); + private final RegisteredReader registeredReader; private final MetricDescriptor metricDescriptor; + private final AggregationTemporality aggregationTemporality; private final TemporalMetricStorage metricStorage; private final Aggregator aggregator; private final AttributesProcessor attributesProcessor; private Map accumulations = new HashMap<>(); private AsynchronousMetricStorage( + RegisteredReader registeredReader, MetricDescriptor metricDescriptor, Aggregator aggregator, AttributesProcessor attributesProcessor) { + this.registeredReader = registeredReader; this.metricDescriptor = metricDescriptor; + this.aggregationTemporality = + registeredReader + .getReader() + .getAggregationTemporality(metricDescriptor.getSourceInstrument().getType()); this.metricStorage = new TemporalMetricStorage<>(aggregator, /* isSynchronous= */ false); this.aggregator = aggregator; this.attributesProcessor = attributesProcessor; @@ -60,7 +68,9 @@ private AsynchronousMetricStorage( */ // TODO(anuraaga): The cast to generic type here looks suspicious. static AsynchronousMetricStorage create( - RegisteredView registeredView, InstrumentDescriptor instrumentDescriptor) { + RegisteredReader registeredReader, + RegisteredView registeredView, + InstrumentDescriptor instrumentDescriptor) { View view = registeredView.getView(); MetricDescriptor metricDescriptor = MetricDescriptor.create(view, registeredView.getViewSourceInfo(), instrumentDescriptor); @@ -68,7 +78,10 @@ static AsynchronousMetricStorage create( ((AggregatorFactory) view.getAggregation()) .createAggregator(instrumentDescriptor, ExemplarFilter.neverSample()); return new AsynchronousMetricStorage<>( - metricDescriptor, aggregator, registeredView.getViewAttributesProcessor()); + registeredReader, + metricDescriptor, + aggregator, + registeredView.getViewAttributesProcessor()); } /** Record callback long measurements from {@link ObservableLongMeasurement}. */ @@ -119,24 +132,25 @@ public MetricDescriptor getMetricDescriptor() { return metricDescriptor; } + @Override + public RegisteredReader getRegisteredReader() { + return registeredReader; + } + @Override public MetricData collectAndReset( - CollectionInfo collectionInfo, Resource resource, InstrumentationScopeInfo instrumentationScopeInfo, long startEpochNanos, - long epochNanos, - boolean suppressSynchronousCollection) { - AggregationTemporality temporality = - collectionInfo.getAggregationTemporality(metricDescriptor.getSourceInstrument().getType()); + long epochNanos) { Map currentAccumulations = accumulations; accumulations = new HashMap<>(); return metricStorage.buildMetricFor( - collectionInfo.getCollector(), + registeredReader, resource, instrumentationScopeInfo, getMetricDescriptor(), - temporality, + aggregationTemporality, currentAccumulations, startEpochNanos, epochNanos); diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/CallbackRegistration.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/CallbackRegistration.java index 5221ff858b0..88714fdc50e 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/CallbackRegistration.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/CallbackRegistration.java @@ -12,10 +12,13 @@ import io.opentelemetry.api.metrics.ObservableLongMeasurement; 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.function.Consumer; +import java.util.function.Function; import java.util.logging.Level; import java.util.logging.Logger; +import javax.annotation.Nullable; /** * A registered callback of an asynchronous instrument. @@ -30,17 +33,18 @@ public class CallbackRegistration { private final InstrumentDescriptor instrumentDescriptor; private final Consumer callback; private final T measurement; - private final boolean noStoragesRegistered; + private final List> storages; + @Nullable private volatile RegisteredReader activeReader; private CallbackRegistration( InstrumentDescriptor instrumentDescriptor, Consumer callback, - T measurement, - List> storages) { + List> storages, + Function, T> measurementProvider) { this.instrumentDescriptor = instrumentDescriptor; this.callback = callback; - this.measurement = measurement; - this.noStoragesRegistered = storages.size() == 0; + this.measurement = measurementProvider.apply(this); + this.storages = storages; } /** Create a {@link CallbackRegistration} for a {@code double} asynchronous instrument. */ @@ -48,10 +52,11 @@ public static CallbackRegistration createDouble( InstrumentDescriptor instrumentDescriptor, Consumer callback, List> asyncMetricStorages) { - ObservableDoubleMeasurement measurement = - new ObservableDoubleMeasurementImpl(asyncMetricStorages); return new CallbackRegistration<>( - instrumentDescriptor, callback, measurement, asyncMetricStorages); + instrumentDescriptor, + callback, + asyncMetricStorages, + callbackRegistration -> callbackRegistration.new ObservableDoubleMeasurementImpl()); } /** Create a {@link CallbackRegistration} for a {@code long} asynchronous instrument. */ @@ -59,21 +64,25 @@ public static CallbackRegistration createLong( InstrumentDescriptor instrumentDescriptor, Consumer callback, List> asyncMetricStorages) { - ObservableLongMeasurement measurement = new ObservableLongMeasurementImpl(asyncMetricStorages); return new CallbackRegistration<>( - instrumentDescriptor, callback, measurement, asyncMetricStorages); + instrumentDescriptor, + callback, + asyncMetricStorages, + callbackRegistration -> callbackRegistration.new ObservableLongMeasurementImpl()); } public InstrumentDescriptor getInstrumentDescriptor() { return instrumentDescriptor; } - void invokeCallback() { + void invokeCallback(RegisteredReader reader) { // Return early if no storages are registered - if (noStoragesRegistered) { + if (storages.isEmpty()) { return; } try { + // Set the active reader so that measurements are only recorded to relevant storages + activeReader = reader; callback.accept(measurement); } catch (Throwable e) { propagateIfFatal(e); @@ -83,17 +92,12 @@ void invokeCallback() { + instrumentDescriptor.getName() + ".", e); + } finally { + activeReader = null; } } - private static class ObservableDoubleMeasurementImpl implements ObservableDoubleMeasurement { - - private final List> asyncMetricStorages; - - private ObservableDoubleMeasurementImpl( - List> asyncMetricStorages) { - this.asyncMetricStorages = asyncMetricStorages; - } + private class ObservableDoubleMeasurementImpl implements ObservableDoubleMeasurement { @Override public void record(double value) { @@ -102,20 +106,15 @@ public void record(double value) { @Override public void record(double value, Attributes attributes) { - for (AsynchronousMetricStorage asyncMetricStorage : asyncMetricStorages) { - asyncMetricStorage.recordDouble(value, attributes); + for (AsynchronousMetricStorage asyncMetricStorage : storages) { + if (asyncMetricStorage.getRegisteredReader().equals(activeReader)) { + asyncMetricStorage.recordDouble(value, attributes); + } } } } - private static class ObservableLongMeasurementImpl implements ObservableLongMeasurement { - - private final List> asyncMetricStorages; - - private ObservableLongMeasurementImpl( - List> asyncMetricStorages) { - this.asyncMetricStorages = asyncMetricStorages; - } + private class ObservableLongMeasurementImpl implements ObservableLongMeasurement { @Override public void record(long value) { @@ -124,8 +123,10 @@ public void record(long value) { @Override public void record(long value, Attributes attributes) { - for (AsynchronousMetricStorage asyncMetricStorage : asyncMetricStorages) { - asyncMetricStorage.recordLong(value, attributes); + for (AsynchronousMetricStorage asyncMetricStorage : storages) { + if (asyncMetricStorage.getRegisteredReader().equals(activeReader)) { + asyncMetricStorage.recordLong(value, attributes); + } } } } diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DefaultSynchronousMetricStorage.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DefaultSynchronousMetricStorage.java index 58ce8c33e8b..38182895840 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DefaultSynchronousMetricStorage.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DefaultSynchronousMetricStorage.java @@ -5,19 +5,27 @@ package io.opentelemetry.sdk.metrics.internal.state; +import static io.opentelemetry.sdk.metrics.internal.state.MetricStorageUtils.MAX_ACCUMULATIONS; + import io.opentelemetry.api.common.Attributes; import io.opentelemetry.context.Context; import io.opentelemetry.sdk.common.InstrumentationScopeInfo; +import io.opentelemetry.sdk.internal.ThrottlingLogger; import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.metrics.data.ExemplarData; import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.metrics.internal.aggregator.Aggregator; +import io.opentelemetry.sdk.metrics.internal.aggregator.AggregatorHandle; import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor; -import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; import io.opentelemetry.sdk.metrics.internal.view.AttributesProcessor; import io.opentelemetry.sdk.resources.Resource; +import java.util.HashMap; import java.util.Map; import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.logging.Level; +import java.util.logging.Logger; /** * Stores aggregated {@link MetricData} for synchronous instruments. @@ -27,20 +35,34 @@ */ public final class DefaultSynchronousMetricStorage implements SynchronousMetricStorage { + + private static final ThrottlingLogger logger = + new ThrottlingLogger(Logger.getLogger(DefaultSynchronousMetricStorage.class.getName())); + private static final BoundStorageHandle NOOP_STORAGE_HANDLE = new NoopBoundHandle(); + + private final RegisteredReader registeredReader; private final MetricDescriptor metricDescriptor; - private final DeltaMetricStorage deltaMetricStorage; + private final AggregationTemporality aggregationTemporality; + private final Aggregator aggregator; + private final ConcurrentHashMap> activeCollectionStorage = + new ConcurrentHashMap<>(); private final TemporalMetricStorage temporalMetricStorage; private final AttributesProcessor attributesProcessor; DefaultSynchronousMetricStorage( + RegisteredReader registeredReader, MetricDescriptor metricDescriptor, Aggregator aggregator, AttributesProcessor attributesProcessor) { - this.attributesProcessor = attributesProcessor; + this.registeredReader = registeredReader; this.metricDescriptor = metricDescriptor; - this.deltaMetricStorage = - new DeltaMetricStorage<>(aggregator, metricDescriptor.getSourceInstrument()); + this.aggregationTemporality = + registeredReader + .getReader() + .getAggregationTemporality(metricDescriptor.getSourceInstrument().getType()); + this.aggregator = aggregator; this.temporalMetricStorage = new TemporalMetricStorage<>(aggregator, /* isSynchronous= */ true); + this.attributesProcessor = attributesProcessor; } // This is a storage handle to use when the attributes processor requires @@ -67,7 +89,43 @@ public BoundStorageHandle bind(Attributes attributes) { // We cannot pre-bind attributes because we need to pull attributes from context. return lateBoundStorageHandle; } - return deltaMetricStorage.bind(attributesProcessor.process(attributes, Context.current())); + return doBind(attributesProcessor.process(attributes, Context.current())); + } + + private BoundStorageHandle doBind(Attributes attributes) { + AggregatorHandle aggregatorHandle = activeCollectionStorage.get(attributes); + if (aggregatorHandle != null && aggregatorHandle.acquire()) { + // At this moment it is guaranteed that the Bound is in the map and will not be removed. + return aggregatorHandle; + } + + // Missing entry or no longer mapped. Try to add a new one if not exceeded cardinality limits. + aggregatorHandle = aggregator.createHandle(); + while (true) { + if (activeCollectionStorage.size() >= MAX_ACCUMULATIONS) { + logger.log( + Level.WARNING, + "Instrument " + + metricDescriptor.getSourceInstrument().getName() + + " has exceeded the maximum allowed accumulations (" + + MAX_ACCUMULATIONS + + ")."); + return NOOP_STORAGE_HANDLE; + } + AggregatorHandle boundAggregatorHandle = + activeCollectionStorage.putIfAbsent(attributes, aggregatorHandle); + if (boundAggregatorHandle != null) { + if (boundAggregatorHandle.acquire()) { + // At this moment it is guaranteed that the Bound is in the map and will not be removed. + return boundAggregatorHandle; + } + // Try to remove the boundAggregator. This will race with the collect method, but only one + // will succeed. + activeCollectionStorage.remove(attributes, boundAggregatorHandle); + continue; + } + return aggregatorHandle; + } } // Overridden to make sure attributes processor can pull baggage. @@ -75,7 +133,7 @@ public BoundStorageHandle bind(Attributes attributes) { public void recordLong(long value, Attributes attributes, Context context) { Objects.requireNonNull(attributes, "attributes"); attributes = attributesProcessor.process(attributes, context); - BoundStorageHandle handle = deltaMetricStorage.bind(attributes); + BoundStorageHandle handle = doBind(attributes); try { handle.recordLong(value, attributes, context); } finally { @@ -88,7 +146,7 @@ public void recordLong(long value, Attributes attributes, Context context) { public void recordDouble(double value, Attributes attributes, Context context) { Objects.requireNonNull(attributes, "attributes"); attributes = attributesProcessor.process(attributes, context); - BoundStorageHandle handle = deltaMetricStorage.bind(attributes); + BoundStorageHandle handle = doBind(attributes); try { handle.recordDouble(value, attributes, context); } finally { @@ -98,27 +156,33 @@ public void recordDouble(double value, Attributes attributes, Context context) { @Override public MetricData collectAndReset( - CollectionInfo collectionInfo, Resource resource, InstrumentationScopeInfo instrumentationScopeInfo, long startEpochNanos, - long epochNanos, - boolean suppressSynchronousCollection) { - AggregationTemporality temporality = - collectionInfo.getAggregationTemporality( - getMetricDescriptor().getSourceInstrument().getType()); - Map result = - deltaMetricStorage.collectFor( - collectionInfo.getCollector(), - collectionInfo.getAllCollectors(), - suppressSynchronousCollection); + long epochNanos) { + // Grab accumulated measurements. + Map accumulations = new HashMap<>(); + for (Map.Entry> entry : activeCollectionStorage.entrySet()) { + boolean unmappedEntry = entry.getValue().tryUnmap(); + if (unmappedEntry) { + // If able to unmap then remove the record from the current Map. This can race with the + // acquire but because we requested a specific value only one will succeed. + activeCollectionStorage.remove(entry.getKey(), entry.getValue()); + } + T accumulation = entry.getValue().accumulateThenReset(entry.getKey()); + if (accumulation == null) { + continue; + } + accumulations.put(entry.getKey(), accumulation); + } + return temporalMetricStorage.buildMetricFor( - collectionInfo.getCollector(), + registeredReader, resource, instrumentationScopeInfo, getMetricDescriptor(), - temporality, - result, + aggregationTemporality, + accumulations, startEpochNanos, epochNanos); } @@ -127,4 +191,22 @@ public MetricData collectAndReset( public MetricDescriptor getMetricDescriptor() { return metricDescriptor; } + + @Override + public RegisteredReader getRegisteredReader() { + return registeredReader; + } + + /** An implementation of {@link BoundStorageHandle} that does not record. */ + private static class NoopBoundHandle implements BoundStorageHandle { + + @Override + public void recordLong(long value, Attributes attributes, Context context) {} + + @Override + public void recordDouble(double value, Attributes attributes, Context context) {} + + @Override + public void release() {} + } } diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaAccumulation.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaAccumulation.java deleted file mode 100644 index c6c5ad990d5..00000000000 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaAccumulation.java +++ /dev/null @@ -1,47 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - */ - -package io.opentelemetry.sdk.metrics.internal.state; - -import io.opentelemetry.api.common.Attributes; -import io.opentelemetry.sdk.metrics.internal.export.CollectionHandle; -import java.util.Map; -import java.util.Set; - -/** - * Synchronous recording of delta-accumulated measurements. - * - *

This stores in-progress metric values that haven't been exported yet. - */ -class DeltaAccumulation { - private final Map recording; - private final Set readers; - - DeltaAccumulation(Map recording) { - this.recording = recording; - this.readers = CollectionHandle.mutableSet(); - } - - /** Returns true if this accumulation was read by the {@link CollectionHandle}. */ - boolean wasReadBy(CollectionHandle handle) { - return readers.contains(handle); - } - - /** Returns true if all readers in the given set have read this accumulation. */ - boolean wasReadByAll(Set handles) { - return readers.containsAll(handles); - } - - /** - * Reads the current delta accumulation. - * - * @param handle The reader of the accumulation. - * @return the accumulation. - */ - Map read(CollectionHandle handle) { - readers.add(handle); - return recording; - } -} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaMetricStorage.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaMetricStorage.java deleted file mode 100644 index fb2d3206736..00000000000 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaMetricStorage.java +++ /dev/null @@ -1,162 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - */ - -package io.opentelemetry.sdk.metrics.internal.state; - -import static io.opentelemetry.sdk.metrics.internal.state.MetricStorageUtils.MAX_ACCUMULATIONS; - -import io.opentelemetry.api.common.Attributes; -import io.opentelemetry.context.Context; -import io.opentelemetry.sdk.internal.ThrottlingLogger; -import io.opentelemetry.sdk.metrics.data.ExemplarData; -import io.opentelemetry.sdk.metrics.internal.aggregator.Aggregator; -import io.opentelemetry.sdk.metrics.internal.aggregator.AggregatorHandle; -import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor; -import io.opentelemetry.sdk.metrics.internal.export.CollectionHandle; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.logging.Level; -import java.util.logging.Logger; -import javax.annotation.concurrent.ThreadSafe; - -/** - * Allows synchronous collection of metrics and reports delta values isolated by collection handle. - * - *

This storage should allow allocation of new aggregation cells for metrics and unique reporting - * of delta accumulations per-collection-handle. - */ -@ThreadSafe -class DeltaMetricStorage { - - private static final ThrottlingLogger logger = - new ThrottlingLogger(Logger.getLogger(DeltaMetricStorage.class.getName())); - private static final BoundStorageHandle NOOP_STORAGE_HANDLE = new NoopBoundHandle(); - - private final Aggregator aggregator; - private final InstrumentDescriptor instrument; - private final ConcurrentHashMap> activeCollectionStorage = - new ConcurrentHashMap<>(); - private final List> unreportedDeltas = new ArrayList<>(); - - DeltaMetricStorage(Aggregator aggregator, InstrumentDescriptor instrument) { - this.aggregator = aggregator; - this.instrument = instrument; - } - - /** - * Allocates memory for a new metric stream, and returns a handle for synchronous recordings. - * - * @param attributes The identifying attributes for the metric stream. - * @return A handle that will (efficiently) record synchronous measurements. - */ - public BoundStorageHandle bind(Attributes attributes) { - AggregatorHandle aggregatorHandle = activeCollectionStorage.get(attributes); - if (aggregatorHandle != null && aggregatorHandle.acquire()) { - // At this moment it is guaranteed that the Bound is in the map and will not be removed. - return aggregatorHandle; - } - - // Missing entry or no longer mapped. Try to add a new one if not exceeded cardinality limits. - aggregatorHandle = aggregator.createHandle(); - while (true) { - if (activeCollectionStorage.size() >= MAX_ACCUMULATIONS) { - logger.log( - Level.WARNING, - "Instrument " - + instrument.getName() - + " has exceeded the maximum allowed accumulations (" - + MAX_ACCUMULATIONS - + ")."); - return NOOP_STORAGE_HANDLE; - } - AggregatorHandle boundAggregatorHandle = - activeCollectionStorage.putIfAbsent(attributes, aggregatorHandle); - if (boundAggregatorHandle != null) { - if (boundAggregatorHandle.acquire()) { - // At this moment it is guaranteed that the Bound is in the map and will not be removed. - return boundAggregatorHandle; - } - // Try to remove the boundAggregator. This will race with the collect method, but only one - // will succeed. - activeCollectionStorage.remove(attributes, boundAggregatorHandle); - continue; - } - return aggregatorHandle; - } - } - - /** - * Returns the latest delta accumulation for a specific collection handle. - * - * @param collector The current reader of metrics. - * @param collectors All possible readers of metrics. - * @param suppressCollection If true, don't actively pull synchronous instruments, measurements - * should be up to date. - * @return The delta accumulation of metrics since the last read of the specified reader. - */ - public synchronized Map collectFor( - CollectionHandle collector, Set collectors, boolean suppressCollection) { - // First we force a collection - if (!suppressCollection) { - collectSynchronousDeltaAccumulationAndReset(); - } - // Now build a delta result. - Map result = new HashMap<>(); - for (DeltaAccumulation point : unreportedDeltas) { - if (!point.wasReadBy(collector)) { - MetricStorageUtils.mergeInPlace(result, point.read(collector), aggregator); - } - } - // Now run a quick cleanup of deltas before returning. - unreportedDeltas.removeIf(delta -> delta.wasReadByAll(collectors)); - return result; - } - - /** - * Collects the currently accumulated measurements from the concurrent-friendly synchronous - * storage. - * - *

All synchronous handles will be collected + reset during this method. Additionally cleanup - * related stale concurrent-map handles will occur. Any {@code null} measurements are ignored. - */ - private synchronized void collectSynchronousDeltaAccumulationAndReset() { - // Grab accumulated measurements. - Map result = new HashMap<>(); - for (Map.Entry> entry : activeCollectionStorage.entrySet()) { - boolean unmappedEntry = entry.getValue().tryUnmap(); - if (unmappedEntry) { - // If able to unmap then remove the record from the current Map. This can race with the - // acquire but because we requested a specific value only one will succeed. - activeCollectionStorage.remove(entry.getKey(), entry.getValue()); - } - T accumulation = entry.getValue().accumulateThenReset(entry.getKey()); - if (accumulation == null) { - continue; - } - // Feed latest batch to the aggregator. - result.put(entry.getKey(), accumulation); - } - if (!result.isEmpty()) { - unreportedDeltas.add(new DeltaAccumulation<>(result)); - } - } - - /** An implementation of {@link BoundStorageHandle} that does not record. */ - private static class NoopBoundHandle implements BoundStorageHandle { - - @Override - public void recordLong(long value, Attributes attributes, Context context) {} - - @Override - public void recordDouble(double value, Attributes attributes, Context context) {} - - @Override - public void release() {} - } -} diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/EmptyMetricStorage.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/EmptyMetricStorage.java index ba99c234f40..15ba0fd4afc 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/EmptyMetricStorage.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/EmptyMetricStorage.java @@ -7,11 +7,16 @@ import io.opentelemetry.api.common.Attributes; import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.common.CompletableResultCode; import io.opentelemetry.sdk.common.InstrumentationScopeInfo; +import io.opentelemetry.sdk.metrics.InstrumentType; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.export.CollectionRegistration; +import io.opentelemetry.sdk.metrics.export.MetricReader; import io.opentelemetry.sdk.metrics.internal.aggregator.EmptyMetricData; import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor; -import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; import io.opentelemetry.sdk.resources.Resource; final class EmptyMetricStorage implements SynchronousMetricStorage { @@ -31,12 +36,38 @@ public void recordDouble(double value, Attributes attributes, Context context) { @Override public void release() {} }; + private final MetricReader emptyReader = + new MetricReader() { + @Override + public void register(CollectionRegistration registration) {} + + @Override + public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) { + return AggregationTemporality.CUMULATIVE; + } + + @Override + public CompletableResultCode forceFlush() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + return CompletableResultCode.ofFailure(); + } + }; + private final RegisteredReader registeredReader = RegisteredReader.create(emptyReader); @Override public MetricDescriptor getMetricDescriptor() { return descriptor; } + @Override + public RegisteredReader getRegisteredReader() { + return registeredReader; + } + @Override public BoundStorageHandle bind(Attributes attributes) { return emptyHandle; @@ -44,12 +75,10 @@ public BoundStorageHandle bind(Attributes attributes) { @Override public MetricData collectAndReset( - CollectionInfo collectionInfo, Resource resource, InstrumentationScopeInfo instrumentationScopeInfo, long startEpochNanos, - long epochNanos, - boolean suppressSynchronousCollection) { + long epochNanos) { return EmptyMetricData.getInstance(); } } diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MeterSharedState.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MeterSharedState.java index 44c87e95b4f..6c7aa839010 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MeterSharedState.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MeterSharedState.java @@ -5,19 +5,24 @@ package io.opentelemetry.sdk.metrics.internal.state; -import static java.util.stream.Collectors.toList; +import static java.util.stream.Collectors.toMap; import io.opentelemetry.api.internal.GuardedBy; import io.opentelemetry.api.metrics.ObservableDoubleMeasurement; import io.opentelemetry.api.metrics.ObservableLongMeasurement; import io.opentelemetry.sdk.common.InstrumentationScopeInfo; +import io.opentelemetry.sdk.metrics.Aggregation; import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor; -import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; +import io.opentelemetry.sdk.metrics.internal.view.RegisteredView; import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.Map; +import java.util.Objects; import java.util.function.Consumer; +import java.util.function.Function; /** * State for a {@code Meter}. @@ -33,18 +38,21 @@ public class MeterSharedState { @GuardedBy("callbackLock") private final List> callbackRegistrations = new ArrayList<>(); + private final Map readerStorageRegistries; + private final InstrumentationScopeInfo instrumentationScopeInfo; - private final MetricStorageRegistry metricStorageRegistry; private MeterSharedState( - InstrumentationScopeInfo instrumentationScopeInfo, - MetricStorageRegistry metricStorageRegistry) { + InstrumentationScopeInfo instrumentationScopeInfo, List registeredReaders) { this.instrumentationScopeInfo = instrumentationScopeInfo; - this.metricStorageRegistry = metricStorageRegistry; + this.readerStorageRegistries = + registeredReaders.stream() + .collect(toMap(Function.identity(), unused -> new MetricStorageRegistry())); } - public static MeterSharedState create(InstrumentationScopeInfo instrumentationScopeInfo) { - return new MeterSharedState(instrumentationScopeInfo, new MetricStorageRegistry()); + public static MeterSharedState create( + InstrumentationScopeInfo instrumentationScopeInfo, List registeredReaders) { + return new MeterSharedState(instrumentationScopeInfo, registeredReaders); } /** @@ -67,17 +75,11 @@ public InstrumentationScopeInfo getInstrumentationScopeInfo() { return instrumentationScopeInfo; } - /** Returns the metric storage for metrics in this {@code Meter}. */ - MetricStorageRegistry getMetricStorageRegistry() { - return metricStorageRegistry; - } - /** Collects all accumulated metric stream points. */ public List collectAll( - CollectionInfo collectionInfo, + RegisteredReader registeredReader, MeterProviderSharedState meterProviderSharedState, - long epochNanos, - boolean suppressSynchronousCollection) { + long epochNanos) { List> currentRegisteredCallbacks; synchronized (callbackLock) { currentRegisteredCallbacks = new ArrayList<>(callbackRegistrations); @@ -85,20 +87,19 @@ public List collectAll( // Collections across all readers are sequential synchronized (collectLock) { for (CallbackRegistration callbackRegistration : currentRegisteredCallbacks) { - callbackRegistration.invokeCallback(); + callbackRegistration.invokeCallback(registeredReader); } - Collection metrics = getMetricStorageRegistry().getMetrics(); - List result = new ArrayList<>(metrics.size()); - for (MetricStorage metric : metrics) { + Collection storages = + Objects.requireNonNull(readerStorageRegistries.get(registeredReader)).getStorages(); + List result = new ArrayList<>(storages.size()); + for (MetricStorage storage : storages) { MetricData current = - metric.collectAndReset( - collectionInfo, + storage.collectAndReset( meterProviderSharedState.getResource(), getInstrumentationScopeInfo(), meterProviderSharedState.getStartEpochNanos(), - epochNanos, - suppressSynchronousCollection); + epochNanos); // Ignore if the metric data doesn't have any data points, for example when aggregation is // Aggregation#drop() if (!current.isEmpty()) { @@ -115,7 +116,7 @@ public void resetForTest() { synchronized (callbackLock) { callbackRegistrations.clear(); } - this.metricStorageRegistry.resetForTest(); + this.readerStorageRegistries.values().forEach(MetricStorageRegistry::resetForTest); } } @@ -123,26 +124,32 @@ public void resetForTest() { public final WriteableMetricStorage registerSynchronousMetricStorage( InstrumentDescriptor instrument, MeterProviderSharedState meterProviderSharedState) { - List storages = + List registeredStorages = new ArrayList<>(); + for (RegisteredView registeredView : meterProviderSharedState .getViewRegistry() - .findViews(instrument, getInstrumentationScopeInfo()) - .stream() - .map( - view -> - SynchronousMetricStorage.create( - view, instrument, meterProviderSharedState.getExemplarFilter())) - .filter(m -> !m.isEmpty()) - .collect(toList()); - - List registeredStorages = new ArrayList<>(storages.size()); - for (SynchronousMetricStorage storage : storages) { - registeredStorages.add(getMetricStorageRegistry().register(storage)); + .findViews(instrument, getInstrumentationScopeInfo())) { + if (Aggregation.drop() == registeredView.getView().getAggregation()) { + continue; + } + for (Map.Entry entry : + readerStorageRegistries.entrySet()) { + RegisteredReader reader = entry.getKey(); + MetricStorageRegistry registry = entry.getValue(); + registeredStorages.add( + registry.register( + SynchronousMetricStorage.create( + reader, + registeredView, + instrument, + meterProviderSharedState.getExemplarFilter()))); + } } if (registeredStorages.size() == 1) { return registeredStorages.get(0); } + return new MultiWritableMetricStorage(registeredStorages); } @@ -194,18 +201,23 @@ public final CallbackRegistration registerLongAsynchr private List> registerAsynchronousInstrument( InstrumentDescriptor instrumentDescriptor, MeterProviderSharedState meterProviderSharedState) { - List> storages = + + List> registeredStorages = new ArrayList<>(); + for (RegisteredView registeredView : meterProviderSharedState .getViewRegistry() - .findViews(instrumentDescriptor, getInstrumentationScopeInfo()) - .stream() - .map(view -> AsynchronousMetricStorage.create(view, instrumentDescriptor)) - .filter(storage -> !storage.isEmpty()) - .collect(toList()); - - List> registeredStorages = new ArrayList<>(storages.size()); - for (AsynchronousMetricStorage storage : storages) { - registeredStorages.add(getMetricStorageRegistry().register(storage)); + .findViews(instrumentDescriptor, getInstrumentationScopeInfo())) { + if (Aggregation.drop() == registeredView.getView().getAggregation()) { + continue; + } + for (Map.Entry entry : + readerStorageRegistries.entrySet()) { + RegisteredReader reader = entry.getKey(); + MetricStorageRegistry registry = entry.getValue(); + registeredStorages.add( + registry.register( + AsynchronousMetricStorage.create(reader, registeredView, instrumentDescriptor))); + } } return registeredStorages; diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorage.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorage.java index 794f292c6dd..86d9aa201b0 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorage.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorage.java @@ -8,7 +8,7 @@ import io.opentelemetry.sdk.common.InstrumentationScopeInfo; import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor; -import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; import io.opentelemetry.sdk.resources.Resource; /** @@ -22,28 +22,26 @@ public interface MetricStorage { /** Returns a description of the metric produced in this storage. */ MetricDescriptor getMetricDescriptor(); + /** Returns the registered reader this storage is associated with. */ + RegisteredReader getRegisteredReader(); + /** * Collects the metrics from this storage and resets for the next collection period. * *

Note: This is a stateful operation and will reset any interval-related state for the {@code * collector}. * - * @param collectionInfo The identity of the current reader of metrics and other information. * @param resource The resource associated with the metrics. * @param instrumentationScopeInfo The instrumentation scope generating the metrics. * @param startEpochNanos The start timestamp for this SDK. * @param epochNanos The timestamp for this collection. - * @param suppressSynchronousCollection Whether or not to suppress active (blocking) collection of - * metrics, meaning recently collected data is "fresh enough" * @return The {@link MetricData} from this collection period. */ MetricData collectAndReset( - CollectionInfo collectionInfo, Resource resource, InstrumentationScopeInfo instrumentationScopeInfo, long startEpochNanos, - long epochNanos, - boolean suppressSynchronousCollection); + long epochNanos); /** * Determines whether this storage is an empty metric storage. diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorageRegistry.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorageRegistry.java index 8310b826034..62684c27a5c 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorageRegistry.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorageRegistry.java @@ -38,7 +38,7 @@ public class MetricStorageRegistry { private final Map registry = new HashMap<>(); /** Returns a {@link Collection} of the registered {@link MetricStorage}. */ - public Collection getMetrics() { + public Collection getStorages() { synchronized (lock) { return Collections.unmodifiableCollection(new ArrayList<>(registry.values())); } diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/SynchronousMetricStorage.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/SynchronousMetricStorage.java index e2befa7df87..ba54bb0752d 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/SynchronousMetricStorage.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/SynchronousMetricStorage.java @@ -13,6 +13,7 @@ import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor; import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor; import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarFilter; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; import io.opentelemetry.sdk.metrics.internal.view.RegisteredView; /** @@ -35,6 +36,7 @@ static SynchronousMetricStorage empty() { * recorded. */ static SynchronousMetricStorage create( + RegisteredReader registeredReader, RegisteredView registeredView, InstrumentDescriptor instrumentDescriptor, ExemplarFilter exemplarFilter) { @@ -49,6 +51,9 @@ static SynchronousMetricStorage create( return empty(); } return new DefaultSynchronousMetricStorage<>( - metricDescriptor, aggregator, registeredView.getViewAttributesProcessor()); + registeredReader, + metricDescriptor, + aggregator, + registeredView.getViewAttributesProcessor()); } } diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/TemporalMetricStorage.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/TemporalMetricStorage.java index bc4ed3e88df..ee738434a81 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/TemporalMetricStorage.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/TemporalMetricStorage.java @@ -13,7 +13,7 @@ import io.opentelemetry.sdk.metrics.internal.aggregator.Aggregator; import io.opentelemetry.sdk.metrics.internal.aggregator.EmptyMetricData; import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor; -import io.opentelemetry.sdk.metrics.internal.export.CollectionHandle; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; import io.opentelemetry.sdk.resources.Resource; import java.util.HashMap; import java.util.Map; @@ -24,7 +24,7 @@ class TemporalMetricStorage { private final Aggregator aggregator; private final boolean isSynchronous; - private final Map> reportHistory = new HashMap<>(); + private final Map> reportHistory = new HashMap<>(); TemporalMetricStorage(Aggregator aggregator, boolean isSynchronous) { this.aggregator = aggregator; @@ -34,7 +34,6 @@ class TemporalMetricStorage { /** * Builds the {@link MetricData} streams to report against a specific metric reader. * - * @param collector The handle of the metric reader. * @param resource The resource to attach these metrics against. * @param instrumentationScopeInfo The instrumentation scope that generated these metrics. * @param temporality The aggregation temporality requested by the reader. @@ -45,7 +44,7 @@ class TemporalMetricStorage { * @return The {@link MetricData} points. */ synchronized MetricData buildMetricFor( - CollectionHandle collector, + RegisteredReader registeredReader, Resource resource, InstrumentationScopeInfo instrumentationScopeInfo, MetricDescriptor descriptor, @@ -58,8 +57,8 @@ synchronized MetricData buildMetricFor( long lastCollectionEpoch = startEpochNanos; Map result = currentAccumulation; // Check our last report time. - if (reportHistory.containsKey(collector)) { - LastReportedAccumulation last = reportHistory.get(collector); + if (reportHistory.containsKey(registeredReader)) { + LastReportedAccumulation last = reportHistory.get(registeredReader); lastCollectionEpoch = last.getEpochNanos(); // Use aggregation temporality + instrument to determine if we do a merge or a diff of // previous. We have the following four scenarios: @@ -92,10 +91,11 @@ synchronized MetricData buildMetricFor( // could be optimised to not record results for cases 3+4 listed above. if (isSynchronous) { // Sync instruments remember the full recording. - reportHistory.put(collector, new LastReportedAccumulation<>(result, epochNanos)); + reportHistory.put(registeredReader, new LastReportedAccumulation<>(result, epochNanos)); } else { // Async instruments record the raw measurement. - reportHistory.put(collector, new LastReportedAccumulation<>(currentAccumulation, epochNanos)); + reportHistory.put( + registeredReader, new LastReportedAccumulation<>(currentAccumulation, epochNanos)); } if (result.isEmpty()) { return EmptyMetricData.getInstance(); diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/CardinalityTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/CardinalityTest.java index 2f2451355c8..ca1b4ddec63 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/CardinalityTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/CardinalityTest.java @@ -14,8 +14,8 @@ import io.opentelemetry.internal.testing.slf4j.SuppressLogger; import io.opentelemetry.sdk.metrics.data.LongPointData; import io.opentelemetry.sdk.metrics.data.SumData; +import io.opentelemetry.sdk.metrics.internal.state.DefaultSynchronousMetricStorage; import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; -import java.time.Duration; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; import org.junit.jupiter.api.BeforeEach; @@ -23,7 +23,7 @@ @SuppressLogger( loggerName = "io.opentelemetry.sdk.metrics.internal.state.AsynchronousMetricStorage") -@SuppressLogger(loggerName = "io.opentelemetry.sdk.metrics.internal.state.DeltaMetricStorage") +@SuppressLogger(DefaultSynchronousMetricStorage.class) class CardinalityTest { /** Traces {@code MetricStorageUtils#MAX_ACCUMULATIONS}. */ @@ -41,7 +41,6 @@ void setup() { SdkMeterProvider.builder() .registerMetricReader(deltaReader) .registerMetricReader(cumulativeReader) - .setMinimumCollectionInterval(Duration.ofSeconds(0)) .build(); meter = sdkMeterProvider.get(CardinalityTest.class.getName()); } diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/IdentityTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/IdentityTest.java index ef804752743..f9e76215f34 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/IdentityTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/IdentityTest.java @@ -13,7 +13,6 @@ import io.opentelemetry.sdk.metrics.internal.state.MetricStorageRegistry; import io.opentelemetry.sdk.metrics.internal.view.ViewRegistry; import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; -import java.time.Duration; import java.util.concurrent.atomic.AtomicLong; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -34,10 +33,7 @@ class IdentityTest { @BeforeEach void setup() { reader = InMemoryMetricReader.createDelta(); - builder = - SdkMeterProvider.builder() - .registerMetricReader(reader) - .setMinimumCollectionInterval(Duration.ZERO); + builder = SdkMeterProvider.builder().registerMetricReader(reader); } @Test diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkMeterProviderBuilderTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkMeterProviderBuilderTest.java index eed87e92c0c..74e5e5dcc12 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkMeterProviderBuilderTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkMeterProviderBuilderTest.java @@ -5,15 +5,10 @@ package io.opentelemetry.sdk.metrics; -import static org.assertj.core.api.Assertions.as; import static org.assertj.core.api.Assertions.assertThat; -import io.opentelemetry.sdk.metrics.internal.SdkMeterProviderUtil; import io.opentelemetry.sdk.resources.Resource; import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; -import java.time.Duration; -import java.util.concurrent.TimeUnit; -import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; class SdkMeterProviderBuilderTest { @@ -28,19 +23,4 @@ void defaultResource() { .extracting("sharedState") .hasFieldOrPropertyWithValue("resource", Resource.getDefault()); } - - @Test - void setMinimumCollectionInterval() { - assertThat(SdkMeterProvider.builder().setMinimumCollectionInterval(Duration.ofSeconds(10))) - .extracting( - "minimumCollectionIntervalNanos", as(InstanceOfAssertFactories.type(Long.class))) - .isEqualTo(TimeUnit.SECONDS.toNanos(10)); - - SdkMeterProviderBuilder builder = SdkMeterProvider.builder(); - SdkMeterProviderUtil.setMinimumCollectionInterval(builder, Duration.ofSeconds(10)); - assertThat(builder) - .extracting( - "minimumCollectionIntervalNanos", as(InstanceOfAssertFactories.type(Long.class))) - .isEqualTo(TimeUnit.SECONDS.toNanos(10)); - } } diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkMeterProviderTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkMeterProviderTest.java index 910d0a82bcf..406b65ef4ff 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkMeterProviderTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkMeterProviderTest.java @@ -734,22 +734,21 @@ void viewSdk_capturesBaggageFromContext() { } @Test - void sdkMeterProvider_supportsMultipleCollectorsCumulative() { - InMemoryMetricReader collector1 = InMemoryMetricReader.create(); - InMemoryMetricReader collector2 = InMemoryMetricReader.create(); + void sdkMeterProvider_supportsMultipleReadersCumulative() { + InMemoryMetricReader reader1 = InMemoryMetricReader.create(); + InMemoryMetricReader reader2 = InMemoryMetricReader.create(); SdkMeterProvider meterProvider = - sdkMeterProviderBuilder - .registerMetricReader(collector1) - .registerMetricReader(collector2) - .build(); + sdkMeterProviderBuilder.registerMetricReader(reader1).registerMetricReader(reader2).build(); Meter sdkMeter = meterProvider.get(SdkMeterProviderTest.class.getName()); LongCounter counter = sdkMeter.counterBuilder("testSum").build(); long startTime = testClock.now(); + Attributes attributes = Attributes.builder().put("key", "value").build(); counter.add(1L); + counter.add(1L, attributes); testClock.advance(Duration.ofSeconds(1)); - assertThat(collector1.collectAllMetrics()) + assertThat(reader1.collectAllMetrics()) .satisfiesExactly( metric -> assertThat(metric) @@ -763,13 +762,20 @@ void sdkMeterProvider_supportsMultipleCollectorsCumulative() { point .hasStartEpochNanos(startTime) .hasEpochNanos(testClock.now()) - .hasValue(1)))); + .hasValue(1) + .hasAttributes(Attributes.empty()), + point -> + point + .hasStartEpochNanos(startTime) + .hasEpochNanos(testClock.now()) + .hasValue(1) + .hasAttributes(attributes)))); counter.add(1L); testClock.advance(Duration.ofSeconds(1)); - // Make sure collector 2 sees the value collector 1 saw - assertThat(collector2.collectAllMetrics()) + // Reader 2 should see the measurements of Reader 1 plus the additional measurement + assertThat(reader2.collectAllMetrics()) .satisfiesExactly( metric -> assertThat(metric) @@ -783,10 +789,17 @@ void sdkMeterProvider_supportsMultipleCollectorsCumulative() { point .hasStartEpochNanos(startTime) .hasEpochNanos(testClock.now()) - .hasValue(2)))); + .hasValue(2) + .hasAttributes(Attributes.empty()), + point -> + point + .hasStartEpochNanos(startTime) + .hasEpochNanos(testClock.now()) + .hasValue(1) + .hasAttributes(attributes)))); - // Make sure Collector 1 sees the same point as 2 - assertThat(collector1.collectAllMetrics()) + // Reader 1 should see updated cumulative values + assertThat(reader1.collectAllMetrics()) .satisfiesExactly( metric -> assertThat(metric) @@ -800,34 +813,32 @@ void sdkMeterProvider_supportsMultipleCollectorsCumulative() { point .hasStartEpochNanos(startTime) .hasEpochNanos(testClock.now()) - .hasValue(2)))); + .hasValue(2) + .hasAttributes(Attributes.empty()), + point -> + point + .hasStartEpochNanos(startTime) + .hasEpochNanos(testClock.now()) + .hasValue(1) + .hasAttributes(attributes)))); } @Test - void sdkMeterProvider_supportsMultipleCollectorsDelta() { - // Note: we use a view to do delta aggregation, but any view ALWAYS uses double-precision right - // now. - InMemoryMetricReader collector1 = InMemoryMetricReader.createDelta(); - InMemoryMetricReader collector2 = InMemoryMetricReader.createDelta(); + void sdkMeterProvider_supportsMultipleReadersDelta() { + InMemoryMetricReader reader1 = InMemoryMetricReader.createDelta(); + InMemoryMetricReader reader2 = InMemoryMetricReader.createDelta(); SdkMeterProvider meterProvider = - sdkMeterProviderBuilder - .registerMetricReader(collector1) - .registerMetricReader(collector2) - .registerView( - InstrumentSelector.builder() - .setType(InstrumentType.COUNTER) - .setName("testSum") - .build(), - View.builder().setAggregation(Aggregation.sum()).build()) - .build(); + sdkMeterProviderBuilder.registerMetricReader(reader1).registerMetricReader(reader2).build(); Meter sdkMeter = meterProvider.get(SdkMeterProviderTest.class.getName()); LongCounter counter = sdkMeter.counterBuilder("testSum").build(); long startTime = testClock.now(); + Attributes attributes = Attributes.builder().put("key", "value").build(); counter.add(1L); + counter.add(1L, attributes); testClock.advance(Duration.ofSeconds(1)); - assertThat(collector1.collectAllMetrics()) + assertThat(reader1.collectAllMetrics()) .satisfiesExactly( metric -> assertThat(metric) @@ -841,14 +852,21 @@ void sdkMeterProvider_supportsMultipleCollectorsDelta() { point .hasStartEpochNanos(startTime) .hasEpochNanos(testClock.now()) - .hasValue(1)))); + .hasValue(1) + .hasAttributes(Attributes.empty()), + point -> + point + .hasStartEpochNanos(startTime) + .hasEpochNanos(testClock.now()) + .hasValue(1) + .hasAttributes(attributes)))); long collectorOneTimeOne = testClock.now(); counter.add(1L); testClock.advance(Duration.ofSeconds(1)); - // Make sure collector 2 sees the value collector 1 saw - assertThat(collector2.collectAllMetrics()) + // Reader 2 should see the measurements of Reader 1 plus the additional measurement + assertThat(reader2.collectAllMetrics()) .satisfiesExactly( metric -> assertThat(metric) @@ -862,10 +880,17 @@ void sdkMeterProvider_supportsMultipleCollectorsDelta() { point .hasStartEpochNanos(startTime) .hasEpochNanos(testClock.now()) - .hasValue(2)))); + .hasValue(2) + .hasAttributes(Attributes.empty()), + point -> + point + .hasStartEpochNanos(startTime) + .hasEpochNanos(testClock.now()) + .hasValue(1) + .hasAttributes(attributes)))); - // Make sure Collector 1 sees the same point as 2, when it collects. - assertThat(collector1.collectAllMetrics()) + // Reader 1 should only see diff since its last collect + assertThat(reader1.collectAllMetrics()) .satisfiesExactly( metric -> assertThat(metric) diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkObservableInstrumentTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkObservableInstrumentTest.java index 0e4587362b0..e90ad73af8e 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkObservableInstrumentTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkObservableInstrumentTest.java @@ -29,7 +29,7 @@ class SdkObservableInstrumentTest { @SuppressLogger(SdkObservableInstrument.class) void close() { MeterSharedState meterSharedState = - spy(MeterSharedState.create(InstrumentationScopeInfo.empty())); + spy(MeterSharedState.create(InstrumentationScopeInfo.empty(), Collections.emptyList())); CallbackRegistration callbackRegistration = CallbackRegistration.createDouble( InstrumentDescriptor.create( diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/export/RegisteredReaderTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/export/RegisteredReaderTest.java new file mode 100644 index 00000000000..47cdc4ab930 --- /dev/null +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/export/RegisteredReaderTest.java @@ -0,0 +1,39 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.metrics.internal.export; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.opentelemetry.sdk.metrics.export.MetricReader; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class RegisteredReaderTest { + + @Mock private MetricReader reader; + + @Test + void create_UniqueIdentity() { + RegisteredReader registeredReader1 = RegisteredReader.create(reader); + RegisteredReader registeredReader2 = RegisteredReader.create(reader); + + assertThat(registeredReader1).isEqualTo(registeredReader1); + assertThat(registeredReader1).isNotEqualTo(registeredReader2); + + assertThat(registeredReader1.hashCode()).isEqualTo(registeredReader1.hashCode()); + assertThat(registeredReader1.hashCode()).isNotEqualTo(registeredReader2.hashCode()); + } + + @Test + void getReader() { + RegisteredReader registeredReader = RegisteredReader.create(reader); + + assertThat(registeredReader.getReader()).isSameAs(reader); + } +} diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/export/TestCollectionHandle.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/export/TestCollectionHandle.java deleted file mode 100644 index f3f15eae798..00000000000 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/export/TestCollectionHandle.java +++ /dev/null @@ -1,80 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - */ - -package io.opentelemetry.sdk.metrics.internal.export; - -import static org.assertj.core.api.Assertions.assertThat; - -import java.util.Iterator; -import java.util.Set; -import java.util.function.Supplier; -import org.junit.jupiter.api.Test; - -public class TestCollectionHandle { - - @Test - public void created_haveUniqueIdentity() { - Supplier supplier = CollectionHandle.createSupplier(); - CollectionHandle one = supplier.get(); - CollectionHandle two = supplier.get(); - - assertThat(one).isEqualTo(one); - assertThat(one).isNotEqualTo(two); - } - - @Test - public void mutableSet_allowsAddAndContains() { - Supplier supplier = CollectionHandle.createSupplier(); - Set mutable = CollectionHandle.mutableSet(); - CollectionHandle one = supplier.get(); - assertThat(mutable).hasSize(0); - assertThat(mutable.contains(one)).isFalse(); - mutable.add(one); - assertThat(mutable).hasSize(1); - assertThat(mutable.contains(one)).isTrue(); - - CollectionHandle two = supplier.get(); - assertThat(mutable.contains(two)).isFalse(); - mutable.add(two); - assertThat(mutable).hasSize(2); - assertThat(mutable.contains(two)).isTrue(); - } - - @Test - public void mutableSet_allowsContainsAll() { - Supplier supplier = CollectionHandle.createSupplier(); - CollectionHandle one = supplier.get(); - CollectionHandle two = supplier.get(); - CollectionHandle three = supplier.get(); - Set mutable = CollectionHandle.mutableSet(); - mutable.add(one); - mutable.add(two); - Set mutableCopy = CollectionHandle.of(one, two); - Set mutablePlus = CollectionHandle.of(one, two, three); - - assertThat(mutable.containsAll(mutableCopy)).isTrue(); - assertThat(mutable.containsAll(mutablePlus)).isFalse(); - assertThat(mutablePlus.containsAll(mutable)).isTrue(); - } - - @Test - public void mutableSet_iteratingWorks() { - Supplier supplier = CollectionHandle.createSupplier(); - CollectionHandle one = supplier.get(); - CollectionHandle two = supplier.get(); - CollectionHandle three = supplier.get(); - Set set = CollectionHandle.of(one, two, three); - assertThat(set).hasSize(3); - Iterator iterator = set.iterator(); - assertThat(iterator.hasNext()).isTrue(); - assertThat(iterator.next()).isEqualTo(one); - assertThat(iterator.hasNext()).isTrue(); - assertThat(iterator.next()).isEqualTo(two); - assertThat(iterator.hasNext()).isTrue(); - assertThat(iterator.next()).isEqualTo(three); - assertThat(iterator.hasNext()).isFalse(); - // TODO: Verify next throws. - } -} diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorageTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorageTest.java index 211bb14ce3b..205185f4785 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorageTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorageTest.java @@ -22,13 +22,11 @@ import io.opentelemetry.sdk.metrics.export.MetricReader; import io.opentelemetry.sdk.metrics.internal.debug.SourceInfo; import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor; -import io.opentelemetry.sdk.metrics.internal.export.CollectionHandle; -import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; import io.opentelemetry.sdk.metrics.internal.view.AttributesProcessor; import io.opentelemetry.sdk.metrics.internal.view.RegisteredView; import io.opentelemetry.sdk.resources.Resource; import io.opentelemetry.sdk.testing.time.TestClock; -import java.util.Set; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -43,8 +41,6 @@ class AsynchronousMetricStorageTest { @RegisterExtension LogCapturer logs = LogCapturer.create().captureForType(AsynchronousMetricStorage.class); - @Mock private MetricReader reader; - private final TestClock testClock = TestClock.create(); private final Resource resource = Resource.empty(); private final InstrumentationScopeInfo scope = InstrumentationScopeInfo.empty(); @@ -52,21 +48,21 @@ class AsynchronousMetricStorageTest { private final RegisteredView registeredView = RegisteredView.create( selector, View.builder().build(), AttributesProcessor.noop(), SourceInfo.noSourceInfo()); - private CollectionInfo collectionInfo; + + @Mock private MetricReader reader; + private RegisteredReader registeredReader; @BeforeEach void setup() { - CollectionHandle handle = CollectionHandle.createSupplier().get(); - Set all = CollectionHandle.mutableSet(); - all.add(handle); - collectionInfo = CollectionInfo.create(handle, all, reader); when(reader.getAggregationTemporality(any())).thenReturn(AggregationTemporality.CUMULATIVE); + registeredReader = RegisteredReader.create(reader); } @Test void recordLong() { AsynchronousMetricStorage storage = AsynchronousMetricStorage.create( + registeredReader, registeredView, InstrumentDescriptor.create( "name", "description", "unit", InstrumentType.COUNTER, InstrumentValueType.LONG)); @@ -75,14 +71,7 @@ void recordLong() { storage.recordLong(2, Attributes.builder().put("key", "b").build()); storage.recordLong(3, Attributes.builder().put("key", "c").build()); - assertThat( - storage.collectAndReset( - collectionInfo, - resource, - scope, - 0, - testClock.nanoTime(), - /* suppressSynchronousCollection= */ false)) + assertThat(storage.collectAndReset(resource, scope, 0, testClock.nanoTime())) .satisfies( metricData -> assertThat(metricData) @@ -102,6 +91,7 @@ void recordLong() { void recordDouble() { AsynchronousMetricStorage storage = AsynchronousMetricStorage.create( + registeredReader, registeredView, InstrumentDescriptor.create( "name", "description", "unit", InstrumentType.COUNTER, InstrumentValueType.DOUBLE)); @@ -110,14 +100,7 @@ void recordDouble() { storage.recordDouble(2.2, Attributes.builder().put("key", "b").build()); storage.recordDouble(3.3, Attributes.builder().put("key", "c").build()); - assertThat( - storage.collectAndReset( - collectionInfo, - resource, - scope, - 0, - testClock.nanoTime(), - /* suppressSynchronousCollection= */ false)) + assertThat(storage.collectAndReset(resource, scope, 0, testClock.nanoTime())) .satisfies( metricData -> assertThat(metricData) @@ -139,6 +122,7 @@ void recordDouble() { void record_ProcessesAttributes() { AsynchronousMetricStorage storage = AsynchronousMetricStorage.create( + registeredReader, RegisteredView.create( selector, View.builder().build(), @@ -149,14 +133,7 @@ void record_ProcessesAttributes() { storage.recordLong(1, Attributes.builder().put("key1", "a").put("key2", "b").build()); - assertThat( - storage.collectAndReset( - collectionInfo, - resource, - scope, - 0, - testClock.nanoTime(), - /* suppressSynchronousCollection= */ false)) + assertThat(storage.collectAndReset(resource, scope, 0, testClock.nanoTime())) .satisfies( metricData -> assertThat(metricData) @@ -172,6 +149,7 @@ void record_ProcessesAttributes() { void record_MaxAccumulations() { AsynchronousMetricStorage storage = AsynchronousMetricStorage.create( + registeredReader, registeredView, InstrumentDescriptor.create( "name", "description", "unit", InstrumentType.COUNTER, InstrumentValueType.LONG)); @@ -180,14 +158,7 @@ void record_MaxAccumulations() { storage.recordLong(1, Attributes.builder().put("key" + i, "val").build()); } - assertThat( - storage.collectAndReset( - collectionInfo, - resource, - scope, - 0, - testClock.nanoTime(), - /* suppressSynchronousCollection= */ false)) + assertThat(storage.collectAndReset(resource, scope, 0, testClock.nanoTime())) .satisfies( metricData -> assertThat(metricData.getLongSumData().getPoints()) @@ -199,6 +170,7 @@ void record_MaxAccumulations() { void record_DuplicateAttributes() { AsynchronousMetricStorage storage = AsynchronousMetricStorage.create( + registeredReader, registeredView, InstrumentDescriptor.create( "name", "description", "unit", InstrumentType.COUNTER, InstrumentValueType.LONG)); @@ -206,14 +178,7 @@ void record_DuplicateAttributes() { storage.recordLong(1, Attributes.builder().put("key1", "a").build()); storage.recordLong(2, Attributes.builder().put("key1", "a").build()); - assertThat( - storage.collectAndReset( - collectionInfo, - resource, - scope, - 0, - testClock.nanoTime(), - /* suppressSynchronousCollection= */ false)) + assertThat(storage.collectAndReset(resource, scope, 0, testClock.nanoTime())) .satisfies( metricData -> assertThat(metricData) diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/CallbackRegistrationTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/CallbackRegistrationTest.java index 88c3603a89c..12a8f29d028 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/CallbackRegistrationTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/CallbackRegistrationTest.java @@ -10,6 +10,7 @@ import static org.mockito.ArgumentMatchers.anyDouble; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import com.google.common.util.concurrent.AtomicDouble; import io.github.netmikey.logunit.api.LogCapturer; @@ -18,11 +19,14 @@ import io.opentelemetry.internal.testing.slf4j.SuppressLogger; import io.opentelemetry.sdk.metrics.InstrumentType; import io.opentelemetry.sdk.metrics.InstrumentValueType; +import io.opentelemetry.sdk.metrics.export.MetricReader; import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; import java.util.Arrays; import java.util.Collections; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.RegisterExtension; @@ -51,11 +55,22 @@ class CallbackRegistrationTest { @RegisterExtension LogCapturer logs = LogCapturer.create().captureForType(CallbackRegistration.class); + @Mock private MetricReader reader; @Mock private AsynchronousMetricStorage storage1; @Mock private AsynchronousMetricStorage storage2; + @Mock private AsynchronousMetricStorage storage3; + + private RegisteredReader registeredReader; + + @BeforeEach + void setup() { + registeredReader = RegisteredReader.create(reader); + } @Test void invokeCallback_Double() { + when(storage1.getRegisteredReader()).thenReturn(registeredReader); + when(storage2.getRegisteredReader()).thenReturn(registeredReader); AtomicDouble counter = new AtomicDouble(); Consumer callback = measurement -> @@ -63,17 +78,20 @@ void invokeCallback_Double() { counter.addAndGet(1.1), Attributes.builder().put("key", "val").build()); CallbackRegistration callbackRegistration = CallbackRegistration.createDouble( - DOUBLE_INSTRUMENT, callback, Arrays.asList(storage1, storage2)); + DOUBLE_INSTRUMENT, callback, Arrays.asList(storage1, storage2, storage3)); - callbackRegistration.invokeCallback(); + callbackRegistration.invokeCallback(registeredReader); assertThat(counter.get()).isEqualTo(1.1); verify(storage1).recordDouble(1.1, Attributes.builder().put("key", "val").build()); verify(storage2).recordDouble(1.1, Attributes.builder().put("key", "val").build()); + verify(storage3, never()).recordDouble(anyDouble(), any()); } @Test void invokeCallback_Long() { + when(storage1.getRegisteredReader()).thenReturn(registeredReader); + when(storage2.getRegisteredReader()).thenReturn(registeredReader); AtomicInteger counter = new AtomicInteger(); Consumer callback = measurement -> @@ -81,13 +99,14 @@ void invokeCallback_Long() { counter.incrementAndGet(), Attributes.builder().put("key", "val").build()); CallbackRegistration callbackRegistration = CallbackRegistration.createDouble( - LONG_INSTRUMENT, callback, Arrays.asList(storage1, storage2)); + LONG_INSTRUMENT, callback, Arrays.asList(storage1, storage2, storage3)); - callbackRegistration.invokeCallback(); + callbackRegistration.invokeCallback(registeredReader); assertThat(counter.get()).isEqualTo(1); verify(storage1).recordDouble(1, Attributes.builder().put("key", "val").build()); verify(storage2).recordDouble(1, Attributes.builder().put("key", "val").build()); + verify(storage3, never()).recordDouble(anyDouble(), any()); } @Test @@ -100,7 +119,7 @@ void invokeCallback_NoStorage() { CallbackRegistration callbackRegistration = CallbackRegistration.createDouble(LONG_INSTRUMENT, callback, Collections.emptyList()); - callbackRegistration.invokeCallback(); + callbackRegistration.invokeCallback(registeredReader); assertThat(counter.get()).isEqualTo(0); } @@ -115,10 +134,11 @@ void invokeCallback_ThrowsException() { CallbackRegistration.createDouble( LONG_INSTRUMENT, callback, Arrays.asList(storage1, storage2)); - callbackRegistration.invokeCallback(); + callbackRegistration.invokeCallback(registeredReader); verify(storage1, never()).recordDouble(anyDouble(), any()); verify(storage2, never()).recordDouble(anyDouble(), any()); + verify(storage3, never()).recordDouble(anyDouble(), any()); logs.assertContains("An exception occurred invoking callback for instrument name"); } } diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/DeltaAccumulationTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/DeltaAccumulationTest.java deleted file mode 100644 index 018a1c2b3ef..00000000000 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/DeltaAccumulationTest.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - */ - -package io.opentelemetry.sdk.metrics.internal.state; - -import static org.assertj.core.api.Assertions.assertThat; - -import io.opentelemetry.api.common.Attributes; -import io.opentelemetry.sdk.metrics.internal.export.CollectionHandle; -import java.util.HashMap; -import java.util.Map; -import java.util.Set; -import java.util.function.Supplier; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -class DeltaAccumulationTest { - private CollectionHandle handle1; - private CollectionHandle handle2; - private Set all; - - @BeforeEach - void setup() { - Supplier supplier = CollectionHandle.createSupplier(); - handle1 = supplier.get(); - handle2 = supplier.get(); - all = CollectionHandle.mutableSet(); - all.add(handle1); - all.add(handle2); - } - - @Test - void wasReadBy_works() { - Map measurement = new HashMap<>(); - measurement.put(Attributes.empty(), 1L); - DeltaAccumulation accumlation = new DeltaAccumulation<>(measurement); - assertThat(accumlation.wasReadBy(handle1)).isFalse(); - assertThat(accumlation.wasReadBy(handle2)).isFalse(); - assertThat(accumlation.wasReadByAll(all)).isFalse(); - - // Read and check. - assertThat(accumlation.read(handle1)).isEqualTo(measurement); - assertThat(accumlation.wasReadBy(handle1)).isTrue(); - assertThat(accumlation.wasReadBy(handle2)).isFalse(); - assertThat(accumlation.wasReadByAll(all)).isFalse(); - - // Read and check. - assertThat(accumlation.read(handle2)).isEqualTo(measurement); - assertThat(accumlation.wasReadBy(handle1)).isTrue(); - assertThat(accumlation.wasReadBy(handle2)).isTrue(); - assertThat(accumlation.wasReadByAll(all)).isTrue(); - } -} diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/DeltaMetricStorageTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/DeltaMetricStorageTest.java deleted file mode 100644 index 7dc3d8e1096..00000000000 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/DeltaMetricStorageTest.java +++ /dev/null @@ -1,101 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - */ - -package io.opentelemetry.sdk.metrics.internal.state; - -import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat; - -import io.opentelemetry.api.common.Attributes; -import io.opentelemetry.context.Context; -import io.opentelemetry.sdk.metrics.Aggregation; -import io.opentelemetry.sdk.metrics.InstrumentType; -import io.opentelemetry.sdk.metrics.InstrumentValueType; -import io.opentelemetry.sdk.metrics.data.DoubleExemplarData; -import io.opentelemetry.sdk.metrics.internal.aggregator.AggregatorFactory; -import io.opentelemetry.sdk.metrics.internal.aggregator.DoubleAccumulation; -import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor; -import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarFilter; -import io.opentelemetry.sdk.metrics.internal.export.CollectionHandle; -import java.util.Set; -import java.util.function.Supplier; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -class DeltaMetricStorageTest { - private static final InstrumentDescriptor DESCRIPTOR = - InstrumentDescriptor.create( - "name", "description", "unit", InstrumentType.COUNTER, InstrumentValueType.DOUBLE); - - private CollectionHandle collector1; - private CollectionHandle collector2; - private Set allCollectors; - private DeltaMetricStorage storage; - - @BeforeEach - void setup() { - Supplier supplier = CollectionHandle.createSupplier(); - collector1 = supplier.get(); - collector2 = supplier.get(); - allCollectors = CollectionHandle.mutableSet(); - allCollectors.add(collector1); - allCollectors.add(collector2); - storage = - new DeltaMetricStorage<>( - ((AggregatorFactory) Aggregation.sum()) - .createAggregator(DESCRIPTOR, ExemplarFilter.neverSample()), - DESCRIPTOR); - } - - @Test - void collectionDeltaForMultiReader() { - BoundStorageHandle bound = storage.bind(Attributes.empty()); - bound.recordDouble(1, Attributes.empty(), Context.root()); - // First collector only sees first recording. - assertThat(storage.collectFor(collector1, allCollectors, /* suppressCollection=*/ false)) - .hasSize(1) - .hasEntrySatisfying(Attributes.empty(), value -> assertThat(value.getValue()).isEqualTo(1)); - - bound.recordDouble(2, Attributes.empty(), Context.root()); - // First collector only sees second recording. - assertThat(storage.collectFor(collector1, allCollectors, /* suppressCollection=*/ false)) - .hasSize(1) - .hasEntrySatisfying(Attributes.empty(), value -> assertThat(value.getValue()).isEqualTo(2)); - - // First collector no longer sees a recording. - assertThat(storage.collectFor(collector1, allCollectors, /* suppressCollection=*/ false)) - .isEmpty(); - - // Second collector gets merged recordings - assertThat(storage.collectFor(collector2, allCollectors, /* suppressCollection=*/ false)) - .hasSize(1) - .hasEntrySatisfying(Attributes.empty(), value -> assertThat(value.getValue()).isEqualTo(3)); - - // Second collector no longer sees a recording. - assertThat(storage.collectFor(collector2, allCollectors, /* suppressCollection=*/ false)) - .isEmpty(); - } - - @Test - void avoidCollectionInRapidSuccession() { - BoundStorageHandle bound = storage.bind(Attributes.empty()); - bound.recordDouble(1, Attributes.empty(), Context.root()); - // First collector only sees first recording. - assertThat(storage.collectFor(collector1, allCollectors, /* suppressCollection=*/ false)) - .hasSize(1) - .hasEntrySatisfying(Attributes.empty(), value -> assertThat(value.getValue()).isEqualTo(1)); - // Add some data immediately after read, but pretent it hasn't been long. - bound.recordDouble(2, Attributes.empty(), Context.root()); - // Collector1 doesn't see new data, because we don't recollect, but collector2 sees old delta. - assertThat(storage.collectFor(collector1, allCollectors, /* suppressCollection=*/ true)) - .isEmpty(); - assertThat(storage.collectFor(collector2, allCollectors, /* suppressCollection=*/ true)) - .hasSize(1) - .hasEntrySatisfying(Attributes.empty(), value -> assertThat(value.getValue()).isEqualTo(1)); - // After enough time passes, collector1 sees new data - assertThat(storage.collectFor(collector1, allCollectors, /* suppressCollection=*/ false)) - .hasSize(1) - .hasEntrySatisfying(Attributes.empty(), value -> assertThat(value.getValue()).isEqualTo(2)); - } -} diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorageRegistryTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorageRegistryTest.java index c00d41fe23c..2217189eb39 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorageRegistryTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/MetricStorageRegistryTest.java @@ -15,15 +15,21 @@ import io.opentelemetry.sdk.metrics.InstrumentValueType; import io.opentelemetry.sdk.metrics.View; import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.export.MetricReader; import io.opentelemetry.sdk.metrics.internal.debug.SourceInfo; import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor; import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor; -import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; import io.opentelemetry.sdk.resources.Resource; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.RegisterExtension; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; /** Unit tests for {@link MetricStorageRegistry}. */ +@ExtendWith(MockitoExtension.class) @SuppressLogger(MetricStorageRegistry.class) class MetricStorageRegistryTest { private static final MetricDescriptor SYNC_DESCRIPTOR = @@ -35,45 +41,61 @@ class MetricStorageRegistryTest { private static final MetricDescriptor OTHER_ASYNC_DESCRIPTOR = descriptor("async", "other_description", InstrumentType.OBSERVABLE_GAUGE); + private final MetricStorageRegistry metricStorageRegistry = new MetricStorageRegistry(); + @RegisterExtension LogCapturer logs = LogCapturer.create().captureForType(MetricStorageRegistry.class); - private final MetricStorageRegistry metricStorageRegistry = new MetricStorageRegistry(); + @Mock private MetricReader reader; + private RegisteredReader registeredReader; + + @BeforeEach + void setup() { + registeredReader = RegisteredReader.create(reader); + } @Test void register_Sync() { - TestMetricStorage storage = new TestMetricStorage(SYNC_DESCRIPTOR); + TestMetricStorage storage = new TestMetricStorage(SYNC_DESCRIPTOR, registeredReader); assertThat(metricStorageRegistry.register(storage)).isSameAs(storage); assertThat(metricStorageRegistry.register(storage)).isSameAs(storage); - assertThat(metricStorageRegistry.register(new TestMetricStorage(SYNC_DESCRIPTOR))) + assertThat( + metricStorageRegistry.register( + new TestMetricStorage(SYNC_DESCRIPTOR, registeredReader))) .isSameAs(storage); } @Test void register_SyncIncompatibleDescriptor() { - TestMetricStorage storage = new TestMetricStorage(SYNC_DESCRIPTOR); + TestMetricStorage storage = new TestMetricStorage(SYNC_DESCRIPTOR, registeredReader); assertThat(metricStorageRegistry.register(storage)).isSameAs(storage); assertThat(logs.getEvents()).isEmpty(); - assertThat(metricStorageRegistry.register(new TestMetricStorage(OTHER_SYNC_DESCRIPTOR))) + assertThat( + metricStorageRegistry.register( + new TestMetricStorage(OTHER_SYNC_DESCRIPTOR, registeredReader))) .isNotSameAs(storage); logs.assertContains("Found duplicate metric definition"); } @Test void register_Async() { - TestMetricStorage storage = new TestMetricStorage(ASYNC_DESCRIPTOR); + TestMetricStorage storage = new TestMetricStorage(ASYNC_DESCRIPTOR, registeredReader); assertThat(metricStorageRegistry.register(storage)).isSameAs(storage); assertThat(metricStorageRegistry.register(storage)).isSameAs(storage); - assertThat(metricStorageRegistry.register(new TestMetricStorage(ASYNC_DESCRIPTOR))) + assertThat( + metricStorageRegistry.register( + new TestMetricStorage(ASYNC_DESCRIPTOR, registeredReader))) .isSameAs(storage); } @Test void register_AsyncIncompatibleDescriptor() { - TestMetricStorage storage = new TestMetricStorage(ASYNC_DESCRIPTOR); + TestMetricStorage storage = new TestMetricStorage(ASYNC_DESCRIPTOR, registeredReader); assertThat(metricStorageRegistry.register(storage)).isSameAs(storage); assertThat(logs.getEvents()).isEmpty(); - assertThat(metricStorageRegistry.register(new TestMetricStorage(OTHER_ASYNC_DESCRIPTOR))) + assertThat( + metricStorageRegistry.register( + new TestMetricStorage(OTHER_ASYNC_DESCRIPTOR, registeredReader))) .isNotSameAs(storage); logs.assertContains("Found duplicate metric definition"); } @@ -89,9 +111,11 @@ private static MetricDescriptor descriptor( private static final class TestMetricStorage implements MetricStorage, WriteableMetricStorage { private final MetricDescriptor descriptor; + private final RegisteredReader registeredReader; - TestMetricStorage(MetricDescriptor descriptor) { + TestMetricStorage(MetricDescriptor descriptor, RegisteredReader registeredReader) { this.descriptor = descriptor; + this.registeredReader = registeredReader; } @Override @@ -99,14 +123,17 @@ public MetricDescriptor getMetricDescriptor() { return descriptor; } + @Override + public RegisteredReader getRegisteredReader() { + return registeredReader; + } + @Override public MetricData collectAndReset( - CollectionInfo collectionInfo, Resource resource, InstrumentationScopeInfo instrumentationScopeInfo, long startEpochNanos, - long epochNanos, - boolean suppressSynchronousCollection) { + long epochNanos) { return null; } diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/SynchronousMetricStorageTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/SynchronousMetricStorageTest.java index 45f38559421..a5e48527f9f 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/SynchronousMetricStorageTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/SynchronousMetricStorageTest.java @@ -23,12 +23,10 @@ import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor; import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor; 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.RegisteredReader; import io.opentelemetry.sdk.metrics.internal.view.AttributesProcessor; import io.opentelemetry.sdk.resources.Resource; import io.opentelemetry.sdk.testing.time.TestClock; -import java.util.Set; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -51,16 +49,13 @@ public class SynchronousMetricStorageTest { ((AggregatorFactory) Aggregation.lastValue()) .createAggregator(DESCRIPTOR, ExemplarFilter.neverSample()); private final AttributesProcessor attributesProcessor = AttributesProcessor.noop(); - private CollectionHandle collector; - private Set allCollectors; @Mock private MetricReader reader; + private RegisteredReader registeredReader; @BeforeEach void setup() { - collector = CollectionHandle.createSupplier().get(); - allCollectors = CollectionHandle.mutableSet(); - allCollectors.add(collector); + registeredReader = RegisteredReader.create(reader); } @Test @@ -68,7 +63,7 @@ void attributesProcessor_used() { AttributesProcessor spyAttributesProcessor = Mockito.spy(this.attributesProcessor); SynchronousMetricStorage accumulator = new DefaultSynchronousMetricStorage<>( - METRIC_DESCRIPTOR, aggregator, spyAttributesProcessor); + registeredReader, METRIC_DESCRIPTOR, aggregator, spyAttributesProcessor); accumulator.bind(Attributes.empty()); Mockito.verify(spyAttributesProcessor).process(Attributes.empty(), Context.current()); } @@ -80,17 +75,12 @@ void attributesProcessor_applied() { AttributesProcessor.append(Attributes.builder().put("modifiedK", "modifiedV").build()); AttributesProcessor spyLabelsProcessor = Mockito.spy(attributesProcessor); SynchronousMetricStorage accumulator = - new DefaultSynchronousMetricStorage<>(METRIC_DESCRIPTOR, aggregator, spyLabelsProcessor); + new DefaultSynchronousMetricStorage<>( + registeredReader, METRIC_DESCRIPTOR, aggregator, spyLabelsProcessor); BoundStorageHandle handle = accumulator.bind(labels); handle.recordDouble(1, labels, Context.root()); MetricData md = - accumulator.collectAndReset( - CollectionInfo.create(collector, allCollectors, reader), - RESOURCE, - INSTRUMENTATION_SCOPE_INFO, - 0, - testClock.now(), - false); + accumulator.collectAndReset(RESOURCE, INSTRUMENTATION_SCOPE_INFO, 0, testClock.now()); assertThat(md) .hasDoubleGaugeSatisfying( gauge -> @@ -103,19 +93,14 @@ void attributesProcessor_applied() { @Test void sameAggregator_ForSameAttributes() { SynchronousMetricStorage accumulator = - new DefaultSynchronousMetricStorage<>(METRIC_DESCRIPTOR, aggregator, attributesProcessor); + new DefaultSynchronousMetricStorage<>( + registeredReader, METRIC_DESCRIPTOR, aggregator, attributesProcessor); BoundStorageHandle handle = accumulator.bind(Attributes.builder().put("K", "V").build()); BoundStorageHandle duplicateHandle = accumulator.bind(Attributes.builder().put("K", "V").build()); try { assertThat(duplicateHandle).isSameAs(handle); - accumulator.collectAndReset( - CollectionInfo.create(collector, allCollectors, reader), - RESOURCE, - INSTRUMENTATION_SCOPE_INFO, - 0, - testClock.now(), - false); + accumulator.collectAndReset(RESOURCE, INSTRUMENTATION_SCOPE_INFO, 0, testClock.now()); BoundStorageHandle anotherDuplicateAggregatorHandle = accumulator.bind(Attributes.builder().put("K", "V").build()); try { @@ -131,13 +116,7 @@ void sameAggregator_ForSameAttributes() { // If we try to collect once all bound references are gone AND no recordings have occurred, we // should not see any labels (or metric). assertThat( - accumulator.collectAndReset( - CollectionInfo.create(collector, allCollectors, reader), - RESOURCE, - INSTRUMENTATION_SCOPE_INFO, - 0, - testClock.now(), - false)) + accumulator.collectAndReset(RESOURCE, INSTRUMENTATION_SCOPE_INFO, 0, testClock.now())) .isEqualTo(EmptyMetricData.getInstance()); } } diff --git a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/TemporalMetricStorageTest.java b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/TemporalMetricStorageTest.java index 55a35e75cc0..fc44ca77d3b 100644 --- a/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/TemporalMetricStorageTest.java +++ b/sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/TemporalMetricStorageTest.java @@ -14,21 +14,24 @@ import io.opentelemetry.sdk.metrics.InstrumentValueType; import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.metrics.data.DoubleExemplarData; +import io.opentelemetry.sdk.metrics.export.MetricReader; import io.opentelemetry.sdk.metrics.internal.aggregator.Aggregator; import io.opentelemetry.sdk.metrics.internal.aggregator.AggregatorFactory; import io.opentelemetry.sdk.metrics.internal.aggregator.DoubleAccumulation; import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor; import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor; import io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarFilter; -import io.opentelemetry.sdk.metrics.internal.export.CollectionHandle; +import io.opentelemetry.sdk.metrics.internal.export.RegisteredReader; import io.opentelemetry.sdk.resources.Resource; import java.util.HashMap; import java.util.Map; -import java.util.Set; -import java.util.function.Supplier; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +@ExtendWith(MockitoExtension.class) class TemporalMetricStorageTest { private static final InstrumentDescriptor DESCRIPTOR = InstrumentDescriptor.create( @@ -50,18 +53,14 @@ class TemporalMetricStorageTest { ((AggregatorFactory) Aggregation.sum()) .createAggregator(ASYNC_DESCRIPTOR, ExemplarFilter.neverSample()); - private CollectionHandle collector1; - private CollectionHandle collector2; - private Set allCollectors; + @Mock private MetricReader reader; + private RegisteredReader registeredReader1; + private RegisteredReader registeredReader2; @BeforeEach void setup() { - Supplier supplier = CollectionHandle.createSupplier(); - collector1 = supplier.get(); - collector2 = supplier.get(); - allCollectors = CollectionHandle.mutableSet(); - allCollectors.add(collector1); - allCollectors.add(collector2); + registeredReader1 = RegisteredReader.create(reader); + registeredReader2 = RegisteredReader.create(reader); } private static Map createMeasurement(double value) { @@ -78,7 +77,7 @@ void synchronousCumulative_joinsWithLastMeasurementForCumulative() { // Send in new measurement at time 10 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -94,7 +93,7 @@ void synchronousCumulative_joinsWithLastMeasurementForCumulative() { // Send in new measurement at time 30 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -110,7 +109,7 @@ void synchronousCumulative_joinsWithLastMeasurementForCumulative() { // Send in new measurement at time 40 for collector 2 assertThat( storage.buildMetricFor( - collector2, + registeredReader2, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -126,7 +125,7 @@ void synchronousCumulative_joinsWithLastMeasurementForCumulative() { // Send in new measurement at time 35 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -154,7 +153,7 @@ void synchronousCumulative_dropsStaleAtLimit() { } assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -185,7 +184,7 @@ void synchronousCumulative_dropsStaleAtLimit() { measurement2.put(attr2, DoubleAccumulation.create(3)); assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -208,7 +207,7 @@ void synchronousDelta_dropsStale() { measurement1.put(attr1, DoubleAccumulation.create(3)); assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -229,7 +228,7 @@ void synchronousDelta_dropsStale() { measurement2.put(attr2, DoubleAccumulation.create(7)); assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -257,7 +256,7 @@ void synchronousDelta_useLastTimestamp() { // Send in new measurement at time 10 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -272,7 +271,7 @@ void synchronousDelta_useLastTimestamp() { // Send in new measurement at time 30 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -287,7 +286,7 @@ void synchronousDelta_useLastTimestamp() { // Send in new measurement at time 40 for collector 2 assertThat( storage.buildMetricFor( - collector2, + registeredReader2, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -303,7 +302,7 @@ void synchronousDelta_useLastTimestamp() { // Send in new measurement at time 35 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -325,7 +324,7 @@ void synchronous_deltaAndCumulative() { // Send in new measurement at time 10 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -341,7 +340,7 @@ void synchronous_deltaAndCumulative() { // Send in new measurement at time 30 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -357,7 +356,7 @@ void synchronous_deltaAndCumulative() { // Send in new measurement at time 40 for collector 2 assertThat( storage.buildMetricFor( - collector2, + registeredReader2, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -373,7 +372,7 @@ void synchronous_deltaAndCumulative() { // Send in new measurement at time 35 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -389,7 +388,7 @@ void synchronous_deltaAndCumulative() { // Send in new measurement at time 60 for collector 2 assertThat( storage.buildMetricFor( - collector2, + registeredReader2, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -412,7 +411,7 @@ void asynchronousCumulative_doesNotJoin() { // Send in new measurement at time 10 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -428,7 +427,7 @@ void asynchronousCumulative_doesNotJoin() { // Send in new measurement at time 30 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -443,7 +442,7 @@ void asynchronousCumulative_doesNotJoin() { // Send in new measurement at time 40 for collector 2 assertThat( storage.buildMetricFor( - collector2, + registeredReader2, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -459,7 +458,7 @@ void asynchronousCumulative_doesNotJoin() { // Send in new measurement at time 35 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -485,7 +484,7 @@ void asynchronousCumulative_dropsStale() { measurement1.put(attr1, DoubleAccumulation.create(3)); assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -511,7 +510,7 @@ void asynchronousCumulative_dropsStale() { measurement2.put(attr2, DoubleAccumulation.create(7)); assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -542,7 +541,7 @@ void asynchronousDelta_dropsStale() { measurement1.put(attr1, DoubleAccumulation.create(3)); assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -568,7 +567,7 @@ void asynchronousDelta_dropsStale() { measurement2.put(attr2, DoubleAccumulation.create(7)); assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -596,7 +595,7 @@ void asynchronousDelta_diffsLastTimestamp() { // Send in new measurement at time 10 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -612,7 +611,7 @@ void asynchronousDelta_diffsLastTimestamp() { // Send in new measurement at time 30 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -628,7 +627,7 @@ void asynchronousDelta_diffsLastTimestamp() { // Send in new measurement at time 40 for collector 2 assertThat( storage.buildMetricFor( - collector2, + registeredReader2, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -644,7 +643,7 @@ void asynchronousDelta_diffsLastTimestamp() { // Send in new measurement at time 35 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -667,7 +666,7 @@ void asynchronous_DeltaAndCumulative() { // Send in new measurement at time 10 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -683,7 +682,7 @@ void asynchronous_DeltaAndCumulative() { // Send in new measurement at time 30 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -699,7 +698,7 @@ void asynchronous_DeltaAndCumulative() { // Send in new measurement at time 40 for collector 2 assertThat( storage.buildMetricFor( - collector2, + registeredReader2, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -715,7 +714,7 @@ void asynchronous_DeltaAndCumulative() { // Send in new measurement at time 35 for collector 1 assertThat( storage.buildMetricFor( - collector1, + registeredReader1, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR, @@ -732,7 +731,7 @@ void asynchronous_DeltaAndCumulative() { // Send in new measurement at time 60 for collector 2 assertThat( storage.buildMetricFor( - collector2, + registeredReader2, Resource.empty(), InstrumentationScopeInfo.empty(), METRIC_DESCRIPTOR,