Merge pull request #149 from meiliang86/metrics2

expose metrics scope in workflow and de-dup metrics in replay mode

Author: meiliang86

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/NoopScope.java

@@ -19,6 +19,7 @@
 
 import com.uber.m3.tally.Buckets;
 import com.uber.m3.tally.Capabilities;
+import com.uber.m3.tally.CapableOf;
 import com.uber.m3.tally.RootScopeBuilder;
 import com.uber.m3.tally.Scope;
 import com.uber.m3.tally.StatsReporter;
@@ -57,17 +58,7 @@ public void reportHistogramDurationSamples(
 
     @Override
     public Capabilities capabilities() {
-      return new Capabilities() {
-        @Override
-        public boolean reporting() {
-          return true;
-        }
-
-        @Override
-        public boolean tagging() {
-          return true;
-        }
-      };
+      return CapableOf.REPORTING_TAGGING;
     }
 
     @Override

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

@@ -0,0 +1,199 @@
+/*
+ *  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.internal.metrics;
+
+import com.uber.cadence.internal.replay.ReplayAware;
+import com.uber.m3.tally.Buckets;
+import com.uber.m3.tally.Capabilities;
+import com.uber.m3.tally.Counter;
+import com.uber.m3.tally.Gauge;
+import com.uber.m3.tally.Histogram;
+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 {
+  private Scope scope;
+  private ReplayAware context;
+  private Supplier<Long> clock;
+
+  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 class ReplayAwareCounter implements Counter {
+    Counter counter;
+
+    ReplayAwareCounter(Counter counter) {
+      this.counter = Objects.requireNonNull(counter);
+    }
+
+    @Override
+    public void inc(long delta) {
+      if (context.isReplaying()) {
+        return;
+      }
+
+      counter.inc(delta);
+    }
+  }
+
+  private class ReplayAwareGauge implements Gauge {
+    Gauge gauge;
+
+    ReplayAwareGauge(Gauge gauge) {
+      this.gauge = Objects.requireNonNull(gauge);
+    }
+
+    @Override
+    public void update(double value) {
+      if (context.isReplaying()) {
+        return;
+      }
+
+      gauge.update(value);
+    }
+  }
+
+  private class ReplayAwareTimer implements Timer, DurationRecorder {
+    Timer timer;
+
+    ReplayAwareTimer(Timer timer) {
+      this.timer = Objects.requireNonNull(timer);
+    }
+
+    @Override
+    public void record(Duration interval) {
+      if (context.isReplaying()) {
+        return;
+      }
+
+      timer.record(interval);
+    }
+
+    @Override
+    public Stopwatch 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 class ReplayAwareHistogram implements Histogram, DurationRecorder {
+    Histogram histogram;
+
+    ReplayAwareHistogram(Histogram histogram) {
+      this.histogram = Objects.requireNonNull(histogram);
+    }
+
+    @Override
+    public void recordValue(double value) {
+      if (context.isReplaying()) {
+        return;
+      }
+
+      histogram.recordValue(value);
+    }
+
+    @Override
+    public void recordDuration(Duration value) {
+      if (context.isReplaying()) {
+        return;
+      }
+
+      histogram.recordDuration(value);
+    }
+
+    @Override
+    public Stopwatch 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));
+  }
+
+  @Override
+  public Gauge gauge(String name) {
+    return new ReplayAwareGauge(scope.gauge(name));
+  }
+
+  @Override
+  public Timer timer(String name) {
+    return new ReplayAwareTimer(scope.timer(name));
+  }
+
+  @Override
+  public Histogram histogram(String name, Buckets buckets) {
+    return new ReplayAwareHistogram(scope.histogram(name, buckets));
+  }
+
+  @Override
+  public Scope tagged(Map<String, String> tags) {
+    return new ReplayAwareScope(scope.tagged(tags), context, clock);
+  }
+
+  @Override
+  public Scope subScope(String name) {
+    return new ReplayAwareScope(scope.subScope(name), context, clock);
+  }
+
+  @Override
+  public Capabilities capabilities() {
+    return scope.capabilities();
+  }
+
+  @Override
+  public void close() throws ScopeCloseException {
+    scope.close();
+  }
+}

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

@@ -20,6 +20,7 @@
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowType;
 import com.uber.cadence.workflow.Promise;
+import com.uber.m3.tally.Scope;
 import java.time.Duration;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -28,7 +29,7 @@
  * Represents the context for decider. Should only be used within the scope of workflow definition
  * code, meaning any code which is not part of activity implementations.
  */
-public interface DecisionContext {
+public interface DecisionContext extends ReplayAware {
 
   WorkflowExecution getWorkflowExecution();
 
@@ -101,13 +102,6 @@ Consumer<Exception> signalWorkflowExecution(
    */
   long currentTimeMillis();
 
-  /**
-   * <code>true</code> indicates if workflow is replaying already processed events to reconstruct it
-   * state. <code>false</code> indicates that code is making forward process for the first time. For
-   * example can be used to avoid duplicating log records due to replay.
-   */
-  boolean isReplaying();
-
   /**
    * Create a Value that becomes ready after the specified delay.
    *
@@ -117,4 +111,6 @@ Consumer<Exception> signalWorkflowExecution(
    * @return cancellation handle. Invoke {@link Consumer#accept(Object)} to cancel timer.
    */
   Consumer<Exception> createTimer(long delaySeconds, Consumer<Exception> callback);
+
+  Scope getMetricsScope();
 }

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

@@ -24,8 +24,10 @@
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowExecutionStartedEventAttributes;
 import com.uber.cadence.WorkflowType;
+import com.uber.cadence.internal.metrics.ReplayAwareScope;
 import com.uber.cadence.workflow.Promise;
 import com.uber.cadence.workflow.Workflow;
+import com.uber.m3.tally.Scope;
 import java.time.Duration;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -40,6 +42,8 @@ final class DecisionContextImpl implements DecisionContext, HistoryEventHandler
 
   private final WorkflowContext workflowContext;
 
+  private Scope metricsScope;
+
   DecisionContextImpl(
       DecisionsHelper decisionsHelper,
       String domain,
@@ -51,6 +55,15 @@ final class DecisionContextImpl implements DecisionContext, HistoryEventHandler
     this.workflowClock = new ClockDecisionContext(decisionsHelper);
   }
 
+  public void setMetricsScope(Scope metricsScope) {
+    this.metricsScope = new ReplayAwareScope(metricsScope, this, workflowClock::currentTimeMillis);
+  }
+
+  @Override
+  public Scope getMetricsScope() {
+    return metricsScope;
+  }
+
   @Override
   public WorkflowExecution getWorkflowExecution() {
     return workflowContext.getWorkflowExecution();

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

@@ -0,0 +1,27 @@
+/*
+ *  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.internal.replay;
+
+public interface ReplayAware {
+  /**
+   * <code>true</code> indicates if workflow is replaying already processed events to reconstruct it
+   * state. <code>false</code> indicates that code is making forward process for the first time. For
+   * example can be used to avoid duplicating log records due to replay.
+   */
+  boolean isReplaying();
+}

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

@@ -88,6 +88,7 @@ class ReplayDecider {
             domain,
             decisionTask,
             historyHelper.getWorkflowExecutionStartedEventAttributes());
+    context.setMetricsScope(metricsScope);
   }
 
   public boolean isCancelRequested() {

src/main/java/com/uber/cadence/internal/sync/DeterministicRunnerImpl.java

@@ -27,6 +27,7 @@
 import com.uber.cadence.internal.replay.SignalExternalWorkflowParameters;
 import com.uber.cadence.internal.replay.StartChildWorkflowExecutionParameters;
 import com.uber.cadence.workflow.Promise;
+import com.uber.m3.tally.Scope;
 import java.time.Duration;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
@@ -516,5 +517,10 @@ public boolean isReplaying() {
     public Consumer<Exception> createTimer(long delaySeconds, Consumer<Exception> callback) {
       throw new UnsupportedOperationException("not implemented");
     }
+
+    @Override
+    public Scope getMetricsScope() {
+      throw new UnsupportedOperationException("not implemented");
+    }
   }
 }

src/main/java/com/uber/cadence/internal/sync/POJOWorkflowImplementationFactory.java

@@ -176,7 +176,7 @@ public void setDataConverter(DataConverter dataConverter) {
   }
 
   @Override
-  public ReplayWorkflow getWorkflow(WorkflowType workflowType) throws Exception {
+  public ReplayWorkflow getWorkflow(WorkflowType workflowType) {
     SyncWorkflowDefinition workflow = getWorkflowDefinition(workflowType);
     return new SyncWorkflow(workflow, dataConverter, threadPool, interceptorFactory);
   }

src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java

@@ -47,6 +47,7 @@
 import com.uber.cadence.workflow.SignalExternalWorkflowException;
 import com.uber.cadence.workflow.Workflow;
 import com.uber.cadence.workflow.WorkflowInterceptor;
+import com.uber.m3.tally.Scope;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
@@ -479,4 +480,8 @@ private RuntimeException mapSignalWorkflowException(Exception failure) {
     }
     return (SignalExternalWorkflowException) failure;
   }
+
+  public Scope getMetricsScope() {
+    return context.getMetricsScope();
+  }
 }