Fix time skew exceptions crashing disruptor thread (#2071)

* Add a fix to metric views when a time skew is detected rather than crashing the application/thread.
* Add test for time-rewind issues, and fix problems discovered.

Author: jsuereth

impl_core/src/main/java/io/opencensus/implcore/stats/MutableViewData.java

@@ -132,6 +132,7 @@ private CumulativeMutableViewData(View view, Timestamp start) {
     @javax.annotation.Nullable
     @Override
     Metric toMetric(Timestamp now, State state) {
+      handleTimeRewinds(now);
       if (state == State.DISABLED) {
         return null;
       }
@@ -166,6 +167,7 @@ void record(
 
     @Override
     ViewData toViewData(Timestamp now, State state) {
+      handleTimeRewinds(now);
       if (state == State.ENABLED) {
         return ViewData.create(
             super.view,
@@ -180,6 +182,18 @@ ViewData toViewData(Timestamp now, State state) {
       }
     }
 
+    /**
+     * This method attemps to migrate this view into a reasonable state in the event of time going
+     * backwards.
+     */
+    private void handleTimeRewinds(Timestamp now) {
+      if (now.compareTo(start) < 0) {
+        // Time went backwards, physics is broken, forget what we know.
+        clearStats();
+        start = now;
+      }
+    }
+
     @Override
     void clearStats() {
       tagValueAggregationMap.clear();
@@ -304,10 +318,18 @@ private void refreshBucketList(Timestamp now) {
       }
       Timestamp startOfLastBucket =
           CheckerFrameworkUtils.castNonNull(buckets.peekLast()).getStart();
-      // TODO(songya): decide what to do when time goes backwards
-      checkArgument(
-          now.compareTo(startOfLastBucket) >= 0,
-          "Current time must be within or after the last bucket.");
+      // Time went backwards!  Physics has failed us!  drop everything we know and relearn.
+      // Prioritize:  Report data we're confident is correct.
+      if (now.compareTo(startOfLastBucket) < 0) {
+        // TODO: configurable time-skew handling with options:
+        // - Drop events in the future, keep others within a duration.
+        // - Drop all events on skew
+        // - Guess at time-skew and "fix" events
+        // - Reset our "start" time to now if necessary.
+        buckets.clear();
+        shiftBucketList(N + 1, now);
+        return;
+      }
       long elapsedTimeMillis = now.subtractTimestamp(startOfLastBucket).toMillis();
       long numOfPadBuckets = elapsedTimeMillis / bucketDuration.toMillis();
 

impl_core/src/test/java/io/opencensus/implcore/stats/MutableViewDataTest.java

@@ -19,6 +19,19 @@
 import static com.google.common.truth.Truth.assertThat;
 
 import io.opencensus.common.Timestamp;
+import io.opencensus.implcore.internal.CurrentState;
+import io.opencensus.implcore.tags.TagMapImpl;
+import io.opencensus.metrics.data.AttachmentValue;
+import io.opencensus.stats.Aggregation;
+import io.opencensus.stats.Aggregation.Count;
+import io.opencensus.stats.Aggregation.Distribution;
+import io.opencensus.stats.BucketBoundaries;
+import io.opencensus.stats.Measure.MeasureDouble;
+import io.opencensus.stats.View;
+import io.opencensus.stats.ViewData;
+import io.opencensus.tags.TagKey;
+import java.util.Arrays;
+import java.util.Collections;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -31,4 +44,53 @@ public class MutableViewDataTest {
   public void testConstants() {
     assertThat(MutableViewData.ZERO_TIMESTAMP).isEqualTo(Timestamp.create(0, 0));
   }
+
+  @Test
+  public void testTimeRewindsOnCountViewNoThrow() {
+    // First we set up some buckets THEN we rewind time for giggles.
+    View tester =
+        View.create(
+            View.Name.create("view"),
+            "Description",
+            MeasureDouble.create("name", "desc", "us"),
+            Count.create(),
+            Collections.singletonList(TagKey.create("KEY")));
+    Timestamp start = Timestamp.create(10000000, 0);
+    Timestamp validPointTime = Timestamp.create(10000010, 0);
+    CurrentState.State state = CurrentState.State.ENABLED;
+    MutableViewData viewData = MutableViewData.create(tester, start);
+    // Create a data points to get thrown away.
+    viewData.record(
+        TagMapImpl.EMPTY, 1.0, validPointTime, Collections.<String, AttachmentValue>emptyMap());
+    // Rewind time and look for explosions.
+    Timestamp thePast = Timestamp.create(0, 0);
+    ViewData result = viewData.toViewData(thePast, state);
+    assertThat(result.getAggregationMap()).isEmpty();
+  }
+
+  @Test
+  public void testTimeRewindsOnDistributionViewNoThrow() {
+    // First we set up some buckets THEN we rewind time for giggles.
+    Aggregation latencyDistribution =
+        Distribution.create(
+            BucketBoundaries.create(Arrays.asList(0.0, 25.0, 100.0, 200.0, 400.0, 800.0, 10000.0)));
+    View tester =
+        View.create(
+            View.Name.create("view"),
+            "Description",
+            MeasureDouble.create("name", "desc", "us"),
+            latencyDistribution,
+            Collections.singletonList(TagKey.create("KEY")));
+    Timestamp start = Timestamp.create(10000000, 0);
+    Timestamp validPointTime = Timestamp.create(10000010, 0);
+    CurrentState.State state = CurrentState.State.ENABLED;
+    MutableViewData viewData = MutableViewData.create(tester, start);
+    // Create a data points to get thrown away.
+    viewData.record(
+        TagMapImpl.EMPTY, 1.0, validPointTime, Collections.<String, AttachmentValue>emptyMap());
+    // Rewind time and look for explosions.
+    Timestamp thePast = Timestamp.create(0, 0);
+    ViewData result = viewData.toViewData(thePast, state);
+    assertThat(result.getAggregationMap()).isEmpty();
+  }
 }