From 87d2f5056a276d80a29d933e1753e4176f8caf13 Mon Sep 17 00:00:00 2001 From: Devyansh Malhotra <94832704+DevyanshMalhotra@users.noreply.github.com> Date: Sat, 31 Oct 2026 08:54:26 +0000 Subject: [PATCH] Fix start timestamps for dynamically created metric instruments --- .../state/AsynchronousMetricStorage.java | 12 +- .../state/DeltaSynchronousMetricStorage.java | 8 +- .../state/AsynchronousMetricStorageTest.java | 147 ++++++++++++++++-- .../state/SynchronousMetricStorageTest.java | 63 +++++++- 4 files changed, 205 insertions(+), 25 deletions(-) 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 2aa28cb8ade..e54ea5ed54b 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 @@ -187,7 +187,9 @@ private AggregatorHandle getAggregatorHandle(Attributes attributes) { protected AggregatorHandle createAggregatorHandle() { return aggregator.createHandle( - registeredReader.getLastCollectEpochNanosOrDefault(instrumentCreationEpochNanos)); + Math.max( + registeredReader.getLastCollectEpochNanosOrDefault(instrumentCreationEpochNanos), + instrumentCreationEpochNanos)); } @Nullable @@ -274,10 +276,12 @@ Collection doCollect(long epochNanos) { currentPoints = new HashMap<>(); } - // Start time for asynchronous delta instruments is the time of the last collection, or if no - // collection has yet taken place, the time the instrument was created. + // Start time for asynchronous delta instruments is the later of the previous collection + // and instrument creation, so newly created instruments never predate their creation. long startEpochNanos = - registeredReader.getLastCollectEpochNanosOrDefault(instrumentCreationEpochNanos); + Math.max( + registeredReader.getLastCollectEpochNanosOrDefault(instrumentCreationEpochNanos), + instrumentCreationEpochNanos); aggregatorHandles.forEach( (attributes, handle) -> { diff --git a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaSynchronousMetricStorage.java b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaSynchronousMetricStorage.java index 51edfe0321d..5076f50bf9b 100644 --- a/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaSynchronousMetricStorage.java +++ b/sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaSynchronousMetricStorage.java @@ -295,10 +295,12 @@ public MetricData collect( } } - // Start time for synchronous delta instruments is the time of the last collection, or if no - // collection has yet taken place, the time the instrument was created. + // Start time for synchronous delta instruments is the later of the previous collection + // and instrument creation, so newly created instruments never predate their creation. long startEpochNanos = - registeredReader.getLastCollectEpochNanosOrDefault(instrumentCreationEpochNanos); + Math.max( + registeredReader.getLastCollectEpochNanosOrDefault(instrumentCreationEpochNanos), + instrumentCreationEpochNanos); // Grab aggregated points. aggregatorHandles.forEach( 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 f1dd1a13d13..77d59b04f17 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 @@ -279,6 +279,115 @@ void record_DuplicateAttributes(MemoryMode memoryMode) { point.hasValue(3).hasAttributes(attributeEntry("key1", "a"))))); } + @ParameterizedTest + @EnumSource(MemoryMode.class) + void collect_DeltaInstrumentCreatedAfterPreviousCollection(MemoryMode memoryMode) { + setup(memoryMode, AggregationTemporality.DELTA); + testClock.advance(Duration.ofSeconds(10)); + longCounterStorage.record(Attributes.empty(), 1); + longCounterStorage.collect(resource, scope, testClock.now()); + registeredReader.setLastCollectEpochNanos(testClock.now()); + + testClock.advance(Duration.ofSeconds(5)); + long creationEpochNanos = testClock.now(); + AsynchronousMetricStorage newStorage = + AsynchronousMetricStorage.create( + registeredReader, + registeredView, + testClock, + InstrumentDescriptor.create( + "new-counter", + "description", + "unit", + InstrumentType.COUNTER, + InstrumentValueType.LONG, + Advice.empty()), + asExemplarFilterInternal(ExemplarFilter.alwaysOff()), + /* enabled= */ true); + + testClock.advance(Duration.ofSeconds(5)); + newStorage.record(Attributes.empty(), 3); + assertThat(newStorage.collect(resource, scope, testClock.now())) + .hasLongSumSatisfying( + sum -> + sum.isDelta() + .hasPointsSatisfying( + point -> + point + .hasStartEpochNanos(creationEpochNanos) + .hasEpochNanos(testClock.now()) + .hasValue(3))); + long firstNewCollectionEpochNanos = testClock.now(); + registeredReader.setLastCollectEpochNanos(firstNewCollectionEpochNanos); + + testClock.advance(Duration.ofSeconds(5)); + newStorage.record(Attributes.empty(), 7); + assertThat(newStorage.collect(resource, scope, testClock.now())) + .hasLongSumSatisfying( + sum -> + sum.isDelta() + .hasPointsSatisfying( + point -> + point + .hasStartEpochNanos(firstNewCollectionEpochNanos) + .hasEpochNanos(testClock.now()) + .hasValue(4))); + } + + @ParameterizedTest + @EnumSource(MemoryMode.class) + void collect_CumulativeInstrumentCreatedAfterPreviousCollection(MemoryMode memoryMode) { + setup(memoryMode, AggregationTemporality.CUMULATIVE); + testClock.advance(Duration.ofSeconds(10)); + longCounterStorage.record(Attributes.empty(), 1); + longCounterStorage.collect(resource, scope, testClock.now()); + registeredReader.setLastCollectEpochNanos(testClock.now()); + + testClock.advance(Duration.ofSeconds(5)); + long creationEpochNanos = testClock.now(); + AsynchronousMetricStorage newStorage = + AsynchronousMetricStorage.create( + registeredReader, + registeredView, + testClock, + InstrumentDescriptor.create( + "new-counter", + "description", + "unit", + InstrumentType.COUNTER, + InstrumentValueType.LONG, + Advice.empty()), + asExemplarFilterInternal(ExemplarFilter.alwaysOff()), + /* enabled= */ true); + + testClock.advance(Duration.ofSeconds(5)); + newStorage.record(Attributes.empty(), 3); + assertThat(newStorage.collect(resource, scope, testClock.now())) + .hasLongSumSatisfying( + sum -> + sum.isCumulative() + .hasPointsSatisfying( + point -> + point + .hasStartEpochNanos(creationEpochNanos) + .hasEpochNanos(testClock.now()) + .hasValue(3))); + registeredReader.setLastCollectEpochNanos(testClock.now()); + + testClock.advance(Duration.ofSeconds(5)); + newStorage.record(Attributes.empty(), 7); + assertThat(newStorage.collect(resource, scope, testClock.now())) + .hasLongSumSatisfying( + sum -> + sum.isCumulative() + .hasPointsSatisfying( + point -> + point + .hasStartEpochNanos(creationEpochNanos) + .hasEpochNanos(testClock.now()) + .hasValue(7))); + } + @ParameterizedTest @EnumSource(MemoryMode.class) void collect_CumulativeReportsCumulativeObservations(MemoryMode memoryMode) { @@ -473,9 +582,11 @@ void collect_DeltaComputesDiff(MemoryMode memoryMode) { asExemplarFilterInternal(ExemplarFilter.alwaysOff()), /* enabled= */ true); - // Record measurement and collect at time 10 + // Collection timestamps must be after the instrument creation time. + long firstCollectEpochNanos = testClock.now() + 10; + // Record measurement and collect 10ns after creation longCounterStorage.record(Attributes.empty(), 3); - assertThat(longCounterStorage.collect(resource, scope, 10)) + assertThat(longCounterStorage.collect(resource, scope, firstCollectEpochNanos)) .hasLongSumSatisfying( sum -> sum.isDelta() @@ -483,51 +594,53 @@ void collect_DeltaComputesDiff(MemoryMode memoryMode) { point -> point .hasStartEpochNanos(testClock.now()) - .hasEpochNanos(10) + .hasEpochNanos(firstCollectEpochNanos) .hasValue(3) .hasAttributes(Attributes.empty()))); - registeredReader.setLastCollectEpochNanos(10); + registeredReader.setLastCollectEpochNanos(firstCollectEpochNanos); - // Record measurement and collect at time 30 + long secondCollectEpochNanos = testClock.now() + 30; + // Record measurement and collect 30ns after creation longCounterStorage.record(Attributes.empty(), 3); longCounterStorage.record(Attributes.builder().put("key", "value1").build(), 6); - assertThat(longCounterStorage.collect(resource, scope, 30)) + assertThat(longCounterStorage.collect(resource, scope, secondCollectEpochNanos)) .hasLongSumSatisfying( sum -> sum.isDelta() .hasPointsSatisfying( point -> point - .hasStartEpochNanos(10) - .hasEpochNanos(30) + .hasStartEpochNanos(firstCollectEpochNanos) + .hasEpochNanos(secondCollectEpochNanos) .hasValue(0) .hasAttributes(Attributes.empty()), point -> point - .hasStartEpochNanos(10) - .hasEpochNanos(30) + .hasStartEpochNanos(firstCollectEpochNanos) + .hasEpochNanos(secondCollectEpochNanos) .hasValue(6) .hasAttributes(Attributes.builder().put("key", "value1").build()))); - registeredReader.setLastCollectEpochNanos(30); + registeredReader.setLastCollectEpochNanos(secondCollectEpochNanos); - // Record measurement and collect at time 35 + long thirdCollectEpochNanos = testClock.now() + 35; + // Record measurement and collect 35ns after creation longCounterStorage.record(Attributes.empty(), 4); longCounterStorage.record(Attributes.builder().put("key", "value2").build(), 5); - assertThat(longCounterStorage.collect(resource, scope, 35)) + assertThat(longCounterStorage.collect(resource, scope, thirdCollectEpochNanos)) .hasLongSumSatisfying( sum -> sum.isDelta() .hasPointsSatisfying( point -> point - .hasStartEpochNanos(30) - .hasEpochNanos(35) + .hasStartEpochNanos(secondCollectEpochNanos) + .hasEpochNanos(thirdCollectEpochNanos) .hasValue(1) .hasAttributes(Attributes.empty()), point -> point - .hasStartEpochNanos(30) - .hasEpochNanos(35) + .hasStartEpochNanos(secondCollectEpochNanos) + .hasEpochNanos(thirdCollectEpochNanos) .hasValue(5) .hasAttributes(Attributes.builder().put("key", "value2").build()))); } 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 d5d382d4809..1253964dec2 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 @@ -45,6 +45,7 @@ import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; import io.opentelemetry.sdk.testing.time.TestClock; import java.time.Duration; +import java.time.Instant; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; @@ -75,7 +76,7 @@ public class SynchronousMetricStorageTest { private RegisteredReader deltaReader; private RegisteredReader cumulativeReader; - private final TestClock testClock = TestClock.create(); + private final TestClock testClock = TestClock.create(Instant.EPOCH); private Aggregator aggregator; private void initialize(MemoryMode memoryMode) { @@ -262,6 +263,66 @@ void recordAndCollect_CumulativeNewSeriesAfterFirstCollection(MemoryMode memoryM .hasAttributes(Attributes.builder().put("series", "B").build()))); } + @ParameterizedTest + @EnumSource(MemoryMode.class) + void recordAndCollect_DeltaInstrumentCreatedAfterPreviousCollection(MemoryMode memoryMode) { + initialize(memoryMode); + + DefaultSynchronousMetricStorage existingStorage = + DefaultSynchronousMetricStorage.create( + deltaReader, + METRIC_DESCRIPTOR, + aggregator, + AttributesFilters.ALLOW_ALL, + CARDINALITY_LIMIT, + testClock, + /* enabled= */ true); + existingStorage.recordDouble(1, Attributes.empty(), Context.current()); + testClock.advance(Duration.ofSeconds(10)); + existingStorage.collect(RESOURCE, INSTRUMENTATION_SCOPE_INFO, testClock.now()); + deltaReader.setLastCollectEpochNanos(testClock.now()); + + testClock.advance(Duration.ofSeconds(5)); + long creationEpochNanos = testClock.now(); + DefaultSynchronousMetricStorage newStorage = + DefaultSynchronousMetricStorage.create( + deltaReader, + METRIC_DESCRIPTOR, + aggregator, + AttributesFilters.ALLOW_ALL, + CARDINALITY_LIMIT, + testClock, + /* enabled= */ true); + + testClock.advance(Duration.ofSeconds(5)); + newStorage.recordDouble(3, Attributes.empty(), Context.current()); + assertThat(newStorage.collect(RESOURCE, INSTRUMENTATION_SCOPE_INFO, testClock.now())) + .hasDoubleSumSatisfying( + sum -> + sum.isDelta() + .hasPointsSatisfying( + point -> + point + .hasStartEpochNanos(creationEpochNanos) + .hasEpochNanos(testClock.now()) + .hasValue(3))); + long firstNewCollectionEpochNanos = testClock.now(); + deltaReader.setLastCollectEpochNanos(firstNewCollectionEpochNanos); + + testClock.advance(Duration.ofSeconds(5)); + newStorage.recordDouble(4, Attributes.empty(), Context.current()); + assertThat(newStorage.collect(RESOURCE, INSTRUMENTATION_SCOPE_INFO, testClock.now())) + .hasDoubleSumSatisfying( + sum -> + sum.isDelta() + .hasPointsSatisfying( + point -> + point + .hasStartEpochNanos(firstNewCollectionEpochNanos) + .hasEpochNanos(testClock.now()) + .hasValue(4))); + } + @Test void recordAndCollect_DeltaResets_ImmutableData() { initialize(IMMUTABLE_DATA);