Skip to content

Fix start timestamps for dynamically created metric instruments #8992

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
DevyanshMalhotra wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Fix start timestamps for dynamically created metric instruments #8992

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions .../src/main/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,9 @@ private AggregatorHandle<T> getAggregatorHandle(Attributes attributes) {

protected AggregatorHandle<T> createAggregatorHandle() {
return aggregator.createHandle(
registeredReader.getLastCollectEpochNanosOrDefault(instrumentCreationEpochNanos));
Math.max(
registeredReader.getLastCollectEpochNanosOrDefault(instrumentCreationEpochNanos),
instrumentCreationEpochNanos));
}

@Nullable
Expand Down Expand Up @@ -274,10 +276,12 @@ Collection<T> 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) -> {
Expand Down
8 changes: 5 additions & 3 deletions .../main/java/io/opentelemetry/sdk/metrics/internal/state/DeltaSynchronousMetricStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
147 changes: 130 additions & 17 deletions .../test/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -473,61 +582,65 @@ 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()
.hasPointsSatisfying(
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())));
}
Expand Down
63 changes: 62 additions & 1 deletion ...c/test/java/io/opentelemetry/sdk/metrics/internal/state/SynchronousMetricStorageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<LongPointData> aggregator;

private void initialize(MemoryMode memoryMode) {
Expand Down Expand Up @@ -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);
Expand Down
Loading