Use workflow clock for timer metrics calculation

Author: Liang Mei

build.gradle

@@ -42,7 +42,7 @@ dependencies {
     compile group: 'org.slf4j', name: 'slf4j-api', version: '1.7.25'
     compile group: 'org.apache.thrift', name: 'libthrift', version: '0.9.3'
     compile group: 'com.google.code.gson', name: 'gson', version: '2.8.2'
-    compile group: 'com.uber.m3', name: 'tally-core', version: '0.2.0'
+    compile group: 'com.uber.m3', name: 'tally-core', version: '0.2.1'
     testCompile group: 'junit', name: 'junit', version: '4.12'
     testCompile group: 'com.googlecode.junit-toolbox', name: 'junit-toolbox', version: '2.4'
     testCompile group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.3'

src/main/java/com/uber/cadence/internal/metrics/ReplayAwareScope.java

@@ -26,27 +26,30 @@
 import com.uber.m3.tally.Scope;
 import com.uber.m3.tally.ScopeCloseException;
 import com.uber.m3.tally.Stopwatch;
+import com.uber.m3.tally.StopwatchRecorder;
 import com.uber.m3.tally.Timer;
 import com.uber.m3.util.Duration;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 
 public class ReplayAwareScope implements Scope {
-  Scope scope;
-  ReplayAware context;
+  private Scope scope;
+  private ReplayAware context;
+  private Supplier<Long> clock;
 
-  public ReplayAwareScope(Scope scope, ReplayAware context) {
+  public ReplayAwareScope(Scope scope, ReplayAware context, Supplier<Long> clock) {
     this.scope = Objects.requireNonNull(scope);
     this.context = Objects.requireNonNull(context);
+    this.clock = Objects.requireNonNull(clock);
   }
 
-  private static class ReplayAwareCounter implements Counter {
+  private class ReplayAwareCounter implements Counter {
     Counter counter;
-    ReplayAware context;
 
-    public ReplayAwareCounter(Counter counter, ReplayAware context) {
+    ReplayAwareCounter(Counter counter) {
       this.counter = Objects.requireNonNull(counter);
-      this.context = Objects.requireNonNull(context);
     }
 
     @Override
@@ -59,13 +62,11 @@ public void inc(long delta) {
     }
   }
 
-  private static class ReplayAwareGauge implements Gauge {
+  private class ReplayAwareGauge implements Gauge {
     Gauge gauge;
-    ReplayAware context;
 
-    public ReplayAwareGauge(Gauge gauge, ReplayAware context) {
+    ReplayAwareGauge(Gauge gauge) {
       this.gauge = Objects.requireNonNull(gauge);
-      this.context = Objects.requireNonNull(context);
     }
 
     @Override
@@ -78,13 +79,11 @@ public void update(double value) {
     }
   }
 
-  private static class ReplayAwareTimer implements Timer {
+  private class ReplayAwareTimer implements Timer, DurationRecorder {
     Timer timer;
-    ReplayAware context;
 
-    public ReplayAwareTimer(Timer timer, ReplayAware context) {
+    ReplayAwareTimer(Timer timer) {
       this.timer = Objects.requireNonNull(timer);
-      this.context = Objects.requireNonNull(context);
     }
 
     @Override
@@ -98,20 +97,39 @@ public void record(Duration interval) {
 
     @Override
     public Stopwatch start() {
-      // TODO: Stopwatch is a concrete class with non-public constructor so there's no
-      // way to extend it. Need to fix this when the following issue is resolved.
-      // https://github.com/uber-java/tally/issues/26
-      return timer.start();
+      long startNanos = TimeUnit.MILLISECONDS.toNanos(clock.get());
+      return new Stopwatch(startNanos, new ReplayAwareStopwatchRecorder(this));
+    }
+
+    @Override
+    public void recordDuration(Duration interval) {
+      record(interval);
+    }
+  }
+
+  interface DurationRecorder {
+    void recordDuration(Duration interval);
+  }
+
+  private class ReplayAwareStopwatchRecorder implements StopwatchRecorder {
+    DurationRecorder recorder;
+
+    ReplayAwareStopwatchRecorder(DurationRecorder recorder) {
+      this.recorder = recorder;
+    }
+
+    @Override
+    public void recordStopwatch(long startNanos) {
+      long endNanos = TimeUnit.MILLISECONDS.toNanos(clock.get());
+      recorder.recordDuration(Duration.between(startNanos, endNanos));
     }
   }
 
-  private static class ReplayAwareHistogram implements Histogram {
+  private class ReplayAwareHistogram implements Histogram, DurationRecorder {
     Histogram histogram;
-    ReplayAware context;
 
-    public ReplayAwareHistogram(Histogram histogram, ReplayAware context) {
+    ReplayAwareHistogram(Histogram histogram) {
       this.histogram = Objects.requireNonNull(histogram);
-      this.context = Objects.requireNonNull(context);
     }
 
     @Override
@@ -134,41 +152,39 @@ public void recordDuration(Duration value) {
 
     @Override
     public Stopwatch start() {
-      // TODO: Stopwatch is a concrete class with non-public constructor so there's no
-      // way to extend it. Need to fix this when the following issue is resolved.
-      // https://github.com/uber-java/tally/issues/26
-      return histogram.start();
+      long startNanos = TimeUnit.MILLISECONDS.toNanos(clock.get());
+      return new Stopwatch(startNanos, new ReplayAwareStopwatchRecorder(this));
     }
   }
 
   @Override
   public Counter counter(String name) {
-    return new ReplayAwareCounter(scope.counter(name), context);
+    return new ReplayAwareCounter(scope.counter(name));
   }
 
   @Override
   public Gauge gauge(String name) {
-    return new ReplayAwareGauge(scope.gauge(name), context);
+    return new ReplayAwareGauge(scope.gauge(name));
   }
 
   @Override
   public Timer timer(String name) {
-    return new ReplayAwareTimer(scope.timer(name), context);
+    return new ReplayAwareTimer(scope.timer(name));
   }
 
   @Override
   public Histogram histogram(String name, Buckets buckets) {
-    return new ReplayAwareHistogram(scope.histogram(name, buckets), context);
+    return new ReplayAwareHistogram(scope.histogram(name, buckets));
   }
 
   @Override
   public Scope tagged(Map<String, String> tags) {
-    return new ReplayAwareScope(scope.tagged(tags), context);
+    return new ReplayAwareScope(scope.tagged(tags), context, clock);
   }
 
   @Override
   public Scope subScope(String name) {
-    return new ReplayAwareScope(scope.subScope(name), context);
+    return new ReplayAwareScope(scope.subScope(name), context, clock);
   }
 
   @Override

src/main/java/com/uber/cadence/internal/replay/DecisionContextImpl.java

@@ -56,7 +56,7 @@ final class DecisionContextImpl implements DecisionContext, HistoryEventHandler
   }
 
   public void setMetricsScope(Scope metricsScope) {
-    this.metricsScope = new ReplayAwareScope(metricsScope, this);
+    this.metricsScope = new ReplayAwareScope(metricsScope, this, workflowClock::currentTimeMillis);
   }
 
   @Override

src/main/java/com/uber/cadence/workflow/Workflow.java

@@ -21,6 +21,7 @@
 import com.uber.cadence.activity.ActivityOptions;
 import com.uber.cadence.common.RetryOptions;
 import com.uber.cadence.internal.sync.WorkflowInternal;
+import com.uber.cadence.worker.WorkerOptions;
 import com.uber.cadence.workflow.Functions.Func;
 import com.uber.cadence.workflow.Functions.Func1;
 import com.uber.cadence.workflow.Functions.Func2;
@@ -709,6 +710,12 @@ public static boolean isReplaying() {
     return WorkflowInternal.isReplaying();
   }
 
+  /**
+   * Get scope for reporting business metrics in workflow logic. This should be used instead of
+   * creating new metrics scopes as it is able to dedup metrics during replay.
+   *
+   * <p>The original metrics scope is set through {@link WorkerOptions} when a worker starts up.
+   */
   public static Scope getMetricsScope() {
     return WorkflowInternal.getMetricsScope();
   }

src/test/java/com/uber/cadence/metrics/ReplayAwareScopeTest.java

@@ -1,3 +1,20 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file 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 com.uber.cadence.metrics;
 
 import static org.mockito.Mockito.mock;
@@ -13,9 +30,11 @@
 import com.uber.m3.tally.Gauge;
 import com.uber.m3.tally.Histogram;
 import com.uber.m3.tally.Scope;
+import com.uber.m3.tally.Stopwatch;
 import com.uber.m3.tally.Timer;
 import com.uber.m3.tally.ValueBuckets;
 import com.uber.m3.util.Duration;
+import java.util.function.Supplier;
 import org.junit.Test;
 
 public class ReplayAwareScopeTest {
@@ -48,7 +67,7 @@ public void testReplayAwareScopeReplaying() {
     when(scope.histogram("test-histogram", buckets)).thenReturn(histogram);
 
     TestContext context = new TestContext(true);
-    Scope replayAwareScope = new ReplayAwareScope(scope, context);
+    Scope replayAwareScope = new ReplayAwareScope(scope, context, System::currentTimeMillis);
 
     replayAwareScope.counter("test-counter").inc(1);
     replayAwareScope.gauge("test-gauge").update(100.0);
@@ -78,7 +97,7 @@ public void testReplayAwareScopeNotReplaying() {
     when(scope.histogram("test-histogram", buckets)).thenReturn(histogram);
 
     TestContext context = new TestContext(false);
-    Scope replayAwareScope = new ReplayAwareScope(scope, context);
+    Scope replayAwareScope = new ReplayAwareScope(scope, context, System::currentTimeMillis);
 
     replayAwareScope.counter("test-counter").inc(1);
     replayAwareScope.gauge("test-gauge").update(100.0);
@@ -92,4 +111,43 @@ public void testReplayAwareScopeNotReplaying() {
     verify(histogram, times(1)).recordValue(10);
     verify(histogram, times(1)).recordDuration(Duration.ofHours(1));
   }
+
+  static class TestClock implements Supplier<Long> {
+    private long currTime;
+
+    @Override
+    public Long get() {
+      return currTime;
+    }
+
+    void setTime(long currTime) {
+      this.currTime = currTime;
+    }
+  }
+
+  @Test
+  public void testCustomClockForTimer() {
+    Scope scope = mock(Scope.class);
+    Timer timer = mock(Timer.class);
+    Histogram histogram = mock(Histogram.class);
+
+    Buckets buckets = ValueBuckets.linear(0, 10, 10);
+    when(scope.timer("test-timer")).thenReturn(timer);
+    when(scope.histogram("test-histogram", buckets)).thenReturn(histogram);
+
+    TestContext context = new TestContext(false);
+    TestClock clock = new TestClock();
+    clock.setTime(0);
+    Scope replayAwareScope = new ReplayAwareScope(scope, context, clock);
+    Stopwatch sw = replayAwareScope.timer("test-timer").start();
+    clock.setTime(100);
+    sw.stop();
+
+    sw = replayAwareScope.histogram("test-histogram", buckets).start();
+    clock.setTime(150);
+    sw.stop();
+
+    verify(timer, times(1)).record(Duration.ofMillis(100));
+    verify(histogram, times(1)).recordDuration(Duration.ofMillis(50));
+  }
 }

src/test/java/com/uber/cadence/workflow/MetricsTest.java

@@ -1,5 +1,25 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file 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 com.uber.cadence.workflow;
 
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -13,8 +33,10 @@
 import com.uber.m3.tally.RootScopeBuilder;
 import com.uber.m3.tally.Scope;
 import com.uber.m3.tally.StatsReporter;
+import com.uber.m3.tally.Stopwatch;
 import java.time.Duration;
 import org.junit.Test;
+import org.mockito.ArgumentCaptor;
 
 public class MetricsTest {
   private static final String taskList = "metrics-test";
@@ -48,6 +70,11 @@ public static class TestMetricsInChildWorkflow implements TestChildWorkflow {
     @Override
     public void executeChild() {
       Workflow.getMetricsScope().counter("test-child-started").inc(1);
+
+      Stopwatch sw = Workflow.getMetricsScope().timer("test-timer").start();
+      Workflow.sleep(3000);
+      sw.stop();
+
       Workflow.getMetricsScope().counter("test-child-done").inc(1);
     }
   }
@@ -83,5 +110,13 @@ public void testWorkflowMetrics() throws InterruptedException {
     verify(reporter, times(1)).reportCounter("test-done", null, 1);
     verify(reporter, times(1)).reportCounter("test-child-started", null, 1);
     verify(reporter, times(1)).reportCounter("test-child-done", null, 1);
+
+    ArgumentCaptor<com.uber.m3.util.Duration> sleepDurationCaptor =
+        ArgumentCaptor.forClass(com.uber.m3.util.Duration.class);
+    verify(reporter, times(1)).reportTimer(eq("test-timer"), any(), sleepDurationCaptor.capture());
+
+    com.uber.m3.util.Duration sleepDuration = sleepDurationCaptor.getValue();
+    assertTrue(sleepDuration.compareTo(com.uber.m3.util.Duration.ofSeconds(3)) > 0);
+    assertTrue(sleepDuration.compareTo(com.uber.m3.util.Duration.ofMillis(3100)) < 0);
   }
 }