Register an OTel attributes reader for each injected/extension class-loader

Author: mcculls

dd-java-agent/agent-otel/otel-bootstrap/src/main/java/datadog/trace/bootstrap/otel/metrics/data/OtelMetricStorage.java

@@ -11,9 +11,11 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.WeakHashMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.function.BiConsumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import org.slf4j.Logger;
@@ -33,6 +35,9 @@ public final class OtelMetricStorage {
   private static final Attributes CARDINALITY_OVERFLOW =
       Attributes.builder().put("otel.metric.overflow", true).build();
 
+  private static final Map<ClassLoader, BiConsumer<Object, BiConsumer<String, Object>>>
+      ATTRIBUTE_READERS = new WeakHashMap<>();
+
   private final OtelInstrumentDescriptor descriptor;
   private final boolean resetOnCollect;
   private final Function<Object, OtelAggregator> aggregatorSupplier;
@@ -157,13 +162,27 @@ public void collectMetric(OtelMetricVisitor visitor) {
     }
   }
 
+  public static void registerAttributeReader(
+      ClassLoader cl, BiConsumer<Object, BiConsumer<String, Object>> reader) {
+    ATTRIBUTE_READERS.put(cl, reader);
+  }
+
+  private static void visitAttributes(Object attributes, OtelMetricVisitor visitor) {
+    ClassLoader cl = attributes.getClass().getClassLoader();
+    BiConsumer<Object, BiConsumer<String, Object>> reader = ATTRIBUTE_READERS.get(cl);
+    if (reader != null) {
+      reader.accept(attributes, visitor::visitAttribute);
+    }
+  }
+
   /** Collect data for CUMULATIVE temporality, keeping aggregators for future writes. */
   private void doCollect(OtelMetricVisitor visitor) {
     // no need to hold writers back if we are not resetting metrics on collect
     currentRecording.aggregators.forEach(
         (attributes, aggregator) -> {
           if (!aggregator.isEmpty()) {
-            visitor.visitPoint(attributes, aggregator.collect());
+            visitAttributes(attributes, visitor);
+            visitor.visitPoint(aggregator.collect());
           }
         });
   }
@@ -197,7 +216,8 @@ private void doCollectAndReset(OtelMetricVisitor visitor) {
     aggregators.forEach(
         (attributes, aggregator) -> {
           if (!aggregator.isEmpty()) {
-            visitor.visitPoint(attributes, aggregator.collectAndReset());
+            visitAttributes(attributes, visitor);
+            visitor.visitPoint(aggregator.collectAndReset());
           }
         });
 

dd-java-agent/agent-otel/otel-bootstrap/src/main/java/datadog/trace/bootstrap/otel/metrics/export/OtelMetricVisitor.java

@@ -2,8 +2,15 @@
 
 import datadog.trace.bootstrap.otel.metrics.data.OtelPoint;
 
-/** A visitor to visit a metric in an instrumentation scope. */
+/**
+ * A visitor to visit a metric in an instrumentation scope.
+ *
+ * <p>Methods must be called in the following order: ( visitAttribute* visitPoint )*
+ */
 public interface OtelMetricVisitor {
+  /** Visits an attribute of the upcoming data point. */
+  void visitAttribute(String key, Object value);
+
   /** Visits a data point in the metric. */
-  void visitPoint(Object attributes, OtelPoint point);
+  void visitPoint(OtelPoint point);
 }

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

@@ -1,7 +1,9 @@
 package datadog.opentelemetry.shim.metrics;
 
 import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
+import datadog.trace.bootstrap.otel.metrics.data.OtelMetricStorage;
 import datadog.trace.util.Strings;
+import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.metrics.MeterBuilder;
 import io.opentelemetry.api.metrics.MeterProvider;
@@ -22,6 +24,15 @@ public final class OtelMeterProvider implements MeterProvider {
   /** Meter shims, indexed by instrumentation scope. */
   private final Map<OtelInstrumentationScope, OtelMeter> meters = new ConcurrentHashMap<>();
 
+  private OtelMeterProvider() {
+    // register attribute reader for class-loader where this provider is being used/injected
+    OtelMetricStorage.registerAttributeReader(
+        Attributes.class.getClassLoader(),
+        (attributes, consumer) ->
+            ((Attributes) attributes)
+                .forEach((attribute, value) -> consumer.accept(attribute.getKey(), value)));
+  }
+
   @Override
   public Meter get(String instrumentationScopeName) {
     return getMeterShim(instrumentationScopeName, null, null);

dd-java-agent/instrumentation/opentelemetry/opentelemetry-1.47/src/test/groovy/opentelemetry147/metrics/MetricsTest.groovy

@@ -429,6 +429,7 @@ class MetricsTest extends InstrumentationSpecification {
   class MeterReader implements OtelMetricsVisitor, OtelScopedMetricsVisitor, OtelMetricVisitor {
     def scopeName
     def instrumentName
+    def attributes = [:]
 
     @Override
     OtelScopedMetricsVisitor visitScopedMetrics(OtelInstrumentationScope scope) {
@@ -443,10 +444,16 @@ class MetricsTest extends InstrumentationSpecification {
     }
 
     @Override
-    void visitPoint(Object attributes, OtelPoint point) {
+    void visitAttribute(String key, Object value) {
+      attributes.put(key, value)
+    }
+
+    @Override
+    void visitPoint(OtelPoint point) {
       def key = scopeName + ':' + instrumentName
       if (!attributes.isEmpty()) {
-        key = key + '@' + attributes.asMap()
+        key = key + '@' + attributes
+        attributes.clear()
       }
       switch (point.class) {
         case OtelLongPoint: