Add Aggregation.LastValue and AggregationData.LastValueData to support Gauge (#1055)

* Add LastValue and LastValueData

* Support LastValue and LastValueData in impl

* Use Utils instead of Precondition

* Add LastValue and remove Mean from match() method.

* Support LastValue and LastValueData in exporters and zpages.

* Update MutableLastValue, add comments on why Mean is still supported.

2 files changed, 120 insertions, 15 deletions

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 deabdce7..51504cff 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
@@ -38,6 +38,7 @@ abstract class MutableAggregation {
    */
   abstract void add(double value);
 
+  // TODO(songya): remove this method once interval stats is completely removed.
   /**
    * Combine the internal values of this MutableAggregation and value of the given
    * MutableAggregation, with the given fraction. Then set the internal value of this
@@ -55,7 +56,8 @@ abstract class MutableAggregation {
       Function<? super MutableSum, T> p0,
       Function<? super MutableCount, T> p1,
       Function<? super MutableMean, T> p2,
-      Function<? super MutableDistribution, T> p3);
+      Function<? super MutableDistribution, T> p3,
+      Function<? super MutableLastValue, T> p4);
 
   /** Calculate sum on aggregated {@code MeasureValue}s. */
   static final class MutableSum extends MutableAggregation {
@@ -98,7 +100,8 @@ abstract class MutableAggregation {
         Function<? super MutableSum, T> p0,
         Function<? super MutableCount, T> p1,
         Function<? super MutableMean, T> p2,
-        Function<? super MutableDistribution, T> p3) {
+        Function<? super MutableDistribution, T> p3,
+        Function<? super MutableLastValue, T> p4) {
       return p0.apply(this);
     }
   }
@@ -144,7 +147,8 @@ abstract class MutableAggregation {
         Function<? super MutableSum, T> p0,
         Function<? super MutableCount, T> p1,
         Function<? super MutableMean, T> p2,
-        Function<? super MutableDistribution, T> p3) {
+        Function<? super MutableDistribution, T> p3,
+        Function<? super MutableLastValue, T> p4) {
       return p1.apply(this);
     }
   }
@@ -212,7 +216,8 @@ abstract class MutableAggregation {
         Function<? super MutableSum, T> p0,
         Function<? super MutableCount, T> p1,
         Function<? super MutableMean, T> p2,
-        Function<? super MutableDistribution, T> p3) {
+        Function<? super MutableDistribution, T> p3,
+        Function<? super MutableLastValue, T> p4) {
       return p2.apply(this);
     }
   }
@@ -354,8 +359,66 @@ abstract class MutableAggregation {
         Function<? super MutableSum, T> p0,
         Function<? super MutableCount, T> p1,
         Function<? super MutableMean, T> p2,
-        Function<? super MutableDistribution, T> p3) {
+        Function<? super MutableDistribution, T> p3,
+        Function<? super MutableLastValue, T> p4) {
       return p3.apply(this);
     }
   }
+
+  /** Calculate last value on aggregated {@code MeasureValue}s. */
+  static final class MutableLastValue extends MutableAggregation {
+
+    // Initial value that will get reset as soon as first value is added.
+    private double lastValue = Double.NaN;
+    // TODO(songya): remove this once interval stats is completely removed.
+    private boolean initialized = false;
+
+    private MutableLastValue() {}
+
+    /**
+     * Construct a {@code MutableLastValue}.
+     *
+     * @return an empty {@code MutableLastValue}.
+     */
+    static MutableLastValue create() {
+      return new MutableLastValue();
+    }
+
+    @Override
+    void add(double value) {
+      lastValue = value;
+      // TODO(songya): remove this once interval stats is completely removed.
+      if (!initialized) {
+        initialized = true;
+      }
+    }
+
+    @Override
+    void combine(MutableAggregation other, double fraction) {
+      checkArgument(other instanceof MutableLastValue, "MutableLastValue expected.");
+      MutableLastValue otherValue = (MutableLastValue) other;
+      // Assume other is always newer than this, because we combined interval buckets in time order.
+      // If there's a newer value, overwrite current value.
+      this.lastValue = otherValue.initialized ? otherValue.getLastValue() : this.lastValue;
+    }
+
+    /**
+     * Returns the last value.
+     *
+     * @return the last value.
+     */
+    double getLastValue() {
+      return lastValue;
+    }
+
+    @Override
+    final <T> T match(
+        Function<? super MutableSum, T> p0,
+        Function<? super MutableCount, T> p1,
+        Function<? super MutableMean, T> p2,
+        Function<? super MutableDistribution, T> p3,
+        Function<? super MutableLastValue, T> p4) {
+      return p4.apply(this);
+    }
+  }
 }
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 d60715b5..f4136616 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
@@ -19,7 +19,7 @@ package io.opencensus.implcore.stats;
 import static com.google.common.base.Preconditions.checkArgument;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.HashMultimap;
+import com.google.common.collect.LinkedHashMultimap;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Multimap;
 import io.opencensus.common.Duration;
@@ -29,16 +29,20 @@ import io.opencensus.common.Timestamp;
 import io.opencensus.implcore.internal.CheckerFrameworkUtils;
 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.implcore.tags.TagContextImpl;
 import io.opencensus.stats.Aggregation;
 import io.opencensus.stats.Aggregation.Count;
 import io.opencensus.stats.Aggregation.Distribution;
+import io.opencensus.stats.Aggregation.LastValue;
 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.LastValueDataDouble;
+import io.opencensus.stats.AggregationData.LastValueDataLong;
 import io.opencensus.stats.AggregationData.SumDataDouble;
 import io.opencensus.stats.AggregationData.SumDataLong;
 import io.opencensus.stats.Measure;
@@ -157,9 +161,9 @@ abstract class MutableViewData {
     return aggregation.match(
         CreateMutableSum.INSTANCE,
         CreateMutableCount.INSTANCE,
-        CreateMutableMean.INSTANCE,
         CreateMutableDistribution.INSTANCE,
-        Functions.<MutableAggregation>throwIllegalArgumentException());
+        CreateMutableLastValue.INSTANCE,
+        AggregationDefaultFunction.INSTANCE);
   }
 
   /**
@@ -175,7 +179,8 @@ abstract class MutableViewData {
         new CreateSumData(measure),
         CreateCountData.INSTANCE,
         CreateMeanData.INSTANCE,
-        CreateDistributionData.INSTANCE);
+        CreateDistributionData.INSTANCE,
+        new CreateLastValueData(measure));
   }
 
   // Covert a mapping from TagValues to MutableAggregation, to a mapping from TagValues to
@@ -389,7 +394,9 @@ abstract class MutableViewData {
     // tag values to aggregation data.
     private Map<List</*@Nullable*/ TagValue>, AggregationData> combineBucketsAndGetAggregationMap(
         Timestamp now) {
-      Multimap<List</*@Nullable*/ TagValue>, MutableAggregation> multimap = HashMultimap.create();
+      // Need to maintain the order of inserted MutableAggregations (inserted based on time order).
+      Multimap<List</*@Nullable*/ TagValue>, MutableAggregation> multimap =
+          LinkedHashMultimap.create();
 
       // TODO(sebright): Decide whether to use a different class instead of LinkedList.
       @SuppressWarnings("JdkObsolete")
@@ -491,14 +498,20 @@ abstract class MutableViewData {
     private static final CreateMutableCount INSTANCE = new CreateMutableCount();
   }
 
-  private static final class CreateMutableMean
-      implements Function<Aggregation.Mean, MutableAggregation> {
+  // TODO(songya): remove this once Mean aggregation is completely removed. Before that
+  // we need to continue supporting Mean, since it could still be used by users and some
+  // deprecated RPC views.
+  private static final class AggregationDefaultFunction
+      implements Function<Aggregation, MutableAggregation> {
     @Override
-    public MutableAggregation apply(Aggregation.Mean arg) {
-      return MutableMean.create();
+    public MutableAggregation apply(Aggregation arg) {
+      if (arg instanceof Aggregation.Mean) {
+        return MutableMean.create();
+      }
+      throw new IllegalArgumentException("Unknown Aggregation.");
     }
 
-    private static final CreateMutableMean INSTANCE = new CreateMutableMean();
+    private static final AggregationDefaultFunction INSTANCE = new AggregationDefaultFunction();
   }
 
   private static final class CreateMutableDistribution
@@ -511,6 +524,16 @@ abstract class MutableViewData {
     private static final CreateMutableDistribution INSTANCE = new CreateMutableDistribution();
   }
 
+  private static final class CreateMutableLastValue
+      implements Function<LastValue, MutableAggregation> {
+    @Override
+    public MutableAggregation apply(LastValue arg) {
+      return MutableLastValue.create();
+    }
+
+    private static final CreateMutableLastValue INSTANCE = new CreateMutableLastValue();
+  }
+
   private static final class CreateSumData implements Function<MutableSum, AggregationData> {
 
     private final Measure measure;
@@ -566,6 +589,25 @@ abstract class MutableViewData {
     private static final CreateDistributionData INSTANCE = new CreateDistributionData();
   }
 
+  private static final class CreateLastValueData
+      implements Function<MutableLastValue, AggregationData> {
+
+    private final Measure measure;
+
+    private CreateLastValueData(Measure measure) {
+      this.measure = measure;
+    }
+
+    @Override
+    public AggregationData apply(final MutableLastValue arg) {
+      return measure.match(
+          Functions.<AggregationData>returnConstant(LastValueDataDouble.create(arg.getLastValue())),
+          Functions.<AggregationData>returnConstant(
+              LastValueDataLong.create(Math.round(arg.getLastValue()))),
+          Functions.<AggregationData>throwAssertionError());
+    }
+  }
+
   private static final class CreateCumulative
       implements Function<View.AggregationWindow.Cumulative, MutableViewData> {
     @Override