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.

5 files changed, 202 insertions, 11 deletions

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);
   }