Merge pull request #180 from halakaraki/add-metrics-corrupted-signals

add metric for corrupted signal

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

@@ -121,6 +121,9 @@ public class MetricsType {
   public static final String UNHANDLED_SIGNALS_COUNTER =
       CADENCE_METRICS_PREFIX + "unhandled-signals";
 
+  public static final String CORRUPTED_SIGNALS_COUNTER =
+      CADENCE_METRICS_PREFIX + "corrupted-signals";
+
   public static final String WORKER_START_COUNTER = CADENCE_METRICS_PREFIX + "worker-start";
   public static final String POLLER_START_COUNTER = CADENCE_METRICS_PREFIX + "poller-start";
 

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

@@ -23,6 +23,7 @@
 import com.uber.cadence.converter.DataConverterException;
 import com.uber.cadence.internal.common.CheckedExceptionWrapper;
 import com.uber.cadence.internal.common.InternalUtils;
+import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.internal.replay.ReplayWorkflow;
 import com.uber.cadence.internal.replay.ReplayWorkflowFactory;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
@@ -35,6 +36,7 @@
 import com.uber.cadence.workflow.WorkflowInfo;
 import com.uber.cadence.workflow.WorkflowInterceptor;
 import com.uber.cadence.workflow.WorkflowMethod;
+import com.uber.m3.tally.Scope;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Collections;
@@ -64,14 +66,17 @@ final class POJOWorkflowImplementationFactory implements ReplayWorkflowFactory {
       Collections.synchronizedMap(new HashMap<>());
 
   private final ExecutorService threadPool;
+  private final Scope metricsScope;
 
   POJOWorkflowImplementationFactory(
       DataConverter dataConverter,
       ExecutorService threadPool,
-      Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory) {
+      Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
+      Scope metricsScope) {
     this.dataConverter = Objects.requireNonNull(dataConverter);
     this.threadPool = Objects.requireNonNull(threadPool);
     this.interceptorFactory = Objects.requireNonNull(interceptorFactory);
+    this.metricsScope = metricsScope;
   }
 
   void setWorkflowImplementationTypes(Class<?>[] workflowImplementationTypes) {
@@ -279,22 +284,19 @@ public void processSignal(String signalName, byte[] input, long eventId) {
                 + signalHandlers.keySet());
         return;
       }
-      Object[] args = dataConverter.fromDataArray(input, signalMethod.getGenericParameterTypes());
+
       try {
+        Object[] args = dataConverter.fromDataArray(input, signalMethod.getGenericParameterTypes());
         newInstance();
         signalMethod.invoke(workflow, args);
       } catch (IllegalAccessException e) {
         throw new Error("Failure processing \"" + signalName + "\" at eventID " + eventId, e);
+      } catch (DataConverterException e){
+        logSerializationException(signalName, eventId, e);
       } catch (InvocationTargetException e) {
         Throwable targetException = e.getTargetException();
         if (targetException instanceof DataConverterException) {
-          log.error(
-              "Failure deserializing signal input for \""
-                  + signalName
-                  + "\" at eventID "
-                  + eventId
-                  + ". Dropping it.",
-              targetException);
+          logSerializationException(signalName, eventId, (DataConverterException)targetException);
         } else {
           throw new Error(
               "Failure processing \"" + signalName + "\" at eventID " + eventId, targetException);
@@ -303,6 +305,17 @@ public void processSignal(String signalName, byte[] input, long eventId) {
     }
   }
 
+  void logSerializationException(String signalName, Long eventId, DataConverterException exception){
+    log.error(
+            "Failure deserializing signal input for \""
+                    + signalName
+                    + "\" at eventID "
+                    + eventId
+                    + ". Dropping it.",
+            exception);
+    metricsScope.counter(MetricsType.CORRUPTED_SIGNALS_COUNTER).inc(1);
+  }
+
   static WorkflowExecutionException mapToWorkflowExecutionException(
       Exception failure, DataConverter dataConverter) {
     failure = CheckedExceptionWrapper.unwrap(failure);

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

@@ -55,7 +55,10 @@ public SyncWorkflowWorker(
             new SynchronousQueue<>());
     factory =
         new POJOWorkflowImplementationFactory(
-            options.getDataConverter(), workflowThreadPool, interceptorFactory);
+            options.getDataConverter(),
+            workflowThreadPool,
+            interceptorFactory,
+            options.getMetricsScope());
     DecisionTaskHandler taskHandler = new ReplayDecisionTaskHandler(domain, factory, options);
     worker = new WorkflowWorker(service, domain, taskList, options, taskHandler);
     this.options = options;

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

@@ -86,6 +86,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.function.Function;
 import org.apache.thrift.TException;
 import org.apache.thrift.async.AsyncMethodCallback;
 import org.slf4j.Logger;
@@ -111,16 +112,22 @@ public TestWorkflowEnvironmentInternal(TestEnvironmentOptions options) {
 
   @Override
   public Worker newWorker(String taskList) {
+    return newWorker(taskList, x->x);
+  }
+
+  @Override
+  public Worker newWorker(String taskList, Function<WorkerOptions.Builder,WorkerOptions.Builder> overrideOptions) {
     WorkerOptions.Builder builder =
-        new WorkerOptions.Builder()
-            .setInterceptorFactory(testEnvironmentOptions.getInterceptorFactory())
-            .setMetricsScope(testEnvironmentOptions.getMetricsScope())
-            .setEnableLoggingInReplay(testEnvironmentOptions.isLoggingEnabledInReplay());
+            new WorkerOptions.Builder()
+                    .setInterceptorFactory(testEnvironmentOptions.getInterceptorFactory())
+                    .setMetricsScope(testEnvironmentOptions.getMetricsScope())
+                    .setEnableLoggingInReplay(testEnvironmentOptions.isLoggingEnabledInReplay());
     if (testEnvironmentOptions.getDataConverter() != null) {
       builder.setDataConverter(testEnvironmentOptions.getDataConverter());
     }
+    builder = overrideOptions.apply(builder);
     Worker result =
-        new Worker(service, testEnvironmentOptions.getDomain(), taskList, builder.build());
+            new Worker(service, testEnvironmentOptions.getDomain(), taskList, builder.build());
     workers.add(result);
     return result;
   }

src/main/java/com/uber/cadence/testing/TestWorkflowEnvironment.java

@@ -23,7 +23,9 @@
 import com.uber.cadence.internal.sync.TestWorkflowEnvironmentInternal;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.cadence.worker.Worker;
+import com.uber.cadence.worker.WorkerOptions;
 import java.time.Duration;
+import java.util.function.Function;
 
 /**
  * TestWorkflowEnvironment provides workflow unit testing capabilities.
@@ -104,6 +106,15 @@ static TestWorkflowEnvironment newInstance(TestEnvironmentOptions options) {
    */
   Worker newWorker(String taskList);
 
+  /**
+   * Creates a new Worker instance that is connected to the in-memory test Cadence service. {@link
+   * #close()} calls {@link Worker#shutdown(Duration)} for all workers created through this method.
+   *
+   * @param taskList task list to poll.
+   * @param overrideOptions is used to override the default worker options.
+   */
+  Worker newWorker(String taskList, Function<WorkerOptions.Builder,WorkerOptions.Builder> overrideOptions);
+
   /** Creates a WorkflowClient that is connected to the in-memory test Cadence service. */
   WorkflowClient newWorkflowClient();
 

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

@@ -20,25 +20,26 @@
 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;
+import static org.mockito.Mockito.*;
 
 import com.uber.cadence.client.WorkflowClient;
 import com.uber.cadence.client.WorkflowOptions;
 import com.uber.cadence.internal.metrics.MetricsTag;
+import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.testing.TestEnvironmentOptions;
 import com.uber.cadence.testing.TestEnvironmentOptions.Builder;
 import com.uber.cadence.testing.TestWorkflowEnvironment;
 import com.uber.cadence.worker.Worker;
+import com.uber.cadence.workflow.interceptors.SignalWorkflowInterceptor;
 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 com.uber.m3.util.ImmutableMap;
 import java.time.Duration;
 import java.util.Map;
-import org.junit.Before;
+import java.util.function.Function;
+
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TestWatcher;
@@ -103,21 +104,79 @@ public void executeChild() {
     }
   }
 
-  @Before
-  public void setUp() {
+  public interface ReceiveSignalObjectChildWorkflow {
+
+    @WorkflowMethod
+    String execute();
+
+    @SignalMethod(name = "testSignal")
+    void signal(Signal arg);
+
+    @SignalMethod(name = "endWorkflow")
+    void close();
+  }
+
+  public static class ReceiveSignalObjectChildWorkflowImpl implements ReceiveSignalObjectChildWorkflow {
+    private String receivedSignal = "Initial State";
+    // Keep workflow open so that we can send signal
+    CompletablePromise<Void> promise = Workflow.newPromise();
+    @Override
+    public String execute() {
+      promise.get();
+      return receivedSignal;
+    }
+
+    @Override
+    public void signal(Signal arg) {
+      receivedSignal = arg.value;
+    }
+
+    @Override
+    public void close() {
+      promise.complete(null);
+    }
+  }
+
+  public interface SendSignalObjectWorkflow {
+
+    @WorkflowMethod
+    String execute();
+  }
+
+  public static class SendSignalObjectWorkflowImpl implements SendSignalObjectWorkflow {
+    @Override
+    public String execute() {
+      ReceiveSignalObjectChildWorkflow child =
+              Workflow.newChildWorkflowStub(ReceiveSignalObjectChildWorkflow.class);
+      Promise<String> greeting = Async.function(child::execute);
+      Signal sig = new Signal();
+      sig.value = "Hello World";
+      child.signal(sig);
+      child.close();
+      return greeting.get();
+    }
+  }
+
+  public static class Signal {
+
+    public String value;
+  }
+
+  public void setUp(com.uber.m3.util.Duration reportingFrequecy){
     reporter = mock(StatsReporter.class);
     Scope scope =
-        new RootScopeBuilder()
-            .reporter(reporter)
-            .reportEvery(com.uber.m3.util.Duration.ofMillis(10));
+            new RootScopeBuilder()
+                    .reporter(reporter)
+                    .reportEvery(reportingFrequecy);
 
     TestEnvironmentOptions testOptions =
-        new Builder().setDomain(WorkflowTest.DOMAIN).setMetricsScope(scope).build();
+            new Builder().setDomain(WorkflowTest.DOMAIN).setMetricsScope(scope).build();
     testEnvironment = TestWorkflowEnvironment.newInstance(testOptions);
   }
 
   @Test
   public void testWorkflowMetrics() throws InterruptedException {
+    setUp(com.uber.m3.util.Duration.ofMillis(10));
 
     Worker worker = testEnvironment.newWorker(taskList);
     worker.registerWorkflowImplementationTypes(
@@ -158,4 +217,50 @@ public void testWorkflowMetrics() throws InterruptedException {
         sleepDuration.toString(),
         sleepDuration.compareTo(com.uber.m3.util.Duration.ofMillis(3100)) < 0);
   }
+
+  @Test
+  public void testCorruptedSignalMetrics() throws InterruptedException {
+    setUp(com.uber.m3.util.Duration.ofMillis(300));
+
+    Worker worker = testEnvironment.newWorker(taskList, builder ->
+            builder.setInterceptorFactory(new CorruptedSignalWorkflowInterceptorFactory()));
+
+    worker.registerWorkflowImplementationTypes(
+            SendSignalObjectWorkflowImpl.class, ReceiveSignalObjectChildWorkflowImpl.class);
+    worker.start();
+
+    WorkflowOptions options =
+            new WorkflowOptions.Builder()
+                    .setExecutionStartToCloseTimeout(Duration.ofSeconds(1000))
+                    .setTaskList(taskList)
+                    .build();
+
+    WorkflowClient workflowClient = testEnvironment.newWorkflowClient();
+    SendSignalObjectWorkflow workflow = workflowClient.newWorkflowStub(SendSignalObjectWorkflow.class, options);
+    workflow.execute();
+
+    //Wait for reporter
+    Thread.sleep(600);
+
+    Map<String, String> tags =
+            new ImmutableMap.Builder<String, String>(2)
+                    .put(MetricsTag.DOMAIN, WorkflowTest.DOMAIN)
+                    .put(MetricsTag.TASK_LIST, taskList)
+                    .build();
+    verify(reporter, times(1)).reportCounter(MetricsType.CORRUPTED_SIGNALS_COUNTER, tags, 2);
+  }
+
+  private static class CorruptedSignalWorkflowInterceptorFactory
+          implements Function<WorkflowInterceptor, WorkflowInterceptor> {
+
+      @Override
+      public WorkflowInterceptor apply(WorkflowInterceptor next) {
+          return new SignalWorkflowInterceptor(args -> {
+                      if(args != null && args.length > 0){
+                          return new Object [] {"Corrupted Signal"};
+                      }
+                      return args;
+                  }, sig->sig, next);
+      }
+  }
 }