implement async long and double counters

elastic · Aug 27, 2024 · b9c0bd0 · b9c0bd0
1 parent e633802
commit b9c0bd0
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 12 deletions.
diff --git a/test/framework/src/main/java/org/elasticsearch/telemetry/RecordingInstruments.java b/test/framework/src/main/java/org/elasticsearch/telemetry/RecordingInstruments.java
@@ -173,21 +173,29 @@ public void incrementBy(long inc, Map<String, Object> attributes) {
 
     public static class RecordingAsyncLongCounter extends CallbackRecordingInstrument implements LongAsyncCounter {
 
-        public RecordingAsyncLongCounter(String name, Supplier<LongWithAttributes> observer, MetricRecorder<Instrument> recorder) {
+        public RecordingAsyncLongCounter(
+            String name,
+            Supplier<Collection<LongWithAttributes>> observer,
+            MetricRecorder<Instrument> recorder
+        ) {
             super(name, () -> {
                 var observation = observer.get();
-                return Collections.singletonList(new Tuple<>(observation.value(), observation.attributes()));
+                return observation.stream().map(o -> new Tuple<>((Number) o.value(), o.attributes())).toList();
             }, recorder);
         }
 
     }
 
     public static class RecordingAsyncDoubleCounter extends CallbackRecordingInstrument implements DoubleAsyncCounter {
 
-        public RecordingAsyncDoubleCounter(String name, Supplier<DoubleWithAttributes> observer, MetricRecorder<Instrument> recorder) {
+        public RecordingAsyncDoubleCounter(
+            String name,
+            Supplier<Collection<DoubleWithAttributes>> observer,
+            MetricRecorder<Instrument> recorder
+        ) {
             super(name, () -> {
                 var observation = observer.get();
-                return Collections.singletonList(new Tuple<>(observation.value(), observation.attributes()));
+                return observation.stream().map(o -> new Tuple<>((Number) o.value(), o.attributes())).toList();
             }, recorder);
         }
 

diff --git a/test/framework/src/main/java/org/elasticsearch/telemetry/RecordingMeterRegistry.java b/test/framework/src/main/java/org/elasticsearch/telemetry/RecordingMeterRegistry.java
@@ -127,9 +127,7 @@ public LongCounter registerLongCounter(String name, String description, String u
 
     @Override
     public LongAsyncCounter registerLongAsyncCounter(String name, String description, String unit, Supplier<LongWithAttributes> observer) {
-        LongAsyncCounter instrument = new RecordingInstruments.RecordingAsyncLongCounter(name, observer, recorder);
-        recorder.register(instrument, InstrumentType.fromInstrument(instrument), name, description, unit);
-        return instrument;
+        return registerLongsAsyncCounter(name, description, unit, () -> Collections.singleton(observer.get()));
     }
 
     @Override
@@ -139,7 +137,9 @@ public LongAsyncCounter registerLongsAsyncCounter(
         String unit,
         Supplier<Collection<LongWithAttributes>> observer
     ) {
-        throw new UnsupportedOperationException("not implemented");
+        LongAsyncCounter instrument = new RecordingInstruments.RecordingAsyncLongCounter(name, observer, recorder);
+        recorder.register(instrument, InstrumentType.fromInstrument(instrument), name, description, unit);
+        return instrument;
     }
 
     @Override
@@ -154,9 +154,7 @@ public DoubleAsyncCounter registerDoubleAsyncCounter(
         String unit,
         Supplier<DoubleWithAttributes> observer
     ) {
-        DoubleAsyncCounter instrument = new RecordingInstruments.RecordingAsyncDoubleCounter(name, observer, recorder);
-        recorder.register(instrument, InstrumentType.fromInstrument(instrument), name, description, unit);
-        return instrument;
+        return registerDoublesAsyncCounter(name, description, unit, () -> Collections.singleton(observer.get()));
     }
 
     @Override
@@ -166,7 +164,9 @@ public DoubleAsyncCounter registerDoublesAsyncCounter(
         String unit,
         Supplier<Collection<DoubleWithAttributes>> observer
     ) {
-        throw new UnsupportedOperationException("not implemented");
+        DoubleAsyncCounter instrument = new RecordingInstruments.RecordingAsyncDoubleCounter(name, observer, recorder);
+        recorder.register(instrument, InstrumentType.fromInstrument(instrument), name, description, unit);
+        return instrument;
     }
 
     @Override