implement replay aware logging (#153)

meiliang86 · web-flow · commit f6bf98a4d8a0 · 2018-05-01T13:30:48.000-07:00
* implement replay aware logging

* review comments

* use list appender in logging test

* remove an unnecessary line in logger test
diff --git a/src/main/java/com/uber/cadence/internal/logging/ReplayAwareLogger.java b/src/main/java/com/uber/cadence/internal/logging/ReplayAwareLogger.java
diff --git a/src/main/java/com/uber/cadence/internal/replay/DecisionContext.java b/src/main/java/com/uber/cadence/internal/replay/DecisionContext.java
@@ -112,5 +112,9 @@ Consumer<Exception> signalWorkflowExecution(
    */
   Consumer<Exception> createTimer(long delaySeconds, Consumer<Exception> callback);
 
+  /** @return scope to be used for metrics reporting. */
   Scope getMetricsScope();
+
+  /** @return whether we do logging during decision replay. */
+  boolean getEnableLoggingInReplay();
 }
diff --git a/src/main/java/com/uber/cadence/internal/replay/DecisionContextImpl.java b/src/main/java/com/uber/cadence/internal/replay/DecisionContextImpl.java
@@ -44,21 +44,30 @@ final class DecisionContextImpl implements DecisionContext, HistoryEventHandler
 
   private Scope metricsScope;
 
+  private final boolean enableLoggingInReplay;
+
   DecisionContextImpl(
       DecisionsHelper decisionsHelper,
       String domain,
       PollForDecisionTaskResponse decisionTask,
-      WorkflowExecutionStartedEventAttributes startedAttributes) {
+      WorkflowExecutionStartedEventAttributes startedAttributes,
+      boolean enableLoggingInReplay) {
     this.activityClient = new ActivityDecisionContext(decisionsHelper);
     this.workflowContext = new WorkflowContext(domain, decisionTask, startedAttributes);
     this.workflowClient = new WorkflowDecisionContext(decisionsHelper, workflowContext);
     this.workflowClock = new ClockDecisionContext(decisionsHelper);
+    this.enableLoggingInReplay = enableLoggingInReplay;
   }
 
   public void setMetricsScope(Scope metricsScope) {
     this.metricsScope = new ReplayAwareScope(metricsScope, this, workflowClock::currentTimeMillis);
   }
 
+  @Override
+  public boolean getEnableLoggingInReplay() {
+    return enableLoggingInReplay;
+  }
+
   @Override
   public Scope getMetricsScope() {
     return metricsScope;
diff --git a/src/main/java/com/uber/cadence/internal/replay/ReplayDecider.java b/src/main/java/com/uber/cadence/internal/replay/ReplayDecider.java
@@ -76,7 +76,8 @@ class ReplayDecider {
       ReplayWorkflow workflow,
       HistoryHelper historyHelper,
       DecisionsHelper decisionsHelper,
-      Scope metricsScope) {
+      Scope metricsScope,
+      boolean enableLoggingInReplay) {
     this.workflow = workflow;
     this.historyHelper = historyHelper;
     this.decisionsHelper = decisionsHelper;
@@ -87,7 +88,8 @@ class ReplayDecider {
             decisionsHelper,
             domain,
             decisionTask,
-            historyHelper.getWorkflowExecutionStartedEventAttributes());
+            historyHelper.getWorkflowExecutionStartedEventAttributes(),
+            enableLoggingInReplay);
     context.setMetricsScope(metricsScope);
   }
 
diff --git a/src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java b/src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java
@@ -29,6 +29,7 @@
 import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.internal.worker.DecisionTaskHandler;
 import com.uber.cadence.internal.worker.DecisionTaskWithHistoryIterator;
+import com.uber.cadence.internal.worker.SingleWorkerOptions;
 import com.uber.m3.tally.Scope;
 import java.io.PrintWriter;
 import java.io.StringWriter;
@@ -44,12 +45,14 @@ public final class ReplayDecisionTaskHandler implements DecisionTaskHandler {
   private final ReplayWorkflowFactory workflowFactory;
   private final String domain;
   private final Scope metricsScope;
+  private final boolean enableLoggingInReplay;
 
   public ReplayDecisionTaskHandler(
-      String domain, ReplayWorkflowFactory asyncWorkflowFactory, Scope metricsScope) {
+      String domain, ReplayWorkflowFactory asyncWorkflowFactory, SingleWorkerOptions options) {
     this.domain = domain;
     this.workflowFactory = asyncWorkflowFactory;
-    this.metricsScope = metricsScope;
+    this.metricsScope = options.getMetricsScope();
+    this.enableLoggingInReplay = options.getEnableLoggingInReplay();
   }
 
   @Override
@@ -161,6 +164,7 @@ private ReplayDecider createDecider(HistoryHelper historyHelper) throws Exceptio
     WorkflowType workflowType = decisionTask.getWorkflowType();
     DecisionsHelper decisionsHelper = new DecisionsHelper(decisionTask);
     ReplayWorkflow workflow = workflowFactory.getWorkflow(workflowType);
-    return new ReplayDecider(domain, workflow, historyHelper, decisionsHelper, metricsScope);
+    return new ReplayDecider(
+        domain, workflow, historyHelper, decisionsHelper, metricsScope, enableLoggingInReplay);
   }
 }
diff --git a/src/main/java/com/uber/cadence/internal/sync/DeterministicRunnerImpl.java b/src/main/java/com/uber/cadence/internal/sync/DeterministicRunnerImpl.java
@@ -522,5 +522,10 @@ public Consumer<Exception> createTimer(long delaySeconds, Consumer<Exception> ca
     public Scope getMetricsScope() {
       throw new UnsupportedOperationException("not implemented");
     }
+
+    @Override
+    public boolean getEnableLoggingInReplay() {
+      return false;
+    }
   }
 }
diff --git a/src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java b/src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java
@@ -484,4 +484,8 @@ private RuntimeException mapSignalWorkflowException(Exception failure) {
   public Scope getMetricsScope() {
     return context.getMetricsScope();
   }
+
+  public boolean isLoggingEnabledInReplay() {
+    return context.getEnableLoggingInReplay();
+  }
 }
diff --git a/src/main/java/com/uber/cadence/internal/sync/SyncWorkflowWorker.java b/src/main/java/com/uber/cadence/internal/sync/SyncWorkflowWorker.java
@@ -55,8 +55,7 @@ public SyncWorkflowWorker(
     factory =
         new POJOWorkflowImplementationFactory(
             options.getDataConverter(), workflowThreadPool, interceptorFactory);
-    DecisionTaskHandler taskHandler =
-        new ReplayDecisionTaskHandler(domain, factory, options.getMetricsScope());
+    DecisionTaskHandler taskHandler = new ReplayDecisionTaskHandler(domain, factory, options);
     worker = new WorkflowWorker(service, domain, taskList, options, taskHandler);
     this.options = options;
   }
diff --git a/src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java b/src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java
@@ -118,6 +118,7 @@ public Worker newWorker(String taskList) {
             new WorkerOptions.Builder()
                 .setInterceptorFactory(testEnvironmentOptions.getInterceptorFactory())
                 .setMetricsScope(testEnvironmentOptions.getMetricsScope())
+                .setEnableLoggingInReplay(testEnvironmentOptions.isLoggingEnabledInReplay())
                 .build());
     workers.add(result);
     return result;
diff --git a/src/main/java/com/uber/cadence/internal/sync/WorkflowInternal.java b/src/main/java/com/uber/cadence/internal/sync/WorkflowInternal.java
@@ -25,6 +25,7 @@
 import com.uber.cadence.common.RetryOptions;
 import com.uber.cadence.internal.common.CheckedExceptionWrapper;
 import com.uber.cadence.internal.common.InternalUtils;
+import com.uber.cadence.internal.logging.ReplayAwareLogger;
 import com.uber.cadence.workflow.ActivityStub;
 import com.uber.cadence.workflow.CancellationScope;
 import com.uber.cadence.workflow.ChildWorkflowOptions;
@@ -48,6 +49,8 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Never reference directly. It is public only because Java doesn't have internal package support.
@@ -312,4 +315,20 @@ public static void sleep(Duration duration) {
   public static Scope getMetricsScope() {
     return getRootDecisionContext().getMetricsScope();
   }
+
+  private static boolean isLoggingEnabledInReplay() {
+    return getRootDecisionContext().isLoggingEnabledInReplay();
+  }
+
+  public static Logger getLogger(Class<?> clazz) {
+    Logger logger = LoggerFactory.getLogger(clazz);
+    return new ReplayAwareLogger(
+        logger, WorkflowInternal::isReplaying, WorkflowInternal::isLoggingEnabledInReplay);
+  }
+
+  public static Logger getLogger(String name) {
+    Logger logger = LoggerFactory.getLogger(name);
+    return new ReplayAwareLogger(
+        logger, WorkflowInternal::isReplaying, WorkflowInternal::isLoggingEnabledInReplay);
+  }
 }
diff --git a/src/main/java/com/uber/cadence/internal/worker/SingleWorkerOptions.java b/src/main/java/com/uber/cadence/internal/worker/SingleWorkerOptions.java
@@ -43,6 +43,8 @@ public static final class Builder {
 
     private Scope metricsScope;
 
+    private boolean enableLoggingInReplay;
+
     public Builder setIdentity(String identity) {
       this.identity = identity;
       return this;
@@ -68,6 +70,11 @@ public Builder setMetricsScope(Scope metricsScope) {
       return this;
     }
 
+    public Builder setEnableLoggingInReplay(boolean enableLoggingInReplay) {
+      this.enableLoggingInReplay = enableLoggingInReplay;
+      return this;
+    }
+
     public SingleWorkerOptions build() {
       if (reportCompletionRetryOptions == null) {
         reportCompletionRetryOptions =
@@ -111,7 +118,8 @@ public SingleWorkerOptions build() {
           pollerOptions,
           reportCompletionRetryOptions,
           reportFailureRetryOptions,
-          metricsScope);
+          metricsScope,
+          enableLoggingInReplay);
     }
 
     public Builder setReportCompletionRetryOptions(RetryOptions reportCompletionRetryOptions) {
@@ -139,21 +147,25 @@ public Builder setReportFailureRetryOptions(RetryOptions reportFailureRetryOptio
 
   private final Scope metricsScope;
 
+  private final boolean enableLoggingInReplay;
+
   private SingleWorkerOptions(
       String identity,
       DataConverter dataConverter,
       int taskExecutorThreadPoolSize,
       PollerOptions pollerOptions,
       RetryOptions reportCompletionRetryOptions,
       RetryOptions reportFailureRetryOptions,
-      Scope metricsScope) {
+      Scope metricsScope,
+      boolean enableLoggingInReplay) {
     this.identity = identity;
     this.dataConverter = dataConverter;
     this.taskExecutorThreadPoolSize = taskExecutorThreadPoolSize;
     this.pollerOptions = pollerOptions;
     this.reportCompletionRetryOptions = reportCompletionRetryOptions;
     this.reportFailureRetryOptions = reportFailureRetryOptions;
     this.metricsScope = metricsScope;
+    this.enableLoggingInReplay = enableLoggingInReplay;
   }
 
   public String getIdentity() {
@@ -183,4 +195,8 @@ public RetryOptions getReportFailureRetryOptions() {
   public Scope getMetricsScope() {
     return metricsScope;
   }
+
+  public boolean getEnableLoggingInReplay() {
+    return enableLoggingInReplay;
+  }
 }
diff --git a/src/main/java/com/uber/cadence/testing/TestEnvironmentOptions.java b/src/main/java/com/uber/cadence/testing/TestEnvironmentOptions.java
@@ -39,6 +39,8 @@ public static final class Builder {
 
     private Scope metricsScope;
 
+    private boolean enableLoggingInReplay;
+
     /** Sets data converter to use for unit-tests. Default is {@link JsonDataConverter}. */
     public Builder setDataConverter(DataConverter dataConverter) {
       this.dataConverter = Objects.requireNonNull(dataConverter);
@@ -70,32 +72,39 @@ public Builder setMetricsScope(Scope metricsScope) {
       return this;
     }
 
+    /** Set whether to log during decision replay. */
+    public Builder setEnableLoggingInReplay(boolean enableLoggingInReplay) {
+      this.enableLoggingInReplay = enableLoggingInReplay;
+      return this;
+    }
+
     public TestEnvironmentOptions build() {
       if (metricsScope == null) {
         metricsScope = NoopScope.getInstance();
       }
 
-      return new TestEnvironmentOptions(dataConverter, domain, interceptorFactory, metricsScope);
+      return new TestEnvironmentOptions(
+          dataConverter, domain, interceptorFactory, metricsScope, enableLoggingInReplay);
     }
   }
 
   private final DataConverter dataConverter;
-
   private final String domain;
-
   private final Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory;
-
   private final Scope metricsScope;
+  private final boolean enableLoggingInReplay;
 
   private TestEnvironmentOptions(
       DataConverter dataConverter,
       String domain,
       Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
-      Scope metricsScope) {
+      Scope metricsScope,
+      boolean enableLoggingInReplay) {
     this.dataConverter = dataConverter;
     this.domain = domain;
     this.interceptorFactory = interceptorFactory;
     this.metricsScope = metricsScope;
+    this.enableLoggingInReplay = enableLoggingInReplay;
   }
 
   public DataConverter getDataConverter() {
@@ -114,6 +123,10 @@ public Scope getMetricsScope() {
     return metricsScope;
   }
 
+  public boolean isLoggingEnabledInReplay() {
+    return enableLoggingInReplay;
+  }
+
   @Override
   public String toString() {
     return "TestEnvironmentOptions{"
diff --git a/src/main/java/com/uber/cadence/worker/Worker.java b/src/main/java/com/uber/cadence/worker/Worker.java
@@ -145,6 +145,7 @@ private SingleWorkerOptions toActivityOptions(WorkerOptions options) {
         .setReportFailureRetryOptions(options.getReportActivityFailureRetryOptions())
         .setTaskExecutorThreadPoolSize(options.getMaxConcurrentActivityExecutionSize())
         .setMetricsScope(options.getMetricsScope())
+        .setEnableLoggingInReplay(options.getEnableLoggingInReplay())
         .build();
   }
 
@@ -157,6 +158,7 @@ private SingleWorkerOptions toWorkflowOptions(WorkerOptions options) {
         .setReportFailureRetryOptions(options.getReportWorkflowFailureRetryOptions())
         .setTaskExecutorThreadPoolSize(options.getMaxConcurrentWorklfowExecutionSize())
         .setMetricsScope(options.getMetricsScope())
+        .setEnableLoggingInReplay(options.getEnableLoggingInReplay())
         .build();
   }
 
diff --git a/src/main/java/com/uber/cadence/worker/WorkerOptions.java b/src/main/java/com/uber/cadence/worker/WorkerOptions.java
@@ -48,6 +48,7 @@ public static final class Builder {
     private RetryOptions reportWorkflowFailureRetryOptions;
     private Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory = (n) -> n;
     private Scope metricsScope;
+    private boolean enableLoggingInReplay;
 
     /**
      * When set to true doesn't poll on workflow task list even if there are registered workflows
@@ -178,6 +179,11 @@ public Builder setMetricsScope(Scope metricsScope) {
       return this;
     }
 
+    public Builder setEnableLoggingInReplay(boolean enableLoggingInReplay) {
+      this.enableLoggingInReplay = enableLoggingInReplay;
+      return this;
+    }
+
     public WorkerOptions build() {
       if (identity == null) {
         identity = ManagementFactory.getRuntimeMXBean().getName();
@@ -203,7 +209,8 @@ public WorkerOptions build() {
           reportWorkflowCompletionRetryOptions,
           reportWorkflowFailureRetryOptions,
           interceptorFactory,
-          metricsScope);
+          metricsScope,
+          enableLoggingInReplay);
     }
   }
 
@@ -223,6 +230,7 @@ public WorkerOptions build() {
   private final RetryOptions reportWorkflowFailureRetryOptions;
   private final Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory;
   private final Scope metricsScope;
+  private final boolean enableLoggingInReplay;
 
   private WorkerOptions(
       boolean disableWorkflowWorker,
@@ -240,7 +248,8 @@ private WorkerOptions(
       RetryOptions reportWorkflowCompletionRetryOptions,
       RetryOptions reportWorkflowFailureRetryOptions,
       Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
-      Scope metricsScope) {
+      Scope metricsScope,
+      boolean enableLoggingInReplay) {
     this.disableWorkflowWorker = disableWorkflowWorker;
     this.disableActivityWorker = disableActivityWorker;
     this.workerActivitiesPerSecond = workerActivitiesPerSecond;
@@ -257,6 +266,7 @@ private WorkerOptions(
     this.reportWorkflowFailureRetryOptions = reportWorkflowFailureRetryOptions;
     this.interceptorFactory = interceptorFactory;
     this.metricsScope = metricsScope;
+    this.enableLoggingInReplay = enableLoggingInReplay;
   }
 
   public boolean isDisableWorkflowWorker() {
@@ -323,6 +333,10 @@ public Scope getMetricsScope() {
     return metricsScope;
   }
 
+  public boolean getEnableLoggingInReplay() {
+    return enableLoggingInReplay;
+  }
+
   @Override
   public String toString() {
     return "WorkerOptions{"
diff --git a/src/main/java/com/uber/cadence/workflow/Workflow.java b/src/main/java/com/uber/cadence/workflow/Workflow.java
diff --git a/src/test/java/com/uber/cadence/workflow/LoggerTest.java b/src/test/java/com/uber/cadence/workflow/LoggerTest.java

Original file line number	Diff line number	Diff line change
`@@ -112,5 +112,9 @@ Consumer<Exception> signalWorkflowExecution(`
`112`	`112`	`*/`
`113`	`113`	`Consumer<Exception> createTimer(long delaySeconds, Consumer<Exception> callback);`
`114`	`114`
	`115`	`+ /** @return scope to be used for metrics reporting. */`
`115`	`116`	`Scope getMetricsScope();`
	`117`	`+`
	`118`	`+ /** @return whether we do logging during decision replay. */`
	`119`	`+ boolean getEnableLoggingInReplay();`
`116`	`120`	`}`
Original file line number	Diff line number	Diff line change
`@@ -522,5 +522,10 @@ public Consumer<Exception> createTimer(long delaySeconds, Consumer<Exception> ca`
`522`	`522`	`public Scope getMetricsScope() {`
`523`	`523`	`throw new UnsupportedOperationException("not implemented");`
`524`	`524`	`}`
	`525`	`+`
	`526`	`+ @Override`
	`527`	`+ public boolean getEnableLoggingInReplay() {`
	`528`	`+ return false;`
	`529`	`+ }`
`525`	`530`	`}`
`526`	`531`	`}`
Original file line number	Diff line number	Diff line change
`@@ -484,4 +484,8 @@ private RuntimeException mapSignalWorkflowException(Exception failure) {`
`484`	`484`	`public Scope getMetricsScope() {`
`485`	`485`	`return context.getMetricsScope();`
`486`	`486`	`}`
	`487`	`+`
	`488`	`+ public boolean isLoggingEnabledInReplay() {`
	`489`	`+ return context.getEnableLoggingInReplay();`
	`490`	`+ }`
`487`	`491`	`}`