code: 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.HashMap;
  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 AtomicDouble 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 = new AtomicDouble();
    }
  
    // 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,
          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 metric value without tags. Update metric info for user.
     *
     * @param value latest value for updating
     */
    public abstract void update(double value);
  
    /**
     * Update metric value with dynamic tag values.
     *
     * @param value latest value for updating
     * @param tags tag map
     */
    public void update(double value, Map<TagKey, String> tags) {
      update(value);
      this.tags = tags;
    }
  
    /**
     * Update metric value with dynamic tag values in pair string format.
     *
     * @param tags tag map
     * @param value latest value for updating
     */
    public void update(Map<String, String> tags, double value) {
      update(value);
      this.tags = TagKey.tagsFromMap(tags);
    }
  
    /**
     * Convert tagkey list to map in string format.
     *
     * @return tags.
     */
    public Map<String, String> getTagMap() {
      Map<String, String> tagMap = new HashMap<>();
      for (Map.Entry<TagKey, String> entry : tags.entrySet()) {
        tagMap.put(entry.getKey().getTagKey(), entry.getValue());
      }
      return tagMap;
    }
  
    /** Deallocate object from stats and reset native pointer in null. */
    public void unregister() {
      if (0 != metricNativePointer) {
        NativeMetric.unregisterMetricNative(metricNativePointer);
      }
      metricNativePointer = 0;
    }
  }
  