Fix the behaviour of asynchronous counters and up-down counters.

Unlike synchronous counters which take the increment/delta value,
asynchronous callbacks report the absolute value of the counter.

Author: mcculls

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

@@ -79,7 +79,7 @@ public DoubleCounter build() {
 
     @Override
     public ObservableDoubleMeasurement buildObserver() {
-      return meter.registerObservableStorage(builder, OtelMetricStorage::newDoubleSumStorage);
+      return meter.registerObservableStorage(builder, OtelMetricStorage::newDoubleValueStorage);
     }
 
     @Override

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

@@ -65,7 +65,7 @@ public DoubleUpDownCounter build() {
 
     @Override
     public ObservableDoubleMeasurement buildObserver() {
-      return meter.registerObservableStorage(builder, OtelMetricStorage::newDoubleSumStorage);
+      return meter.registerObservableStorage(builder, OtelMetricStorage::newDoubleValueStorage);
     }
 
     @Override

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

@@ -85,7 +85,7 @@ public LongCounter build() {
 
     @Override
     public ObservableLongMeasurement buildObserver() {
-      return meter.registerObservableStorage(builder, OtelMetricStorage::newLongSumStorage);
+      return meter.registerObservableStorage(builder, OtelMetricStorage::newLongValueStorage);
     }
 
     @Override

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

@@ -71,7 +71,7 @@ public LongUpDownCounter build() {
 
     @Override
     public ObservableLongMeasurement buildObserver() {
-      return meter.registerObservableStorage(builder, OtelMetricStorage::newLongSumStorage);
+      return meter.registerObservableStorage(builder, OtelMetricStorage::newLongValueStorage);
     }
 
     @Override

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

@@ -212,6 +212,13 @@ void testObservableLongCounter() {
     assertEquals(1L, points.get("test:observable-long-counter"));
     assertEquals(2L, points.get("test:observable-long-counter" + WITH_ATTRS));
 
+    // second collect: absolute values are reported, not accumulated
+    points.clear();
+    OtelMetricRegistry.INSTANCE.collectMetrics(meterReader);
+
+    assertEquals(1L, points.get("test:observable-long-counter"));
+    assertEquals(2L, points.get("test:observable-long-counter" + WITH_ATTRS));
+
     closeQuietly(observable);
   }
 
@@ -231,6 +238,13 @@ void testObservableDoubleCounter() {
     assertEquals(1.2, points.get("test:observable-double-counter"));
     assertEquals(3.4, points.get("test:observable-double-counter" + WITH_ATTRS));
 
+    // second collect: absolute values are reported, not accumulated
+    points.clear();
+    OtelMetricRegistry.INSTANCE.collectMetrics(meterReader);
+
+    assertEquals(1.2, points.get("test:observable-double-counter"));
+    assertEquals(3.4, points.get("test:observable-double-counter" + WITH_ATTRS));
+
     closeQuietly(observable);
   }
 
@@ -249,6 +263,13 @@ void testObservableLongUpDownCounter() {
     assertEquals(1L, points.get("test:observable-long-up-down-counter"));
     assertEquals(2L, points.get("test:observable-long-up-down-counter" + WITH_ATTRS));
 
+    // second collect: absolute values are reported, not accumulated
+    points.clear();
+    OtelMetricRegistry.INSTANCE.collectMetrics(meterReader);
+
+    assertEquals(1L, points.get("test:observable-long-up-down-counter"));
+    assertEquals(2L, points.get("test:observable-long-up-down-counter" + WITH_ATTRS));
+
     closeQuietly(observable);
   }
 
@@ -268,6 +289,13 @@ void testObservableDoubleUpDownCounter() {
     assertEquals(1.2, points.get("test:observable-double-up-down-counter"));
     assertEquals(3.4, points.get("test:observable-double-up-down-counter" + WITH_ATTRS));
 
+    // second collect: absolute values are reported, not accumulated
+    points.clear();
+    OtelMetricRegistry.INSTANCE.collectMetrics(meterReader);
+
+    assertEquals(1.2, points.get("test:observable-double-up-down-counter"));
+    assertEquals(3.4, points.get("test:observable-double-up-down-counter" + WITH_ATTRS));
+
     closeQuietly(observable);
   }
 
@@ -385,18 +413,18 @@ void testBatchCallback() {
     points.clear();
     OtelMetricRegistry.INSTANCE.collectMetrics(meterReader);
 
-    // delta mode: counters show values added during last collect
+    // async counters show the count _set_ during last collect
     assertEquals(1L, points.get("test:long-counter-observer"));
     assertEquals(10L, points.get("test:long-counter-observer" + WITH_ATTRS));
     assertEquals(2.3, points.get("test:double-counter-observer"));
     assertEquals(20.3, points.get("test:double-counter-observer" + WITH_ATTRS));
-    // up-down counters stay cumulative: they show the running total
-    assertEquals(8L, points.get("test:long-up-down-counter-observer"));
-    assertEquals(80L, points.get("test:long-up-down-counter-observer" + WITH_ATTRS));
-    assertEquals(11.2, (double) points.get("test:double-up-down-counter-observer"), 0.001);
+    // async up-down counters also show the count _set_ during last collect
+    assertEquals(4L, points.get("test:long-up-down-counter-observer"));
+    assertEquals(40L, points.get("test:long-up-down-counter-observer" + WITH_ATTRS));
+    assertEquals(5.6, (double) points.get("test:double-up-down-counter-observer"), 0.001);
     assertEquals(
-        101.2, (double) points.get("test:double-up-down-counter-observer" + WITH_ATTRS), 0.001);
-    // gauges also stay cumulative: they only show latest value
+        50.6, (double) points.get("test:double-up-down-counter-observer" + WITH_ATTRS), 0.001);
+    // gauges continue to only show the latest value
     assertEquals(7L, points.get("test:long-gauge-observer"));
     assertEquals(70L, points.get("test:long-gauge-observer" + WITH_ATTRS));
     assertEquals(8.9, points.get("test:double-gauge-observer"));
@@ -407,18 +435,18 @@ void testBatchCallback() {
     points.clear();
     OtelMetricRegistry.INSTANCE.collectMetrics(meterReader);
 
-    // delta mode: no values were added as batchCallback is closed
+    // delta mode: no counts were set as batchCallback is closed, so no data point
     assertNull(points.get("test:long-counter-observer"));
     assertNull(points.get("test:long-counter-observer" + WITH_ATTRS));
     assertNull(points.get("test:double-counter-observer"));
     assertNull(points.get("test:double-counter-observer" + WITH_ATTRS));
-    // up-down counters stay cumulative: they show the running total
-    assertEquals(8L, points.get("test:long-up-down-counter-observer"));
-    assertEquals(80L, points.get("test:long-up-down-counter-observer" + WITH_ATTRS));
-    assertEquals(11.2, (double) points.get("test:double-up-down-counter-observer"), 0.001);
+    // up-down counters stay cumulative: they continue to show the last count set
+    assertEquals(4L, points.get("test:long-up-down-counter-observer"));
+    assertEquals(40L, points.get("test:long-up-down-counter-observer" + WITH_ATTRS));
+    assertEquals(5.6, (double) points.get("test:double-up-down-counter-observer"), 0.001);
     assertEquals(
-        101.2, (double) points.get("test:double-up-down-counter-observer" + WITH_ATTRS), 0.001);
-    // gauges also stay cumulative: they only show latest value
+        50.6, (double) points.get("test:double-up-down-counter-observer" + WITH_ATTRS), 0.001);
+    // gauges also stay cumulative: they continue to show the latest value set
     assertEquals(7L, points.get("test:long-gauge-observer"));
     assertEquals(70L, points.get("test:long-gauge-observer" + WITH_ATTRS));
     assertEquals(8.9, points.get("test:double-gauge-observer"));