[Metrics] Java metric API (#9377)

java/generate_jni_header_files.sh

@@ -40,7 +40,8 @@ generate_one io.ray.runtime.actor.NativeActorHandle
 generate_one io.ray.runtime.object.NativeObjectStore
 generate_one io.ray.runtime.task.NativeTaskExecutor
 generate_one io.ray.runtime.gcs.GlobalStateAccessor
+generate_one io.ray.runtime.metric.NativeMetric
 
 # Remove empty files
 rm -f io_ray_runtime_RayNativeRuntime_AsyncContext.h
-rm -f io_ray_runtime_task_NativeTaskExecutor_NativeActorContext.h
+rm -f io_ray_runtime_task_NativeTaskExecutor_NativeActorContext.h

java/runtime/src/main/java/io/ray/runtime/gcs/GlobalStateAccessor.java

@@ -135,8 +135,6 @@ public class GlobalStateAccessor {
 
   private native boolean nativeConnect(long nativePtr);
 
-  private native void nativeDisconnect(long nativePtr);
-
   private native List<byte[]> nativeGetAllJobInfo(long nativePtr);
 
   private native List<byte[]> nativeGetAllNodeInfo(long nativePtr);

java/runtime/src/main/java/io/ray/runtime/metric/Count.java

@@ -0,0 +1,47 @@
+package io.ray.runtime.metric;
+
+import com.google.common.base.Preconditions;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class Count extends Metric {
+
+  private double count;
+
+  public Count(String name, String description, String unit, Map<TagKey, String> tags) {
+    super(name, tags);
+    count = 0.0d;
+    metricNativePointer = NativeMetric.registerCountNative(name, description, unit,
+      tags.keySet().stream().map(TagKey::getTagKey).collect(Collectors.toList()));
+    Preconditions.checkState(metricNativePointer != 0, "Count native pointer must not be 0.");
+  }
+
+  @Override
+  public void update(double value) {
+    super.update(value);
+    count += value;
+  }
+
+  @Override
+  public void update(double value, Map<TagKey, String> tags) {
+    super.update(value, tags);
+    count += value;
+  }
+
+  @Override
+  public void reset() {
+
+  }
+
+  public double getCount() {
+    return count;
+  }
+
+  /**
+   * @param delta add delta for counter
+   */
+  public void inc(double delta) {
+    update(delta);
+  }
+}

java/runtime/src/main/java/io/ray/runtime/metric/Gauge.java

@@ -0,0 +1,25 @@
+package io.ray.runtime.metric;
+
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+
+/**
+ * Gauge metric for recording last value and mapping object from stats.
+ */
+public class Gauge extends Metric {
+
+  public Gauge(String name, String description, String unit, Map<TagKey, String> tags) {
+    super(name, tags);
+    metricNativePointer = NativeMetric.registerGaugeNative(name, description, unit,
+      tags.keySet().stream().map(TagKey::getTagKey).collect(Collectors.toList()));
+    Preconditions.checkState(metricNativePointer != 0, "Gauge native pointer must not be 0.");
+  }
+
+  @Override
+  public void reset() {
+
+  }
+}
+

java/runtime/src/main/java/io/ray/runtime/metric/Histogram.java

@@ -0,0 +1,58 @@
+package io.ray.runtime.metric;
+
+import com.google.common.base.Preconditions;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Histogram measurement is mapped to histogram object in stats.
+ * In order to reduce JNI calls overhead, a memory historical window is used
+ * for storing transient value and we assume its max size is 100.
+ */
+public class Histogram extends Metric {
+
+  private List<Double> histogramWindow;
+  public static final int HISTOGRAM_WINDOW_SIZE = 100;
+
+  public Histogram(String name, String description, String unit, List<Double> boundaries,
+                   Map<TagKey, String> tags) {
+    super(name, tags);
+    metricNativePointer = NativeMetric.registerHistogramNative(name, description, unit,
+      boundaries.stream().mapToDouble(Double::doubleValue).toArray(),
+      tags.keySet().stream().map(TagKey::getTagKey).collect(Collectors.toList()));
+    Preconditions.checkState(metricNativePointer != 0,
+        "Histogram native pointer must not be 0.");
+    histogramWindow = new ArrayList<>();
+  }
+
+  private void updateForWindow(double value) {
+    if (histogramWindow.size() == HISTOGRAM_WINDOW_SIZE) {
+      histogramWindow.remove(0);
+    }
+    histogramWindow.add(value);
+  }
+
+  @Override
+  public void update(double value) {
+    super.update(value);
+    updateForWindow(value);
+  }
+
+  @Override
+  public void update(double value, Map<TagKey, String> tags) {
+    super.update(value, tags);
+    updateForWindow(value);
+  }
+
+  @Override
+  public void reset() {
+
+  }
+
+  public List<Double> getHistogramWindow() {
+    return histogramWindow;
+  }
+}

java/runtime/src/main/java/io/ray/runtime/metric/Metric.java

@@ -0,0 +1,91 @@
+package io.ray.runtime.metric;
+
+import com.google.common.base.Preconditions;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Class metric is mapped to stats metric object in core worker.
+ * it must be in categories set [Gague, Count, Sum, Histogram].
+ */
+public abstract class Metric {
+  protected String name;
+
+  protected double value;
+  // Native pointer mapping to gauge object of stats.
+  protected long metricNativePointer = 0L;
+
+  protected Map<TagKey, String> tags;
+
+  public Metric(String name, Map<TagKey, String> tags) {
+    Preconditions.checkNotNull(tags, "Metric tags map must not be null.");
+    Preconditions.checkNotNull(name, "Metric name must not be null.");
+    this.name = name;
+    this.tags = tags;
+    this.value = 0.0d;
+  }
+
+  // Sync metric with core worker stats for registry.
+  // Metric data will be flushed into stats view data inside core worker immediately after
+  // record is called.
+  /**
+   * Flush records to stats in last aggregator.
+   */
+  public void record() {
+    Preconditions.checkState(metricNativePointer != 0, "Metric native pointer must not be 0.");
+    // Get tag key list from map;
+    List<TagKey> nativeTagKeyList = new ArrayList<>();
+    List<String> tagValues = new ArrayList<>();
+    for (Map.Entry<TagKey, String> entry : tags.entrySet()) {
+      nativeTagKeyList.add(entry.getKey());
+      tagValues.add(entry.getValue());
+    }
+    // Get tag value list from map;
+    NativeMetric.recordNative(metricNativePointer, value, nativeTagKeyList.stream()
+        .map(TagKey::getTagKey).collect(Collectors.toList()), tagValues);
+  }
+
+  /** Update gauge value without tags.
+   * Update metric info for user.
+   * @param value lastest value for updating
+   */
+  public void update(double value) {
+    this.value = value;
+
+  }
+
+  /** Update gauge value with dynamic tag values.
+   * @param value lastest value for updating
+   * @param tags tag map
+   */
+  public void update(double value, Map<TagKey, String> tags) {
+    this.value = value;
+    this.tags = tags;
+  }
+
+  /**
+   * Deallocate object from stats and reset native pointer in null.
+   */
+  public void unregister() {
+    if (0 != metricNativePointer) {
+      NativeMetric.unregisterMetricNative(metricNativePointer);
+    }
+    metricNativePointer = 0;
+  }
+
+  /**
+   * @return lastest updating value.
+   */
+  public double getValue() {
+    return value;
+  }
+
+  /**
+   * It's abstract method for each metric measurements, so metric registry can store transient
+   * value and aggregate historical data for flushing.
+   */
+  public abstract void reset();
+}

java/runtime/src/main/java/io/ray/runtime/metric/NativeMetric.java

@@ -0,0 +1,31 @@
+package io.ray.runtime.metric;
+
+import java.util.List;
+
+/**
+ * Native metric provide a native interface to register tag or metric for current metric package.
+ */
+class NativeMetric {
+  public static native void registerTagkeyNative(String tagKey);
+
+  public static native long registerCountNative(String name, String description,
+                                                String unit, List<String> tagKeys);
+
+  public static native long registerGaugeNative(String name, String description,
+                                                String unit, List<String> tagKeys);
+
+  public static native long registerHistogramNative(String name, String description,
+                                                    String unit, double[] boundaries,
+                                                    List<String> tagKeys);
+
+  public static native long registerSumNative(String name, String description,
+                                              String unit, List<String> tagKeys);
+
+  public static native void recordNative(long metricNativePointer, double value,
+                                         List tagKeys, List<String> tagValues);
+
+  public static native void unregisterMetricNative(long gaugePtr);
+
+
+
+}

java/runtime/src/main/java/io/ray/runtime/metric/Sum.java

@@ -0,0 +1,44 @@
+package io.ray.runtime.metric;
+
+
+import com.google.common.base.Preconditions;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Sum measurement is mapped to sum object in stats.
+ * Property sum is used for storing transient sum for registry aggregation.
+ */
+public class Sum extends Metric {
+  private double sum;
+
+  public Sum(String name, String description, String unit, Map<TagKey, String> tags) {
+    super(name, tags);
+    metricNativePointer = NativeMetric.registerSumNative(name, description, unit,
+      tags.keySet().stream().map(TagKey::getTagKey).collect(Collectors.toList()));
+    Preconditions.checkState(metricNativePointer != 0,"Count native pointer must not be 0.");
+    this.sum = 0.0d;
+  }
+
+  @Override
+  public void update(double value) {
+    super.update(value);
+    sum += value;
+  }
+
+  @Override
+  public void update(double value, Map<TagKey, String> tags) {
+    super.update(value, tags);
+    sum += value;
+  }
+
+  @Override
+  public void reset() {
+
+  }
+
+  public double getSum() {
+    return sum;
+  }
+}

java/runtime/src/main/java/io/ray/runtime/metric/TagKey.java

@@ -0,0 +1,44 @@
+package io.ray.runtime.metric;
+
+import java.util.Objects;
+
+/**
+ * Tagkey is mapping java object to stats tagkey object.
+ */
+public class TagKey {
+
+  private String tagKey;
+
+  public TagKey(String key) {
+    tagKey =  key;
+    NativeMetric.registerTagkeyNative(key);
+  }
+
+  public String getTagKey() {
+    return tagKey;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof TagKey)) {
+      return false;
+    }
+    TagKey tagKey1 = (TagKey) o;
+    return Objects.equals(tagKey, tagKey1.tagKey);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(tagKey);
+  }
+
+  @Override
+  public String toString() {
+    return "TagKey{" +
+      ", tagKey='" + tagKey + '\'' +
+      '}';
+  }
+}

java/test/src/main/java/io/ray/test/MetricTest.java

@@ -0,0 +1,87 @@
+package io.ray.test;
+
+import io.ray.runtime.metric.Count;
+import io.ray.runtime.metric.Gauge;
+import io.ray.runtime.metric.Histogram;
+import io.ray.runtime.metric.Sum;
+import io.ray.runtime.metric.TagKey;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class MetricTest extends BaseTest {
+
+  boolean doubleEqual(double value, double other) {
+    return value <= other + 1e-5 && value >= other - 1e-5;
+  }
+
+  @Test
+  public void testAddGauge() {
+    TestUtils.skipTestUnderSingleProcess();
+    Map<TagKey, String> tags = new HashMap<>();
+    tags.put(new TagKey("tag1"), "value1");
+
+    Gauge gauge = new Gauge("metric1", "", "", tags);
+    gauge.update(2);
+    gauge.record();
+    Assert.assertTrue(doubleEqual(gauge.getValue(), 2.0));
+    gauge.unregister();
+  }
+
+  @Test
+  public void testAddCount() {
+    TestUtils.skipTestUnderSingleProcess();
+    Map<TagKey, String> tags = new HashMap<>();
+    tags.put(new TagKey("tag1"), "value1");
+    tags.put(new TagKey("count_tag"), "default");
+
+    Count count = new Count("metric_count", "counter", "1pc", tags);
+    count.inc(10.0);
+    count.inc(20.0);
+    count.record();
+    Assert.assertTrue(doubleEqual(count.getValue(), 20.0));
+    Assert.assertTrue(doubleEqual(count.getCount(), 30.0));
+  }
+
+  @Test
+  public void testAddSum() {
+    TestUtils.skipTestUnderSingleProcess();
+    Map<TagKey, String> tags = new HashMap<>();
+    tags.put(new TagKey("tag1"), "value1");
+    tags.put(new TagKey("sum_tag"), "default");
+
+    Sum sum = new Sum("metric_sum", "sum", "sum", tags);
+    sum.update(10.0);
+    sum.update(20.0);
+    sum.record();
+    Assert.assertTrue(doubleEqual(sum.getValue(), 20.0));
+    Assert.assertTrue(doubleEqual(sum.getSum(), 30.0));
+  }
+
+  @Test
+  public void testAddHistogram() {
+    TestUtils.skipTestUnderSingleProcess();
+    Map<TagKey, String> tags = new HashMap<>();
+    tags.put(new TagKey("tag1"), "value1");
+    tags.put(new TagKey("histogram_tag"), "default");
+    List<Double> boundaries = new ArrayList<>();
+    boundaries.add(10.0);
+    boundaries.add(15.0);
+    boundaries.add(12.0);
+    Histogram histogram = new Histogram("metric_histogram", "histogram", "1pc",
+        boundaries, tags);
+    for (int i = 1; i <= 200; ++i) {
+      histogram.update(i * 1.0d);
+      histogram.record();
+    }
+    List<Double> window = histogram.getHistogramWindow();
+    for (int i = 0; i < Histogram.HISTOGRAM_WINDOW_SIZE; ++i) {
+      Assert.assertTrue(doubleEqual(i + 101.0d, window.get(i)));
+    }
+  }
+}