initial impl complete

Author: jack-berg

sdk/all/src/jmh/java/io/opentelemetry/sdk/MetricRecordBenchmark.java

@@ -12,6 +12,22 @@
 
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.incubator.metrics.BoundDoubleCounter;
+import io.opentelemetry.api.incubator.metrics.BoundDoubleGauge;
+import io.opentelemetry.api.incubator.metrics.BoundDoubleHistogram;
+import io.opentelemetry.api.incubator.metrics.BoundDoubleUpDownCounter;
+import io.opentelemetry.api.incubator.metrics.BoundLongCounter;
+import io.opentelemetry.api.incubator.metrics.BoundLongGauge;
+import io.opentelemetry.api.incubator.metrics.BoundLongHistogram;
+import io.opentelemetry.api.incubator.metrics.BoundLongUpDownCounter;
+import io.opentelemetry.api.incubator.metrics.ExtendedDoubleCounter;
+import io.opentelemetry.api.incubator.metrics.ExtendedDoubleGauge;
+import io.opentelemetry.api.incubator.metrics.ExtendedDoubleHistogram;
+import io.opentelemetry.api.incubator.metrics.ExtendedDoubleUpDownCounter;
+import io.opentelemetry.api.incubator.metrics.ExtendedLongCounter;
+import io.opentelemetry.api.incubator.metrics.ExtendedLongGauge;
+import io.opentelemetry.api.incubator.metrics.ExtendedLongHistogram;
+import io.opentelemetry.api.incubator.metrics.ExtendedLongUpDownCounter;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.Tracer;
@@ -52,6 +68,8 @@
  *       {@link Aggregation}, including all relevant combinations for synchronous instruments.
  *   <li>{@link BenchmarkState#aggregationTemporality}
  *   <li>{@link BenchmarkState#cardinality}
+ *   <li>{@link BenchmarkState#bound} whether recording goes through bound instruments ({@code
+ *       Extended*#bind(Attributes)}) or unbound instruments.
  *   <li>thread count
  *   <li>{@link BenchmarkState#instrumentValueType}, {@link BenchmarkState#memoryMode}, and {@link
  *       BenchmarkState#exemplars} are disabled to reduce combinatorial explosion.
@@ -62,16 +80,16 @@
  *
  * <p>The cardinality and thread count dimensions partially overlap. Cardinality dictates how many
  * unique attribute sets (i.e. series) are recorded to, and thread count dictates how many threads
- * are simultaneously recording to those series. In all cases, the record path needs to look up an
- * aggregation handle for the series corresponding to the measurement's {@link Attributes} in a
- * {@link java.util.concurrent.ConcurrentHashMap}. That will be the case until otel adds support for
- * <a href="https://github.com/open-telemetry/opentelemetry-specification/issues/4126">bound
- * instruments</a>. The cardinality dictates the size of this map, which has some impact on
- * performance. However, by far the dominant bottleneck is contention. That is, the number of
- * threads simultaneously trying to record to the same series. Increasing the threads increases
- * contention. Increasing cardinality decreases contention, as the threads are now spreading their
- * record activities over more distinct series. The highest contention scenario is cardinality=1,
- * threads=4. Any scenario with threads=1 has zero contention.
+ * are simultaneously recording to those series. For unbound instruments, the record path looks up
+ * an aggregation handle for the series corresponding to the measurement's {@link Attributes} in a
+ * {@link java.util.concurrent.ConcurrentHashMap}; for bound instruments ({@link
+ * BenchmarkState#bound}) that handle is resolved once at bind time, so the record path skips the
+ * lookup and attribute processing entirely. The cardinality dictates the size of this map, which
+ * has some impact on performance. However, by far the dominant bottleneck is contention. That is,
+ * the number of threads simultaneously trying to record to the same series. Increasing the threads
+ * increases contention. Increasing cardinality decreases contention, as the threads are now
+ * spreading their record activities over more distinct series. The highest contention scenario is
+ * cardinality=1, threads=4. Any scenario with threads=1 has zero contention.
  *
  * <p>It's useful to characterize the performance of the metrics system under contention, as some
  * high-performance applications may have many threads trying to record to the same series. It's
@@ -99,6 +117,12 @@ public static class BenchmarkState {
     @Param({"1", "128"})
     int cardinality;
 
+    // Whether to record through bound instruments (Extended*#bind(Attributes)), which resolve the
+    // timeseries once up front, or unbound instruments, which look up the timeseries by Attributes
+    // on every record.
+    @Param({"false", "true"})
+    boolean bound;
+
     // The following parameters are excluded from the benchmark to reduce combinatorial explosion
     // but can optionally be enabled for adhoc evaluation.
 
@@ -120,7 +144,10 @@ public static class BenchmarkState {
     boolean exemplars = false;
 
     OpenTelemetrySdk openTelemetry;
+    // Populated when bound == false.
     private Instrument instrument;
+    // Populated when bound == true; parallel to attributesList (one bound instrument per series).
+    private List<BoundInstrument> boundInstruments;
     List<Long> measurements;
     List<Attributes> attributesList;
     Span span;
@@ -151,7 +178,6 @@ public void setup() {
               .build();
 
       Meter meter = openTelemetry.getMeter("benchmark");
-      instrument = getInstrument(meter, instrumentType, instrumentValueType);
       Tracer tracer = openTelemetry.getTracer("benchmark");
       span = tracer.spanBuilder("benchmark").startSpan();
       // We suppress warnings on closing here, as we rely on tests to make sure context is closed.
@@ -169,6 +195,13 @@ public void setup() {
       }
       Collections.shuffle(attributesList);
 
+      if (bound) {
+        boundInstruments =
+            bindInstruments(meter, instrumentType, instrumentValueType, attributesList);
+      } else {
+        instrument = getInstrument(meter, instrumentType, instrumentValueType);
+      }
+
       measurements = new ArrayList<>(RECORDS_PER_INVOCATION);
       for (int i = 0; i < RECORDS_PER_INVOCATION; i++) {
         measurements.add((long) random.nextInt(2000));
@@ -207,11 +240,21 @@ public void record_MultipleThreads(BenchmarkState benchmarkState) {
   }
 
   private static void record(BenchmarkState benchmarkState) {
-    for (int i = 0; i < RECORDS_PER_INVOCATION; i++) {
-      Attributes attributes =
-          benchmarkState.attributesList.get(i % benchmarkState.attributesList.size());
-      long value = benchmarkState.measurements.get(i % benchmarkState.measurements.size());
-      benchmarkState.instrument.record(value, attributes);
+    if (benchmarkState.bound) {
+      // Bound: record straight to the pre-resolved bound instrument for the series, indexed in
+      // lockstep with attributesList — no per-record Attributes lookup.
+      List<BoundInstrument> boundInstruments = benchmarkState.boundInstruments;
+      for (int i = 0; i < RECORDS_PER_INVOCATION; i++) {
+        long value = benchmarkState.measurements.get(i % benchmarkState.measurements.size());
+        boundInstruments.get(i % boundInstruments.size()).record(value);
+      }
+    } else {
+      for (int i = 0; i < RECORDS_PER_INVOCATION; i++) {
+        Attributes attributes =
+            benchmarkState.attributesList.get(i % benchmarkState.attributesList.size());
+        long value = benchmarkState.measurements.get(i % benchmarkState.measurements.size());
+        benchmarkState.instrument.record(value, attributes);
+      }
     }
   }
 
@@ -262,4 +305,95 @@ private static Instrument getInstrument(
     }
     throw new IllegalArgumentException();
   }
+
+  @FunctionalInterface
+  private interface BoundInstrument {
+    void record(long value);
+  }
+
+  /**
+   * Builds the instrument, then binds one {@link BoundInstrument} per series in {@code
+   * attributesList}, returned in the same order so the record loop can index it in lockstep.
+   */
+  private static List<BoundInstrument> bindInstruments(
+      Meter meter,
+      InstrumentType instrumentType,
+      InstrumentValueType instrumentValueType,
+      List<Attributes> attributesList) {
+    boolean isDouble = instrumentValueType == InstrumentValueType.DOUBLE;
+    String name = "instrument";
+    List<BoundInstrument> result = new ArrayList<>(attributesList.size());
+    switch (instrumentType) {
+      case COUNTER:
+        if (isDouble) {
+          ExtendedDoubleCounter instrument =
+              (ExtendedDoubleCounter) meter.counterBuilder(name).ofDoubles().build();
+          for (Attributes attributes : attributesList) {
+            BoundDoubleCounter bound = instrument.bind(attributes);
+            result.add(bound::add);
+          }
+        } else {
+          ExtendedLongCounter instrument = (ExtendedLongCounter) meter.counterBuilder(name).build();
+          for (Attributes attributes : attributesList) {
+            BoundLongCounter bound = instrument.bind(attributes);
+            result.add(bound::add);
+          }
+        }
+        return result;
+      case UP_DOWN_COUNTER:
+        if (isDouble) {
+          ExtendedDoubleUpDownCounter instrument =
+              (ExtendedDoubleUpDownCounter) meter.upDownCounterBuilder(name).ofDoubles().build();
+          for (Attributes attributes : attributesList) {
+            BoundDoubleUpDownCounter bound = instrument.bind(attributes);
+            result.add(bound::add);
+          }
+        } else {
+          ExtendedLongUpDownCounter instrument =
+              (ExtendedLongUpDownCounter) meter.upDownCounterBuilder(name).build();
+          for (Attributes attributes : attributesList) {
+            BoundLongUpDownCounter bound = instrument.bind(attributes);
+            result.add(bound::add);
+          }
+        }
+        return result;
+      case HISTOGRAM:
+        if (isDouble) {
+          ExtendedDoubleHistogram instrument =
+              (ExtendedDoubleHistogram) meter.histogramBuilder(name).build();
+          for (Attributes attributes : attributesList) {
+            BoundDoubleHistogram bound = instrument.bind(attributes);
+            result.add(bound::record);
+          }
+        } else {
+          ExtendedLongHistogram instrument =
+              (ExtendedLongHistogram) meter.histogramBuilder(name).ofLongs().build();
+          for (Attributes attributes : attributesList) {
+            BoundLongHistogram bound = instrument.bind(attributes);
+            result.add(bound::record);
+          }
+        }
+        return result;
+      case GAUGE:
+        if (isDouble) {
+          ExtendedDoubleGauge instrument = (ExtendedDoubleGauge) meter.gaugeBuilder(name).build();
+          for (Attributes attributes : attributesList) {
+            BoundDoubleGauge bound = instrument.bind(attributes);
+            result.add(bound::set);
+          }
+        } else {
+          ExtendedLongGauge instrument =
+              (ExtendedLongGauge) meter.gaugeBuilder(name).ofLongs().build();
+          for (Attributes attributes : attributesList) {
+            BoundLongGauge bound = instrument.bind(attributes);
+            result.add(bound::set);
+          }
+        }
+        return result;
+      case OBSERVABLE_COUNTER:
+      case OBSERVABLE_UP_DOWN_COUNTER:
+      case OBSERVABLE_GAUGE:
+    }
+    throw new IllegalArgumentException();
+  }
 }

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

@@ -10,23 +10,57 @@
 import io.opentelemetry.api.incubator.metrics.BoundDoubleCounter;
 import io.opentelemetry.api.incubator.metrics.ExtendedDoubleCounter;
 import io.opentelemetry.api.incubator.metrics.ExtendedDoubleCounterBuilder;
+import io.opentelemetry.context.Context;
 import io.opentelemetry.sdk.metrics.internal.descriptor.Advice;
 import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor;
+import io.opentelemetry.sdk.metrics.internal.state.BoundStorageHandle;
 import io.opentelemetry.sdk.metrics.internal.state.WriteableMetricStorage;
 import java.util.List;
+import javax.annotation.Nullable;
 
-final class ExtendedSdkDoubleCounter extends SdkDoubleCounter implements ExtendedDoubleCounter {
+final class ExtendedSdkDoubleCounter extends SdkDoubleCounter
+    implements ExtendedDoubleCounter, BoundDoubleCounter {
+
+  // Non-null only when this is a bound instance returned from bind(); null for the instrument
+  // itself. When set, the add() methods record straight to this handle instead of resolving the
+  // series from the storage on each call.
+  @Nullable private final BoundStorageHandle boundHandle;
 
   private ExtendedSdkDoubleCounter(
       InstrumentDescriptor descriptor, SdkMeter sdkMeter, WriteableMetricStorage storage) {
+    this(descriptor, sdkMeter, storage, null);
+  }
+
+  private ExtendedSdkDoubleCounter(
+      InstrumentDescriptor descriptor,
+      SdkMeter sdkMeter,
+      WriteableMetricStorage storage,
+      @Nullable BoundStorageHandle boundHandle) {
     super(descriptor, sdkMeter, storage);
+    this.boundHandle = boundHandle;
   }
 
   @Override
   public BoundDoubleCounter bind(Attributes attributes) {
-    // TODO(bound-instruments): implement against WriteableMetricStorage in the implementation phase
-    // (includes fresh-eyes work on DeltaSynchronousMetricStorage).
-    throw new UnsupportedOperationException("bind is not yet implemented");
+    return new ExtendedSdkDoubleCounter(
+        getDescriptor(), sdkMeter, storage, storage.bind(attributes));
+  }
+
+  @Override
+  public void add(double value) {
+    add(value, Context.current());
+  }
+
+  @Override
+  public void add(double value, Context context) {
+    if (!validateNonNegative(value)) {
+      return;
+    }
+    if (boundHandle != null) {
+      boundHandle.recordDouble(value, context);
+    } else {
+      storage.recordDouble(value, Attributes.empty(), context);
+    }
   }
 
   static final class ExtendedSdkDoubleCounterBuilder extends SdkDoubleCounterBuilder

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

@@ -11,22 +11,52 @@
 import io.opentelemetry.api.incubator.metrics.ExtendedDoubleGauge;
 import io.opentelemetry.api.incubator.metrics.ExtendedDoubleGaugeBuilder;
 import io.opentelemetry.api.incubator.metrics.ExtendedLongGaugeBuilder;
+import io.opentelemetry.context.Context;
 import io.opentelemetry.sdk.metrics.internal.descriptor.InstrumentDescriptor;
+import io.opentelemetry.sdk.metrics.internal.state.BoundStorageHandle;
 import io.opentelemetry.sdk.metrics.internal.state.WriteableMetricStorage;
 import java.util.List;
+import javax.annotation.Nullable;
 
-final class ExtendedSdkDoubleGauge extends SdkDoubleGauge implements ExtendedDoubleGauge {
+final class ExtendedSdkDoubleGauge extends SdkDoubleGauge
+    implements ExtendedDoubleGauge, BoundDoubleGauge {
+
+  // Non-null only when this is a bound instance returned from bind(); null for the instrument
+  // itself. When set, the set() methods record straight to this handle instead of resolving the
+  // series from the storage on each call.
+  @Nullable private final BoundStorageHandle boundHandle;
 
   private ExtendedSdkDoubleGauge(
       InstrumentDescriptor descriptor, SdkMeter sdkMeter, WriteableMetricStorage storage) {
+    this(descriptor, sdkMeter, storage, null);
+  }
+
+  private ExtendedSdkDoubleGauge(
+      InstrumentDescriptor descriptor,
+      SdkMeter sdkMeter,
+      WriteableMetricStorage storage,
+      @Nullable BoundStorageHandle boundHandle) {
     super(descriptor, sdkMeter, storage);
+    this.boundHandle = boundHandle;
   }
 
   @Override
   public BoundDoubleGauge bind(Attributes attributes) {
-    // TODO(bound-instruments): implement against WriteableMetricStorage in the implementation phase
-    // (includes fresh-eyes work on DeltaSynchronousMetricStorage).
-    throw new UnsupportedOperationException("bind is not yet implemented");
+    return new ExtendedSdkDoubleGauge(getDescriptor(), sdkMeter, storage, storage.bind(attributes));
+  }
+
+  @Override
+  public void set(double value) {
+    set(value, Context.current());
+  }
+
+  @Override
+  public void set(double value, Context context) {
+    if (boundHandle != null) {
+      boundHandle.recordDouble(value, context);
+    } else {
+      storage.recordDouble(value, Attributes.empty(), context);
+    }
   }
 
   static final class ExtendedSdkDoubleGaugeBuilder extends SdkDoubleGaugeBuilder