Implement OpenTelemetry meter storage and aggregations

dd-java-agent/agent-otel/otel-shim/build.gradle

@@ -7,5 +7,6 @@ dependencies {
   // minimum OpenTelemetry API version this shim is compatible with
   compileOnly group: 'io.opentelemetry', name: 'opentelemetry-api', version: '1.47.0'
 
+  implementation project(':products:metrics:metrics-api')
   implementation project(':internal-api')
 }

dd-java-agent/agent-otel/otel-shim/src/main/java/datadog/opentelemetry/shim/OtelInstrumentationScope.java

@@ -17,6 +17,20 @@ public OtelInstrumentationScope(
     this.schemaUrl = schemaUrl;
   }
 
+  public String getName() {
+    return scopeName;
+  }
+
+  @Nullable
+  public String getVersion() {
+    return scopeVersion;
+  }
+
+  @Nullable
+  public String getSchemaUrl() {
+    return schemaUrl;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (!(o instanceof OtelInstrumentationScope)) {
@@ -36,4 +50,15 @@ public int hashCode() {
     result = 31 * result + Objects.hashCode(schemaUrl);
     return result;
   }
+
+  @Override
+  public String toString() {
+    // use same property names as OTel in toString
+    return "OtelInstrumentationScope{"
+        + "name='"
+        + scopeName
+        + (scopeVersion != null ? "', version='" + scopeVersion : "")
+        + (schemaUrl != null ? "', schemaUrl='" + schemaUrl : "")
+        + "'}";
+  }
 }

dd-java-agent/agent-otel/otel-shim/src/main/java/datadog/opentelemetry/shim/metrics/OtelDoubleCounter.java

@@ -5,55 +5,76 @@
 import static datadog.opentelemetry.shim.metrics.OtelMeter.NOOP_INSTRUMENT_NAME;
 import static datadog.opentelemetry.shim.metrics.OtelMeter.NOOP_METER;
 
+import datadog.opentelemetry.shim.metrics.data.OtelMetricStorage;
+import datadog.trace.relocate.api.RatelimitedLogger;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.metrics.DoubleCounter;
 import io.opentelemetry.api.metrics.DoubleCounterBuilder;
 import io.opentelemetry.api.metrics.ObservableDoubleCounter;
 import io.opentelemetry.api.metrics.ObservableDoubleMeasurement;
 import io.opentelemetry.context.Context;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import javax.annotation.ParametersAreNonnullByDefault;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @ParametersAreNonnullByDefault
-final class OtelDoubleCounter implements DoubleCounter {
+final class OtelDoubleCounter extends OtelInstrument implements DoubleCounter {
+  private static final Logger LOGGER = LoggerFactory.getLogger(OtelDoubleCounter.class);
+  private static final RatelimitedLogger RATELIMITED_LOGGER =
+      new RatelimitedLogger(LOGGER, 5, TimeUnit.MINUTES);
+
+  OtelDoubleCounter(OtelMetricStorage storage) {
+    super(storage);
+  }
 
   @Override
   public void add(double value) {
-    // FIXME: implement recording
+    add(value, Attributes.empty());
   }
 
   @Override
   public void add(double value, Attributes attributes) {
-    // FIXME: implement recording
+    if (value < 0) {
+      RATELIMITED_LOGGER.warn(
+          "Counters can only increase. Instrument {} has recorded a negative value.",
+          storage.getInstrumentName());
+    } else {
+      storage.recordDouble(value, attributes);
+    }
   }
 
   @Override
-  public void add(double value, Attributes attributes, Context context) {
-    // FIXME: implement recording
+  public void add(double value, Attributes attributes, Context unused) {
+    add(value, attributes);
   }
 
   static final class Builder implements DoubleCounterBuilder {
-    private final OtelInstrumentBuilder instrumentBuilder;
+    private final OtelMeter meter;
+    private final OtelInstrumentBuilder builder;
 
-    Builder(OtelInstrumentBuilder builder) {
-      this.instrumentBuilder = ofDoubles(builder, COUNTER);
+    Builder(OtelMeter meter, OtelInstrumentBuilder builder) {
+      this.meter = meter;
+      this.builder = ofDoubles(builder, COUNTER);
     }
 
     @Override
     public DoubleCounterBuilder setDescription(String description) {
-      instrumentBuilder.setDescription(description);
+      builder.setDescription(description);
       return this;
     }
 
     @Override
     public DoubleCounterBuilder setUnit(String unit) {
-      instrumentBuilder.setUnit(unit);
+      builder.setUnit(unit);
       return this;
     }
 
     @Override
     public DoubleCounter build() {
-      return new OtelDoubleCounter();
+      return new OtelDoubleCounter(
+          meter.registerStorage(builder.descriptor(), OtelMetricStorage::newDoubleSumStorage));
     }
 
     @Override

dd-java-agent/agent-otel/otel-shim/src/main/java/datadog/opentelemetry/shim/metrics/OtelDoubleGauge.java

@@ -5,6 +5,7 @@
 import static datadog.opentelemetry.shim.metrics.OtelMeter.NOOP_INSTRUMENT_NAME;
 import static datadog.opentelemetry.shim.metrics.OtelMeter.NOOP_METER;
 
+import datadog.opentelemetry.shim.metrics.data.OtelMetricStorage;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.metrics.DoubleGauge;
 import io.opentelemetry.api.metrics.DoubleGaugeBuilder;
@@ -16,50 +17,56 @@
 import javax.annotation.ParametersAreNonnullByDefault;
 
 @ParametersAreNonnullByDefault
-final class OtelDoubleGauge implements DoubleGauge {
+final class OtelDoubleGauge extends OtelInstrument implements DoubleGauge {
+  OtelDoubleGauge(OtelMetricStorage storage) {
+    super(storage);
+  }
 
   @Override
   public void set(double value) {
-    // FIXME: implement recording
+    set(value, Attributes.empty());
   }
 
   @Override
   public void set(double value, Attributes attributes) {
-    // FIXME: implement recording
+    storage.recordDouble(value, attributes);
   }
 
   @Override
-  public void set(double value, Attributes attributes, Context context) {
-    // FIXME: implement recording
+  public void set(double value, Attributes attributes, Context unused) {
+    set(value, attributes);
   }
 
   static final class Builder implements DoubleGaugeBuilder {
-    private final OtelInstrumentBuilder instrumentBuilder;
+    private final OtelMeter meter;
+    private final OtelInstrumentBuilder builder;
 
     Builder(OtelMeter meter, String instrumentName) {
-      this.instrumentBuilder = ofDoubles(meter, instrumentName, GAUGE);
+      this.meter = meter;
+      this.builder = ofDoubles(instrumentName, GAUGE);
     }
 
     @Override
     public DoubleGaugeBuilder setDescription(String description) {
-      instrumentBuilder.setDescription(description);
+      builder.setDescription(description);
       return this;
     }
 
     @Override
     public DoubleGaugeBuilder setUnit(String unit) {
-      instrumentBuilder.setUnit(unit);
+      builder.setUnit(unit);
       return this;
     }
 
     @Override
     public LongGaugeBuilder ofLongs() {
-      return new OtelLongGauge.Builder(instrumentBuilder);
+      return new OtelLongGauge.Builder(meter, builder);
     }
 
     @Override
     public DoubleGauge build() {
-      return new OtelDoubleGauge();
+      return new OtelDoubleGauge(
+          meter.registerStorage(builder.descriptor(), OtelMetricStorage::newDoubleValueStorage));
     }
 
     @Override

dd-java-agent/agent-otel/otel-shim/src/main/java/datadog/opentelemetry/shim/metrics/OtelDoubleHistogram.java

@@ -2,66 +2,135 @@
 
 import static datadog.opentelemetry.shim.metrics.OtelInstrumentBuilder.ofDoubles;
 import static datadog.opentelemetry.shim.metrics.OtelInstrumentType.HISTOGRAM;
+import static datadog.opentelemetry.shim.metrics.data.OtelMetricStorage.newHistogramStorage;
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
 
+import datadog.opentelemetry.shim.metrics.data.OtelMetricStorage;
+import datadog.trace.relocate.api.RatelimitedLogger;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.metrics.DoubleHistogram;
 import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
 import io.opentelemetry.api.metrics.LongHistogramBuilder;
 import io.opentelemetry.context.Context;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
 import javax.annotation.ParametersAreNonnullByDefault;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @ParametersAreNonnullByDefault
-final class OtelDoubleHistogram implements DoubleHistogram {
+final class OtelDoubleHistogram extends OtelInstrument implements DoubleHistogram {
+  private static final Logger LOGGER = LoggerFactory.getLogger(OtelDoubleHistogram.class);
+  private static final RatelimitedLogger RATELIMITED_LOGGER =
+      new RatelimitedLogger(LOGGER, 5, TimeUnit.MINUTES);
+
+  OtelDoubleHistogram(OtelMetricStorage storage) {
+    super(storage);
+  }
 
   @Override
   public void record(double value) {
-    // FIXME: implement recording
+    record(value, Attributes.empty());
   }
 
   @Override
   public void record(double value, Attributes attributes) {
-    // FIXME: implement recording
+    if (value < 0) {
+      RATELIMITED_LOGGER.warn(
+          "Histograms can only record non-negative values. Instrument {} has recorded a negative value.",
+          storage.getInstrumentName());
+    } else {
+      storage.recordDouble(value, attributes);
+    }
   }
 
   @Override
-  public void record(double value, Attributes attributes, Context context) {
-    // FIXME: implement recording
+  public void record(double value, Attributes attributes, Context unused) {
+    record(value, attributes);
   }
 
   static final class Builder implements DoubleHistogramBuilder {
-    private final OtelInstrumentBuilder instrumentBuilder;
+    private static final List<Double> DEFAULT_BOUNDARIES =
+        asList(
+            0d, 5d, 10d, 25d, 50d, 75d, 100d, 250d, 500d, 750d, 1_000d, 2_500d, 5_000d, 7_500d,
+            10_000d);
+
+    private final OtelMeter meter;
+    private final OtelInstrumentBuilder builder;
+    private List<Double> bucketBoundaries;
 
     Builder(OtelMeter meter, String instrumentName) {
-      this.instrumentBuilder = ofDoubles(meter, instrumentName, HISTOGRAM);
+      this.meter = meter;
+      this.builder = ofDoubles(instrumentName, HISTOGRAM);
+      this.bucketBoundaries = DEFAULT_BOUNDARIES;
     }
 
     @Override
     public DoubleHistogramBuilder setDescription(String description) {
-      instrumentBuilder.setDescription(description);
+      builder.setDescription(description);
       return this;
     }
 
     @Override
     public DoubleHistogramBuilder setUnit(String unit) {
-      instrumentBuilder.setUnit(unit);
+      builder.setUnit(unit);
       return this;
     }
 
     @Override
+    @SuppressFBWarnings("DCN") // match OTel in catching and logging NPE
     public DoubleHistogramBuilder setExplicitBucketBoundariesAdvice(List<Double> bucketBoundaries) {
-      // FIXME: implement boundary advice
+      try {
+        Objects.requireNonNull(bucketBoundaries, "bucketBoundaries must not be null");
+        this.bucketBoundaries = validateBoundaries(new ArrayList<>(bucketBoundaries));
+      } catch (IllegalArgumentException | NullPointerException e) {
+        LOGGER.warn("Error setting explicit bucket boundaries advice: {}", e.getMessage());
+      }
       return this;
     }
 
     @Override
     public LongHistogramBuilder ofLongs() {
-      return new OtelLongHistogram.Builder(instrumentBuilder);
+      return new OtelLongHistogram.Builder(meter, builder, bucketBoundaries);
     }
 
     @Override
     public DoubleHistogram build() {
-      return new OtelDoubleHistogram();
+      return new OtelDoubleHistogram(
+          meter.registerStorage(
+              builder.descriptor(),
+              descriptor -> newHistogramStorage(descriptor, bucketBoundaries)));
+    }
+
+    static List<Double> validateBoundaries(List<Double> boundaries) {
+      if (boundaries.isEmpty()) {
+        return emptyList();
+      }
+      if (boundaries.get(0) == Double.NEGATIVE_INFINITY) {
+        throw new IllegalArgumentException("invalid bucket boundary: -Inf");
+      }
+      if (boundaries.get(boundaries.size() - 1) == Double.POSITIVE_INFINITY) {
+        throw new IllegalArgumentException("invalid bucket boundary: +Inf");
+      }
+      Double previousBoundary = null;
+      for (Double boundary : boundaries) {
+        if (boundary.isNaN()) {
+          throw new IllegalArgumentException("invalid bucket boundary: NaN");
+        }
+        if (previousBoundary != null && previousBoundary >= boundary) {
+          throw new IllegalArgumentException(
+              "Bucket boundaries must be in increasing order: "
+                  + previousBoundary
+                  + " >= "
+                  + boundary);
+        }
+        previousBoundary = boundary;
+      }
+      return boundaries;
     }
   }
 }