Stats: Support recording Exemplars in the impl. (#1294)

* Stats: Support recording Exemplars in the impl.

* Add more comments and tests.

* Exemplar array will be null with no histogram.

10 files changed, 283 insertions, 62 deletions

diff --git a/impl_core/src/main/java/io/opencensus/implcore/stats/IntervalBucket.java b/impl_core/src/main/java/io/opencensus/implcore/stats/IntervalBucket.java
index 65b768da..12cb319e 100644
--- a/impl_core/src/main/java/io/opencensus/implcore/stats/IntervalBucket.java
+++ b/impl_core/src/main/java/io/opencensus/implcore/stats/IntervalBucket.java
@@ -61,11 +61,15 @@ final class IntervalBucket {
   }
 
   // Puts a new value into the internal MutableAggregations, based on the TagValues.
-  void record(List</*@Nullable*/ TagValue> tagValues, double value) {
+  void record(
+      List</*@Nullable*/ TagValue> tagValues,
+      double value,
+      Map<String, String> attachments,
+      Timestamp timestamp) {
     if (!tagValueAggregationMap.containsKey(tagValues)) {
       tagValueAggregationMap.put(tagValues, MutableViewData.createMutableAggregation(aggregation));
     }
-    tagValueAggregationMap.get(tagValues).add(value);
+    tagValueAggregationMap.get(tagValues).add(value, attachments, timestamp);
   }
 
   /*
diff --git a/impl_core/src/main/java/io/opencensus/implcore/stats/MeasureMapInternal.java b/impl_core/src/main/java/io/opencensus/implcore/stats/MeasureMapInternal.java
index 304ff8c4..d867b342 100644
--- a/impl_core/src/main/java/io/opencensus/implcore/stats/MeasureMapInternal.java
+++ b/impl_core/src/main/java/io/opencensus/implcore/stats/MeasureMapInternal.java
@@ -26,7 +26,6 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.NoSuchElementException;
-import javax.annotation.Nullable;
 
 // TODO(songya): consider combining MeasureMapImpl and this class.
 /** A map from {@link Measure}'s to measured values. */
@@ -51,15 +50,11 @@ final class MeasureMapInternal {
   }
 
   private final ArrayList<Measurement> measurements;
-  @Nullable private final Map<String, String> attachments;
+  private final Map<String, String> attachments;
 
-  private MeasureMapInternal(
-      ArrayList<Measurement> measurements, @Nullable Map<String, String> attachments) {
+  private MeasureMapInternal(ArrayList<Measurement> measurements, Map<String, String> attachments) {
     this.measurements = measurements;
-    this.attachments =
-        attachments == null
-            ? null
-            : Collections.unmodifiableMap(new HashMap<String, String>(attachments));
+    this.attachments = Collections.unmodifiableMap(new HashMap<String, String>(attachments));
   }
 
   /** Builder for the {@link MeasureMapInternal} class. */
@@ -91,9 +86,6 @@ final class MeasureMapInternal {
     }
 
     Builder putAttachment(String key, String value) {
-      if (this.attachments == null) {
-        this.attachments = new HashMap<String, String>();
-      }
       this.attachments.put(key, value);
       return this;
     }
@@ -115,7 +107,7 @@ final class MeasureMapInternal {
     }
 
     private final ArrayList<Measurement> measurements = new ArrayList<Measurement>();
-    @Nullable private Map<String, String> attachments;
+    private final Map<String, String> attachments = new HashMap<String, String>();
 
     private Builder() {}
   }
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 58d41c0d..b5917289 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
@@ -192,13 +192,13 @@ final class MeasureToViewMap {
     private final TagContext tags;
     private final MutableViewData view;
     private final Timestamp timestamp;
-    @javax.annotation.Nullable private final Map<String, String> attachments;
+    private final Map<String, String> attachments;
 
     private RecordDoubleValueFunc(
         TagContext tags,
         MutableViewData view,
         Timestamp timestamp,
-        @javax.annotation.Nullable Map<String, String> attachments) {
+        Map<String, String> attachments) {
       this.tags = tags;
       this.view = view;
       this.timestamp = timestamp;
@@ -216,13 +216,13 @@ final class MeasureToViewMap {
     private final TagContext tags;
     private final MutableViewData view;
     private final Timestamp timestamp;
-    @javax.annotation.Nullable private final Map<String, String> attachments;
+    private final Map<String, String> attachments;
 
     private RecordLongValueFunc(
         TagContext tags,
         MutableViewData view,
         Timestamp timestamp,
-        @javax.annotation.Nullable Map<String, String> attachments) {
+        Map<String, String> attachments) {
       this.tags = tags;
       this.view = view;
       this.timestamp = timestamp;
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 51504cff..a7fa0da4 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
@@ -20,8 +20,11 @@ import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import io.opencensus.common.Function;
+import io.opencensus.common.Timestamp;
 import io.opencensus.stats.Aggregation;
+import io.opencensus.stats.AggregationData.DistributionData.Exemplar;
 import io.opencensus.stats.BucketBoundaries;
+import java.util.Map;
 
 /** Mutable version of {@link Aggregation} that supports adding values. */
 abstract class MutableAggregation {
@@ -35,8 +38,10 @@ abstract class MutableAggregation {
    * Put a new value into the MutableAggregation.
    *
    * @param value new value to be added to population
+   * @param attachments the contextual information on an {@link Exemplar}
+   * @param timestamp the timestamp when the value is recorded
    */
-  abstract void add(double value);
+  abstract void add(double value, Map<String, String> attachments, Timestamp timestamp);
 
   // TODO(songya): remove this method once interval stats is completely removed.
   /**
@@ -76,7 +81,7 @@ abstract class MutableAggregation {
     }
 
     @Override
-    void add(double value) {
+    void add(double value, Map<String, String> attachments, Timestamp timestamp) {
       sum += value;
     }
 
@@ -123,7 +128,7 @@ abstract class MutableAggregation {
     }
 
     @Override
-    void add(double value) {
+    void add(double value, Map<String, String> attachments, Timestamp timestamp) {
       count++;
     }
 
@@ -171,7 +176,7 @@ abstract class MutableAggregation {
     }
 
     @Override
-    void add(double value) {
+    void add(double value, Map<String, String> attachments, Timestamp timestamp) {
       count++;
       sum += value;
     }
@@ -236,10 +241,19 @@ abstract class MutableAggregation {
 
     private final BucketBoundaries bucketBoundaries;
     private final long[] bucketCounts;
+    // If there's a histogram (i.e bucket boundaries are not empty) in this MutableDistribution,
+    // exemplars will have the same size to bucketCounts; otherwise exemplars are null.
+    // Only the newest exemplar will be kept at each index.
+    @javax.annotation.Nullable private final Exemplar[] exemplars;
 
     private MutableDistribution(BucketBoundaries bucketBoundaries) {
       this.bucketBoundaries = bucketBoundaries;
-      this.bucketCounts = new long[bucketBoundaries.getBoundaries().size() + 1];
+      int buckets = bucketBoundaries.getBoundaries().size() + 1;
+      this.bucketCounts = new long[buckets];
+      // In the implementation, each histogram bucket can have up to one exemplar, and the exemplar
+      // array is guaranteed to be in ascending order.
+      // If there's no histogram, don't record exemplars.
+      this.exemplars = bucketBoundaries.getBoundaries().isEmpty() ? null : new Exemplar[buckets];
     }
 
     /**
@@ -253,7 +267,7 @@ abstract class MutableAggregation {
     }
 
     @Override
-    void add(double value) {
+    void add(double value, Map<String, String> attachments, Timestamp timestamp) {
       sum += value;
       count++;
 
@@ -277,13 +291,19 @@ abstract class MutableAggregation {
         max = value;
       }
 
-      for (int i = 0; i < bucketBoundaries.getBoundaries().size(); i++) {
-        if (value < bucketBoundaries.getBoundaries().get(i)) {
-          bucketCounts[i]++;
-          return;
+      int bucket = 0;
+      for (; bucket < bucketBoundaries.getBoundaries().size(); bucket++) {
+        if (value < bucketBoundaries.getBoundaries().get(bucket)) {
+          break;
         }
       }
-      bucketCounts[bucketCounts.length - 1]++;
+      bucketCounts[bucket]++;
+
+      // No implicit recording for exemplars - if there are no attachments (contextual information),
+      // don't record exemplars.
+      if (!attachments.isEmpty() && exemplars != null) {
+        exemplars[bucket] = Exemplar.create(value, timestamp, attachments);
+      }
     }
 
     // We don't compute fractional MutableDistribution, it's either whole or none.
@@ -327,6 +347,18 @@ abstract class MutableAggregation {
       for (int i = 0; i < bucketCounts.length; i++) {
         this.bucketCounts[i] += bucketCounts[i];
       }
+
+      if (exemplars != null) {
+        for (int i = 0; i < mutableDistribution.getExemplars().length; i++) {
+          Exemplar exemplar = mutableDistribution.getExemplars()[i];
+          // Assume other is always newer than this, because we combined interval buckets in time
+          // order.
+          // If there's a newer exemplar, overwrite current value.
+          if (exemplar != null) {
+            this.exemplars[i] = exemplar;
+          }
+        }
+      }
     }
 
     double getMean() {
@@ -354,6 +386,11 @@ abstract class MutableAggregation {
       return bucketCounts;
     }
 
+    @javax.annotation.Nullable
+    Exemplar[] getExemplars() {
+      return exemplars;
+    }
+
     @Override
     final <T> T match(
         Function<? super MutableSum, T> p0,
@@ -385,7 +422,7 @@ abstract class MutableAggregation {
     }
 
     @Override
-    void add(double value) {
+    void add(double value, Map<String, String> attachments, Timestamp timestamp) {
       lastValue = value;
       // TODO(songya): remove this once interval stats is completely removed.
       if (!initialized) {
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 932da745..df99ecfb 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
@@ -41,6 +41,7 @@ import io.opencensus.stats.Aggregation.Sum;
 import io.opencensus.stats.AggregationData;
 import io.opencensus.stats.AggregationData.CountData;
 import io.opencensus.stats.AggregationData.DistributionData;
+import io.opencensus.stats.AggregationData.DistributionData.Exemplar;
 import io.opencensus.stats.AggregationData.LastValueDataDouble;
 import io.opencensus.stats.AggregationData.LastValueDataLong;
 import io.opencensus.stats.AggregationData.SumDataDouble;
@@ -101,19 +102,11 @@ abstract class MutableViewData {
   }
 
   /** Record double stats with the given tags. */
-  // TODO(songya): store the attachments in MutableDistribution.
   abstract void record(
-      TagContext context,
-      double value,
-      Timestamp timestamp,
-      @javax.annotation.Nullable Map<String, String> attachments);
+      TagContext context, double value, Timestamp timestamp, Map<String, String> attachments);
 
   /** Record long stats with the given tags. */
-  void record(
-      TagContext tags,
-      long value,
-      Timestamp timestamp,
-      @javax.annotation.Nullable Map<String, String> attachments) {
+  void record(TagContext tags, long value, Timestamp timestamp, Map<String, String> attachments) {
     // TODO(songya): shall we check for precision loss here?
     record(tags, (double) value, timestamp, attachments);
   }
@@ -216,17 +209,14 @@ abstract class MutableViewData {
 
     @Override
     void record(
-        TagContext context,
-        double value,
-        Timestamp timestamp,
-        @javax.annotation.Nullable Map<String, String> attachments) {
+        TagContext context, double value, Timestamp timestamp, Map<String, String> attachments) {
       List</*@Nullable*/ TagValue> tagValues =
           getTagValues(getTagMap(context), super.view.getColumns());
       if (!tagValueAggregationMap.containsKey(tagValues)) {
         tagValueAggregationMap.put(
             tagValues, createMutableAggregation(super.view.getAggregation()));
       }
-      tagValueAggregationMap.get(tagValues).add(value);
+      tagValueAggregationMap.get(tagValues).add(value, attachments, timestamp);
     }
 
     @Override
@@ -314,15 +304,13 @@ abstract class MutableViewData {
 
     @Override
     void record(
-        TagContext context,
-        double value,
-        Timestamp timestamp,
-        @javax.annotation.Nullable Map<String, String> attachments) {
+        TagContext context, double value, Timestamp timestamp, Map<String, String> attachments) {
       List</*@Nullable*/ TagValue> tagValues =
           getTagValues(getTagMap(context), super.view.getColumns());
       refreshBucketList(timestamp);
       // It is always the last bucket that does the recording.
-      CheckerFrameworkUtils.castNonNull(buckets.peekLast()).record(tagValues, value);
+      CheckerFrameworkUtils.castNonNull(buckets.peekLast())
+          .record(tagValues, value, attachments, timestamp);
     }
 
     @Override
@@ -594,13 +582,22 @@ abstract class MutableViewData {
       for (long bucketCount : arg.getBucketCounts()) {
         boxedBucketCounts.add(bucketCount);
       }
+      List<Exemplar> exemplars = new ArrayList<Exemplar>();
+      if (arg.getExemplars() != null) {
+        for (Exemplar exemplar : arg.getExemplars()) {
+          if (exemplar != null) {
+            exemplars.add(exemplar);
+          }
+        }
+      }
       return DistributionData.create(
           arg.getMean(),
           arg.getCount(),
           arg.getMin(),
           arg.getMax(),
           arg.getSumOfSquaredDeviations(),
-          boxedBucketCounts);
+          boxedBucketCounts,
+          exemplars);
     }
 
     private static final CreateDistributionData INSTANCE = new CreateDistributionData();
diff --git a/impl_core/src/test/java/io/opencensus/implcore/stats/IntervalBucketTest.java b/impl_core/src/test/java/io/opencensus/implcore/stats/IntervalBucketTest.java
index 10bce262..34cd434c 100644
--- a/impl_core/src/test/java/io/opencensus/implcore/stats/IntervalBucketTest.java
+++ b/impl_core/src/test/java/io/opencensus/implcore/stats/IntervalBucketTest.java
@@ -24,6 +24,7 @@ import io.opencensus.implcore.stats.MutableAggregation.MutableMean;
 import io.opencensus.stats.Aggregation.Mean;
 import io.opencensus.tags.TagValue;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import org.junit.Rule;
 import org.junit.Test;
@@ -82,9 +83,9 @@ public class IntervalBucketTest {
     IntervalBucket bucket = new IntervalBucket(START, MINUTE, MEAN);
     List<TagValue> tagValues1 = Arrays.<TagValue>asList(TagValue.create("VALUE1"));
     List<TagValue> tagValues2 = Arrays.<TagValue>asList(TagValue.create("VALUE2"));
-    bucket.record(tagValues1, 5.0);
-    bucket.record(tagValues1, 15.0);
-    bucket.record(tagValues2, 10.0);
+    bucket.record(tagValues1, 5.0, Collections.<String, String>emptyMap(), START);
+    bucket.record(tagValues1, 15.0, Collections.<String, String>emptyMap(), START);
+    bucket.record(tagValues2, 10.0, Collections.<String, String>emptyMap(), START);
     assertThat(bucket.getTagValueAggregationMap().keySet()).containsExactly(tagValues1, tagValues2);
     MutableMean mutableMean1 = (MutableMean) bucket.getTagValueAggregationMap().get(tagValues1);
     MutableMean mutableMean2 = (MutableMean) bucket.getTagValueAggregationMap().get(tagValues2);
diff --git a/impl_core/src/test/java/io/opencensus/implcore/stats/MeasureMapInternalTest.java b/impl_core/src/test/java/io/opencensus/implcore/stats/MeasureMapInternalTest.java
index c57baa0b..19e8a6c5 100644
--- a/impl_core/src/test/java/io/opencensus/implcore/stats/MeasureMapInternalTest.java
+++ b/impl_core/src/test/java/io/opencensus/implcore/stats/MeasureMapInternalTest.java
@@ -47,6 +47,18 @@ public class MeasureMapInternalTest {
   }
 
   @Test
+  public void testPutAttachment() {
+    MeasureMapInternal metrics =
+        MeasureMapInternal.builder()
+            .putAttachment("k1", "v1")
+            .putAttachment("k2", "v2")
+            .putAttachment("k1", "v3")
+            .build();
+    assertThat(metrics.getAttachments()).containsExactly("k1", "v3", "k2", "v2");
+    assertContains(metrics);
+  }
+
+  @Test
   public void testCombination() {
     MeasureMapInternal metrics =
         MeasureMapInternal.builder()
diff --git a/impl_core/src/test/java/io/opencensus/implcore/stats/MutableAggregationTest.java b/impl_core/src/test/java/io/opencensus/implcore/stats/MutableAggregationTest.java
index 5508588a..e135dcf5 100644
--- a/impl_core/src/test/java/io/opencensus/implcore/stats/MutableAggregationTest.java
+++ b/impl_core/src/test/java/io/opencensus/implcore/stats/MutableAggregationTest.java
@@ -18,17 +18,22 @@ package io.opencensus.implcore.stats;
 
 import static com.google.common.truth.Truth.assertThat;
 
+import com.google.common.collect.ImmutableList;
 import io.opencensus.common.Function;
 import io.opencensus.common.Functions;
+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.stats.AggregationData.DistributionData.Exemplar;
 import io.opencensus.stats.BucketBoundaries;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
@@ -44,6 +49,9 @@ public class MutableAggregationTest {
   private static final double TOLERANCE = 1e-6;
   private static final BucketBoundaries BUCKET_BOUNDARIES =
       BucketBoundaries.create(Arrays.asList(-10.0, 0.0, 10.0));
+  private static final BucketBoundaries BUCKET_BOUNDARIES_EMPTY =
+      BucketBoundaries.create(Collections.<Double>emptyList());
+  private static final Timestamp TIMESTAMP = Timestamp.create(60, 0);
 
   @Test
   public void testCreateEmpty() {
@@ -60,6 +68,11 @@ public class MutableAggregationTest {
     assertThat(mutableDistribution.getMax()).isNegativeInfinity();
     assertThat(mutableDistribution.getSumOfSquaredDeviations()).isWithin(TOLERANCE).of(0);
     assertThat(mutableDistribution.getBucketCounts()).isEqualTo(new long[4]);
+    assertThat(mutableDistribution.getExemplars()).isEqualTo(new Exemplar[4]);
+
+    MutableDistribution mutableDistributionNoHistogram =
+        MutableDistribution.create(BUCKET_BOUNDARIES_EMPTY);
+    assertThat(mutableDistributionNoHistogram.getExemplars()).isNull();
   }
 
   @Test
@@ -91,7 +104,7 @@ public class MutableAggregationTest {
 
     for (double value : values) {
       for (MutableAggregation aggregation : aggregations) {
-        aggregation.add(value);
+        aggregation.add(value, Collections.<String, String>emptyMap(), TIMESTAMP);
       }
     }
 
@@ -136,6 +149,46 @@ public class MutableAggregationTest {
   }
 
   @Test
+  public void testAdd_DistributionWithExemplarAttachments() {
+    MutableDistribution mutableDistribution = MutableDistribution.create(BUCKET_BOUNDARIES);
+    MutableDistribution mutableDistributionNoHistogram =
+        MutableDistribution.create(BUCKET_BOUNDARIES_EMPTY);
+    List<Double> values = Arrays.asList(-1.0, 1.0, -5.0, 20.0, 5.0);
+    List<Map<String, String>> attachmentsList =
+        ImmutableList.<Map<String, String>>of(
+            Collections.<String, String>singletonMap("k1", "v1"),
+            Collections.<String, String>singletonMap("k2", "v2"),
+            Collections.<String, String>singletonMap("k3", "v3"),
+            Collections.<String, String>singletonMap("k4", "v4"),
+            Collections.<String, String>singletonMap("k5", "v5"));
+    List<Timestamp> timestamps =
+        Arrays.asList(
+            Timestamp.fromMillis(500),
+            Timestamp.fromMillis(1000),
+            Timestamp.fromMillis(2000),
+            Timestamp.fromMillis(3000),
+            Timestamp.fromMillis(4000));
+    for (int i = 0; i < values.size(); i++) {
+      mutableDistribution.add(values.get(i), attachmentsList.get(i), timestamps.get(i));
+      mutableDistributionNoHistogram.add(values.get(i), attachmentsList.get(i), timestamps.get(i));
+    }
+
+    // Each bucket can only have up to one exemplar. If there are more than one exemplars in a
+    // bucket, only the last one will be kept.
+    List<Exemplar> expected =
+        Arrays.<Exemplar>asList(
+            null,
+            Exemplar.create(values.get(2), timestamps.get(2), attachmentsList.get(2)),
+            Exemplar.create(values.get(4), timestamps.get(4), attachmentsList.get(4)),
+            Exemplar.create(values.get(3), timestamps.get(3), attachmentsList.get(3)));
+    assertThat(mutableDistribution.getExemplars())
+        .asList()
+        .containsExactlyElementsIn(expected)
+        .inOrder();
+    assertThat(mutableDistributionNoHistogram.getExemplars()).isNull();
+  }
+
+  @Test
   public void testMatch() {
     List<MutableAggregation> aggregations =
         Arrays.asList(
@@ -170,12 +223,12 @@ public class MutableAggregationTest {
 
     for (double val : Arrays.asList(-1.0, -5.0)) {
       for (MutableAggregation aggregation : aggregations1) {
-        aggregation.add(val);
+        aggregation.add(val, Collections.<String, String>emptyMap(), TIMESTAMP);
       }
     }
     for (double val : Arrays.asList(10.0, 50.0)) {
       for (MutableAggregation aggregation : aggregations2) {
-        aggregation.add(val);
+        aggregation.add(val, Collections.<String, String>emptyMap(), TIMESTAMP);
       }
     }
 
@@ -201,13 +254,13 @@ public class MutableAggregationTest {
     MutableDistribution distribution3 = MutableDistribution.create(BUCKET_BOUNDARIES);
 
     for (double val : Arrays.asList(5.0, -5.0)) {
-      distribution1.add(val);
+      distribution1.add(val, Collections.<String, String>emptyMap(), TIMESTAMP);
     }
     for (double val : Arrays.asList(10.0, 20.0)) {
-      distribution2.add(val);
+      distribution2.add(val, Collections.<String, String>emptyMap(), TIMESTAMP);
     }
     for (double val : Arrays.asList(-10.0, 15.0, -15.0, -20.0)) {
-      distribution3.add(val);
+      distribution3.add(val, Collections.<String, String>emptyMap(), TIMESTAMP);
     }
 
     MutableDistribution combined = MutableDistribution.create(BUCKET_BOUNDARIES);
diff --git a/impl_core/src/test/java/io/opencensus/implcore/stats/StatsRecorderImplTest.java b/impl_core/src/test/java/io/opencensus/implcore/stats/StatsRecorderImplTest.java
index 3479bbae..3c8d7af3 100644
--- a/impl_core/src/test/java/io/opencensus/implcore/stats/StatsRecorderImplTest.java
+++ b/impl_core/src/test/java/io/opencensus/implcore/stats/StatsRecorderImplTest.java
@@ -23,8 +23,16 @@ import static io.opencensus.implcore.stats.StatsTestUtil.createEmptyViewData;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import io.grpc.Context;
+import io.opencensus.common.Duration;
+import io.opencensus.common.Timestamp;
 import io.opencensus.implcore.internal.SimpleEventQueue;
+import io.opencensus.stats.Aggregation.Count;
+import io.opencensus.stats.Aggregation.Distribution;
 import io.opencensus.stats.Aggregation.Sum;
+import io.opencensus.stats.AggregationData.CountData;
+import io.opencensus.stats.AggregationData.DistributionData;
+import io.opencensus.stats.AggregationData.DistributionData.Exemplar;
+import io.opencensus.stats.BucketBoundaries;
 import io.opencensus.stats.Measure.MeasureDouble;
 import io.opencensus.stats.MeasureMap;
 import io.opencensus.stats.StatsCollectionState;
@@ -62,9 +70,17 @@ public final class StatsRecorderImplTest {
   private static final MeasureDouble MEASURE_DOUBLE_NO_VIEW_2 =
       MeasureDouble.create("my measurement no view 2", "description", "us");
   private static final View.Name VIEW_NAME = View.Name.create("my view");
+  private static final BucketBoundaries BUCKET_BOUNDARIES =
+      BucketBoundaries.create(Arrays.asList(-10.0, 0.0, 10.0));
+  private static final Distribution DISTRIBUTION = Distribution.create(BUCKET_BOUNDARIES);
+  private static final Distribution DISTRIBUTION_NO_HISTOGRAM =
+      Distribution.create(BucketBoundaries.create(Collections.<Double>emptyList()));
+  private static final Timestamp START_TIME = Timestamp.fromMillis(0);
+  private static final Duration ONE_SECOND = Duration.fromMillis(1000);
 
+  private final TestClock testClock = TestClock.create();
   private final StatsComponent statsComponent =
-      new StatsComponentImplBase(new SimpleEventQueue(), TestClock.create());
+      new StatsComponentImplBase(new SimpleEventQueue(), testClock);
 
   private final ViewManager viewManager = statsComponent.getViewManager();
   private final StatsRecorder statsRecorder = statsComponent.getStatsRecorder();
@@ -143,6 +159,112 @@ public final class StatsRecorderImplTest {
   }
 
   @Test
+  public void record_WithAttachments_Distribution() {
+    testClock.setTime(START_TIME);
+    View view =
+        View.create(VIEW_NAME, "description", MEASURE_DOUBLE, DISTRIBUTION, Arrays.asList(KEY));
+    viewManager.registerView(view);
+    recordWithAttachments();
+    ViewData viewData = viewManager.getView(VIEW_NAME);
+    assertThat(viewData).isNotNull();
+    DistributionData distributionData =
+        (DistributionData) viewData.getAggregationMap().get(Collections.singletonList(VALUE));
+    List<Exemplar> expected =
+        Arrays.asList(
+            Exemplar.create(-20.0, Timestamp.create(4, 0), Collections.singletonMap("k3", "v1")),
+            Exemplar.create(-5.0, Timestamp.create(5, 0), Collections.singletonMap("k3", "v3")),
+            Exemplar.create(1.0, Timestamp.create(2, 0), Collections.singletonMap("k2", "v2")),
+            Exemplar.create(12.0, Timestamp.create(3, 0), Collections.singletonMap("k1", "v3")));
+    assertThat(distributionData.getExemplars()).containsExactlyElementsIn(expected).inOrder();
+  }
+
+  @Test
+  public void record_WithAttachments_DistributionNoHistogram() {
+    testClock.setTime(START_TIME);
+    View view =
+        View.create(
+            VIEW_NAME,
+            "description",
+            MEASURE_DOUBLE,
+            DISTRIBUTION_NO_HISTOGRAM,
+            Arrays.asList(KEY));
+    viewManager.registerView(view);
+    recordWithAttachments();
+    ViewData viewData = viewManager.getView(VIEW_NAME);
+    assertThat(viewData).isNotNull();
+    DistributionData distributionData =
+        (DistributionData) viewData.getAggregationMap().get(Collections.singletonList(VALUE));
+    // Recording exemplar has no effect if there's no histogram.
+    assertThat(distributionData.getExemplars()).isEmpty();
+  }
+
+  @Test
+  public void record_WithAttachments_Count() {
+    testClock.setTime(START_TIME);
+    View view =
+        View.create(VIEW_NAME, "description", MEASURE_DOUBLE, Count.create(), Arrays.asList(KEY));
+    viewManager.registerView(view);
+    recordWithAttachments();
+    ViewData viewData = viewManager.getView(VIEW_NAME);
+    assertThat(viewData).isNotNull();
+    CountData countData =
+        (CountData) viewData.getAggregationMap().get(Collections.singletonList(VALUE));
+    // Recording exemplar does not affect views with an aggregation other than distribution.
+    assertThat(countData.getCount()).isEqualTo(6L);
+  }
+
+  private void recordWithAttachments() {
+    TagContext context = new SimpleTagContext(Tag.create(KEY, VALUE));
+
+    // The test Distribution has bucket boundaries [-10.0, 0.0, 10.0].
+
+    testClock.advanceTime(ONE_SECOND); // 1st second.
+    // -1.0 is in the 2nd bucket [-10.0, 0.0).
+    statsRecorder
+        .newMeasureMap()
+        .put(MEASURE_DOUBLE, -1.0)
+        .putAttachment("k1", "v1")
+        .record(context);
+
+    testClock.advanceTime(ONE_SECOND); // 2nd second.
+    // 1.0 is in the 3rd bucket [0.0, 10.0).
+    statsRecorder
+        .newMeasureMap()
+        .put(MEASURE_DOUBLE, 1.0)
+        .putAttachment("k2", "v2")
+        .record(context);
+
+    testClock.advanceTime(ONE_SECOND); // 3rd second.
+    // 12.0 is in the 4th bucket [10.0, +Inf).
+    statsRecorder
+        .newMeasureMap()
+        .put(MEASURE_DOUBLE, 12.0)
+        .putAttachment("k1", "v3")
+        .record(context);
+
+    testClock.advanceTime(ONE_SECOND); // 4th second.
+    // -20.0 is in the 1st bucket [-Inf, -10.0).
+    statsRecorder
+        .newMeasureMap()
+        .put(MEASURE_DOUBLE, -20.0)
+        .putAttachment("k3", "v1")
+        .record(context);
+
+    testClock.advanceTime(ONE_SECOND); // 5th second.
+    // -5.0 is in the 2nd bucket [-10.0, 0), should overwrite the previous exemplar -1.0.
+    statsRecorder
+        .newMeasureMap()
+        .put(MEASURE_DOUBLE, -5.0)
+        .putAttachment("k3", "v3")
+        .record(context);
+
+    testClock.advanceTime(ONE_SECOND); // 6th second.
+    // -3.0 is in the 2nd bucket [-10.0, 0), but this value doesn't come with attachments, so it
+    // shouldn't overwrite the previous exemplar (-5.0).
+    statsRecorder.newMeasureMap().put(MEASURE_DOUBLE, -3.0).record(context);
+  }
+
+  @Test
   public void recordTwice() {
     View view =
         View.create(
diff --git a/impl_core/src/test/java/io/opencensus/implcore/stats/StatsTestUtil.java b/impl_core/src/test/java/io/opencensus/implcore/stats/StatsTestUtil.java
index 88e1df02..6a2e6366 100644
--- a/impl_core/src/test/java/io/opencensus/implcore/stats/StatsTestUtil.java
+++ b/impl_core/src/test/java/io/opencensus/implcore/stats/StatsTestUtil.java
@@ -22,6 +22,7 @@ import static io.opencensus.implcore.stats.MutableViewData.ZERO_TIMESTAMP;
 import com.google.common.collect.Iterables;
 import io.opencensus.common.Function;
 import io.opencensus.common.Functions;
+import io.opencensus.common.Timestamp;
 import io.opencensus.stats.Aggregation;
 import io.opencensus.stats.AggregationData;
 import io.opencensus.stats.AggregationData.CountData;
@@ -48,6 +49,8 @@ import javax.annotation.Nullable;
 /** Stats test utilities. */
 final class StatsTestUtil {
 
+  private static final Timestamp EMPTY = Timestamp.create(0, 0);
+
   private StatsTestUtil() {}
 
   /**
@@ -62,7 +65,7 @@ final class StatsTestUtil {
       Aggregation aggregation, Measure measure, double... values) {
     MutableAggregation mutableAggregation = MutableViewData.createMutableAggregation(aggregation);
     for (double value : values) {
-      mutableAggregation.add(value);
+      mutableAggregation.add(value, Collections.<String, String>emptyMap(), EMPTY);
     }
     return MutableViewData.createAggregationData(mutableAggregation, measure);
   }