Finish wiring exemplars into metrics SDK (#3599)

* Initial implementation of simple ExemplarReservoirs

- Add an implementation of naive fixed-bucket reservoir sampling
- Add an implementation of FixedSizeHistogram reservoir sampling where latest-measurement per-histogram-bucket is kept.
- Add assertions for exemplars.

* Remove explicit locks for mild performance boost.

* Create a new random holder abstraction.

- Create new random holder abstraction that allows
  - ThreadLocalRandom when efficient
  - simple detection of Android + workaround for TLR
  - Overridable with mocked random for testing.
- Update tracing RandomIdGenerator to leverage new RnadomHolder
- Expand exemplar reservoir tests to use new random holder
- Remove android-only random id generator test

* Add api diff + fix silly naming thing.

* Move random hodler to be Supplier, other updates from code review.

* Fixes from review.

* Fix bytebuddy issue with mocking.

* 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.

* Fix merge conflict.

* Wire exemplar filter config into autoconfigure module.

* 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.

* Fix merge conflict.

* Wire exemplar filter config into autoconfigure module.

* Migrate exemplar reservoir instantiation into aggregations.  Start migrating off AggregatorFactory -> Aggregation.

* Small cleanup.

* Fixes from review (and report of clsasloader fixes from other laptop.

* Migrate to null-object pattern.

* Add internal boilerplate.

* Run spotless.

* Fix toString of aggregation values.

* Fix ==

Author: jsuereth

sdk-extensions/autoconfigure/README.md

@@ -191,6 +191,12 @@ These properties can be used to control the maximum size of recordings per span.
 | otel.span.event.count.limit            | OTEL_SPAN_EVENT_COUNT_LIMIT            | The maximum number of events per span. Default is `128`.               |
 | otel.span.link.count.limit             | OTEL_SPAN_LINK_COUNT_LIMIT             | The maximum number of links per span. Default is `128`                 |
 
+## Exemplars
+
+| System property          | Environment variable     | Description                                                                       |
+|--------------------------|--------------------------|-----------------------------------------------------------------------------------|
+| otel.metrics.exemplar.filter | OTEL_METRICS_EXEMPLAR_FILTER | The filter for exemplar sampling.  Can be `NONE`, `ALL` or `WITH_SAMPLED_TRACE`. Default is `WITH_SAMPLED_TRACE`.|
+
 ## Interval metric reader
 
 | System property          | Environment variable     | Description                                                                       |

sdk-extensions/autoconfigure/src/main/java/io/opentelemetry/sdk/autoconfigure/OpenTelemetrySdkAutoConfiguration.java

@@ -12,6 +12,7 @@
 import io.opentelemetry.sdk.autoconfigure.spi.metrics.SdkMeterProviderConfigurer;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder;
+import io.opentelemetry.sdk.metrics.exemplar.ExemplarFilter;
 import io.opentelemetry.sdk.resources.Resource;
 import io.opentelemetry.sdk.trace.SdkTracerProvider;
 import java.util.ServiceLoader;
@@ -90,6 +91,24 @@ private static void configureMeterProvider(Resource resource, ConfigProperties c
 
     SdkMeterProviderBuilder meterProviderBuilder = SdkMeterProvider.builder().setResource(resource);
 
+    // Configure default exemplar filters.
+    String exemplarFilter = config.getString("otel.metrics.exemplar.filter");
+    if (exemplarFilter == null) {
+      exemplarFilter = "WITH_SAMPLED_TRACE";
+    }
+    switch (exemplarFilter) {
+      case "NONE":
+        meterProviderBuilder.setExemplarFilter(ExemplarFilter.neverSample());
+        break;
+      case "ALL":
+        meterProviderBuilder.setExemplarFilter(ExemplarFilter.alwaysSample());
+        break;
+      case "WITH_SAMPLED_TRACE":
+      default:
+        meterProviderBuilder.setExemplarFilter(ExemplarFilter.sampleWithTraces());
+        break;
+    }
+
     for (SdkMeterProviderConfigurer configurer :
         ServiceLoader.load(SdkMeterProviderConfigurer.class)) {
       configurer.configure(meterProviderBuilder, config);

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,12 @@
 
 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.resources.Resource;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.samplers.Sampler;
 
 @SuppressWarnings("ImmutableEnumChecker")
 public enum TestSdk {
@@ -19,6 +23,18 @@ 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())
+              .setExemplarFilter(ExemplarFilter.neverSample())
+              .build()
+              .get("io.opentelemetry.sdk.metrics");
+        }
+      }),
   SDK(
       new SdkBuilder() {
         @Override
@@ -41,7 +57,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/SdkMeterProvider.java

@@ -11,7 +11,7 @@
 import io.opentelemetry.sdk.common.Clock;
 import io.opentelemetry.sdk.internal.ComponentRegistry;
 import io.opentelemetry.sdk.metrics.data.MetricData;
-import io.opentelemetry.sdk.metrics.exemplar.ExemplarSampler;
+import io.opentelemetry.sdk.metrics.exemplar.ExemplarFilter;
 import io.opentelemetry.sdk.metrics.export.MetricProducer;
 import io.opentelemetry.sdk.metrics.internal.state.MeterProviderSharedState;
 import io.opentelemetry.sdk.metrics.internal.view.ViewRegistry;
@@ -42,9 +42,9 @@ public final class SdkMeterProvider implements MeterProvider, MetricProducer {
   private final MeterProviderSharedState sharedState;
 
   SdkMeterProvider(
-      Clock clock, Resource resource, ViewRegistry viewRegistry, ExemplarSampler exemplarSampler) {
+      Clock clock, Resource resource, ViewRegistry viewRegistry, ExemplarFilter exemplarFilter) {
     this.sharedState =
-        MeterProviderSharedState.create(clock, resource, viewRegistry, exemplarSampler);
+        MeterProviderSharedState.create(clock, resource, viewRegistry, exemplarFilter);
     this.registry =
         new ComponentRegistry<>(
             instrumentationLibraryInfo -> new SdkMeter(sharedState, instrumentationLibraryInfo));

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

@@ -7,7 +7,7 @@
 
 import io.opentelemetry.api.metrics.GlobalMeterProvider;
 import io.opentelemetry.sdk.common.Clock;
-import io.opentelemetry.sdk.metrics.exemplar.ExemplarSampler;
+import io.opentelemetry.sdk.metrics.exemplar.ExemplarFilter;
 import io.opentelemetry.sdk.metrics.internal.view.ViewRegistry;
 import io.opentelemetry.sdk.metrics.internal.view.ViewRegistryBuilder;
 import io.opentelemetry.sdk.metrics.view.InstrumentSelector;
@@ -25,7 +25,7 @@ public final class SdkMeterProviderBuilder {
   private Resource resource = Resource.getDefault();
   private final ViewRegistryBuilder viewRegistryBuilder = ViewRegistry.builder();
   // Default the sampling strategy.
-  private ExemplarSampler exemplarSampler = ExemplarSampler.builder().build();
+  private ExemplarFilter exemplarFilter = ExemplarFilter.sampleWithTraces();
 
   SdkMeterProviderBuilder() {}
 
@@ -54,12 +54,12 @@ public SdkMeterProviderBuilder setResource(Resource resource) {
   }
 
   /**
-   * Assign an {@link ExemplarSampler} for all metrics created by Meters.
+   * Assign an {@link ExemplarFilter} for all metrics created by Meters.
    *
    * @return this
    */
-  public SdkMeterProviderBuilder setExemplarSampler(ExemplarSampler sampler) {
-    this.exemplarSampler = sampler;
+  public SdkMeterProviderBuilder setExemplarFilter(ExemplarFilter filter) {
+    this.exemplarFilter = filter;
     return this;
   }
 
@@ -118,6 +118,6 @@ public SdkMeterProvider buildAndRegisterGlobal() {
    * @see GlobalMeterProvider
    */
   public SdkMeterProvider build() {
-    return new SdkMeterProvider(clock, resource, viewRegistryBuilder.build(), exemplarSampler);
+    return new SdkMeterProvider(clock, resource, viewRegistryBuilder.build(), exemplarFilter);
   }
 }

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

@@ -27,6 +27,10 @@ public static ExemplarReservoir noSamples() {
 
   /** Wraps a {@link ExemplarReservoir} with a measurement pre-filter. */
   public static ExemplarReservoir filtered(ExemplarFilter filter, ExemplarReservoir original) {
+    // Optimisation on memory usage.
+    if (filter == ExemplarFilter.neverSample()) {
+      return ExemplarReservoir.noSamples();
+    }
     return new FilteredExemplarReservoir(filter, original);
   }
 

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

@@ -39,8 +39,7 @@ public FixedSizeExemplarReservoir(Clock clock, int size, Supplier<Random> random
 
   @Override
   protected int reservoirIndexFor(double value, Attributes attributes, Context context) {
-    // Purposefuly truncate here.
-    int count = (int) numMeasurements.sum() + 1;
+    int count = numMeasurements.intValue() + 1;
     int index = this.randomSupplier.get().nextInt(count > 0 ? count : 1);
     numMeasurements.increment();
     if (index < maxSize()) {

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

@@ -25,6 +25,12 @@
  */
 @Immutable
 public interface Aggregator<T> {
+  /**
+   * Returns the empty aggregator, an aggregator that never records measurements or reports values.
+   */
+  static Aggregator<Void> empty() {
+    return EmptyAggregator.INSTANCE;
+  }
   /**
    * Returns a new {@link AggregatorHandle}. This MUST by used by the synchronous to aggregate
    * recorded measurements during the collection cycle.