Support temporality selectors in InMemoryMetricExporter

docs/apidiffs/current_vs_latest/opentelemetry-sdk-testing.txt

@@ -1,2 +1,4 @@
 Comparing source compatibility of opentelemetry-sdk-testing-1.68.0-SNAPSHOT.jar against opentelemetry-sdk-testing-1.67.0.jar
-No changes.
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporter  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporter create(io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector)

sdk/testing/src/main/java/io/opentelemetry/sdk/testing/exporter/InMemoryMetricExporter.java

@@ -9,11 +9,13 @@
 import io.opentelemetry.sdk.metrics.InstrumentType;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
@@ -47,11 +49,11 @@
  */
 public final class InMemoryMetricExporter implements MetricExporter {
   private final Queue<MetricData> finishedMetricItems = new ConcurrentLinkedQueue<>();
-  private final AggregationTemporality aggregationTemporality;
+  private final AggregationTemporalitySelector aggregationTemporalitySelector;
   private boolean isStopped = false;
 
-  private InMemoryMetricExporter(AggregationTemporality aggregationTemporality) {
-    this.aggregationTemporality = aggregationTemporality;
+  private InMemoryMetricExporter(AggregationTemporalitySelector aggregationTemporalitySelector) {
+    this.aggregationTemporalitySelector = aggregationTemporalitySelector;
   }
 
   /**
@@ -64,7 +66,24 @@ public static InMemoryMetricExporter create() {
 
   /** Returns a new {@link InMemoryMetricExporter} with the given {@code aggregationTemporality}. */
   public static InMemoryMetricExporter create(AggregationTemporality aggregationTemporality) {
-    return new InMemoryMetricExporter(aggregationTemporality);
+    return create(unused -> aggregationTemporality);
+  }
+
+  /**
+   * Returns a new {@link InMemoryMetricExporter} with the given {@link
+   * AggregationTemporalitySelector}.
+   *
+   * <p>For example, {@code
+   * InMemoryMetricExporter.create(AggregationTemporalitySelector.deltaPreferred())} uses delta
+   * temporality for counters and histograms, and cumulative temporality for up-down counters.
+   *
+   * @param aggregationTemporalitySelector selects the aggregation temporality for each instrument
+   *     type
+   */
+  public static InMemoryMetricExporter create(
+      AggregationTemporalitySelector aggregationTemporalitySelector) {
+    Objects.requireNonNull(aggregationTemporalitySelector, "aggregationTemporalitySelector");
+    return new InMemoryMetricExporter(aggregationTemporalitySelector);
   }
 
   /**
@@ -87,7 +106,7 @@ public void reset() {
 
   @Override
   public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
-    return aggregationTemporality;
+    return aggregationTemporalitySelector.getAggregationTemporality(instrumentType);
   }
 
   /**

sdk/testing/src/test/java/io/opentelemetry/sdk/testing/exporter/InMemoryMetricExporterTest.java

@@ -5,22 +5,35 @@
 
 package io.opentelemetry.sdk.testing.exporter;
 
-import static org.assertj.core.api.Assertions.assertThat;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.DoubleHistogram;
+import io.opentelemetry.api.metrics.LongCounter;
+import io.opentelemetry.api.metrics.LongUpDownCounter;
+import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector;
+import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableLongPointData;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableMetricData;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableSumData;
 import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.sdk.testing.time.TestClock;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
 
 /** Unit tests for {@link InMemoryMetricExporter}. */
 class InMemoryMetricExporterTest {
@@ -44,14 +57,112 @@ private static MetricData generateFakeMetric() {
                     startNs, endNs, Attributes.builder().put("k", "v").build(), 5))));
   }
 
-  @Test
-  void getAggregationTemporality() {
-    assertThat(InMemoryMetricExporter.create().getAggregationTemporality(InstrumentType.COUNTER))
+  @ParameterizedTest
+  @EnumSource(InstrumentType.class)
+  void getAggregationTemporality(InstrumentType instrumentType) {
+    assertThat(InMemoryMetricExporter.create().getAggregationTemporality(instrumentType))
         .isEqualTo(AggregationTemporality.CUMULATIVE);
     assertThat(
             InMemoryMetricExporter.create(AggregationTemporality.DELTA)
-                .getAggregationTemporality(InstrumentType.COUNTER))
+                .getAggregationTemporality(instrumentType))
         .isEqualTo(AggregationTemporality.DELTA);
+    assertThat(
+            InMemoryMetricExporter.create(AggregationTemporality.CUMULATIVE)
+                .getAggregationTemporality(instrumentType))
+        .isEqualTo(AggregationTemporality.CUMULATIVE);
+  }
+
+  @ParameterizedTest
+  @EnumSource(InstrumentType.class)
+  void getAggregationTemporality_deltaPreferred(InstrumentType instrumentType) {
+    assertThat(
+            InMemoryMetricExporter.create(AggregationTemporalitySelector.deltaPreferred())
+                .getAggregationTemporality(instrumentType))
+        .isEqualTo(
+            instrumentType == InstrumentType.UP_DOWN_COUNTER
+                    || instrumentType == InstrumentType.OBSERVABLE_UP_DOWN_COUNTER
+                ? AggregationTemporality.CUMULATIVE
+                : AggregationTemporality.DELTA);
+  }
+
+  @Test
+  void create_nullSelector() {
+    assertThatThrownBy(() -> InMemoryMetricExporter.create((AggregationTemporalitySelector) null))
+        .isInstanceOf(NullPointerException.class)
+        .hasMessage("aggregationTemporalitySelector");
+  }
+
+  @Test
+  void export_deltaPreferred() {
+    InMemoryMetricExporter deltaExporter =
+        InMemoryMetricExporter.create(AggregationTemporalitySelector.deltaPreferred());
+    TestClock clock = TestClock.create();
+    PeriodicMetricReader reader =
+        PeriodicMetricReader.builder(deltaExporter).setInterval(Duration.ofDays(1)).build();
+    try (SdkMeterProvider meterProvider =
+        SdkMeterProvider.builder().setClock(clock).registerMetricReader(reader).build()) {
+      Meter meter = meterProvider.get("test");
+      LongCounter counter = meter.counterBuilder("counter").build();
+      DoubleHistogram histogram = meter.histogramBuilder("histogram").build();
+      LongUpDownCounter upDownCounter = meter.upDownCounterBuilder("upDownCounter").build();
+      AtomicLong observedUpDownCounter = new AtomicLong();
+      meter
+          .upDownCounterBuilder("observableUpDownCounter")
+          .buildWithCallback(measurement -> measurement.record(observedUpDownCounter.get()));
+
+      for (int value : new int[] {1, 2, 3}) {
+        counter.add(value);
+        histogram.record(value);
+        upDownCounter.add(value == 3 ? -1 : 1);
+        observedUpDownCounter.set(value == 3 ? 0 : 1);
+        clock.advance(Duration.ofSeconds(1));
+
+        assertThat(reader.forceFlush().join(10, TimeUnit.SECONDS).isSuccess()).isTrue();
+        assertThat(deltaExporter.getFinishedMetricItems())
+            .filteredOn(metric -> metric.getInstrumentationScopeInfo().getName().equals("test"))
+            .satisfiesExactlyInAnyOrder(
+                metric ->
+                    assertThat(metric)
+                        .hasName("counter")
+                        .hasLongSumSatisfying(
+                            sum ->
+                                sum.isDelta().hasPointsSatisfying(point -> point.hasValue(value))),
+                metric ->
+                    assertThat(metric)
+                        .hasName("histogram")
+                        .hasHistogramSatisfying(
+                            data ->
+                                data.isDelta()
+                                    .hasPointsSatisfying(point -> point.hasCount(1).hasSum(value))),
+                metric ->
+                    assertThat(metric)
+                        .hasName("upDownCounter")
+                        .hasLongSumSatisfying(
+                            sum ->
+                                sum.isCumulative()
+                                    .hasPointsSatisfying(
+                                        point -> point.hasValue(value == 2 ? 2 : 1))),
+                metric ->
+                    assertThat(metric)
+                        .hasName("observableUpDownCounter")
+                        .hasLongSumSatisfying(
+                            sum ->
+                                sum.isCumulative()
+                                    .hasPointsSatisfying(
+                                        point -> point.hasValue(value == 3 ? 0 : 1))));
+        deltaExporter.reset();
+      }
+    }
+  }
+
+  @Test
+  void export_preservesMetricData() {
+    InMemoryMetricExporter deltaExporter =
+        InMemoryMetricExporter.create(AggregationTemporalitySelector.deltaPreferred());
+    MetricData metric = generateFakeMetric();
+
+    assertThat(deltaExporter.export(Collections.singletonList(metric)).isSuccess()).isTrue();
+    assertThat(deltaExporter.getFinishedMetricItems()).singleElement().isSameAs(metric);
   }
 
   @Test