[Metrics]Ray java worker metric registry

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

@@ -3,44 +3,39 @@
 import com.google.common.base.Preconditions;
 
 import java.util.Map;
+import java.util.concurrent.atomic.DoubleAdder;
 import java.util.stream.Collectors;
 
+/**
+ * Count measurement is mapped to count object in stats and counts the number.
+ */
 public class Count extends Metric {
 
-  private double count;
+  private DoubleAdder count;
 
   public Count(String name, String description, String unit, Map<TagKey, String> tags) {
     super(name, tags);
-    count = 0.0d;
+    count = new DoubleAdder();
     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;
+    count.add(value);
+    this.value.addAndGet(value);
   }
 
   @Override
-  public void update(double value, Map<TagKey, String> tags) {
-    super.update(value, tags);
-    count += value;
-  }
-
-  @Override
-  public void reset() {
-
+  protected double getAndReset() {
+    return count.sumThenReset();
   }
 
   public double getCount() {
-    return count;
+    return this.value.get();
   }
 
-  /**
-   * @param delta add delta for counter
-   */
   public void inc(double delta) {
     update(delta);
   }

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

@@ -6,7 +6,7 @@
 
 
 /**
- * Gauge metric for recording last value and mapping object from stats.
+ * Gauge measurement is mapped to gauge object in stats and is recording the last value.
  */
 public class Gauge extends Metric {
 
@@ -17,9 +17,18 @@ public Gauge(String name, String description, String unit, Map<TagKey, String> t
     Preconditions.checkState(metricNativePointer != 0, "Gauge native pointer must not be 0.");
   }
 
+  public double getValue() {
+    return value.doubleValue();
+  }
+
   @Override
-  public void reset() {
+  protected double getAndReset() {
+    return value.doubleValue();
+  }
 
+  @Override
+  public void update(double value) {
+    this.value.set(value);
   }
 }
 

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

@@ -3,6 +3,7 @@
 import com.google.common.base.Preconditions;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -25,7 +26,7 @@ public Histogram(String name, String description, String unit, List<Double> boun
       tags.keySet().stream().map(TagKey::getTagKey).collect(Collectors.toList()));
     Preconditions.checkState(metricNativePointer != 0,
         "Histogram native pointer must not be 0.");
-    histogramWindow = new ArrayList<>();
+    histogramWindow = Collections.synchronizedList(new ArrayList<>());
   }
 
   private void updateForWindow(double value) {
@@ -37,22 +38,21 @@ private void updateForWindow(double value) {
 
   @Override
   public void update(double value) {
-    super.update(value);
     updateForWindow(value);
+    this.value.set(value);
   }
 
   @Override
-  public void update(double value, Map<TagKey, String> tags) {
-    super.update(value, tags);
-    updateForWindow(value);
-  }
-
-  @Override
-  public void reset() {
-
+  protected double getAndReset() {
+    histogramWindow.clear();
+    return value.doubleValue();
   }
 
   public List<Double> getHistogramWindow() {
     return histogramWindow;
   }
+
+  public double getValue() {
+    return value.get();
+  }
 }

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

@@ -1,7 +1,7 @@
 package io.ray.runtime.metric;
 
 import com.google.common.base.Preconditions;
-
+import com.google.common.util.concurrent.AtomicDouble;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -14,7 +14,8 @@
 public abstract class Metric {
   protected String name;
 
-  protected double value;
+  protected AtomicDouble value;
+
   // Native pointer mapping to gauge object of stats.
   protected long metricNativePointer = 0L;
 
@@ -25,7 +26,7 @@ public Metric(String name, Map<TagKey, String> tags) {
     Preconditions.checkNotNull(name, "Metric name must not be null.");
     this.name = name;
     this.tags = tags;
-    this.value = 0.0d;
+    this.value = new AtomicDouble();
   }
 
   // Sync metric with core worker stats for registry.
@@ -44,25 +45,28 @@ public void record() {
       tagValues.add(entry.getValue());
     }
     // Get tag value list from map;
-    NativeMetric.recordNative(metricNativePointer, value, nativeTagKeyList.stream()
+    NativeMetric.recordNative(metricNativePointer, getAndReset(), nativeTagKeyList.stream()
         .map(TagKey::getTagKey).collect(Collectors.toList()), tagValues);
   }
 
+  /**
+   * Get the value to record and then reset.
+   * @return latest updating value.
+   */
+  protected abstract double getAndReset();
+
   /** Update gauge value without tags.
    * Update metric info for user.
-   * @param value lastest value for updating
+   * @param value latest value for updating
    */
-  public void update(double value) {
-    this.value = value;
-
-  }
+  public abstract void update(double value);
 
   /** Update gauge value with dynamic tag values.
-   * @param value lastest value for updating
+   * @param value latest value for updating
    * @param tags tag map
    */
   public void update(double value, Map<TagKey, String> tags) {
-    this.value = value;
+    update(value);
     this.tags = tags;
   }
 
@@ -76,16 +80,4 @@ public void unregister() {
     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/MetricConfig.java

@@ -0,0 +1,79 @@
+package io.ray.runtime.metric;
+
+import com.google.common.base.MoreObjects;
+
+/**
+ * Configurations of the metric.
+ */
+public class MetricConfig {
+
+  private static final long DEFAULT_TIME_INTERVAL_MS = 5000L;
+  private static final int DEFAULT_THREAD_POLL_SIZE = 1;
+  private static final long DEFAULT_SHUTDOWN_WAIT_TIME_MS = 3000L;
+
+  public static final MetricConfig DEFAULT_CONFIG =
+    new MetricConfig(DEFAULT_TIME_INTERVAL_MS, DEFAULT_THREAD_POLL_SIZE,
+      DEFAULT_SHUTDOWN_WAIT_TIME_MS);
+
+  private final long timeIntervalMs;
+  private final int threadPoolSize;
+  private final long shutdownWaitTimeMs;
+
+  public MetricConfig(long timeIntervalMs, int threadPoolSize, long shutdownWaitTimeMs) {
+    this.timeIntervalMs = timeIntervalMs;
+    this.threadPoolSize = threadPoolSize;
+    this.shutdownWaitTimeMs = shutdownWaitTimeMs;
+  }
+
+  public long timeIntervalMs() {
+    return timeIntervalMs;
+  }
+
+  public int threadPoolSize() {
+    return threadPoolSize;
+  }
+
+  public long shutdownWaitTimeMs() {
+    return shutdownWaitTimeMs;
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+      .add("timeIntervalMs", timeIntervalMs)
+      .add("threadPoolSize", threadPoolSize)
+      .add("shutdownWaitTimeMs", shutdownWaitTimeMs)
+      .toString();
+  }
+
+  public static MetricConfigBuilder builder() {
+    return new MetricConfigBuilder();
+  }
+
+  public static class MetricConfigBuilder {
+    private long timeIntervalMs = DEFAULT_TIME_INTERVAL_MS;
+    private int threadPooSize = DEFAULT_THREAD_POLL_SIZE;
+    private long shutdownWaitTimeMs = DEFAULT_SHUTDOWN_WAIT_TIME_MS;
+
+    public MetricConfig create() {
+      return new MetricConfig(timeIntervalMs, threadPooSize, shutdownWaitTimeMs);
+    }
+
+    public MetricConfigBuilder timeIntervalMs(long timeIntervalMs) {
+      this.timeIntervalMs = timeIntervalMs;
+      return this;
+    }
+
+    public MetricConfigBuilder threadPoolSize(int threadPooSize) {
+      this.threadPooSize = threadPooSize;
+      return this;
+    }
+
+    public MetricConfigBuilder shutdownWaitTimeMs(long shutdownWaitTimeMs) {
+      this.shutdownWaitTimeMs = shutdownWaitTimeMs;
+      return this;
+    }
+  }
+
+
+}

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

@@ -0,0 +1,64 @@
+package io.ray.runtime.metric;
+
+import com.google.common.base.MoreObjects;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * MetricId represents a metric with a given type, name and tags.
+ * If two metrics have the same type and name but different tags(including key and value), they have
+ * a different MetricId. And in this way, {@link MetricRegistry} can register two metrics with same
+ * name but different tags.
+ */
+public class MetricId {
+
+  private final MetricType type;
+  private final String name;
+  private final Map<TagKey, String> tags;
+
+  public MetricId(MetricType type, String name, Map<TagKey, String> tags) {
+    this.type = type;
+    this.name = name;
+    this.tags = tags;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof MetricId)) {
+      return false;
+    }
+    MetricId metricId = (MetricId) o;
+    return type == metricId.type &&
+      Objects.equals(name, metricId.name) &&
+      Objects.equals(tags, metricId.tags);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(type, name, tags);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+      .add("type", type)
+      .add("name", name)
+      .add("tags", tags)
+      .toString();
+  }
+
+  public MetricType getType() {
+    return type;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public Map<TagKey, String> getTags() {
+    return tags;
+  }
+}