Collect async exemplars when exemplar filter is always_on

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

@@ -318,6 +318,7 @@ SdkObservableMeasurement registerObservableMeasurement(
                     registeredView,
                     meterProviderSharedState.getClock(),
                     instrumentDescriptor,
+                    meterProviderSharedState.getExemplarFilter(),
                     meterEnabled)));
       }
     }

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

@@ -15,7 +15,6 @@
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.common.export.MemoryMode;
 import io.opentelemetry.sdk.common.internal.ThrottlingLogger;
-import io.opentelemetry.sdk.metrics.ExemplarFilter;
 import io.opentelemetry.sdk.metrics.View;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
@@ -106,6 +105,7 @@ public static <T extends PointData> AsynchronousMetricStorage<T> create(
       RegisteredView registeredView,
       Clock clock,
       InstrumentDescriptor instrumentDescriptor,
+      ExemplarFilterInternal exemplarFilter,
       boolean enabled) {
     View view = registeredView.getView();
     MetricDescriptor metricDescriptor =
@@ -117,9 +117,7 @@ public static <T extends PointData> AsynchronousMetricStorage<T> create(
     Aggregator<T> aggregator =
         ((AggregatorFactory) view.getAggregation())
             .createAggregator(
-                instrumentDescriptor,
-                ExemplarFilterInternal.asExemplarFilterInternal(ExemplarFilter.alwaysOff()),
-                registeredReader.getReader().getMemoryMode());
+                instrumentDescriptor, exemplarFilter, registeredReader.getReader().getMemoryMode());
     AttributesProcessor attributesProcessor = registeredView.getViewAttributesProcessor();
     int cardinalityLimit = registeredView.getCardinalityLimit();
     return aggregationTemporality == AggregationTemporality.DELTA

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

@@ -37,6 +37,6 @@ public static MeterProviderSharedState create(
   /** Returns the {@link Resource} to attach telemetry to. */
   public abstract Resource getResource();
 
-  /** Returns the {@link ExemplarFilterInternal} for remembering synchronous measurements. */
+  /** Returns the {@link ExemplarFilterInternal} for remembering measurements. */
   public abstract ExemplarFilterInternal getExemplarFilter();
 }

sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/SdkMeterProviderTest.java

@@ -945,6 +945,37 @@ void collectAll_DropAggregator() {
             metric -> assertThat(metric).hasResource(RESOURCE).hasName("async-counter"));
   }
 
+  @Test
+  void collectAll_AsyncInstrumentAlwaysOnExemplars() {
+    InMemoryMetricReader collector = InMemoryMetricReader.create();
+    Meter meter =
+        sdkMeterProviderBuilder
+            .setExemplarFilter(ExemplarFilter.alwaysOn())
+            .registerMetricReader(collector)
+            .build()
+            .get("my-meter");
+
+    meter
+        .counterBuilder("async-counter")
+        .buildWithCallback(measurement -> measurement.record(1, Attributes.empty()));
+
+    assertThat(collector.collectAllMetrics())
+        .satisfiesExactly(
+            metric ->
+                assertThat(metric)
+                    .hasResource(RESOURCE)
+                    .hasName("async-counter")
+                    .hasLongSumSatisfying(
+                        sum ->
+                            sum.hasPointsSatisfying(
+                                point ->
+                                    point
+                                        .hasValue(1)
+                                        .hasAttributes(Attributes.empty())
+                                        .hasExemplarsSatisfying(
+                                            exemplar -> exemplar.hasValue(1)))));
+  }
+
   @Test
   void shutdown() {
     when(metricReader.getDefaultAggregation(any())).thenCallRealMethod();

sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/internal/state/AsynchronousMetricStorageTest.java

@@ -6,6 +6,7 @@
 package io.opentelemetry.sdk.metrics.internal.state;
 
 import static io.opentelemetry.sdk.common.export.MemoryMode.REUSABLE_DATA;
+import static io.opentelemetry.sdk.metrics.internal.exemplar.ExemplarFilterInternal.asExemplarFilterInternal;
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat;
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.attributeEntry;
 import static org.mockito.ArgumentMatchers.any;
@@ -16,6 +17,7 @@
 import io.opentelemetry.internal.testing.slf4j.SuppressLogger;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.common.export.MemoryMode;
+import io.opentelemetry.sdk.metrics.ExemplarFilter;
 import io.opentelemetry.sdk.metrics.InstrumentSelector;
 import io.opentelemetry.sdk.metrics.InstrumentType;
 import io.opentelemetry.sdk.metrics.InstrumentValueType;
@@ -76,10 +78,15 @@ class AsynchronousMetricStorageTest {
 
   // Not using @BeforeEach since many methods require executing them for each MemoryMode
   void setup(MemoryMode memoryMode) {
-    setup(memoryMode, AggregationTemporality.CUMULATIVE);
+    setup(memoryMode, AggregationTemporality.CUMULATIVE, ExemplarFilter.alwaysOff());
   }
 
   void setup(MemoryMode memoryMode, AggregationTemporality temporality) {
+    setup(memoryMode, temporality, ExemplarFilter.alwaysOff());
+  }
+
+  void setup(
+      MemoryMode memoryMode, AggregationTemporality temporality, ExemplarFilter exemplarFilter) {
     when(reader.getAggregationTemporality(any())).thenReturn(temporality);
     when(reader.getMemoryMode()).thenReturn(memoryMode);
     registeredReader = RegisteredReader.create(reader, ViewRegistry.create());
@@ -96,6 +103,7 @@ void setup(MemoryMode memoryMode, AggregationTemporality temporality) {
                 InstrumentType.COUNTER,
                 InstrumentValueType.LONG,
                 Advice.empty()),
+            asExemplarFilterInternal(exemplarFilter),
             /* enabled= */ true);
     doubleCounterStorage =
         AsynchronousMetricStorage.create(
@@ -109,6 +117,7 @@ void setup(MemoryMode memoryMode, AggregationTemporality temporality) {
                 InstrumentType.COUNTER,
                 InstrumentValueType.DOUBLE,
                 Advice.empty()),
+            asExemplarFilterInternal(exemplarFilter),
             /* enabled= */ true);
   }
 
@@ -186,6 +195,7 @@ void record_ProcessesAttributes(MemoryMode memoryMode) {
                 InstrumentType.COUNTER,
                 InstrumentValueType.LONG,
                 Advice.empty()),
+            asExemplarFilterInternal(ExemplarFilter.alwaysOff()),
             /* enabled= */ true);
 
     storage.record(Attributes.builder().put("key1", "a").put("key2", "b").build(), 1);
@@ -421,6 +431,7 @@ void collect_DeltaComputesDiff(MemoryMode memoryMode) {
                 InstrumentType.COUNTER,
                 InstrumentValueType.LONG,
                 Advice.empty()),
+            asExemplarFilterInternal(ExemplarFilter.alwaysOff()),
             /* enabled= */ true);
 
     // Record measurement and collect at time 10
@@ -482,6 +493,24 @@ void collect_DeltaComputesDiff(MemoryMode memoryMode) {
                                 .hasAttributes(Attributes.builder().put("key", "value2").build())));
   }
 
+  @ParameterizedTest
+  @EnumSource(MemoryMode.class)
+  void recordLong_AlwaysOnCollectsExemplars(MemoryMode memoryMode) {
+    setup(memoryMode, AggregationTemporality.CUMULATIVE, ExemplarFilter.alwaysOn());
+
+    longCounterStorage.record(Attributes.empty(), 3);
+
+    assertThat(longCounterStorage.collect(resource, scope, testClock.now()))
+        .hasLongSumSatisfying(
+            sum ->
+                sum.hasPointsSatisfying(
+                    point ->
+                        point
+                            .hasValue(3)
+                            .hasAttributes(Attributes.empty())
+                            .hasExemplarsSatisfying(exemplar -> exemplar.hasValue(3))));
+  }
+
   @Test
   void collect_reusableData_reusedObjectsAreReturnedOnSecondCall() {
     setup(REUSABLE_DATA);