expose metrics scope in workflow and dedup metrics in replay mode

Author: Liang Mei

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,183 @@
+/*
+ *  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.Timer;
+import com.uber.m3.util.Duration;
+import java.util.Map;
+import java.util.Objects;
+
+public class ReplayAwareScope implements Scope {
+  Scope scope;
+  ReplayAware context;
+
+  public ReplayAwareScope(Scope scope, ReplayAware context) {
+    this.scope = Objects.requireNonNull(scope);
+    this.context = Objects.requireNonNull(context);
+  }
+
+  private static class ReplayAwareCounter implements Counter {
+    Counter counter;
+    ReplayAware context;
+
+    public ReplayAwareCounter(Counter counter, ReplayAware context) {
+      this.counter = Objects.requireNonNull(counter);
+      this.context = Objects.requireNonNull(context);
+    }
+
+    @Override
+    public void inc(long delta) {
+      if (context.isReplaying()) {
+        return;
+      }
+
+      counter.inc(delta);
+    }
+  }
+
+  private static class ReplayAwareGauge implements Gauge {
+    Gauge gauge;
+    ReplayAware context;
+
+    public ReplayAwareGauge(Gauge gauge, ReplayAware context) {
+      this.gauge = Objects.requireNonNull(gauge);
+      this.context = Objects.requireNonNull(context);
+    }
+
+    @Override
+    public void update(double value) {
+      if (context.isReplaying()) {
+        return;
+      }
+
+      gauge.update(value);
+    }
+  }
+
+  private static class ReplayAwareTimer implements Timer {
+    Timer timer;
+    ReplayAware context;
+
+    public ReplayAwareTimer(Timer timer, ReplayAware context) {
+      this.timer = Objects.requireNonNull(timer);
+      this.context = Objects.requireNonNull(context);
+    }
+
+    @Override
+    public void record(Duration interval) {
+      if (context.isReplaying()) {
+        return;
+      }
+
+      timer.record(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();
+    }
+  }
+
+  private static class ReplayAwareHistogram implements Histogram {
+    Histogram histogram;
+    ReplayAware context;
+
+    public ReplayAwareHistogram(Histogram histogram, ReplayAware context) {
+      this.histogram = Objects.requireNonNull(histogram);
+      this.context = Objects.requireNonNull(context);
+    }
+
+    @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() {
+      // 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();
+    }
+  }
+
+  @Override
+  public Counter counter(String name) {
+    return new ReplayAwareCounter(scope.counter(name), context);
+  }
+
+  @Override
+  public Gauge gauge(String name) {
+    return new ReplayAwareGauge(scope.gauge(name), context);
+  }
+
+  @Override
+  public Timer timer(String name) {
+    return new ReplayAwareTimer(scope.timer(name), context);
+  }
+
+  @Override
+  public Histogram histogram(String name, Buckets buckets) {
+    return new ReplayAwareHistogram(scope.histogram(name, buckets), context);
+  }
+
+  @Override
+  public Scope tagged(Map<String, String> tags) {
+    return new ReplayAwareScope(scope.tagged(tags), context);
+  }
+
+  @Override
+  public Scope subScope(String name) {
+    return new ReplayAwareScope(scope.subScope(name), context);
+  }
+
+  @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);
+  }
+
+  @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();
+  }
 }

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

@@ -117,6 +117,7 @@ public Worker newWorker(String taskList) {
             taskList,
             new WorkerOptions.Builder()
                 .setInterceptorFactory(testEnvironmentOptions.getInterceptorFactory())
+                .setMetricsScope(testEnvironmentOptions.getMetricsScope())
                 .build());
     workers.add(result);
     return result;
@@ -128,6 +129,7 @@ public WorkflowClient newWorkflowClient() {
         new WorkflowClientOptions.Builder()
             .setDataConverter(testEnvironmentOptions.getDataConverter())
             .setInterceptors(new TimeLockingInterceptor(service))
+            .setMetricsScope(testEnvironmentOptions.getMetricsScope())
             .build();
     return WorkflowClientInternal.newInstance(service, testEnvironmentOptions.getDomain(), options);
   }