Expose aggregation configuration and wire in exemplar sampler

- Expose classes for aggregation configuration that are visible in exemplar sampler
- Create default exemplar sampler that samples w/ traces
- Expand metric benchmarks to include exemplar + no-exemplar to check overhead.

Author: jsuereth

sdk/metrics/build.gradle.kts

@@ -23,4 +23,6 @@ dependencies {
   testImplementation(project(":sdk:metrics-testing"))
   testImplementation(project(":sdk:testing"))
   testImplementation("com.google.guava:guava")
+
+  jmh(project(":sdk:trace"))
 }

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/MetricsBenchmarks.java

@@ -7,6 +7,8 @@
 
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -18,6 +20,7 @@
 import org.openjdk.jmh.annotations.Scope;
 import org.openjdk.jmh.annotations.Setup;
 import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
 import org.openjdk.jmh.annotations.Threads;
 import org.openjdk.jmh.annotations.Warmup;
 import org.openjdk.jmh.infra.ThreadParams;
@@ -37,16 +40,29 @@ public static class ThreadState {
     @Param MetricsTestOperationBuilder opBuilder;
 
     MetricsTestOperationBuilder.Operation op;
+    Span span;
+    io.opentelemetry.context.Scope contextScope;
     final Attributes sharedLabelSet = Attributes.builder().put("KEY", "VALUE").build();
     Attributes threadUniqueLabelSet;
 
     @Setup
+    @SuppressWarnings("MustBeClosedChecker")
     public void setup(ThreadParams threadParams) {
       Meter meter = sdk.getMeter();
+      Tracer tracer = sdk.getTracer();
+      span = tracer.spanBuilder("benchmark").startSpan();
+      // We suppress warnings on closing here, as we rely on tests to make sure context is closed.
+      contextScope = span.makeCurrent();
       op = opBuilder.build(meter);
       threadUniqueLabelSet =
           Attributes.builder().put("KEY", String.valueOf(threadParams.getThreadIndex())).build();
     }
+
+    @TearDown
+    public void tearDown(ThreadParams threadParms) {
+      contextScope.close();
+      span.end();
+    }
   }
 
   @Benchmark

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/MetricsTestOperationBuilder.java

@@ -15,6 +15,7 @@
 import io.opentelemetry.api.metrics.LongCounter;
 import io.opentelemetry.api.metrics.LongHistogram;
 import io.opentelemetry.api.metrics.Meter;
+import java.util.concurrent.ThreadLocalRandom;
 
 /**
  * This enum allows for iteration over all of the operations that we want to benchmark. To ensure
@@ -80,12 +81,13 @@ public void performBound() {
 
           @Override
           public void perform(Attributes labels) {
-            metric.record(5.0d, labels);
+            // We record different values to try to hit more areas of the histogram buckets.
+            metric.record(ThreadLocalRandom.current().nextDouble(0, 20_000d), labels);
           }
 
           @Override
           public void performBound() {
-            boundMetric.record(5.0d);
+            boundMetric.record(ThreadLocalRandom.current().nextDouble(0, 20_000d));
           }
         };
       }),
@@ -103,12 +105,12 @@ public void performBound() {
 
           @Override
           public void perform(Attributes labels) {
-            metric.record(5L, labels);
+            metric.record(ThreadLocalRandom.current().nextLong(0, 20_000L), labels);
           }
 
           @Override
           public void performBound() {
-            boundMetric.record(5L);
+            boundMetric.record(ThreadLocalRandom.current().nextLong(0, 20_000L));
           }
         };
       });

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/TestSdk.java

@@ -7,8 +7,13 @@
 
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.sdk.common.Clock;
+import io.opentelemetry.sdk.metrics.exemplar.ExemplarFilter;
+import io.opentelemetry.sdk.metrics.exemplar.ExemplarSampler;
 import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.samplers.Sampler;
 
 @SuppressWarnings("ImmutableEnumChecker")
 public enum TestSdk {
@@ -19,6 +24,19 @@ Meter build() {
           return MeterProvider.noop().get("io.opentelemetry.sdk.metrics");
         }
       }),
+  SDK_NO_EXEMPLARS(
+      new SdkBuilder() {
+        @Override
+        Meter build() {
+          return SdkMeterProvider.builder()
+              .setClock(Clock.getDefault())
+              .setResource(Resource.empty())
+              .setExemplarSampler(
+                  ExemplarSampler.builder().setFilter(ExemplarFilter.neverSample()).build())
+              .build()
+              .get("io.opentelemetry.sdk.metrics");
+        }
+      }),
   SDK(
       new SdkBuilder() {
         @Override
@@ -41,7 +59,18 @@ public Meter getMeter() {
     return sdkBuilder.build();
   }
 
+  public Tracer getTracer() {
+    return sdkBuilder.buildTracer();
+  }
+
   private abstract static class SdkBuilder {
     abstract Meter build();
+
+    protected Tracer buildTracer() {
+      return SdkTracerProvider.builder()
+          .setSampler(Sampler.alwaysOn())
+          .build()
+          .get("io.opentelemetry.sdk.metrics");
+    }
   }
 }

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/exemplar/DefaultExemplarReservoirFactory.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.metrics.exemplar;
+
+import io.opentelemetry.sdk.common.Clock;
+import io.opentelemetry.sdk.metrics.view.Aggregation;
+import io.opentelemetry.sdk.metrics.view.ExplicitBucketHistogramAggregation;
+
+class DefaultExemplarReservoirFactory implements ExemplarReservoirFactory {
+
+  static final ExemplarReservoirFactory INSTANCE = new DefaultExemplarReservoirFactory();
+
+  private DefaultExemplarReservoirFactory() {}
+
+  @Override
+  public ExemplarReservoir createReservoir(Aggregation aggregation) {
+    Clock clock = Clock.getDefault();
+    // For histograms, line up reservoirs with buckets.
+    if (aggregation instanceof ExplicitBucketHistogramAggregation) {
+      return HistogramBucketExemplarReservoir.create(
+          clock, ((ExplicitBucketHistogramAggregation) aggregation).getBucketBoundaries());
+    }
+    // By default, reduce threading contention.
+    return new FixedSizeExemplarReservoir(clock, Runtime.getRuntime().availableProcessors());
+  }
+}

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/exemplar/ExemplarReservoirFactory.java

@@ -15,4 +15,9 @@ public interface ExemplarReservoirFactory {
    * @param aggregation The aggregation configuration.
    */
   ExemplarReservoir createReservoir(Aggregation aggregation);
+
+  /** Returns the default factory that collects exemplars. */
+  static ExemplarReservoirFactory defaultFactory() {
+    return DefaultExemplarReservoirFactory.INSTANCE;
+  }
 }

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/exemplar/ExemplarSampler.java

@@ -38,7 +38,7 @@ public ExemplarReservoir createReservoir(Aggregation aggregation) {
   /** Returns a builder with default exemplar sampling configuration. */
   public static Builder builder() {
     return new AutoValue_ExemplarSampler.Builder()
-        .setFactory(ignore -> ExemplarReservoir.noSamples())
+        .setFactory(ExemplarReservoirFactory.defaultFactory())
         .setFilter(ExemplarFilter.sampleWithTraces());
   }
 

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

@@ -44,7 +44,7 @@ public static <T> AsynchronousMetricStorage doubleAsynchronousAccumulator(
     final MetricDescriptor metricDescriptor = MetricDescriptor.create(view, instrument);
     Aggregator<T> aggregator =
         view.getAggregation()
-            .config(instrument)
+            .getFactory(instrument)
             .create(
                 resource,
                 instrumentationLibraryInfo,
@@ -84,7 +84,7 @@ public static <T> AsynchronousMetricStorage longAsynchronousAccumulator(
     final MetricDescriptor metricDescriptor = MetricDescriptor.create(view, instrument);
     Aggregator<T> aggregator =
         view.getAggregation()
-            .config(instrument)
+            .getFactory(instrument)
             .create(
                 resource,
                 instrumentationLibraryInfo,

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

@@ -68,7 +68,7 @@ public final WriteableMetricStorage registerSynchronousMetricStorage(
     List<WriteableMetricStorage> storage = new ArrayList<>(views.size());
     for (View view : views) {
       // TODO - move this in a better location.
-      if (view.getAggregation().config(instrument) == null) {
+      if (view.getAggregation().getFactory(instrument) == null) {
         continue;
       }
       try {
@@ -105,7 +105,7 @@ public final void registerLongAsynchronousInstrument(
             .findViews(instrument, getInstrumentationLibraryInfo());
     for (View view : views) {
       // TODO - move this in a better location.
-      if (view.getAggregation().config(instrument) == null) {
+      if (view.getAggregation().getFactory(instrument) == null) {
         continue;
       }
       try {
@@ -136,7 +136,7 @@ public final void registerDoubleAsynchronousInstrument(
             .findViews(instrument, getInstrumentationLibraryInfo());
     for (View view : views) {
       // TODO - move this in a better location.
-      if (view.getAggregation().config(instrument) == null) {
+      if (view.getAggregation().getFactory(instrument) == null) {
         continue;
       }
       try {

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

@@ -15,6 +15,7 @@
 import io.opentelemetry.sdk.metrics.internal.aggregator.AggregatorHandle;
 import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
 import io.opentelemetry.sdk.metrics.internal.view.AttributesProcessor;
+import io.opentelemetry.sdk.metrics.view.Aggregation;
 import io.opentelemetry.sdk.metrics.view.View;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.Map;
@@ -45,15 +46,16 @@ public static <T> SynchronousMetricStorage<T> create(
       long startEpochNanos,
       ExemplarSampler sampler) {
     final MetricDescriptor metricDescriptor = MetricDescriptor.create(view, instrumentDescriptor);
+    final Aggregation resolved = view.getAggregation().resolve(instrumentDescriptor);
     final Aggregator<T> aggregator =
-        view.getAggregation()
-            .config(instrumentDescriptor)
+        resolved
+            .getFactory(instrumentDescriptor)
             .create(
                 resource,
                 instrumentationLibraryInfo,
                 instrumentDescriptor,
                 metricDescriptor,
-                () -> sampler.createReservoir(view.getAggregation()));
+                () -> sampler.createReservoir(resolved));
     return new SynchronousMetricStorage<>(
         metricDescriptor,
         aggregator,