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 99b2ff46f41..172f1d7dc62 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 @@ -55,9 +55,11 @@ public static SdkMeterProviderBuilder builder() { Resource resource, ViewRegistry viewRegistry, ExemplarFilter exemplarFilter) { + long startEpochNanos = clock.now(); this.registeredReaders = registeredReaders; this.sharedState = - MeterProviderSharedState.create(clock, resource, viewRegistry, exemplarFilter); + MeterProviderSharedState.create( + clock, resource, viewRegistry, exemplarFilter, startEpochNanos); this.registry = new ComponentRegistry<>( instrumentationLibraryInfo -> @@ -65,6 +67,7 @@ public static SdkMeterProviderBuilder builder() { for (RegisteredReader registeredReader : registeredReaders) { MetricProducer producer = new LeasedMetricProducer(registry, sharedState, registeredReader); registeredReader.getReader().register(producer); + registeredReader.setLastCollectEpochNanos(startEpochNanos); } } @@ -159,9 +162,11 @@ private static class LeasedMetricProducer implements MetricProducer { public Collection collectAllMetrics() { Collection meters = registry.getComponents(); List result = new ArrayList<>(); + long collectTime = sharedState.getClock().now(); for (SdkMeter meter : meters) { - result.addAll(meter.collectAll(registeredReader, sharedState.getClock().now())); + result.addAll(meter.collectAll(registeredReader, collectTime)); } + registeredReader.setLastCollectEpochNanos(collectTime); return Collections.unmodifiableCollection(result); } } 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 index 6e8feb41b00..16e3e7d15e9 100644 --- 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 @@ -6,6 +6,8 @@ package io.opentelemetry.sdk.metrics.internal.export; import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; +import io.opentelemetry.sdk.metrics.data.PointData; import io.opentelemetry.sdk.metrics.export.MetricReader; import java.util.concurrent.atomic.AtomicInteger; import javax.annotation.Nullable; @@ -21,6 +23,7 @@ public class RegisteredReader { private static final AtomicInteger ID_COUNTER = new AtomicInteger(1); private final int id = ID_COUNTER.incrementAndGet(); private final MetricReader metricReader; + private volatile long lastCollectEpochNanos; /** Construct a new collection info object storing information for collection against a reader. */ public static RegisteredReader create(MetricReader reader) { @@ -35,6 +38,25 @@ public MetricReader getReader() { return metricReader; } + /** + * Set the time the last collection took place for the reader. + * + *

Called by {@link SdkMeterProvider}'s {@link MetricProducer} after collection. + */ + public void setLastCollectEpochNanos(long epochNanos) { + this.lastCollectEpochNanos = epochNanos; + } + + /** + * Get the time of the last collection for the reader. + * + *

Used to compute the {@link PointData#getStartEpochNanos()} for instruments aggregations with + * {@link AggregationTemporality#DELTA} temporality. + */ + public long getLastCollectEpochNanos() { + return lastCollectEpochNanos; + } + @Override public int hashCode() { return 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 7e6bd67ae46..cc2442ef305 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 @@ -41,7 +41,6 @@ final class AsynchronousMetricStorage implements Metr 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; @@ -54,11 +53,17 @@ private AsynchronousMetricStorage( AttributesProcessor attributesProcessor) { this.registeredReader = registeredReader; this.metricDescriptor = metricDescriptor; - this.aggregationTemporality = + AggregationTemporality aggregationTemporality = registeredReader .getReader() .getAggregationTemporality(metricDescriptor.getSourceInstrument().getType()); - this.metricStorage = new TemporalMetricStorage<>(aggregator, /* isSynchronous= */ false); + this.metricStorage = + new TemporalMetricStorage<>( + aggregator, + /* isSynchronous= */ false, + registeredReader, + aggregationTemporality, + metricDescriptor); this.aggregator = aggregator; this.attributesProcessor = attributesProcessor; } @@ -146,14 +151,7 @@ public MetricData collectAndReset( Map currentAccumulations = accumulations; accumulations = new HashMap<>(); return metricStorage.buildMetricFor( - registeredReader, - resource, - instrumentationScopeInfo, - getMetricDescriptor(), - aggregationTemporality, - currentAccumulations, - startEpochNanos, - epochNanos); + resource, instrumentationScopeInfo, currentAccumulations, startEpochNanos, epochNanos); } @Override 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 38182895840..aa7ced9ee45 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 @@ -42,7 +42,6 @@ public final class DefaultSynchronousMetricStorage private final RegisteredReader registeredReader; private final MetricDescriptor metricDescriptor; - private final AggregationTemporality aggregationTemporality; private final Aggregator aggregator; private final ConcurrentHashMap> activeCollectionStorage = new ConcurrentHashMap<>(); @@ -56,12 +55,18 @@ public final class DefaultSynchronousMetricStorage AttributesProcessor attributesProcessor) { this.registeredReader = registeredReader; this.metricDescriptor = metricDescriptor; - this.aggregationTemporality = + AggregationTemporality aggregationTemporality = registeredReader .getReader() .getAggregationTemporality(metricDescriptor.getSourceInstrument().getType()); this.aggregator = aggregator; - this.temporalMetricStorage = new TemporalMetricStorage<>(aggregator, /* isSynchronous= */ true); + this.temporalMetricStorage = + new TemporalMetricStorage<>( + aggregator, + /* isSynchronous= */ true, + registeredReader, + aggregationTemporality, + metricDescriptor); this.attributesProcessor = attributesProcessor; } @@ -177,14 +182,7 @@ public MetricData collectAndReset( } return temporalMetricStorage.buildMetricFor( - registeredReader, - resource, - instrumentationScopeInfo, - getMetricDescriptor(), - aggregationTemporality, - accumulations, - startEpochNanos, - epochNanos); + resource, instrumentationScopeInfo, accumulations, startEpochNanos, epochNanos); } @Override diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MeterProviderSharedState.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MeterProviderSharedState.java index 1a5352f69dc..28353e87763 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MeterProviderSharedState.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/MeterProviderSharedState.java @@ -22,9 +22,13 @@ @Immutable public abstract class MeterProviderSharedState { public static MeterProviderSharedState create( - Clock clock, Resource resource, ViewRegistry viewRegistry, ExemplarFilter exemplarFilter) { + Clock clock, + Resource resource, + ViewRegistry viewRegistry, + ExemplarFilter exemplarFilter, + long startEpochNanos) { return new AutoValue_MeterProviderSharedState( - clock, resource, viewRegistry, clock.now(), exemplarFilter); + clock, resource, viewRegistry, startEpochNanos, exemplarFilter); } MeterProviderSharedState() {} @@ -42,7 +46,7 @@ public static MeterProviderSharedState create( * Returns the timestamp when this {@code MeterProvider} was started, in nanoseconds since Unix * epoch time. */ - abstract long getStartEpochNanos(); + public abstract long getStartEpochNanos(); /** Returns the {@link ExemplarFilter} for remembering synchronous measurements. */ abstract ExemplarFilter getExemplarFilter(); 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 ee738434a81..1d5544ec35c 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 @@ -24,66 +24,67 @@ class TemporalMetricStorage { private final Aggregator aggregator; private final boolean isSynchronous; - private final Map> reportHistory = new HashMap<>(); + private final RegisteredReader registeredReader; + private Map lastAccumulation = new HashMap<>(); + private final AggregationTemporality temporality; + private final MetricDescriptor metricDescriptor; - TemporalMetricStorage(Aggregator aggregator, boolean isSynchronous) { + TemporalMetricStorage( + Aggregator aggregator, + boolean isSynchronous, + RegisteredReader registeredReader, + AggregationTemporality aggregationTemporality, + MetricDescriptor metricDescriptor) { this.aggregator = aggregator; this.isSynchronous = isSynchronous; + this.registeredReader = registeredReader; + this.temporality = aggregationTemporality; + this.metricDescriptor = metricDescriptor; } /** - * Builds the {@link MetricData} streams to report against a specific metric reader. + * Builds the {@link MetricData} for the {@code currentAccumulation}. * * @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. - * @param currentAccumulation THe current accumulation of metric data from instruments. This might + * @param currentAccumulation The current accumulation of metric data from instruments. This might * be delta (for synchronous) or cumulative (for asynchronous). * @param startEpochNanos The timestamp when the metrics SDK started. * @param epochNanos The current collection timestamp. * @return The {@link MetricData} points. */ synchronized MetricData buildMetricFor( - RegisteredReader registeredReader, Resource resource, InstrumentationScopeInfo instrumentationScopeInfo, - MetricDescriptor descriptor, - // Temporality is requested by the collector. - AggregationTemporality temporality, Map currentAccumulation, long startEpochNanos, long epochNanos) { - // In case it's our first collection, default to start timestamp. - long lastCollectionEpoch = startEpochNanos; + Map result = currentAccumulation; - // Check our last report time. - 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: - // 1. Delta Aggregation (temporality) + Cumulative recording (async instrument). - // Here we diff with last cumulative to get a delta. - // 2. Cumulative Aggregation + Delta recording (sync instrument). - // Here we merge with our last record to get a cumulative aggregation. - // 3. Cumulative Aggregation + Cumulative recording - do nothing - // 4. Delta Aggregation + Delta recording - do nothing. - if (temporality == AggregationTemporality.DELTA && !isSynchronous) { - MetricStorageUtils.diffInPlace(last.getAccumulation(), currentAccumulation, aggregator); - result = last.getAccumulation(); - } else if (temporality == AggregationTemporality.CUMULATIVE && isSynchronous) { - // We need to make sure the current delta recording gets merged into the previous cumulative - // for the next cumulative measurement. - MetricStorageUtils.mergeAndPreserveInPlace( - last.getAccumulation(), currentAccumulation, aggregator); - // Note: We allow going over our hard limit on attribute streams when first merging, but - // preserve after this point. - if (last.getAccumulation().size() > MetricStorageUtils.MAX_ACCUMULATIONS) { - MetricStorageUtils.removeUnseen(last.getAccumulation(), currentAccumulation); - } - result = last.getAccumulation(); + long lastCollectionEpoch = registeredReader.getLastCollectEpochNanos(); + // Use aggregation temporality + instrument to determine if we do a merge or a diff of + // previous. We have the following four scenarios: + // 1. Delta Aggregation (temporality) + Cumulative recording (async instrument). + // Here we diff with last cumulative to get a delta. + // 2. Cumulative Aggregation + Delta recording (sync instrument). + // Here we merge with our last record to get a cumulative aggregation. + // 3. Cumulative Aggregation + Cumulative recording - do nothing + // 4. Delta Aggregation + Delta recording - do nothing. + if (temporality == AggregationTemporality.DELTA && !isSynchronous) { + MetricStorageUtils.diffInPlace(lastAccumulation, currentAccumulation, aggregator); + result = lastAccumulation; + } else if (temporality == AggregationTemporality.CUMULATIVE && isSynchronous) { + // We need to make sure the current delta recording gets merged into the previous cumulative + // for the next cumulative measurement. + MetricStorageUtils.mergeAndPreserveInPlace(lastAccumulation, currentAccumulation, aggregator); + // Note: We allow going over our hard limit on attribute streams when first merging, but + // preserve after this point. + if (lastAccumulation.size() > MetricStorageUtils.MAX_ACCUMULATIONS) { + MetricStorageUtils.removeUnseen(lastAccumulation, currentAccumulation); } + result = lastAccumulation; } + // Update last reported (cumulative) accumulation. // For synchronous instruments, we need the merge result. // For asynchronous instruments, we need the recorded value. @@ -91,11 +92,10 @@ 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(registeredReader, new LastReportedAccumulation<>(result, epochNanos)); + lastAccumulation = result; } else { // Async instruments record the raw measurement. - reportHistory.put( - registeredReader, new LastReportedAccumulation<>(currentAccumulation, epochNanos)); + lastAccumulation = currentAccumulation; } if (result.isEmpty()) { return EmptyMetricData.getInstance(); @@ -103,36 +103,11 @@ synchronized MetricData buildMetricFor( return aggregator.toMetricData( resource, instrumentationScopeInfo, - descriptor, + metricDescriptor, result, temporality, startEpochNanos, lastCollectionEpoch, epochNanos); } - - /** Remembers what was presented to a specific exporter. */ - private static class LastReportedAccumulation { - private final Map accumulation; - private final long epochNanos; - - /** - * Constructs a new reporting record. - * - * @param accumulation The last accumulation of metric data. - * @param epochNanos The timestamp the data was reported. - */ - LastReportedAccumulation(Map accumulation, long epochNanos) { - this.accumulation = accumulation; - this.epochNanos = epochNanos; - } - - long getEpochNanos() { - return epochNanos; - } - - Map getAccumulation() { - return accumulation; - } - } } 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 index 47cdc4ab930..a063b0f69be 100644 --- 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 @@ -36,4 +36,15 @@ void getReader() { assertThat(registeredReader.getReader()).isSameAs(reader); } + + @Test + void setAndGetLastCollectEpochNanos() { + RegisteredReader registeredReader = RegisteredReader.create(reader); + + assertThat(registeredReader.getLastCollectEpochNanos()).isEqualTo(0); + registeredReader.setLastCollectEpochNanos(1); + assertThat(registeredReader.getLastCollectEpochNanos()).isEqualTo(1); + registeredReader.setLastCollectEpochNanos(5); + assertThat(registeredReader.getLastCollectEpochNanos()).isEqualTo(5); + } } 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 fc44ca77d3b..bb2d7618392 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 @@ -54,13 +54,12 @@ class TemporalMetricStorageTest { .createAggregator(ASYNC_DESCRIPTOR, ExemplarFilter.neverSample()); @Mock private MetricReader reader; - private RegisteredReader registeredReader1; - private RegisteredReader registeredReader2; + private RegisteredReader registeredReader; @BeforeEach void setup() { - registeredReader1 = RegisteredReader.create(reader); - registeredReader2 = RegisteredReader.create(reader); + registeredReader = RegisteredReader.create(reader); + registeredReader.setLastCollectEpochNanos(0); } private static Map createMeasurement(double value) { @@ -73,66 +72,34 @@ private static Map createMeasurement(double valu void synchronousCumulative_joinsWithLastMeasurementForCumulative() { AggregationTemporality temporality = AggregationTemporality.CUMULATIVE; TemporalMetricStorage storage = - new TemporalMetricStorage<>(SUM, /* isSynchronous= */ true); - // Send in new measurement at time 10 for collector 1 + new TemporalMetricStorage<>( + SUM, /* isSynchronous= */ true, registeredReader, temporality, METRIC_DESCRIPTOR); + // Send in new measurement at time 10 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(3), - 0, - 10)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(3), 0, 10)) .hasDoubleSumSatisfying( sum -> sum.isCumulative() .hasPointsSatisfying( point -> point.hasStartEpochNanos(0).hasEpochNanos(10).hasValue(3))); - // Send in new measurement at time 30 for collector 1 + registeredReader.setLastCollectEpochNanos(10); + + // Send in new measurement at time 30 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(3), - 0, - 30)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(3), 0, 30)) .hasDoubleSumSatisfying( sum -> sum.isCumulative() .hasPointsSatisfying( point -> point.hasStartEpochNanos(0).hasEpochNanos(30).hasValue(6))); - // Send in new measurement at time 40 for collector 2 - assertThat( - storage.buildMetricFor( - registeredReader2, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(4), - 0, - 60)) - .hasDoubleSumSatisfying( - sum -> - sum.isCumulative() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(60).hasValue(4))); - // Send in new measurement at time 35 for collector 1 + registeredReader.setLastCollectEpochNanos(30); + + // Send in new measurement at time 35 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(2), - 0, - 35)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(2), 0, 35)) .hasDoubleSumSatisfying( sum -> sum.isCumulative() @@ -143,9 +110,14 @@ void synchronousCumulative_joinsWithLastMeasurementForCumulative() { @Test void synchronousCumulative_dropsStaleAtLimit() { TemporalMetricStorage storage = - new TemporalMetricStorage<>(SUM, /* isSynchronous= */ true); + new TemporalMetricStorage<>( + SUM, + /* isSynchronous= */ true, + registeredReader, + AggregationTemporality.CUMULATIVE, + METRIC_DESCRIPTOR); - // Send in new measurement at time 10 for collector 1, with attr1 + // Send in new measurement at time 10, with attr1 Map measurement1 = new HashMap<>(); for (int i = 0; i < MetricStorageUtils.MAX_ACCUMULATIONS; i++) { Attributes attr1 = Attributes.builder().put("key", "value" + i).build(); @@ -153,28 +125,23 @@ void synchronousCumulative_dropsStaleAtLimit() { } assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.CUMULATIVE, - measurement1, - 0, - 10)) + Resource.empty(), InstrumentationScopeInfo.empty(), measurement1, 0, 10)) .hasDoubleSumSatisfying( sum -> sum.isCumulative() .satisfies( - sumPoint -> - assertThat(sumPoint.getPoints()) + sumData -> + assertThat(sumData.getPoints()) .hasSize(MetricStorageUtils.MAX_ACCUMULATIONS) .allSatisfy( - sumPointData -> { - assertThat(sumPointData.getStartEpochNanos()).isEqualTo(0); - assertThat(sumPointData.getEpochNanos()).isEqualTo(10); - assertThat(sumPointData.getValue()).isEqualTo(3); + point -> { + assertThat(point.getStartEpochNanos()).isEqualTo(0); + assertThat(point.getEpochNanos()).isEqualTo(10); + assertThat(point.getValue()).isEqualTo(3); }))); - // Send in new measurement at time 20 for collector 1, with attr2 + registeredReader.setLastCollectEpochNanos(10); + + // Send in new measurement at time 20, with attr2 // Result should drop accumulation for attr1, only reporting accumulation for attr2 Map measurement2 = new HashMap<>(); Attributes attr2 = @@ -184,14 +151,7 @@ void synchronousCumulative_dropsStaleAtLimit() { measurement2.put(attr2, DoubleAccumulation.create(3)); assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.CUMULATIVE, - measurement2, - 0, - 20)) + Resource.empty(), InstrumentationScopeInfo.empty(), measurement2, 0, 20)) .hasDoubleSumSatisfying( sum -> sum.isCumulative().hasPointsSatisfying(point -> point.hasAttributes(attr2))); } @@ -199,43 +159,40 @@ void synchronousCumulative_dropsStaleAtLimit() { @Test void synchronousDelta_dropsStale() { TemporalMetricStorage storage = - new TemporalMetricStorage<>(SUM, /* isSynchronous= */ true); + new TemporalMetricStorage<>( + SUM, + /* isSynchronous= */ true, + registeredReader, + AggregationTemporality.DELTA, + METRIC_DESCRIPTOR); - // Send in new measurement at time 10 for collector 1, with attr1 + // Send in new measurement at time 10, with attr1 Map measurement1 = new HashMap<>(); Attributes attr1 = Attributes.builder().put("key", "value1").build(); measurement1.put(attr1, DoubleAccumulation.create(3)); assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.DELTA, - measurement1, - 0, - 10)) + Resource.empty(), InstrumentationScopeInfo.empty(), measurement1, 0, 10)) .hasDoubleSumSatisfying( sum -> sum.isDelta() .hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(10).hasValue(3))); + point -> + point + .hasStartEpochNanos(0) + .hasEpochNanos(10) + .hasAttributes(attr1) + .hasValue(3))); + registeredReader.setLastCollectEpochNanos(10); - // Send in new measurement at time 20 for collector 1, with attr2 + // Send in new measurement at time 20, with attr2 // Result should drop accumulation for attr1, only reporting accumulation for attr2 Map measurement2 = new HashMap<>(); Attributes attr2 = Attributes.builder().put("key", "value2").build(); measurement2.put(attr2, DoubleAccumulation.create(7)); assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.DELTA, - measurement2, - 0, - 20)) + Resource.empty(), InstrumentationScopeInfo.empty(), measurement2, 0, 20)) .hasDoubleSumSatisfying( sum -> sum.isDelta() @@ -252,220 +209,77 @@ void synchronousDelta_dropsStale() { void synchronousDelta_useLastTimestamp() { AggregationTemporality temporality = AggregationTemporality.DELTA; TemporalMetricStorage storage = - new TemporalMetricStorage<>(SUM, /* isSynchronous= */ true); - // Send in new measurement at time 10 for collector 1 + new TemporalMetricStorage<>( + SUM, /* isSynchronous= */ true, registeredReader, temporality, METRIC_DESCRIPTOR); + // Send in new measurement at time 10 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(3), - 0, - 10)) - .hasDoubleSumSatisfying( - sum -> - sum.hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(10).hasValue(3))); - // Send in new measurement at time 30 for collector 1 - assertThat( - storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(3), - 0, - 30)) - .hasDoubleSumSatisfying( - sum -> - sum.hasPointsSatisfying( - point -> point.hasStartEpochNanos(10).hasEpochNanos(30).hasValue(3))); - // Send in new measurement at time 40 for collector 2 - assertThat( - storage.buildMetricFor( - registeredReader2, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(4), - 0, - 60)) - .hasDoubleSumSatisfying( - sum -> - sum.isDelta() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(60).hasValue(4))); - // Send in new measurement at time 35 for collector 1 - assertThat( - storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(2), - 0, - 35)) - .hasDoubleSumSatisfying( - sum -> - sum.isDelta() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(30).hasEpochNanos(35).hasValue(2))); - } - - @Test - void synchronous_deltaAndCumulative() { - TemporalMetricStorage storage = - new TemporalMetricStorage<>(SUM, /* isSynchronous= */ true); - // Send in new measurement at time 10 for collector 1 - assertThat( - storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.DELTA, - createMeasurement(3), - 0, - 10)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(3), 0, 10)) .hasDoubleSumSatisfying( sum -> sum.isDelta() .hasPointsSatisfying( point -> point.hasStartEpochNanos(0).hasEpochNanos(10).hasValue(3))); - // Send in new measurement at time 30 for collector 1 + registeredReader.setLastCollectEpochNanos(10); + + // Send in new measurement at time 30 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.DELTA, - createMeasurement(3), - 0, - 30)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(3), 0, 30)) .hasDoubleSumSatisfying( sum -> sum.isDelta() .hasPointsSatisfying( point -> point.hasStartEpochNanos(10).hasEpochNanos(30).hasValue(3))); - // Send in new measurement at time 40 for collector 2 - assertThat( - storage.buildMetricFor( - registeredReader2, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.CUMULATIVE, - createMeasurement(4), - 0, - 40)) - .hasDoubleSumSatisfying( - sum -> - sum.isCumulative() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(40).hasValue(4))); - // Send in new measurement at time 35 for collector 1 + registeredReader.setLastCollectEpochNanos(30); + + // Send in new measurement at time 35 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.DELTA, - createMeasurement(2), - 0, - 35)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(2), 0, 35)) .hasDoubleSumSatisfying( sum -> sum.isDelta() .hasPointsSatisfying( point -> point.hasStartEpochNanos(30).hasEpochNanos(35).hasValue(2))); - // Send in new measurement at time 60 for collector 2 - assertThat( - storage.buildMetricFor( - registeredReader2, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.CUMULATIVE, - createMeasurement(4), - 0, - 60)) - .hasDoubleSumSatisfying( - sum -> - sum.isCumulative() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(60).hasValue(8))); } @Test void asynchronousCumulative_doesNotJoin() { AggregationTemporality temporality = AggregationTemporality.CUMULATIVE; TemporalMetricStorage storage = - new TemporalMetricStorage<>(ASYNC_SUM, /* isSynchronous= */ false); - // Send in new measurement at time 10 for collector 1 + new TemporalMetricStorage<>( + ASYNC_SUM, + /* isSynchronous= */ false, + registeredReader, + temporality, + METRIC_DESCRIPTOR); + // Send in new measurement at time 10 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(3), - 0, - 10)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(3), 0, 10)) .hasDoubleSumSatisfying( sum -> sum.isCumulative() .hasPointsSatisfying( point -> point.hasStartEpochNanos(0).hasEpochNanos(10).hasValue(3))); - // Send in new measurement at time 30 for collector 1 - assertThat( - storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(3), - 0, - 30)) - .hasDoubleSumSatisfying( - sum -> - sum.hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(30).hasValue(3))); - // Send in new measurement at time 40 for collector 2 + registeredReader.setLastCollectEpochNanos(10); + + // Send in new measurement at time 30 assertThat( storage.buildMetricFor( - registeredReader2, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(4), - 0, - 60)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(3), 0, 30)) .hasDoubleSumSatisfying( sum -> sum.isCumulative() .hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(60).hasValue(4))); - // Send in new measurement at time 35 for collector 1 + point -> point.hasStartEpochNanos(0).hasEpochNanos(30).hasValue(3))); + registeredReader.setLastCollectEpochNanos(30); + + // Send in new measurement at time 35 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(2), - 0, - 35)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(2), 0, 35)) .hasDoubleSumSatisfying( sum -> sum.isCumulative() @@ -476,22 +290,20 @@ void asynchronousCumulative_doesNotJoin() { @Test void asynchronousCumulative_dropsStale() { TemporalMetricStorage storage = - new TemporalMetricStorage<>(ASYNC_SUM, /* isSynchronous= */ false); + new TemporalMetricStorage<>( + ASYNC_SUM, + /* isSynchronous= */ false, + registeredReader, + AggregationTemporality.CUMULATIVE, + METRIC_DESCRIPTOR); - // Send in new measurement at time 10 for collector 1, with attr1 + // Send in new measurement at time 10, with attr1 Map measurement1 = new HashMap<>(); Attributes attr1 = Attributes.builder().put("key", "value1").build(); measurement1.put(attr1, DoubleAccumulation.create(3)); assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.CUMULATIVE, - measurement1, - 0, - 10)) + Resource.empty(), InstrumentationScopeInfo.empty(), measurement1, 0, 10)) .hasDoubleSumSatisfying( sum -> sum.isCumulative() @@ -502,22 +314,16 @@ void asynchronousCumulative_dropsStale() { .hasEpochNanos(10) .hasAttributes(attr1) .hasValue(3))); + registeredReader.setLastCollectEpochNanos(10); - // Send in new measurement at time 20 for collector 1, with attr2 + // Send in new measurement at time 20, with attr2 // Result should drop accumulation for attr1, only reporting accumulation for attr2 Map measurement2 = new HashMap<>(); Attributes attr2 = Attributes.builder().put("key", "value2").build(); measurement2.put(attr2, DoubleAccumulation.create(7)); assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.CUMULATIVE, - measurement2, - 0, - 20)) + Resource.empty(), InstrumentationScopeInfo.empty(), measurement2, 0, 20)) .hasDoubleSumSatisfying( sum -> sum.isCumulative() @@ -533,22 +339,20 @@ void asynchronousCumulative_dropsStale() { @Test void asynchronousDelta_dropsStale() { TemporalMetricStorage storage = - new TemporalMetricStorage<>(ASYNC_SUM, /* isSynchronous= */ false); + new TemporalMetricStorage<>( + ASYNC_SUM, + /* isSynchronous= */ false, + registeredReader, + AggregationTemporality.DELTA, + METRIC_DESCRIPTOR); - // Send in new measurement at time 10 for collector 1, with attr1 + // Send in new measurement at time 10, with attr1 Map measurement1 = new HashMap<>(); Attributes attr1 = Attributes.builder().put("key", "value1").build(); measurement1.put(attr1, DoubleAccumulation.create(3)); assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.DELTA, - measurement1, - 0, - 10)) + Resource.empty(), InstrumentationScopeInfo.empty(), measurement1, 0, 10)) .hasDoubleSumSatisfying( sum -> sum.isDelta() @@ -559,22 +363,16 @@ void asynchronousDelta_dropsStale() { .hasEpochNanos(10) .hasAttributes(attr1) .hasValue(3))); + registeredReader.setLastCollectEpochNanos(10); - // Send in new measurement at time 20 for collector 1, with attr2 + // Send in new measurement at time 20, with attr2 // Result should drop accumulation for attr1, only reporting accumulation for attr2 Map measurement2 = new HashMap<>(); Attributes attr2 = Attributes.builder().put("key", "value2").build(); measurement2.put(attr2, DoubleAccumulation.create(7)); assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.DELTA, - measurement2, - 0, - 20)) + Resource.empty(), InstrumentationScopeInfo.empty(), measurement2, 0, 20)) .hasDoubleSumSatisfying( sum -> sum.isDelta() @@ -591,158 +389,42 @@ void asynchronousDelta_dropsStale() { void asynchronousDelta_diffsLastTimestamp() { AggregationTemporality temporality = AggregationTemporality.DELTA; TemporalMetricStorage storage = - new TemporalMetricStorage<>(ASYNC_SUM, /* isSynchronous= */ false); - // Send in new measurement at time 10 for collector 1 + new TemporalMetricStorage<>( + ASYNC_SUM, + /* isSynchronous= */ false, + registeredReader, + temporality, + METRIC_DESCRIPTOR); + // Send in new measurement at time 10 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(3), - 0, - 10)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(3), 0, 10)) .hasDoubleSumSatisfying( sum -> sum.isDelta() .hasPointsSatisfying( point -> point.hasStartEpochNanos(0).hasEpochNanos(10).hasValue(3))); - // Send in new measurement at time 30 for collector 1 - assertThat( - storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(3), - 0, - 30)) - .hasDoubleSumSatisfying( - sum -> - sum.isDelta() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(10).hasEpochNanos(30).hasValue(0))); - // Send in new measurement at time 40 for collector 2 - assertThat( - storage.buildMetricFor( - registeredReader2, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(4), - 0, - 60)) - .hasDoubleSumSatisfying( - sum -> - sum.isDelta() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(60).hasValue(4))); - // Send in new measurement at time 35 for collector 1 - assertThat( - storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - temporality, - createMeasurement(2), - 0, - 35)) - .hasDoubleSumSatisfying( - sum -> - sum.isDelta() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(30).hasEpochNanos(35).hasValue(-1))); - } + registeredReader.setLastCollectEpochNanos(10); - @Test - void asynchronous_DeltaAndCumulative() { - TemporalMetricStorage storage = - new TemporalMetricStorage<>(ASYNC_SUM, /* isSynchronous= */ false); - - // Send in new measurement at time 10 for collector 1 + // Send in new measurement at time 30 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.DELTA, - createMeasurement(3), - 0, - 10)) - .hasDoubleSumSatisfying( - sum -> - sum.isDelta() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(10).hasValue(3))); - // Send in new measurement at time 30 for collector 1 - assertThat( - storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.DELTA, - createMeasurement(3), - 0, - 30)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(3), 0, 30)) .hasDoubleSumSatisfying( sum -> sum.isDelta() .hasPointsSatisfying( point -> point.hasStartEpochNanos(10).hasEpochNanos(30).hasValue(0))); - // Send in new measurement at time 40 for collector 2 - assertThat( - storage.buildMetricFor( - registeredReader2, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.CUMULATIVE, - createMeasurement(4), - 0, - 60)) - .hasDoubleSumSatisfying( - sum -> - sum.isCumulative() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(60).hasValue(4))); - // Send in new measurement at time 35 for collector 1 + registeredReader.setLastCollectEpochNanos(30); + + // Send in new measurement at time 35 assertThat( storage.buildMetricFor( - registeredReader1, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.DELTA, - createMeasurement(2), - 0, - 35)) + Resource.empty(), InstrumentationScopeInfo.empty(), createMeasurement(2), 0, 35)) .hasDoubleSumSatisfying( sum -> sum.isDelta() .hasPointsSatisfying( point -> point.hasStartEpochNanos(30).hasEpochNanos(35).hasValue(-1))); - - // Send in new measurement at time 60 for collector 2 - assertThat( - storage.buildMetricFor( - registeredReader2, - Resource.empty(), - InstrumentationScopeInfo.empty(), - METRIC_DESCRIPTOR, - AggregationTemporality.CUMULATIVE, - createMeasurement(5), - 0, - 60)) - .hasDoubleSumSatisfying( - sum -> - sum.isCumulative() - .hasPointsSatisfying( - point -> point.hasStartEpochNanos(0).hasEpochNanos(60).hasValue(5))); } }