Fix metric gauge creation model (elastic#100609)

OTEL gauges should follow the callback model otherwise they will not be sent by
apm java agent. (or use BatchCallback)
This commit changes the gagues creation model to return Observable*Gauge
and uses AtomicLong/Double to store current value which will be polled when
metrics are exported (and callback is called)

docs/changelog/100609.yaml

@@ -0,0 +1,5 @@
+pr: 100609
+summary: Fix metric gauge creation model
+area: Infra/Core
+type: bug
+issues: []

modules/apm/src/main/java/org/elasticsearch/telemetry/apm/internal/metrics/DoubleGaugeAdapter.java

@@ -10,33 +10,46 @@
 
 import io.opentelemetry.api.metrics.Meter;
 
+import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * DoubleGaugeAdapter wraps an otel ObservableDoubleMeasurement
  */
-public class DoubleGaugeAdapter extends AbstractInstrument<io.opentelemetry.api.metrics.ObservableDoubleMeasurement>
+public class DoubleGaugeAdapter extends AbstractInstrument<io.opentelemetry.api.metrics.ObservableDoubleGauge>
     implements
         org.elasticsearch.telemetry.metric.DoubleGauge {
 
+    private final AtomicReference<ValueWithAttributes> valueWithAttributes;
+
     public DoubleGaugeAdapter(Meter meter, String name, String description, String unit) {
         super(meter, name, description, unit);
+        this.valueWithAttributes = new AtomicReference<>(new ValueWithAttributes(0.0, Collections.emptyMap()));
     }
 
     @Override
-    io.opentelemetry.api.metrics.ObservableDoubleMeasurement buildInstrument(Meter meter) {
-        var builder = Objects.requireNonNull(meter).gaugeBuilder(getName());
-        return builder.setDescription(getDescription()).setUnit(getUnit()).buildObserver();
+    io.opentelemetry.api.metrics.ObservableDoubleGauge buildInstrument(Meter meter) {
+        return Objects.requireNonNull(meter)
+            .gaugeBuilder(getName())
+            .setDescription(getDescription())
+            .setUnit(getUnit())
+            .buildWithCallback(measurement -> {
+                var localValueWithAttributed = valueWithAttributes.get();
+                measurement.record(localValueWithAttributed.value(), OtelHelper.fromMap(localValueWithAttributed.attributes()));
+            });
     }
 
     @Override
     public void record(double value) {
-        getInstrument().record(value);
+        record(value, Collections.emptyMap());
     }
 
     @Override
     public void record(double value, Map<String, Object> attributes) {
-        getInstrument().record(value, OtelHelper.fromMap(attributes));
+        this.valueWithAttributes.set(new ValueWithAttributes(value, attributes));
     }
+
+    private record ValueWithAttributes(double value, Map<String, Object> attributes) {}
 }

modules/apm/src/main/java/org/elasticsearch/telemetry/apm/internal/metrics/LongGaugeAdapter.java

@@ -10,37 +10,47 @@
 
 import io.opentelemetry.api.metrics.Meter;
 
+import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * LongGaugeAdapter wraps an otel ObservableLongMeasurement
  */
-public class LongGaugeAdapter extends AbstractInstrument<io.opentelemetry.api.metrics.ObservableLongMeasurement>
+public class LongGaugeAdapter extends AbstractInstrument<io.opentelemetry.api.metrics.ObservableLongGauge>
     implements
         org.elasticsearch.telemetry.metric.LongGauge {
+    private final AtomicReference<ValueWithAttributes> valueWithAttributes;
 
     public LongGaugeAdapter(Meter meter, String name, String description, String unit) {
         super(meter, name, description, unit);
+        this.valueWithAttributes = new AtomicReference<>(new ValueWithAttributes(0L, Collections.emptyMap()));
     }
 
     @Override
-    io.opentelemetry.api.metrics.ObservableLongMeasurement buildInstrument(Meter meter) {
+    io.opentelemetry.api.metrics.ObservableLongGauge buildInstrument(Meter meter) {
+
         return Objects.requireNonNull(meter)
             .gaugeBuilder(getName())
             .ofLongs()
             .setDescription(getDescription())
             .setUnit(getUnit())
-            .buildObserver();
+            .buildWithCallback(measurement -> {
+                var localValueWithAttributed = valueWithAttributes.get();
+                measurement.record(localValueWithAttributed.value(), OtelHelper.fromMap(localValueWithAttributed.attributes()));
+            });
     }
 
     @Override
     public void record(long value) {
-        getInstrument().record(value);
+        record(value, Collections.emptyMap());
     }
 
     @Override
     public void record(long value, Map<String, Object> attributes) {
-        getInstrument().record(value, OtelHelper.fromMap(attributes));
+        this.valueWithAttributes.set(new ValueWithAttributes(value, attributes));
     }
+
+    private record ValueWithAttributes(long value, Map<String, Object> attributes) {}
 }

modules/apm/src/test/java/org/elasticsearch/telemetry/apm/internal/metrics/GaugeAdapterTests.java

@@ -0,0 +1,123 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.telemetry.apm.internal.metrics;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.DoubleGaugeBuilder;
+import io.opentelemetry.api.metrics.LongGaugeBuilder;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.metrics.ObservableDoubleMeasurement;
+import io.opentelemetry.api.metrics.ObservableLongMeasurement;
+
+import org.elasticsearch.test.ESTestCase;
+import org.hamcrest.Matchers;
+import org.junit.Before;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.util.Map;
+import java.util.function.Consumer;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class GaugeAdapterTests extends ESTestCase {
+    Meter testMeter = Mockito.mock(Meter.class);
+    LongGaugeBuilder longGaugeBuilder = Mockito.mock(LongGaugeBuilder.class);
+    DoubleGaugeBuilder mockDoubleGaugeBuilder = Mockito.mock(DoubleGaugeBuilder.class);
+
+    @Before
+    public void init() {
+        when(longGaugeBuilder.setDescription(Mockito.anyString())).thenReturn(longGaugeBuilder);
+        when(longGaugeBuilder.setUnit(Mockito.anyString())).thenReturn(longGaugeBuilder);
+
+
+        when(mockDoubleGaugeBuilder.ofLongs()).thenReturn(longGaugeBuilder);
+        when(mockDoubleGaugeBuilder.setUnit(Mockito.anyString())).thenReturn(mockDoubleGaugeBuilder);
+        when(mockDoubleGaugeBuilder.setDescription(Mockito.anyString())).thenReturn(mockDoubleGaugeBuilder);
+        when(testMeter.gaugeBuilder(anyString())).thenReturn(mockDoubleGaugeBuilder);
+    }
+
+    // testing that a value reported is then used in a callback
+    @SuppressWarnings("unchecked")
+    public void testLongGaugeRecord() {
+        LongGaugeAdapter longGaugeAdapter = new LongGaugeAdapter(testMeter, "name", "desc", "unit");
+
+        // recording a value
+        longGaugeAdapter.record(1L, Map.of("k", 1L));
+
+        // upon metric export, the consumer will be called
+        ArgumentCaptor<Consumer<ObservableLongMeasurement>> captor = ArgumentCaptor.forClass(Consumer.class);
+        verify(longGaugeBuilder).buildWithCallback(captor.capture());
+
+        Consumer<ObservableLongMeasurement> value = captor.getValue();
+        // making sure that a consumer will fetch the value passed down upon recording of a value
+        TestLongMeasurement testLongMeasurement = new TestLongMeasurement();
+        value.accept(testLongMeasurement);
+
+        assertThat(testLongMeasurement.value, Matchers.equalTo(1L));
+        assertThat(testLongMeasurement.attributes, Matchers.equalTo(Attributes.builder().put("k", 1).build()));
+    }
+
+    // testing that a value reported is then used in a callback
+    @SuppressWarnings("unchecked")
+    public void testDoubleGaugeRecord() {
+        DoubleGaugeAdapter doubleGaugeAdapter = new DoubleGaugeAdapter(testMeter, "name", "desc", "unit");
+
+        // recording a value
+        doubleGaugeAdapter.record(1.0, Map.of("k", 1.0));
+
+        // upon metric export, the consumer will be called
+        ArgumentCaptor<Consumer<ObservableDoubleMeasurement>> captor = ArgumentCaptor.forClass(Consumer.class);
+        verify(mockDoubleGaugeBuilder).buildWithCallback(captor.capture());
+
+        Consumer<ObservableDoubleMeasurement> value = captor.getValue();
+        // making sure that a consumer will fetch the value passed down upon recording of a value
+        TestDoubleMeasurement testLongMeasurement = new TestDoubleMeasurement();
+        value.accept(testLongMeasurement);
+
+        assertThat(testLongMeasurement.value, Matchers.equalTo(1.0));
+        assertThat(testLongMeasurement.attributes, Matchers.equalTo(Attributes.builder().put("k", 1.0).build()));
+    }
+
+    private static class TestDoubleMeasurement implements ObservableDoubleMeasurement {
+        double value;
+        Attributes attributes;
+
+        @Override
+        public void record(double value) {
+            this.value = value;
+        }
+
+        @Override
+        public void record(double value, Attributes attributes) {
+            this.value = value;
+            this.attributes = attributes;
+
+        }
+    }
+
+    private static class TestLongMeasurement implements ObservableLongMeasurement {
+        long value;
+        Attributes attributes;
+
+        @Override
+        public void record(long value) {
+            this.value = value;
+        }
+
+        @Override
+        public void record(long value, Attributes attributes) {
+            this.value = value;
+            this.attributes = attributes;
+
+        }
+    }
+}