Support temporality selectors in InMemoryMetricExporter

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

@@ -22,3 +22,11 @@ Comparing source compatibility of opentelemetry-sdk-testing-1.68.0-SNAPSHOT.jar
 	+++  NEW METHOD: PUBLIC(+) FINAL(+) io.opentelemetry.sdk.testing.assertj.MetricAssert hasSummaryPointsSatisfying(java.util.function.Consumer[]<io.opentelemetry.sdk.testing.assertj.SummaryPointAssert>)
 		+++  NEW ANNOTATION: java.lang.SafeVarargs
 	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.testing.assertj.MetricAssert hasSummaryPointsSatisfying(java.lang.Iterable<? extends java.util.function.Consumer<? extends io.opentelemetry.sdk.testing.assertj.SummaryPointAssert>>)
+***  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.InMemoryMetricExporterBuilder builder()
++++  NEW CLASS: PUBLIC(+) FINAL(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporterBuilder  (not serializable)
+	+++  CLASS FILE FORMAT VERSION: 52.0 <- n.a.
+	+++  NEW SUPERCLASS: java.lang.Object
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporter build()
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporterBuilder setAggregationTemporalitySelector(io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector)

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

@@ -9,6 +9,7 @@
 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;
@@ -20,7 +21,7 @@
 /**
  * A {@link MetricExporter} implementation that can be used to test OpenTelemetry integration.
  *
- * <p>Can be created using {@code InMemoryMetricExporter.create()}
+ * <p>Can be created using {@link #create()} or configured using {@link #builder()}.
  *
  * <p>Example usage:
  *
@@ -47,24 +48,33 @@
  */
 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;
+  InMemoryMetricExporter(AggregationTemporalitySelector aggregationTemporalitySelector) {
+    this.aggregationTemporalitySelector = aggregationTemporalitySelector;
   }
 
   /**
-   * Returns a new {@link InMemoryMetricExporter} with a aggregation temporality of {@link
-   * AggregationTemporality#CUMULATIVE}.
+   * Creates an {@link InMemoryMetricExporterBuilder} with defaults.
+   *
+   * @return a builder with always-cumulative {@link AggregationTemporalitySelector}
+   */
+  public static InMemoryMetricExporterBuilder builder() {
+    return new InMemoryMetricExporterBuilder();
+  }
+
+  /**
+   * Returns a new {@link InMemoryMetricExporter} with cumulative aggregation temporality for all
+   * instrument types.
    */
   public static InMemoryMetricExporter create() {
-    return create(AggregationTemporality.CUMULATIVE);
+    return builder().build();
   }
 
   /** Returns a new {@link InMemoryMetricExporter} with the given {@code aggregationTemporality}. */
   public static InMemoryMetricExporter create(AggregationTemporality aggregationTemporality) {
-    return new InMemoryMetricExporter(aggregationTemporality);
+    return builder().setAggregationTemporalitySelector(unused -> aggregationTemporality).build();
   }
 
   /**
@@ -87,7 +97,7 @@ public void reset() {
 
   @Override
   public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
-    return aggregationTemporality;
+    return aggregationTemporalitySelector.getAggregationTemporality(instrumentType);
   }
 
   /**

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

@@ -0,0 +1,39 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.testing.exporter;
+
+import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector;
+import java.util.Objects;
+
+/** Builder for {@link InMemoryMetricExporter}. */
+public final class InMemoryMetricExporterBuilder {
+  private AggregationTemporalitySelector aggregationTemporalitySelector =
+      AggregationTemporalitySelector.alwaysCumulative();
+
+  InMemoryMetricExporterBuilder() {}
+
+  /**
+   * Sets the {@link AggregationTemporalitySelector} used by {@link
+   * InMemoryMetricExporter#getAggregationTemporality(InstrumentType)}.
+   *
+   * <p>If unset, defaults to {@link AggregationTemporalitySelector#alwaysCumulative()}.
+   *
+   * @param aggregationTemporalitySelector the {@link AggregationTemporalitySelector} to set
+   * @return this {@link InMemoryMetricExporterBuilder}
+   */
+  public InMemoryMetricExporterBuilder setAggregationTemporalitySelector(
+      AggregationTemporalitySelector aggregationTemporalitySelector) {
+    Objects.requireNonNull(aggregationTemporalitySelector, "aggregationTemporalitySelector");
+    this.aggregationTemporalitySelector = aggregationTemporalitySelector;
+    return this;
+  }
+
+  /** Constructs an {@link InMemoryMetricExporter} based on the builder's values. */
+  public InMemoryMetricExporter build() {
+    return new InMemoryMetricExporter(aggregationTemporalitySelector);
+  }
+}

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

@@ -5,22 +5,33 @@
 
 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.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 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 +55,114 @@ 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.builder().build().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.builder()
+                .setAggregationTemporalitySelector(AggregationTemporalitySelector.deltaPreferred())
+                .build()
+                .getAggregationTemporality(instrumentType))
+        .isEqualTo(
+            instrumentType == InstrumentType.UP_DOWN_COUNTER
+                    || instrumentType == InstrumentType.OBSERVABLE_UP_DOWN_COUNTER
+                ? AggregationTemporality.CUMULATIVE
+                : AggregationTemporality.DELTA);
+  }
+
+  @Test
+  void setAggregationTemporalitySelector_null() {
+    assertThatThrownBy(
+            () -> InMemoryMetricExporter.builder().setAggregationTemporalitySelector(null))
+        .isInstanceOf(NullPointerException.class)
+        .hasMessage("aggregationTemporalitySelector");
+  }
+
+  @Test
+  void builder_reusable() {
+    InMemoryMetricExporterBuilder builder = InMemoryMetricExporter.builder();
+    InMemoryMetricExporter cumulativeExporter = builder.build();
+    InMemoryMetricExporter deltaExporter =
+        builder
+            .setAggregationTemporalitySelector(AggregationTemporalitySelector.deltaPreferred())
+            .build();
+
+    assertThat(cumulativeExporter.getAggregationTemporality(InstrumentType.COUNTER))
+        .isEqualTo(AggregationTemporality.CUMULATIVE);
+    assertThat(deltaExporter.getAggregationTemporality(InstrumentType.COUNTER))
+        .isEqualTo(AggregationTemporality.DELTA);
+  }
+
+  @Test
+  void export_deltaPreferred() {
+    InMemoryMetricExporter deltaExporter =
+        InMemoryMetricExporter.builder()
+            .setAggregationTemporalitySelector(AggregationTemporalitySelector.deltaPreferred())
+            .build();
+    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();
+      LongUpDownCounter upDownCounter = meter.upDownCounterBuilder("upDownCounter").build();
+
+      for (int value : new int[] {1, 2}) {
+        counter.add(value);
+        upDownCounter.add(value == 1 ? 1 : -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("upDownCounter")
+                        .hasLongSumSatisfying(
+                            sum ->
+                                sum.isCumulative()
+                                    .hasPointsSatisfying(
+                                        point -> point.hasValue(value == 1 ? 1 : 0))));
+        deltaExporter.reset();
+      }
+    }
+  }
+
+  @Test
+  void export_preservesMetricData() {
+    InMemoryMetricExporter deltaExporter =
+        InMemoryMetricExporter.builder()
+            .setAggregationTemporalitySelector(AggregationTemporalitySelector.deltaPreferred())
+            .build();
+    MetricData metric = generateFakeMetric();
+
+    assertThat(deltaExporter.export(Collections.singletonList(metric)).isSuccess()).isTrue();
+    assertThat(deltaExporter.getFinishedMetricItems()).singleElement().isSameAs(metric);
   }
 
   @Test