Metrics: Produce and store Metrics in Stats impl. (#1338)

Support recording `Metric`s in stats impl. This works as the following:
- When a `View` is registered, convert that `View` to a `MetricDescriptor` and register it;
- When a `Measurement` along with a `TagContext` are recorded, convert them into a data row consisting of `LabelValue`s and `Point`s;
- If `getMetrics()` is called, convert the `MetricDescriptor`s and data rows into `Metric`s. Then flush all data rows.

This PR only contains internal support in the impl. None of these changes are user-visible yet.

7 files changed, 712 insertions, 7 deletions

diff --git a/impl_core/src/main/java/io/opencensus/implcore/stats/MeasureToViewMap.java b/impl_core/src/main/java/io/opencensus/implcore/stats/MeasureToViewMap.java
index 79e913db..0cd5045f 100644
--- a/impl_core/src/main/java/io/opencensus/implcore/stats/MeasureToViewMap.java
+++ b/impl_core/src/main/java/io/opencensus/implcore/stats/MeasureToViewMap.java
@@ -22,6 +22,7 @@ import com.google.common.collect.Multimap;
 import com.google.common.collect.Sets;
 import io.opencensus.common.Clock;
 import io.opencensus.common.Timestamp;
+import io.opencensus.metrics.MetricDescriptor;
 import io.opencensus.stats.Measure;
 import io.opencensus.stats.Measurement;
 import io.opencensus.stats.StatsCollectionState;
@@ -60,6 +61,9 @@ final class MeasureToViewMap {
   @GuardedBy("this")
   private final Map<String, Measure> registeredMeasures = Maps.newHashMap();
 
+  @GuardedBy("this")
+  private final MetricMap metricMap = new MetricMap();
+
   // Cached set of exported views. It must be set to null whenever a view is registered or
   // unregistered.
   @javax.annotation.Nullable private volatile Set<View> exportedViews;
@@ -115,7 +119,12 @@ final class MeasureToViewMap {
     if (registeredMeasure == null) {
       registeredMeasures.put(measure.getName(), measure);
     }
-    mutableMap.put(view.getMeasure().getName(), MutableViewData.create(view, clock.now()));
+    Timestamp now = clock.now();
+    mutableMap.put(view.getMeasure().getName(), MutableViewData.create(view, now, metricMap));
+    MetricDescriptor metricDescriptor = MetricUtils.viewToMetricDescriptor(view);
+    if (metricDescriptor != null) {
+      metricMap.registerMetricDescriptor(metricDescriptor, now);
+    }
   }
 
   @javax.annotation.Nullable
@@ -165,6 +174,7 @@ final class MeasureToViewMap {
         mutableViewData.clearStats();
       }
     }
+    metricMap.clearStats();
   }
 
   // Resume stats collection for all MutableViewData.
@@ -174,5 +184,6 @@ final class MeasureToViewMap {
         mutableViewData.resumeStatsCollection(now);
       }
     }
+    metricMap.resumeStatsCollection(now);
   }
 }
diff --git a/impl_core/src/main/java/io/opencensus/implcore/stats/MetricMap.java b/impl_core/src/main/java/io/opencensus/implcore/stats/MetricMap.java
new file mode 100644
index 00000000..5ae2eed5
--- /dev/null
+++ b/impl_core/src/main/java/io/opencensus/implcore/stats/MetricMap.java
@@ -0,0 +1,197 @@
+/*
+ * Copyright 2018, OpenCensus Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opencensus.implcore.stats;
+
+import io.opencensus.common.Timestamp;
+import io.opencensus.metrics.LabelValue;
+import io.opencensus.metrics.Metric;
+import io.opencensus.metrics.MetricDescriptor;
+import io.opencensus.metrics.MetricDescriptor.Type;
+import io.opencensus.metrics.Point;
+import io.opencensus.metrics.TimeSeriesCumulative;
+import io.opencensus.metrics.TimeSeriesGauge;
+import io.opencensus.metrics.TimeSeriesList;
+import io.opencensus.metrics.TimeSeriesList.TimeSeriesCumulativeList;
+import io.opencensus.metrics.TimeSeriesList.TimeSeriesGaugeList;
+import io.opencensus.tags.TagValue;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.concurrent.GuardedBy;
+
+/*>>>
+import org.checkerframework.checker.nullness.qual.Nullable;
+*/
+
+// A class that stores a mapping from MetricDescriptor to lists of MutableMetricRows.
+final class MetricMap {
+
+  @GuardedBy("this")
+  private final Map<MetricDescriptor, MutableMetricRows> map =
+      new HashMap<MetricDescriptor, MutableMetricRows>();
+
+  // Registers a MetricDescriptor, creates an entry in the map.
+  // This method should only be called from MeasureToMap.registerView().
+  synchronized void registerMetricDescriptor(
+      MetricDescriptor metricDescriptor, Timestamp timestamp) {
+    if (map.containsKey(metricDescriptor)) {
+      return;
+    }
+    map.put(metricDescriptor, MutableMetricRows.create(metricDescriptor.getType(), timestamp));
+  }
+
+  // Records a data point to this MetricMap.
+  // This method should only be called from CumulativeMutableViewData.record().
+  synchronized void record(
+      MetricDescriptor metricDescriptor,
+      List</*@Nullable*/ TagValue> tagValues,
+      MutableAggregation mutableAggregation,
+      Timestamp now) {
+    if (!map.containsKey(metricDescriptor)) {
+      return;
+    }
+    map.get(metricDescriptor)
+        .record(tagValues, mutableAggregation, now, metricDescriptor.getType());
+  }
+
+  synchronized void clearStats() {
+    for (Entry<MetricDescriptor, MutableMetricRows> entry : map.entrySet()) {
+      entry.getValue().map.clear();
+    }
+  }
+
+  synchronized void resumeStatsCollection(Timestamp now) {
+    for (Entry<MetricDescriptor, MutableMetricRows> entry : map.entrySet()) {
+      entry.getValue().resumeStatsCollection(now);
+    }
+  }
+
+  synchronized List<Metric> toMetrics() {
+    List<Metric> metrics = new ArrayList<Metric>();
+    for (Entry<MetricDescriptor, MutableMetricRows> entry : map.entrySet()) {
+      MutableMetricRows mutableMetricRows = entry.getValue();
+      if (mutableMetricRows.map.isEmpty()) {
+        continue; // Skip MetricDescriptor with no data.
+      }
+      metrics.add(Metric.create(entry.getKey(), mutableMetricRows.toTimeSeriesList()));
+
+      // Reset the data map once the rows are exported, so that we don't export duplicated Points.
+      mutableMetricRows.map.clear();
+    }
+    return metrics;
+  }
+
+  // A class that stores a mapping from lists of label values to lists of points.
+  // Each MutableMetricRows correspond to one MetricDescriptor.
+  // Think of this class as a set of mutable time series.
+  private abstract static class MutableMetricRows {
+
+    /*
+     * Each entry in this map is a list of rows, for example:
+     *   [v1, v2] -> [1, 5, 10]
+     *   [v1, v3] -> [-5, -8]
+     *   ...
+     */
+    private final Map<List<LabelValue>, List<Point>> map =
+        new LinkedHashMap<List<LabelValue>, List<Point>>();
+
+    // Create MutableMetricRows based on the given type.
+    private static MutableMetricRows create(Type type, Timestamp timestamp) {
+      switch (type) {
+        case GAUGE_INT64:
+        case GAUGE_DOUBLE:
+          return createGauge();
+        case CUMULATIVE_DISTRIBUTION:
+        case CUMULATIVE_DOUBLE:
+        case CUMULATIVE_INT64:
+          return createCumulative(timestamp);
+      }
+      throw new AssertionError();
+    }
+
+    private static MutableMetricRows createCumulative(Timestamp timestamp) {
+      return new MutableMetricRowsCumulative(timestamp);
+    }
+
+    private static MutableMetricRows createGauge() {
+      return new MutableMetricRowsGauge();
+    }
+
+    private void record(
+        List</*@Nullable*/ TagValue> tagValues,
+        MutableAggregation mutableAggregation,
+        Timestamp timestamp,
+        Type type) {
+      List<LabelValue> labelValues = MetricUtils.tagValuesToLabelValues(tagValues);
+      Point point = MetricUtils.mutableAggregationToPoint(mutableAggregation, timestamp, type);
+      if (!map.containsKey(labelValues)) {
+        map.put(labelValues, new ArrayList<Point>());
+      }
+      map.get(labelValues).add(point);
+    }
+
+    abstract TimeSeriesList toTimeSeriesList();
+
+    abstract void resumeStatsCollection(Timestamp now);
+
+    private static final class MutableMetricRowsCumulative extends MutableMetricRows {
+
+      // Only cumulative time series has a start timestamp.
+      private Timestamp startTime;
+
+      private MutableMetricRowsCumulative(Timestamp startTime) {
+        this.startTime = startTime;
+      }
+
+      @Override
+      TimeSeriesList toTimeSeriesList() {
+        List<TimeSeriesCumulative> timeSeriesCumulatives = new ArrayList<TimeSeriesCumulative>();
+        for (Entry<List<LabelValue>, List<Point>> entry : super.map.entrySet()) {
+          timeSeriesCumulatives.add(
+              TimeSeriesCumulative.create(entry.getKey(), entry.getValue(), startTime));
+        }
+        return TimeSeriesCumulativeList.create(timeSeriesCumulatives);
+      }
+
+      @Override
+      void resumeStatsCollection(Timestamp now) {
+        // Reset start time to current time.
+        this.startTime = now;
+      }
+    }
+
+    private static final class MutableMetricRowsGauge extends MutableMetricRows {
+
+      @Override
+      TimeSeriesList toTimeSeriesList() {
+        List<TimeSeriesGauge> timeSeriesGauges = new ArrayList<TimeSeriesGauge>();
+        for (Entry<List<LabelValue>, List<Point>> entry : super.map.entrySet()) {
+          timeSeriesGauges.add(TimeSeriesGauge.create(entry.getKey(), entry.getValue()));
+        }
+        return TimeSeriesGaugeList.create(timeSeriesGauges);
+      }
+
+      @Override
+      void resumeStatsCollection(Timestamp now) {
+        // Do nothing for Gauge stats.
+      }
+    }
+  }
+}
diff --git a/impl_core/src/main/java/io/opencensus/implcore/stats/MetricUtils.java b/impl_core/src/main/java/io/opencensus/implcore/stats/MetricUtils.java
new file mode 100644
index 00000000..8e770458
--- /dev/null
+++ b/impl_core/src/main/java/io/opencensus/implcore/stats/MetricUtils.java
@@ -0,0 +1,185 @@
+/*
+ * Copyright 2018, OpenCensus Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opencensus.implcore.stats;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.opencensus.common.Function;
+import io.opencensus.common.Functions;
+import io.opencensus.common.Timestamp;
+import io.opencensus.metrics.Distribution;
+import io.opencensus.metrics.LabelKey;
+import io.opencensus.metrics.LabelValue;
+import io.opencensus.metrics.MetricDescriptor;
+import io.opencensus.metrics.MetricDescriptor.Type;
+import io.opencensus.metrics.Point;
+import io.opencensus.metrics.Value;
+import io.opencensus.stats.Aggregation;
+import io.opencensus.stats.AggregationData;
+import io.opencensus.stats.Measure;
+import io.opencensus.stats.View;
+import io.opencensus.tags.TagKey;
+import io.opencensus.tags.TagValue;
+import java.util.ArrayList;
+import java.util.List;
+
+/*>>>
+import org.checkerframework.checker.nullness.qual.Nullable;
+*/
+
+@SuppressWarnings("deprecation")
+// Utils to convert Stats data models to Metric data models.
+final class MetricUtils {
+
+  @javax.annotation.Nullable
+  static MetricDescriptor viewToMetricDescriptor(View view) {
+    if (view.getWindow() instanceof View.AggregationWindow.Interval) {
+      // Only creates Metric for cumulative stats.
+      return null;
+    }
+    List<LabelKey> labelKeys = new ArrayList<LabelKey>();
+    for (TagKey tagKey : view.getColumns()) {
+      // TODO: add description
+      labelKeys.add(LabelKey.create(tagKey.getName(), ""));
+    }
+    Measure measure = view.getMeasure();
+    return MetricDescriptor.create(
+        view.getName().asString(),
+        view.getDescription(),
+        measure.getUnit(),
+        getType(measure, view.getAggregation()),
+        labelKeys);
+  }
+
+  @VisibleForTesting
+  static Type getType(Measure measure, Aggregation aggregation) {
+    return aggregation.match(
+        Functions.returnConstant(
+            measure.match(
+                Functions.returnConstant(Type.CUMULATIVE_DOUBLE), // Sum Double
+                Functions.returnConstant(Type.CUMULATIVE_INT64), // Sum Int64
+                Functions.<Type>throwAssertionError())),
+        Functions.returnConstant(Type.CUMULATIVE_INT64), // Count
+        Functions.returnConstant(Type.CUMULATIVE_DISTRIBUTION), // Distribution
+        Functions.returnConstant(
+            measure.match(
+                Functions.returnConstant(Type.GAUGE_DOUBLE), // LastValue Double
+                Functions.returnConstant(Type.GAUGE_INT64), // LastValue Long
+                Functions.<Type>throwAssertionError())),
+        new Function<Aggregation, Type>() {
+          @Override
+          public Type apply(Aggregation arg) {
+            if (arg instanceof Aggregation.Mean) {
+              return Type.CUMULATIVE_DOUBLE; // Mean
+            }
+            throw new AssertionError();
+          }
+        });
+  }
+
+  static List<LabelValue> tagValuesToLabelValues(List</*@Nullable*/ TagValue> tagValues) {
+    List<LabelValue> labelValues = new ArrayList<LabelValue>();
+    for (/*@Nullable*/ TagValue tagValue : tagValues) {
+      labelValues.add(LabelValue.create(tagValue == null ? null : tagValue.asString()));
+    }
+    return labelValues;
+  }
+
+  static Point mutableAggregationToPoint(
+      MutableAggregation mutableAggregation, final Timestamp timestamp, final Type type) {
+    return mutableAggregation.match(
+        new Function<MutableAggregation.MutableSum, Point>() {
+          @Override
+          public Point apply(MutableAggregation.MutableSum arg) {
+            return Point.create(toDoubleOrInt64Value(arg.getSum(), type), timestamp);
+          }
+        },
+        new Function<MutableAggregation.MutableCount, Point>() {
+          @Override
+          public Point apply(MutableAggregation.MutableCount arg) {
+            return Point.create(Value.longValue(arg.getCount()), timestamp);
+          }
+        },
+        new Function<MutableAggregation.MutableMean, Point>() {
+          @Override
+          public Point apply(MutableAggregation.MutableMean arg) {
+            return Point.create(toDoubleOrInt64Value(arg.getMean(), type), timestamp);
+          }
+        },
+        new Function<MutableAggregation.MutableDistribution, Point>() {
+          @Override
+          public Point apply(MutableAggregation.MutableDistribution arg) {
+            return Point.create(toDistributionValue(arg), timestamp);
+          }
+        },
+        new Function<MutableAggregation.MutableLastValue, Point>() {
+          @Override
+          public Point apply(MutableAggregation.MutableLastValue arg) {
+            return Point.create(toDoubleOrInt64Value(arg.getLastValue(), type), timestamp);
+          }
+        });
+  }
+
+  private static Value toDoubleOrInt64Value(double valueInDouble, Type type) {
+    switch (type) {
+      case CUMULATIVE_INT64:
+      case GAUGE_INT64:
+        return Value.longValue(Math.round(valueInDouble));
+      case CUMULATIVE_DOUBLE:
+      case GAUGE_DOUBLE:
+        return Value.doubleValue(valueInDouble);
+      case CUMULATIVE_DISTRIBUTION:
+        throw new AssertionError();
+    }
+    throw new AssertionError();
+  }
+
+  private static Value toDistributionValue(MutableAggregation.MutableDistribution distribution) {
+    List<Distribution.Bucket> buckets = new ArrayList<Distribution.Bucket>();
+    @javax.annotation.Nullable
+    AggregationData.DistributionData.Exemplar[] exemplars = distribution.getExemplars();
+    for (int bucket = 0; bucket < distribution.getBucketCounts().length; bucket++) {
+      long bucketCount = distribution.getBucketCounts()[bucket];
+      @javax.annotation.Nullable AggregationData.DistributionData.Exemplar exemplar = null;
+      if (exemplars != null) {
+        exemplar = exemplars[bucket];
+      }
+
+      Distribution.Bucket metricBucket;
+      if (exemplar != null) {
+        // Bucket with an Exemplar.
+        metricBucket =
+            Distribution.Bucket.create(
+                bucketCount,
+                Distribution.Exemplar.create(
+                    exemplar.getValue(), exemplar.getTimestamp(), exemplar.getAttachments()));
+      } else {
+        // Bucket with no Exemplar.
+        metricBucket = Distribution.Bucket.create(bucketCount);
+      }
+      buckets.add(metricBucket);
+    }
+    return Value.distributionValue(
+        Distribution.create(
+            distribution.getMean(),
+            distribution.getCount(),
+            distribution.getSumOfSquaredDeviations(),
+            distribution.getBucketBoundaries().getBoundaries(),
+            buckets));
+  }
+
+  private MetricUtils() {}
+}
diff --git a/impl_core/src/main/java/io/opencensus/implcore/stats/MutableAggregation.java b/impl_core/src/main/java/io/opencensus/implcore/stats/MutableAggregation.java
index ddfce618..770ec4b2 100644
--- a/impl_core/src/main/java/io/opencensus/implcore/stats/MutableAggregation.java
+++ b/impl_core/src/main/java/io/opencensus/implcore/stats/MutableAggregation.java
@@ -387,6 +387,10 @@ abstract class MutableAggregation {
       return bucketCounts;
     }
 
+    BucketBoundaries getBucketBoundaries() {
+      return bucketBoundaries;
+    }
+
     @javax.annotation.Nullable
     Exemplar[] getExemplars() {
       return exemplars;
diff --git a/impl_core/src/main/java/io/opencensus/implcore/stats/MutableViewData.java b/impl_core/src/main/java/io/opencensus/implcore/stats/MutableViewData.java
index 876415a3..0664681c 100644
--- a/impl_core/src/main/java/io/opencensus/implcore/stats/MutableViewData.java
+++ b/impl_core/src/main/java/io/opencensus/implcore/stats/MutableViewData.java
@@ -31,6 +31,7 @@ import io.opencensus.common.Function;
 import io.opencensus.common.Functions;
 import io.opencensus.common.Timestamp;
 import io.opencensus.implcore.internal.CheckerFrameworkUtils;
+import io.opencensus.metrics.MetricDescriptor;
 import io.opencensus.stats.Aggregation;
 import io.opencensus.stats.AggregationData;
 import io.opencensus.stats.StatsCollectionState;
@@ -65,12 +66,13 @@ abstract class MutableViewData {
    *
    * @param view the {@code View} linked with this {@code MutableViewData}.
    * @param start the start {@code Timestamp}.
+   * @param metricMap a reference to {@code MetricMap}.
    * @return a {@code MutableViewData}.
    */
-  static MutableViewData create(final View view, final Timestamp start) {
+  static MutableViewData create(final View view, final Timestamp start, final MetricMap metricMap) {
     return view.getWindow()
         .match(
-            new CreateCumulative(view, start),
+            new CreateCumulative(view, start, metricMap),
             new CreateInterval(view, start),
             Functions.<MutableViewData>throwAssertionError());
   }
@@ -99,10 +101,22 @@ abstract class MutableViewData {
     private Timestamp start;
     private final Map<List</*@Nullable*/ TagValue>, MutableAggregation> tagValueAggregationMap =
         Maps.newHashMap();
+    private final MetricMap metricMap;
 
-    private CumulativeMutableViewData(View view, Timestamp start) {
+    // Cache a MetricDescriptor to avoid converting View to MetricDescriptor in the future.
+    private final MetricDescriptor metricDescriptor;
+
+    private CumulativeMutableViewData(View view, Timestamp start, MetricMap metricMap) {
       super(view);
       this.start = start;
+      this.metricMap = metricMap;
+      MetricDescriptor metricDescriptor = MetricUtils.viewToMetricDescriptor(view);
+      if (metricDescriptor == null) {
+        throw new AssertionError(
+            "Cumulative view should be converted to a non-null MetricDescriptor.");
+      } else {
+        this.metricDescriptor = metricDescriptor;
+      }
     }
 
     @Override
@@ -114,7 +128,9 @@ abstract class MutableViewData {
         tagValueAggregationMap.put(
             tagValues, createMutableAggregation(super.view.getAggregation()));
       }
-      tagValueAggregationMap.get(tagValues).add(value, attachments, timestamp);
+      MutableAggregation mutableAggregation = tagValueAggregationMap.get(tagValues);
+      mutableAggregation.add(value, attachments, timestamp);
+      metricMap.record(metricDescriptor, tagValues, mutableAggregation, timestamp);
     }
 
     @Override
@@ -385,15 +401,17 @@ abstract class MutableViewData {
       implements Function<View.AggregationWindow.Cumulative, MutableViewData> {
     @Override
     public MutableViewData apply(View.AggregationWindow.Cumulative arg) {
-      return new CumulativeMutableViewData(view, start);
+      return new CumulativeMutableViewData(view, start, metricMap);
     }
 
     private final View view;
     private final Timestamp start;
+    private final MetricMap metricMap;
 
-    private CreateCumulative(View view, Timestamp start) {
+    private CreateCumulative(View view, Timestamp start, MetricMap metricMap) {
       this.view = view;
       this.start = start;
+      this.metricMap = metricMap;
     }
   }
 
diff --git a/impl_core/src/test/java/io/opencensus/implcore/stats/MetricMapTest.java b/impl_core/src/test/java/io/opencensus/implcore/stats/MetricMapTest.java
new file mode 100644
index 00000000..3ab5520e
--- /dev/null
+++ b/impl_core/src/test/java/io/opencensus/implcore/stats/MetricMapTest.java
@@ -0,0 +1,115 @@
+/*
+ * Copyright 2018, OpenCensus Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opencensus.implcore.stats;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import io.opencensus.common.Timestamp;
+import io.opencensus.implcore.stats.MutableAggregation.MutableLastValue;
+import io.opencensus.metrics.LabelKey;
+import io.opencensus.metrics.LabelValue;
+import io.opencensus.metrics.Metric;
+import io.opencensus.metrics.MetricDescriptor;
+import io.opencensus.metrics.MetricDescriptor.Type;
+import io.opencensus.metrics.Point;
+import io.opencensus.metrics.TimeSeriesGauge;
+import io.opencensus.metrics.TimeSeriesList.TimeSeriesGaugeList;
+import io.opencensus.metrics.Value;
+import io.opencensus.tags.TagValue;
+import java.util.Arrays;
+import java.util.Collections;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link MetricMap}. */
+@RunWith(JUnit4.class)
+public class MetricMapTest {
+
+  private static final TagValue VALUE = TagValue.create("VALUE");
+  private static final TagValue VALUE_2 = TagValue.create("VALUE_2");
+  private static final TagValue VALUE_3 = TagValue.create("VALUE_3");
+  private static final Timestamp TIMESTAMP_EMPTY = Timestamp.create(0, 0);
+
+  private static final LabelKey LABEL_KEY = LabelKey.create("KEY", "");
+  private static final LabelValue LABEL_VALUE = LabelValue.create("VALUE");
+  private static final LabelValue LABEL_VALUE_2 = LabelValue.create("VALUE_2");
+  private static final LabelValue LABEL_VALUE_3 = LabelValue.create("VALUE_3");
+  private static final MetricDescriptor DESCRIPTOR =
+      MetricDescriptor.create(
+          "name", "description", "1", Type.GAUGE_DOUBLE, Collections.singletonList(LABEL_KEY));
+
+  @Test
+  public void noDataForMetricDescriptor() {
+    MetricMap metricMap = new MetricMap();
+    metricMap.registerMetricDescriptor(DESCRIPTOR, TIMESTAMP_EMPTY);
+    assertThat(metricMap.toMetrics()).isEmpty();
+  }
+
+  @Test
+  public void recordAndExport() {
+    MetricMap metricMap = new MetricMap();
+    Timestamp timestamp1 = Timestamp.fromMillis(1000);
+    Timestamp timestamp2 = Timestamp.fromMillis(2000);
+    Timestamp timestamp3 = Timestamp.fromMillis(3000);
+    Timestamp timestamp4 = Timestamp.fromMillis(4000);
+    metricMap.registerMetricDescriptor(DESCRIPTOR, timestamp1);
+
+    MutableLastValue lastValue1 = MutableLastValue.create();
+    lastValue1.add(10.0, Collections.<String, String>emptyMap(), timestamp2);
+    metricMap.record(DESCRIPTOR, Collections.singletonList(VALUE), lastValue1, timestamp2);
+    lastValue1.add(19.0, Collections.<String, String>emptyMap(), timestamp3);
+    metricMap.record(DESCRIPTOR, Collections.singletonList(VALUE), lastValue1, timestamp3);
+
+    MutableLastValue lastValue2 = MutableLastValue.create();
+    lastValue2.add(-7.0, Collections.<String, String>emptyMap(), timestamp3);
+    metricMap.record(DESCRIPTOR, Collections.singletonList(VALUE_2), lastValue2, timestamp3);
+
+    TimeSeriesGauge expectedTimeSeries1 =
+        TimeSeriesGauge.create(
+            Collections.singletonList(LABEL_VALUE),
+            Arrays.asList(
+                Point.create(Value.doubleValue(10.0), timestamp2),
+                Point.create(Value.doubleValue(19.0), timestamp3)));
+    TimeSeriesGauge expectedTimeSeries2 =
+        TimeSeriesGauge.create(
+            Collections.singletonList(LABEL_VALUE_2),
+            Collections.singletonList(Point.create(Value.doubleValue(-7.0), timestamp3)));
+    assertThat(metricMap.toMetrics())
+        .containsExactly(
+            Metric.create(
+                DESCRIPTOR,
+                TimeSeriesGaugeList.create(
+                    Arrays.asList(expectedTimeSeries1, expectedTimeSeries2))));
+
+    assertThat(metricMap.toMetrics()).isEmpty(); // Flush stats after exporting.
+
+    // Record another Measurement, should only produce one Point.
+    lastValue1.add(6.0, Collections.<String, String>emptyMap(), timestamp4);
+    metricMap.record(DESCRIPTOR, Collections.singletonList(VALUE_3), lastValue1, timestamp4);
+    assertThat(metricMap.toMetrics())
+        .containsExactly(
+            Metric.create(
+                DESCRIPTOR,
+                TimeSeriesGaugeList.create(
+                    Collections.singletonList(
+                        TimeSeriesGauge.create(
+                            Collections.singletonList(LABEL_VALUE_3),
+                            Collections.singletonList(
+                                Point.create(Value.doubleValue(6.0), timestamp4)))))));
+  }
+}
diff --git a/impl_core/src/test/java/io/opencensus/implcore/stats/MetricUtilsTest.java b/impl_core/src/test/java/io/opencensus/implcore/stats/MetricUtilsTest.java
new file mode 100644
index 00000000..bc6f4ca2
--- /dev/null
+++ b/impl_core/src/test/java/io/opencensus/implcore/stats/MetricUtilsTest.java
@@ -0,0 +1,175 @@
+/*
+ * Copyright 2018, OpenCensus Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opencensus.implcore.stats;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import io.opencensus.common.Duration;
+import io.opencensus.common.Timestamp;
+import io.opencensus.implcore.stats.MutableAggregation.MutableCount;
+import io.opencensus.implcore.stats.MutableAggregation.MutableDistribution;
+import io.opencensus.implcore.stats.MutableAggregation.MutableLastValue;
+import io.opencensus.implcore.stats.MutableAggregation.MutableMean;
+import io.opencensus.implcore.stats.MutableAggregation.MutableSum;
+import io.opencensus.metrics.Distribution.Bucket;
+import io.opencensus.metrics.LabelKey;
+import io.opencensus.metrics.LabelValue;
+import io.opencensus.metrics.MetricDescriptor;
+import io.opencensus.metrics.MetricDescriptor.Type;
+import io.opencensus.metrics.Point;
+import io.opencensus.metrics.Value;
+import io.opencensus.stats.Aggregation.Count;
+import io.opencensus.stats.Aggregation.Distribution;
+import io.opencensus.stats.Aggregation.LastValue;
+import io.opencensus.stats.Aggregation.Mean;
+import io.opencensus.stats.Aggregation.Sum;
+import io.opencensus.stats.BucketBoundaries;
+import io.opencensus.stats.Measure.MeasureDouble;
+import io.opencensus.stats.Measure.MeasureLong;
+import io.opencensus.stats.View;
+import io.opencensus.stats.View.AggregationWindow.Interval;
+import io.opencensus.stats.View.Name;
+import io.opencensus.tags.TagKey;
+import io.opencensus.tags.TagValue;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link MetricUtils}. */
+@RunWith(JUnit4.class)
+public class MetricUtilsTest {
+
+  private static final TagKey KEY = TagKey.create("KEY");
+  private static final TagValue VALUE = TagValue.create("VALUE");
+  private static final TagValue VALUE_2 = TagValue.create("VALUE_2");
+  private static final String MEASURE_NAME = "my measurement";
+  private static final String MEASURE_NAME_2 = "my measurement 2";
+  private static final String MEASURE_UNIT = "us";
+  private static final String MEASURE_DESCRIPTION = "measure description";
+  private static final MeasureDouble MEASURE_DOUBLE =
+      MeasureDouble.create(MEASURE_NAME, MEASURE_DESCRIPTION, MEASURE_UNIT);
+  private static final MeasureLong MEASURE_LONG =
+      MeasureLong.create(MEASURE_NAME_2, MEASURE_DESCRIPTION, MEASURE_UNIT);
+  private static final Name VIEW_NAME = Name.create("my view");
+  private static final Name VIEW_NAME_2 = Name.create("my view 2");
+  private static final String VIEW_DESCRIPTION = "view description";
+  private static final Duration TEN_SECONDS = Duration.create(10, 0);
+  private static final Interval INTERVAL = Interval.create(TEN_SECONDS);
+  private static final BucketBoundaries BUCKET_BOUNDARIES =
+      BucketBoundaries.create(Arrays.asList(-10.0, 0.0, 10.0));
+  private static final Sum SUM = Sum.create();
+  private static final Count COUNT = Count.create();
+  private static final Mean MEAN = Mean.create();
+  private static final Distribution DISTRIBUTION = Distribution.create(BUCKET_BOUNDARIES);
+  private static final LastValue LAST_VALUE = LastValue.create();
+  private static final View VIEW_1 =
+      View.create(
+          VIEW_NAME, VIEW_DESCRIPTION, MEASURE_DOUBLE, LAST_VALUE, Collections.singletonList(KEY));
+  private static final View VIEW_2 =
+      View.create(
+          VIEW_NAME_2,
+          VIEW_DESCRIPTION,
+          MEASURE_DOUBLE,
+          MEAN,
+          Collections.singletonList(KEY),
+          INTERVAL);
+  private static final Timestamp TIMESTAMP = Timestamp.fromMillis(1000);
+
+  @Test
+  public void viewToMetricDescriptor() {
+    MetricDescriptor metricDescriptor = MetricUtils.viewToMetricDescriptor(VIEW_1);
+    assertThat(metricDescriptor).isNotNull();
+    assertThat(metricDescriptor.getName()).isEqualTo(VIEW_NAME.asString());
+    assertThat(metricDescriptor.getUnit()).isEqualTo(MEASURE_UNIT);
+    assertThat(metricDescriptor.getType()).isEqualTo(Type.GAUGE_DOUBLE);
+    assertThat(metricDescriptor.getDescription()).isEqualTo(VIEW_DESCRIPTION);
+    assertThat(metricDescriptor.getLabelKeys()).containsExactly(LabelKey.create(KEY.getName(), ""));
+  }
+
+  @Test
+  public void viewToMetricDescriptor_NoIntervalViews() {
+    MetricDescriptor metricDescriptor = MetricUtils.viewToMetricDescriptor(VIEW_2);
+    assertThat(metricDescriptor).isNull();
+  }
+
+  @Test
+  public void getType() {
+    assertThat(MetricUtils.getType(MEASURE_DOUBLE, LAST_VALUE)).isEqualTo(Type.GAUGE_DOUBLE);
+    assertThat(MetricUtils.getType(MEASURE_LONG, LAST_VALUE)).isEqualTo(Type.GAUGE_INT64);
+    assertThat(MetricUtils.getType(MEASURE_DOUBLE, SUM)).isEqualTo(Type.CUMULATIVE_DOUBLE);
+    assertThat(MetricUtils.getType(MEASURE_LONG, SUM)).isEqualTo(Type.CUMULATIVE_INT64);
+    assertThat(MetricUtils.getType(MEASURE_DOUBLE, MEAN)).isEqualTo(Type.CUMULATIVE_DOUBLE);
+    assertThat(MetricUtils.getType(MEASURE_LONG, MEAN)).isEqualTo(Type.CUMULATIVE_DOUBLE);
+    assertThat(MetricUtils.getType(MEASURE_DOUBLE, COUNT)).isEqualTo(Type.CUMULATIVE_INT64);
+    assertThat(MetricUtils.getType(MEASURE_LONG, COUNT)).isEqualTo(Type.CUMULATIVE_INT64);
+    assertThat(MetricUtils.getType(MEASURE_DOUBLE, DISTRIBUTION))
+        .isEqualTo(Type.CUMULATIVE_DISTRIBUTION);
+    assertThat(MetricUtils.getType(MEASURE_LONG, DISTRIBUTION))
+        .isEqualTo(Type.CUMULATIVE_DISTRIBUTION);
+  }
+
+  @Test
+  public void tagValuesToLabelValues() {
+    List<TagValue> tagValues = Arrays.asList(VALUE, VALUE_2, null);
+    assertThat(MetricUtils.tagValuesToLabelValues(tagValues))
+        .containsExactly(
+            LabelValue.create(VALUE.asString()),
+            LabelValue.create(VALUE_2.asString()),
+            LabelValue.create(null));
+  }
+
+  @Test
+  public void mutableAggregationToPoint() {
+    MutableSum sum = MutableSum.create();
+    MutableCount count = MutableCount.create();
+    MutableMean mean = MutableMean.create();
+    MutableLastValue lastValue = MutableLastValue.create();
+    MutableDistribution distribution = MutableDistribution.create(BUCKET_BOUNDARIES);
+
+    assertThat(MetricUtils.mutableAggregationToPoint(sum, TIMESTAMP, Type.CUMULATIVE_INT64))
+        .isEqualTo(Point.create(Value.longValue(0), TIMESTAMP));
+    assertThat(MetricUtils.mutableAggregationToPoint(sum, TIMESTAMP, Type.CUMULATIVE_DOUBLE))
+        .isEqualTo(Point.create(Value.doubleValue(0), TIMESTAMP));
+    assertThat(MetricUtils.mutableAggregationToPoint(count, TIMESTAMP, Type.CUMULATIVE_INT64))
+        .isEqualTo(Point.create(Value.longValue(0), TIMESTAMP));
+    assertThat(MetricUtils.mutableAggregationToPoint(mean, TIMESTAMP, Type.CUMULATIVE_DOUBLE))
+        .isEqualTo(Point.create(Value.doubleValue(0), TIMESTAMP));
+    assertThat(MetricUtils.mutableAggregationToPoint(lastValue, TIMESTAMP, Type.GAUGE_DOUBLE))
+        .isEqualTo(Point.create(Value.doubleValue(Double.NaN), TIMESTAMP));
+    assertThat(MetricUtils.mutableAggregationToPoint(lastValue, TIMESTAMP, Type.GAUGE_INT64))
+        .isEqualTo(Point.create(Value.longValue(0), TIMESTAMP));
+
+    assertThat(MetricUtils.mutableAggregationToPoint(distribution, TIMESTAMP, Type.GAUGE_DOUBLE))
+        .isEqualTo(
+            Point.create(
+                Value.distributionValue(
+                    io.opencensus.metrics.Distribution.create(
+                        0,
+                        0,
+                        0,
+                        Arrays.asList(-10.0, 0.0, 10.0),
+                        Arrays.asList(
+                            Bucket.create(0),
+                            Bucket.create(0),
+                            Bucket.create(0),
+                            Bucket.create(0)))),
+                TIMESTAMP));
+  }
+}