Fix delta intervals

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

@@ -55,16 +55,19 @@ 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 ->
                 new SdkMeter(sharedState, instrumentationLibraryInfo, registeredReaders));
     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<MetricData> collectAllMetrics() {
       Collection<SdkMeter> meters = registry.getComponents();
       List<MetricData> 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);
     }
   }

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.
+   *
+   * <p>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.
+   *
+   * <p>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;

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorage.java

@@ -41,7 +41,6 @@ final class AsynchronousMetricStorage<T, U extends ExemplarData> implements Metr
   private final ThrottlingLogger throttlingLogger = new ThrottlingLogger(logger);
   private final RegisteredReader registeredReader;
   private final MetricDescriptor metricDescriptor;
-  private final AggregationTemporality aggregationTemporality;
   private final TemporalMetricStorage<T, U> metricStorage;
   private final Aggregator<T, U> 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<Attributes, T> currentAccumulations = accumulations;
     accumulations = new HashMap<>();
     return metricStorage.buildMetricFor(
-        registeredReader,
-        resource,
-        instrumentationScopeInfo,
-        getMetricDescriptor(),
-        aggregationTemporality,
-        currentAccumulations,
-        startEpochNanos,
-        epochNanos);
+        resource, instrumentationScopeInfo, currentAccumulations, startEpochNanos, epochNanos);
   }
 
   @Override

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/DefaultSynchronousMetricStorage.java

@@ -42,7 +42,6 @@ public final class DefaultSynchronousMetricStorage<T, U extends ExemplarData>
 
   private final RegisteredReader registeredReader;
   private final MetricDescriptor metricDescriptor;
-  private final AggregationTemporality aggregationTemporality;
   private final Aggregator<T, U> aggregator;
   private final ConcurrentHashMap<Attributes, AggregatorHandle<T, U>> activeCollectionStorage =
       new ConcurrentHashMap<>();
@@ -56,12 +55,18 @@ public final class DefaultSynchronousMetricStorage<T, U extends ExemplarData>
       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

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();

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/internal/state/TemporalMetricStorage.java

@@ -24,115 +24,90 @@
 class TemporalMetricStorage<T, U extends ExemplarData> {
   private final Aggregator<T, U> aggregator;
   private final boolean isSynchronous;
-  private final Map<RegisteredReader, LastReportedAccumulation<T>> reportHistory = new HashMap<>();
+  private final RegisteredReader registeredReader;
+  private Map<Attributes, T> lastAccumulation = new HashMap<>();
+  private final AggregationTemporality temporality;
+  private final MetricDescriptor metricDescriptor;
 
-  TemporalMetricStorage(Aggregator<T, U> aggregator, boolean isSynchronous) {
+  TemporalMetricStorage(
+      Aggregator<T, U> 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<Attributes, T> currentAccumulation,
       long startEpochNanos,
       long epochNanos) {
-    // In case it's our first collection, default to start timestamp.
-    long lastCollectionEpoch = startEpochNanos;
+
     Map<Attributes, T> result = currentAccumulation;
-    // Check our last report time.
-    if (reportHistory.containsKey(registeredReader)) {
-      LastReportedAccumulation<T> 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.
     // This assumes aggregation remains consistent for the lifetime of a collector, and
     // 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();
     }
     return aggregator.toMetricData(
         resource,
         instrumentationScopeInfo,
-        descriptor,
+        metricDescriptor,
         result,
         temporality,
         startEpochNanos,
         lastCollectionEpoch,
         epochNanos);
   }
-
-  /** Remembers what was presented to a specific exporter. */
-  private static class LastReportedAccumulation<T> {
-    private final Map<Attributes, T> 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<Attributes, T> accumulation, long epochNanos) {
-      this.accumulation = accumulation;
-      this.epochNanos = epochNanos;
-    }
-
-    long getEpochNanos() {
-      return epochNanos;
-    }
-
-    Map<Attributes, T> getAccumulation() {
-      return accumulation;
-    }
-  }
 }

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);
+  }
 }