Refactor OtelMetricStorage so the same instance can aggregate metrics

using OTel attributes from both bootstrap and application classpaths

Author: mcculls

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

@@ -35,7 +35,7 @@ public final class OtelMetricStorage {
 
   private final OtelInstrumentDescriptor descriptor;
   private final boolean resetOnCollect;
-  private final Function<Attributes, OtelAggregator> aggregatorSupplier;
+  private final Function<Object, OtelAggregator> aggregatorSupplier;
   private volatile Recording currentRecording;
 
   // only used with DELTA temporality
@@ -98,7 +98,7 @@ public OtelInstrumentDescriptor getDescriptor() {
     return descriptor;
   }
 
-  public void recordLong(long value, Attributes attributes) {
+  public void recordLong(long value, Object attributes) {
     if (resetOnCollect) {
       Recording recording = acquireRecordingForWrite();
       try {
@@ -112,7 +112,7 @@ public void recordLong(long value, Attributes attributes) {
     }
   }
 
-  public void recordDouble(double value, Attributes attributes) {
+  public void recordDouble(double value, Object attributes) {
     if (Double.isNaN(value)) {
       LOGGER.debug(
           "Instrument {} has recorded measurement Not-a-Number (NaN) value with attributes {}. Dropping measurement.",
@@ -133,8 +133,7 @@ public void recordDouble(double value, Attributes attributes) {
     }
   }
 
-  private OtelAggregator aggregator(
-      Map<Attributes, OtelAggregator> aggregators, Attributes attributes) {
+  private OtelAggregator aggregator(Map<Object, OtelAggregator> aggregators, Object attributes) {
     Objects.requireNonNull(attributes, "attributes");
     OtelAggregator aggregator = aggregators.get(attributes);
     if (null != aggregator) {
@@ -188,7 +187,7 @@ private void doCollectAndReset(OtelInstrumentVisitor visitor) {
       Thread.yield(); // other threads are still writing to this recording
     }
 
-    Map<Attributes, OtelAggregator> aggregators = recording.aggregators;
+    Map<Object, OtelAggregator> aggregators = recording.aggregators;
 
     // avoid churn: only remove empty aggregators if we're over cardinality
     if (aggregators.size() >= CARDINALITY_LIMIT) {
@@ -233,7 +232,7 @@ private void releaseRecordingAfterWrite(Recording recording) {
   private static final int WRITER = 2;
 
   static final class Recording {
-    final Map<Attributes, OtelAggregator> aggregators;
+    final Map<Object, OtelAggregator> aggregators;
 
     transient volatile int activity;
 

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

@@ -1,9 +1,8 @@
 package datadog.opentelemetry.shim.metrics.export;
 
 import datadog.opentelemetry.shim.metrics.data.OtelPoint;
-import io.opentelemetry.api.common.Attributes;
 
 public interface OtelInstrumentVisitor {
   /** Visits a data point collected by the instrument. */
-  void visitPoint(Attributes attributes, OtelPoint point);
+  void visitPoint(Object attributes, OtelPoint point);
 }

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

@@ -442,7 +442,7 @@ class MetricsTest extends InstrumentationSpecification {
     }
 
     @Override
-    void visitPoint(Attributes attributes, OtelPoint point) {
+    void visitPoint(Object attributes, OtelPoint point) {
       def key = scopeName + ':' + instrumentName
       if (!attributes.isEmpty()) {
         key = key + '@' + attributes.asMap()