Update Gauge meter to add tags with value

Signed-off-by: Gagan Juneja <[email protected]>

libs/telemetry/src/main/java/org/opensearch/telemetry/metrics/DefaultMetricsRegistry.java

@@ -8,8 +8,6 @@
 
 package org.opensearch.telemetry.metrics;
 
-import org.opensearch.telemetry.metrics.tags.Tags;
-
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.function.Supplier;
@@ -44,8 +42,8 @@ public Histogram createHistogram(String name, String description, String unit) {
     }
 
     @Override
-    public Closeable createGauge(String name, String description, String unit, Supplier<Double> valueProvider, Tags tags) {
-        return metricsTelemetry.createGauge(name, description, unit, valueProvider, tags);
+    public Closeable createGauge(String name, String description, String unit, Supplier<ObservableMeasurement> valueProvider) {
+        return metricsTelemetry.createGauge(name, description, unit, valueProvider);
     }
 
     @Override

libs/telemetry/src/main/java/org/opensearch/telemetry/metrics/MetricsRegistry.java

@@ -9,7 +9,6 @@
 package org.opensearch.telemetry.metrics;
 
 import org.opensearch.common.annotation.ExperimentalApi;
-import org.opensearch.telemetry.metrics.tags.Tags;
 
 import java.io.Closeable;
 import java.util.function.Supplier;
@@ -54,13 +53,12 @@ public interface MetricsRegistry extends Closeable {
      * Creates the Observable Gauge type of Metric. Where the value provider will be called at a certain frequency
      * to capture the value.
      *
-     * @param name          name of the observable gauge.
-     * @param description   any description about the metric.
-     * @param unit          unit of the metric.
-     * @param valueProvider value provider.
-     * @param tags          attributes/dimensions of the metric.
+     * @param name        name of the observable gauge.
+     * @param description any description about the metric.
+     * @param unit        unit of the metric.
+     * @param value       value provider.
      * @return closeable to dispose/close the Gauge metric.
      */
-    Closeable createGauge(String name, String description, String unit, Supplier<Double> valueProvider, Tags tags);
+    Closeable createGauge(String name, String description, String unit, Supplier<ObservableMeasurement> value);
 
 }

libs/telemetry/src/main/java/org/opensearch/telemetry/metrics/ObservableMeasurement.java

@@ -0,0 +1,53 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.telemetry.metrics;
+
+import org.opensearch.common.annotation.ExperimentalApi;
+import org.opensearch.telemetry.metrics.tags.Tags;
+
+/**
+ * Observable Measurement for the Asynchronous instruments.
+ * @opensearch.experimental
+ */
+@ExperimentalApi
+public class ObservableMeasurement {
+    private final Double value;
+    private final Tags tags;
+
+    /**
+     * Factory method to create the {@link ObservableMeasurement} object.
+     * @param value value.
+     * @param tags tags to be added per value.
+     * @return ObservableMeasurement
+     */
+    public static ObservableMeasurement create(double value, Tags tags) {
+        return new ObservableMeasurement(value, tags);
+    }
+
+    private ObservableMeasurement(double value, Tags tags) {
+        this.value = value;
+        this.tags = tags;
+    }
+
+    /**
+     * Returns the value.
+     * @return value
+     */
+    public Double getValue() {
+        return value;
+    }
+
+    /**
+     * Returns the tags.
+     * @return tags
+     */
+    public Tags getTags() {
+        return tags;
+    }
+}

libs/telemetry/src/main/java/org/opensearch/telemetry/metrics/noop/NoopMetricsRegistry.java

@@ -12,7 +12,7 @@
 import org.opensearch.telemetry.metrics.Counter;
 import org.opensearch.telemetry.metrics.Histogram;
 import org.opensearch.telemetry.metrics.MetricsRegistry;
-import org.opensearch.telemetry.metrics.tags.Tags;
+import org.opensearch.telemetry.metrics.ObservableMeasurement;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -48,7 +48,7 @@ public Histogram createHistogram(String name, String description, String unit) {
     }
 
     @Override
-    public Closeable createGauge(String name, String description, String unit, Supplier<Double> valueProvider, Tags tags) {
+    public Closeable createGauge(String name, String description, String unit, Supplier<ObservableMeasurement> valueProvider) {
         return () -> {};
     }
 

libs/telemetry/src/test/java/org/opensearch/telemetry/metrics/DefaultMetricsRegistryTests.java

@@ -66,15 +66,14 @@ public void testHistogram() {
     @SuppressWarnings("unchecked")
     public void testGauge() {
         Closeable mockCloseable = mock(Closeable.class);
-        when(
-            defaultMeterRegistry.createGauge(any(String.class), any(String.class), any(String.class), any(Supplier.class), any(Tags.class))
-        ).thenReturn(mockCloseable);
+        when(defaultMeterRegistry.createGauge(any(String.class), any(String.class), any(String.class), any(Supplier.class))).thenReturn(
+            mockCloseable
+        );
         Closeable closeable = defaultMeterRegistry.createGauge(
             "org.opensearch.telemetry.metrics.DefaultMeterRegistryTests.testObservableGauge",
             "test observable gauge",
             "ms",
-            () -> 1.0,
-            Tags.EMPTY
+            () -> ObservableMeasurement.create(1.0, Tags.EMPTY)
         );
         assertSame(mockCloseable, closeable);
     }

plugins/telemetry-otel/src/internalClusterTest/java/org/opensearch/telemetry/metrics/TelemetryMetricsEnabledSanityIT.java

@@ -23,10 +23,13 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.sdk.metrics.data.DoublePointData;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableExponentialHistogramPointData;
@@ -130,8 +133,10 @@ public void testGauge() throws Exception {
         InMemorySingletonMetricsExporter.INSTANCE.reset();
         Tags tags = Tags.create().addTag("test", "integ-test");
         final AtomicInteger testValue = new AtomicInteger(0);
-        Supplier<Double> valueProvider = () -> { return Double.valueOf(testValue.incrementAndGet()); };
-        Closeable gaugeCloseable = metricsRegistry.createGauge(metricName, "test", "ms", valueProvider, tags);
+        Supplier<ObservableMeasurement> valueProvider = () -> {
+            return ObservableMeasurement.create(Double.valueOf(testValue.incrementAndGet()), tags);
+        };
+        Closeable gaugeCloseable = metricsRegistry.createGauge(metricName, "test", "ms", valueProvider);
         // Sleep for about 2.2s to wait for metrics to be published.
         Thread.sleep(2200);
 
@@ -140,6 +145,9 @@ public void testGauge() throws Exception {
         assertTrue(getMaxObservableGaugeValue(exporter, metricName) >= 2.0);
         gaugeCloseable.close();
         double observableGaugeValueAfterStop = getMaxObservableGaugeValue(exporter, metricName);
+        Map<AttributeKey<?>, Object> attributes = getMetricAttributes(exporter, metricName);
+
+        assertEquals("integ-test", attributes.get(AttributeKey.stringKey("test")));
 
         // Sleep for about 1.2s to wait for metrics to see that closed observableGauge shouldn't execute the callable.
         Thread.sleep(1200);
@@ -154,11 +162,21 @@ private static double getMaxObservableGaugeValue(InMemorySingletonMetricsExporte
             .collect(Collectors.toList());
         double totalValue = 0;
         for (MetricData metricData : dataPoints) {
+            Attributes attributes = metricData.getDoubleGaugeData().getPoints().stream().findAny().get().getAttributes();
             totalValue = Math.max(totalValue, ((DoublePointData) metricData.getDoubleGaugeData().getPoints().toArray()[0]).getValue());
         }
         return totalValue;
     }
 
+    private static Map<AttributeKey<?>, Object> getMetricAttributes(InMemorySingletonMetricsExporter exporter, String metricName) {
+        List<MetricData> dataPoints = exporter.getFinishedMetricItems()
+            .stream()
+            .filter(a -> a.getName().contains(metricName))
+            .collect(Collectors.toList());
+        Attributes attributes = dataPoints.get(0).getDoubleGaugeData().getPoints().stream().findAny().get().getAttributes();
+        return attributes.asMap();
+    }
+
     @After
     public void reset() {
         InMemorySingletonMetricsExporter.INSTANCE.reset();

plugins/telemetry-otel/src/main/java/org/opensearch/telemetry/metrics/OTelMetricsTelemetry.java

@@ -11,7 +11,6 @@
 import org.opensearch.common.concurrent.RefCountedReleasable;
 import org.opensearch.telemetry.OTelAttributesConverter;
 import org.opensearch.telemetry.OTelTelemetryPlugin;
-import org.opensearch.telemetry.metrics.tags.Tags;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -91,12 +90,14 @@ public Histogram createHistogram(String name, String description, String unit) {
     }
 
     @Override
-    public Closeable createGauge(String name, String description, String unit, Supplier<Double> valueProvider, Tags tags) {
+    public Closeable createGauge(String name, String description, String unit, Supplier<ObservableMeasurement> valueProvider) {
         ObservableDoubleGauge doubleObservableGauge = AccessController.doPrivileged(
             (PrivilegedAction<ObservableDoubleGauge>) () -> otelMeter.gaugeBuilder(name)
                 .setUnit(unit)
                 .setDescription(description)
-                .buildWithCallback(record -> record.record(valueProvider.get(), OTelAttributesConverter.convert(tags)))
+                .buildWithCallback(
+                    record -> record.record(valueProvider.get().getValue(), OTelAttributesConverter.convert(valueProvider.get().getTags()))
+                )
         );
         return () -> doubleObservableGauge.close();
     }

plugins/telemetry-otel/src/test/java/org/opensearch/telemetry/metrics/OTelMetricsTelemetryTests.java

@@ -176,7 +176,12 @@ public void testGauge() throws Exception {
         when(mockOTelDoubleGaugeBuilder.setUnit(unit)).thenReturn(mockOTelDoubleGaugeBuilder);
         when(mockOTelDoubleGaugeBuilder.buildWithCallback(any(Consumer.class))).thenReturn(observableDoubleGauge);
 
-        Closeable closeable = metricsTelemetry.createGauge(observableGaugeName, description, unit, () -> 1.0, Tags.EMPTY);
+        Closeable closeable = metricsTelemetry.createGauge(
+            observableGaugeName,
+            description,
+            unit,
+            () -> ObservableMeasurement.create(1.0, Tags.EMPTY)
+        );
         closeable.close();
         verify(observableDoubleGauge).close();
     }

test/framework/src/main/java/org/opensearch/test/telemetry/MockTelemetry.java

@@ -13,9 +13,9 @@
 import org.opensearch.telemetry.metrics.Counter;
 import org.opensearch.telemetry.metrics.Histogram;
 import org.opensearch.telemetry.metrics.MetricsTelemetry;
+import org.opensearch.telemetry.metrics.ObservableMeasurement;
 import org.opensearch.telemetry.metrics.noop.NoopCounter;
 import org.opensearch.telemetry.metrics.noop.NoopHistogram;
-import org.opensearch.telemetry.metrics.tags.Tags;
 import org.opensearch.telemetry.tracing.TracingTelemetry;
 import org.opensearch.test.telemetry.tracing.MockTracingTelemetry;
 
@@ -58,7 +58,7 @@ public Histogram createHistogram(String name, String description, String unit) {
             }
 
             @Override
-            public Closeable createGauge(String name, String description, String unit, Supplier<Double> valueProvider, Tags tags) {
+            public Closeable createGauge(String name, String description, String unit, Supplier<ObservableMeasurement> valueProvider) {
                 return () -> {};
             }