Merge pull request #148 from meiliang86/metrics

add metrics in java client

Author: meiliang86

build.gradle

@@ -40,8 +40,9 @@ targetCompatibility = 1.8
 dependencies {
     compile group: 'com.uber.tchannel', name: 'tchannel-core', version: '0.7.6'
     compile group: 'org.slf4j', name: 'slf4j-api', version: '1.7.25'
-    compile(group: 'org.apache.thrift', name: 'libthrift', version: '0.9.3')
+    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'
     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/client/WorkflowClientOptions.java

@@ -19,6 +19,8 @@
 
 import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.converter.JsonDataConverter;
+import com.uber.cadence.internal.metrics.NoopScope;
+import com.uber.m3.tally.Scope;
 import java.util.Objects;
 
 /** Options for WorkflowClient configuration. */
@@ -30,6 +32,8 @@ public static final class Builder {
 
     private WorkflowClientInterceptor[] interceptors = EMPTY_INTERCEPTOR_ARRAY;
 
+    private Scope metricsScope;
+
     /**
      * Used to override default (JSON) data converter implementation.
      *
@@ -51,8 +55,21 @@ public Builder setInterceptors(WorkflowClientInterceptor... interceptors) {
       return this;
     }
 
+    /**
+     * Sets the scope to be used for the workflow client for metrics reporting.
+     *
+     * @param metricsScope the scope to be used. Not null.
+     */
+    public Builder setMetricsScope(Scope metricsScope) {
+      this.metricsScope = Objects.requireNonNull(metricsScope);
+      return this;
+    }
+
     public WorkflowClientOptions build() {
-      return new WorkflowClientOptions(dataConverter, interceptors);
+      if (metricsScope == null) {
+        metricsScope = NoopScope.getInstance();
+      }
+      return new WorkflowClientOptions(dataConverter, interceptors, metricsScope);
     }
   }
 
@@ -62,10 +79,13 @@ public WorkflowClientOptions build() {
 
   private final WorkflowClientInterceptor[] interceptors;
 
+  private final Scope metricsScope;
+
   private WorkflowClientOptions(
-      DataConverter dataConverter, WorkflowClientInterceptor[] interceptors) {
+      DataConverter dataConverter, WorkflowClientInterceptor[] interceptors, Scope metricsScope) {
     this.dataConverter = dataConverter;
     this.interceptors = interceptors;
+    this.metricsScope = metricsScope;
   }
 
   public DataConverter getDataConverter() {
@@ -75,4 +95,8 @@ public DataConverter getDataConverter() {
   public WorkflowClientInterceptor[] getInterceptors() {
     return interceptors;
   }
+
+  public Scope getMetricsScope() {
+    return metricsScope;
+  }
 }

src/main/java/com/uber/cadence/internal/external/GenericWorkflowClientExternalImpl.java

@@ -31,9 +31,14 @@
 import com.uber.cadence.internal.common.CheckedExceptionWrapper;
 import com.uber.cadence.internal.common.StartWorkflowExecutionParameters;
 import com.uber.cadence.internal.common.TerminateWorkflowExecutionParameters;
+import com.uber.cadence.internal.metrics.MetricsTag;
+import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.internal.replay.QueryWorkflowParameters;
 import com.uber.cadence.internal.replay.SignalExternalWorkflowParameters;
 import com.uber.cadence.serviceclient.IWorkflowService;
+import com.uber.m3.tally.Scope;
+import com.uber.m3.util.ImmutableMap;
+import java.util.Map;
 import java.util.UUID;
 import org.apache.thrift.TException;
 
@@ -43,9 +48,13 @@ public final class GenericWorkflowClientExternalImpl implements GenericWorkflowC
 
   private final IWorkflowService service;
 
-  public GenericWorkflowClientExternalImpl(IWorkflowService service, String domain) {
+  private final Scope metricsScope;
+
+  public GenericWorkflowClientExternalImpl(
+      IWorkflowService service, String domain, Scope metricsScope) {
     this.service = service;
     this.domain = domain;
+    this.metricsScope = metricsScope;
   }
 
   @Override
@@ -61,6 +70,20 @@ public IWorkflowService getService() {
   @Override
   public WorkflowExecution startWorkflow(StartWorkflowExecutionParameters startParameters)
       throws WorkflowExecutionAlreadyStartedError {
+    try {
+      return startWorkflowInternal(startParameters);
+    } finally {
+      // TODO: can probably cache this
+      Map<String, String> tags =
+          new ImmutableMap.Builder<String, String>(1)
+              .put(MetricsTag.WORKFLOW_TYPE, startParameters.getWorkflowType().getName())
+              .build();
+      metricsScope.tagged(tags).counter(MetricsType.WORKFLOW_START_COUNTER).inc(1);
+    }
+  }
+
+  private WorkflowExecution startWorkflowInternal(StartWorkflowExecutionParameters startParameters)
+      throws WorkflowExecutionAlreadyStartedError {
     StartWorkflowExecutionRequest request = new StartWorkflowExecutionRequest();
     request.setDomain(domain);
 

src/main/java/com/uber/cadence/internal/external/ManualActivityCompletionClientFactoryImpl.java

@@ -20,20 +20,22 @@
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.serviceclient.IWorkflowService;
+import com.uber.m3.tally.Scope;
 
 public class ManualActivityCompletionClientFactoryImpl
     extends ManualActivityCompletionClientFactory {
 
   private final IWorkflowService service;
-
   private final DataConverter dataConverter;
   private final String domain;
+  private final Scope metricsScope;
 
   public ManualActivityCompletionClientFactoryImpl(
-      IWorkflowService service, String domain, DataConverter dataConverter) {
+      IWorkflowService service, String domain, DataConverter dataConverter, Scope metricsScope) {
     this.service = service;
     this.domain = domain;
     this.dataConverter = dataConverter;
+    this.metricsScope = metricsScope;
   }
 
   public IWorkflowService getService() {
@@ -55,7 +57,7 @@ public ManualActivityCompletionClient getClient(byte[] taskToken) {
     if (taskToken == null || taskToken.length == 0) {
       throw new IllegalArgumentException("null or empty task token");
     }
-    return new ManualActivityCompletionClientImpl(service, taskToken, dataConverter);
+    return new ManualActivityCompletionClientImpl(service, taskToken, dataConverter, metricsScope);
   }
 
   @Override
@@ -67,6 +69,6 @@ public ManualActivityCompletionClient getClient(WorkflowExecution execution, Str
       throw new IllegalArgumentException("null activityId");
     }
     return new ManualActivityCompletionClientImpl(
-        service, domain, execution, activityId, dataConverter);
+        service, domain, execution, activityId, dataConverter, metricsScope);
   }
 }

src/main/java/com/uber/cadence/internal/external/ManualActivityCompletionClientImpl.java

@@ -31,7 +31,9 @@
 import com.uber.cadence.client.ActivityCompletionFailureException;
 import com.uber.cadence.client.ActivityNotExistsException;
 import com.uber.cadence.converter.DataConverter;
+import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.serviceclient.IWorkflowService;
+import com.uber.m3.tally.Scope;
 import java.util.concurrent.CancellationException;
 import org.apache.thrift.TException;
 import org.slf4j.Logger;
@@ -51,29 +53,33 @@ class ManualActivityCompletionClientImpl extends ManualActivityCompletionClient
   private final String domain;
   private final WorkflowExecution execution;
   private final String activityId;
+  private final Scope metricsScope;
 
   public ManualActivityCompletionClientImpl(
-      IWorkflowService service, byte[] taskToken, DataConverter dataConverter) {
+      IWorkflowService service, byte[] taskToken, DataConverter dataConverter, Scope metricsScope) {
     this.service = service;
     this.taskToken = taskToken;
     this.dataConverter = dataConverter;
     this.domain = null;
     this.execution = null;
     this.activityId = null;
+    this.metricsScope = metricsScope;
   }
 
   public ManualActivityCompletionClientImpl(
       IWorkflowService service,
       String domain,
       WorkflowExecution execution,
       String activityId,
-      DataConverter dataConverter) {
+      DataConverter dataConverter,
+      Scope metricsScope) {
     this.service = service;
     this.taskToken = null;
     this.domain = domain;
     this.execution = execution;
     this.activityId = activityId;
     this.dataConverter = dataConverter;
+    this.metricsScope = metricsScope;
   }
 
   @Override
@@ -85,6 +91,7 @@ public void complete(Object result) {
       request.setTaskToken(taskToken);
       try {
         service.RespondActivityTaskCompleted(request);
+        metricsScope.counter(MetricsType.ACTIVITY_TASK_COMPLETED_COUNTER).inc(1);
       } catch (EntityNotExistsError e) {
         throw new ActivityNotExistsException(e);
       } catch (TException e) {
@@ -104,6 +111,7 @@ public void complete(Object result) {
       request.setRunID(execution.getRunId());
       try {
         service.RespondActivityTaskCompletedByID(request);
+        metricsScope.counter(MetricsType.ACTIVITY_TASK_COMPLETED_BY_ID_COUNTER).inc(1);
       } catch (EntityNotExistsError e) {
         throw new ActivityNotExistsException(e);
       } catch (TException e) {
@@ -125,6 +133,7 @@ public void fail(Throwable failure) {
       request.setTaskToken(taskToken);
       try {
         service.RespondActivityTaskFailed(request);
+        metricsScope.counter(MetricsType.ACTIVITY_TASK_FAILED_COUNTER).inc(1);
       } catch (EntityNotExistsError e) {
         throw new ActivityNotExistsException(e);
       } catch (TException e) {
@@ -139,6 +148,7 @@ public void fail(Throwable failure) {
       request.setRunID(execution.getRunId());
       try {
         service.RespondActivityTaskFailedByID(request);
+        metricsScope.counter(MetricsType.ACTIVITY_TASK_FAILED_BY_ID_COUNTER).inc(1);
       } catch (EntityNotExistsError e) {
         throw new ActivityNotExistsException(e);
       } catch (TException e) {
@@ -178,6 +188,7 @@ public void reportCancellation(Object details) {
       request.setTaskToken(taskToken);
       try {
         service.RespondActivityTaskCanceled(request);
+        metricsScope.counter(MetricsType.ACTIVITY_TASK_CANCELED_COUNTER).inc(1);
       } catch (TException e) {
         // There is nothing that can be done at this point.
         // so let's just ignore.
@@ -191,6 +202,7 @@ public void reportCancellation(Object details) {
       request.setRunID(execution.getRunId());
       try {
         service.RespondActivityTaskCanceledByID(request);
+        metricsScope.counter(MetricsType.ACTIVITY_TASK_CANCELED_BY_ID_COUNTER).inc(1);
       } catch (TException e) {
         // There is nothing that can be done at this point.
         // so let's just ignore.

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

@@ -0,0 +1,35 @@
+/*
+ *  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;
+
+public class MetricsTag {
+  public static final String ACTIVITY_ID = "ActivityID";
+  public static final String ACTIVITY_TYPE = "ActivityType";
+  public static final String DOMAIN = "Domain";
+  public static final String EVENT_ID = "EventID";
+  public static final String EVENT_TYPE = "EventType";
+  public static final String RUN_ID = "RunID";
+  public static final String TASK_LIST = "TaskList";
+  public static final String TIMER_ID = "TimerID";
+  public static final String WORKFLOW_ID = "WorkflowID";
+  public static final String WORKFLOW_TYPE = "WorkflowType";
+  public static final String WORKER_ID = "WorkerID";
+  public static final String WORKER_TYPE = "WorkerType";
+  public static final String SIDE_EFFECT_ID = "SideEffectID";
+  public static final String CHILD_WORKFLOW_ID = "ChildWorkflowID";
+}