Series start time (#8180)

Signed-off-by: Gregor Zeitlinger <gregor.zeitlinger@grafana.com>
Co-authored-by: Gregor Zeitlinger <gregor.zeitlinger@grafana.com>

Author: jack-berg

exporters/prometheus/src/test/java/io/opentelemetry/exporter/prometheus/PrometheusMetricReaderTest.java

@@ -972,29 +972,23 @@ void createdTimestamp() throws IOException {
 
     LongCounter counter = meter.counterBuilder("requests").build();
     testClock.advance(Duration.ofMillis(1));
+    long bearStartNanos = testClock.now();
     counter.add(3, Attributes.builder().put("animal", "bear").build());
     testClock.advance(Duration.ofMillis(1));
+    long mouseStartNanos = testClock.now();
     counter.add(2, Attributes.builder().put("animal", "mouse").build());
     testClock.advance(Duration.ofMillis(1));
 
-    // There is a curious difference between Prometheus and OpenTelemetry:
-    // In Prometheus metrics the _created timestamp is per data point,
-    // i.e. the _created timestamp says when this specific set of label values
-    // was first observed.
-    // In the OTel Java SDK the _created timestamp is the initialization time
-    // of the SdkMeterProvider, i.e. all data points will have the same _created timestamp.
-    // So we expect the _created timestamp to be the start time of the application,
-    // not the timestamp when the counter or an individual data point was created.
     String expected =
         ""
             + "# TYPE requests counter\n"
             + "requests_total{animal=\"bear\",otel_scope_name=\"test\"} 3.0\n"
             + "requests_created{animal=\"bear\",otel_scope_name=\"test\"} "
-            + createdTimestamp
+            + convertTimestamp(bearStartNanos)
             + "\n"
             + "requests_total{animal=\"mouse\",otel_scope_name=\"test\"} 2.0\n"
             + "requests_created{animal=\"mouse\",otel_scope_name=\"test\"} "
-            + createdTimestamp
+            + convertTimestamp(mouseStartNanos)
             + "\n"
             + "# TYPE target info\n"
             + "target_info{service_name=\"unknown_service:java\",telemetry_sdk_language=\"java\",telemetry_sdk_name=\"opentelemetry\",telemetry_sdk_version=\"1.x.x\"} 1\n"

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/internal/aggregator/HistogramBenchmark.java

@@ -7,6 +7,7 @@
 
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.common.Clock;
 import java.util.concurrent.TimeUnit;
 import java.util.function.DoubleSupplier;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -40,7 +41,7 @@ public static class ThreadState {
 
     @Setup(Level.Trial)
     public final void setup() {
-      aggregatorHandle = aggregation.getAggregator().createHandle();
+      aggregatorHandle = aggregation.getAggregator().createHandle(Clock.getDefault().now());
       valueSupplier = valueGen.supplier();
     }
 

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/internal/aggregator/HistogramScaleBenchmark.java

@@ -7,6 +7,7 @@
 
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.common.Clock;
 import java.util.concurrent.TimeUnit;
 import java.util.function.DoubleSupplier;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -46,7 +47,7 @@ public static class ThreadState {
 
     @Setup(Level.Invocation)
     public final void setup() {
-      aggregatorHandle = aggregation.getAggregator().createHandle();
+      aggregatorHandle = aggregation.getAggregator().createHandle(Clock.getDefault().now());
       valueSupplier = valueGen.supplier();
     }
 

sdk/metrics/src/jmhBasedTest/java/io/opentelemetry/sdk/metrics/internal/state/InstrumentGarbageCollectionBenchmarkTest.java

@@ -30,8 +30,8 @@ public class InstrumentGarbageCollectionBenchmarkTest {
 
   /**
    * This test validates that in {@link MemoryMode#REUSABLE_DATA}, any {@link
-   * MetricStorage#collect(Resource, InstrumentationScopeInfo, long, long)} barely allocates memory
-   * which is then subsequently garbage collected. It is done so comparatively to {@link
+   * MetricStorage#collect(Resource, InstrumentationScopeInfo, long)} barely allocates memory which
+   * is then subsequently garbage collected. It is done so comparatively to {@link
    * MemoryMode#IMMUTABLE_DATA},
    *
    * <p>It runs the JMH test {@link InstrumentGarbageCollectionBenchmark} with GC profiler, and

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

@@ -132,8 +132,7 @@ Collection<MetricData> collectAll(RegisteredReader registeredReader, long epochN
       // Only invoke callbacks if meter is enabled
       if (meterEnabled) {
         for (CallbackRegistration callbackRegistration : currentRegisteredCallbacks) {
-          callbackRegistration.invokeCallback(
-              registeredReader, meterProviderSharedState.getStartEpochNanos(), epochNanos);
+          callbackRegistration.invokeCallback(registeredReader);
         }
       }
 
@@ -145,10 +144,7 @@ Collection<MetricData> collectAll(RegisteredReader registeredReader, long epochN
       for (MetricStorage storage : storages) {
         MetricData current =
             storage.collect(
-                meterProviderSharedState.getResource(),
-                getInstrumentationScopeInfo(),
-                meterProviderSharedState.getStartEpochNanos(),
-                epochNanos);
+                meterProviderSharedState.getResource(), getInstrumentationScopeInfo(), epochNanos);
         // Ignore if the metric data doesn't have any data points, for example when aggregation is
         // Aggregation#drop()
         if (!current.isEmpty()) {
@@ -288,6 +284,7 @@ WriteableMetricStorage registerSynchronousMetricStorage(InstrumentDescriptor ins
                 SynchronousMetricStorage.create(
                     reader,
                     registeredView,
+                    meterProviderSharedState.getClock(),
                     instrument,
                     meterProviderSharedState.getExemplarFilter(),
                     meterEnabled)));
@@ -317,7 +314,11 @@ SdkObservableMeasurement registerObservableMeasurement(
         registeredStorages.add(
             registry.register(
                 AsynchronousMetricStorage.create(
-                    reader, registeredView, instrumentDescriptor, meterEnabled)));
+                    reader,
+                    registeredView,
+                    meterProviderSharedState.getClock(),
+                    instrumentDescriptor,
+                    meterEnabled)));
       }
     }
 

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

@@ -72,7 +72,6 @@ public static SdkMeterProviderBuilder builder() {
       Resource resource,
       ExemplarFilterInternal exemplarFilter,
       ScopeConfigurator<MeterConfig> meterConfigurator) {
-    long startEpochNanos = clock.now();
     this.registeredViews = registeredViews;
     this.registeredReaders =
         metricReaders.entrySet().stream()
@@ -83,8 +82,7 @@ public static SdkMeterProviderBuilder builder() {
                         ViewRegistry.create(entry.getKey(), entry.getValue(), registeredViews)))
             .collect(toList());
     this.metricProducers = metricProducers;
-    this.sharedState =
-        MeterProviderSharedState.create(clock, resource, exemplarFilter, startEpochNanos);
+    this.sharedState = MeterProviderSharedState.create(clock, resource, exemplarFilter);
     this.registry =
         new ComponentRegistry<>(
             instrumentationLibraryInfo ->
@@ -99,7 +97,6 @@ public static SdkMeterProviderBuilder builder() {
       readerMetricProducers.add(new LeasedMetricProducer(registry, sharedState, registeredReader));
       MetricReader reader = registeredReader.getReader();
       reader.register(new SdkCollectionRegistration(readerMetricProducers, sharedState));
-      registeredReader.setLastCollectEpochNanos(startEpochNanos);
       if (reader instanceof PeriodicMetricReader) {
         setReaderMeterProvider((PeriodicMetricReader) reader, this);
       }

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/aggregator/Aggregator.java

@@ -30,12 +30,16 @@ static Aggregator<?> drop() {
   }
 
   /**
-   * Returns a new {@link AggregatorHandle}. This MUST by used by the synchronous to aggregate
-   * recorded measurements during the collection cycle.
+   * Returns a new {@link AggregatorHandle}. Used by both synchronous and asynchronous metric
+   * storage to aggregate recorded measurements during the collection cycle.
    *
+   * @param creationEpochNanos the epoch timestamp (nanos) at which the handle is being created,
+   *     stored via {@link AggregatorHandle#getCreationEpochNanos()}. Whether this value is used as
+   *     the start timestamp of reported data points depends on the instrument and temporality — see
+   *     {@link AggregatorHandle#getCreationEpochNanos()} for details.
    * @return a new {@link AggregatorHandle}.
    */
-  AggregatorHandle<T> createHandle();
+  AggregatorHandle<T> createHandle(long creationEpochNanos);
 
   /**
    * Returns a new DELTA point by computing the difference between two cumulative points.

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/aggregator/AggregatorHandle.java

@@ -35,13 +35,16 @@ public abstract class AggregatorHandle<T extends PointData> {
   private static final String UNSUPPORTED_DOUBLE_MESSAGE =
       "This aggregator does not support double values.";
 
+  private final long creationEpochNanos;
   // A reservoir of sampled exemplars for this time period.
   @Nullable private final DoubleExemplarReservoir doubleReservoirFactory;
   @Nullable private final LongExemplarReservoir longReservoirFactory;
   private final boolean isDoubleType;
   private volatile boolean valuesRecorded = false;
 
-  protected AggregatorHandle(ExemplarReservoirFactory reservoirFactory, boolean isDoubleType) {
+  protected AggregatorHandle(
+      long creationEpochNanos, ExemplarReservoirFactory reservoirFactory, boolean isDoubleType) {
+    this.creationEpochNanos = creationEpochNanos;
     this.isDoubleType = isDoubleType;
     if (isDoubleType) {
       this.doubleReservoirFactory = reservoirFactory.createDoubleExemplarReservoir();
@@ -145,4 +148,22 @@ private static <S> S throwUnsupportedIfNull(@Nullable S value, String message) {
     }
     return value;
   }
+
+  /**
+   * Returns the epoch timestamp (nanos) at which this handle was created.
+   *
+   * <p>For cumulative synchronous instruments, this is the time of the first measurement for the
+   * series and is used as {@link PointData#getStartEpochNanos()}.
+   *
+   * <p>For cumulative asynchronous instruments, this is either the instrument creation time (if the
+   * series first appeared during the first collection cycle) or the preceding collection interval's
+   * timestamp (if the series appeared in a later cycle), and is used as {@link
+   * PointData#getStartEpochNanos()}.
+   *
+   * <p>Not used for delta instruments; their start epoch is computed directly from the reader's
+   * last collection time or instrument creation time.
+   */
+  public long getCreationEpochNanos() {
+    return creationEpochNanos;
+  }
 }

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/aggregator/DoubleBase2ExponentialHistogramAggregator.java

@@ -66,8 +66,9 @@ public DoubleBase2ExponentialHistogramAggregator(
   }
 
   @Override
-  public AggregatorHandle<ExponentialHistogramPointData> createHandle() {
-    return new Handle(reservoirFactory, maxBuckets, maxScale, recordMinMax, memoryMode);
+  public AggregatorHandle<ExponentialHistogramPointData> createHandle(long creationEpochNanos) {
+    return new Handle(
+        creationEpochNanos, reservoirFactory, maxBuckets, maxScale, recordMinMax, memoryMode);
   }
 
   @Override
@@ -104,12 +105,13 @@ static final class Handle extends AggregatorHandle<ExponentialHistogramPointData
     @Nullable private final MutableExponentialHistogramPointData reusablePoint;
 
     Handle(
+        long creationEpochNanos,
         ExemplarReservoirFactory reservoirFactory,
         int maxBuckets,
         int maxScale,
         boolean recordMinMax,
         MemoryMode memoryMode) {
-      super(reservoirFactory, /* isDoubleType= */ true);
+      super(creationEpochNanos, reservoirFactory, /* isDoubleType= */ true);
       this.maxBuckets = maxBuckets;
       this.maxScale = maxScale;
       this.recordMinMax = recordMinMax;

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/aggregator/DoubleExplicitBucketHistogramAggregator.java

@@ -72,8 +72,9 @@ public DoubleExplicitBucketHistogramAggregator(
   }
 
   @Override
-  public AggregatorHandle<HistogramPointData> createHandle() {
-    return new Handle(boundaryList, boundaries, recordMinMax, reservoirFactory, memoryMode);
+  public AggregatorHandle<HistogramPointData> createHandle(long creationEpochNanos) {
+    return new Handle(
+        creationEpochNanos, boundaryList, boundaries, recordMinMax, reservoirFactory, memoryMode);
   }
 
   @Override
@@ -120,12 +121,13 @@ static final class Handle extends AggregatorHandle<HistogramPointData> {
     @Nullable private final MutableHistogramPointData reusablePoint;
 
     Handle(
+        long creationEpochNanos,
         List<Double> boundaryList,
         double[] boundaries,
         boolean recordMinMax,
         ExemplarReservoirFactory reservoirFactory,
         MemoryMode memoryMode) {
-      super(reservoirFactory, /* isDoubleType= */ true);
+      super(creationEpochNanos, reservoirFactory, /* isDoubleType= */ true);
       this.boundaryList = boundaryList;
       this.boundaries = boundaries;
       this.recordMinMax = recordMinMax;