Add activity worker thread count metrics (#607)

Author: Shaddoll

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

@@ -131,6 +131,10 @@ public class MetricsType {
   public static final String ACTIVITY_LOCAL_DISPATCH_SUCCEED_COUNTER =
       CADENCE_METRICS_PREFIX + "activity-local-dispatch-succeed";
   public static final String WORKER_PANIC_COUNTER = CADENCE_METRICS_PREFIX + "worker-panic";
+  public static final String LOCAL_ACTIVITY_ACTIVE_THREAD_COUNT =
+      CADENCE_METRICS_PREFIX + "local_activity_active_thread_count";
+  public static final String ACTIVITY_ACTIVE_THREAD_COUNT =
+      CADENCE_METRICS_PREFIX + "activity_active_thread_count";
 
   public static final String TASK_LIST_QUEUE_LATENCY =
       CADENCE_METRICS_PREFIX + "tasklist-queue-latency";

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

@@ -19,6 +19,7 @@
 
 import com.google.common.base.Joiner;
 import com.google.common.reflect.TypeToken;
+import com.google.common.util.concurrent.RateLimiter;
 import com.uber.cadence.PollForActivityTaskResponse;
 import com.uber.cadence.RespondActivityTaskCompletedRequest;
 import com.uber.cadence.RespondActivityTaskFailedRequest;
@@ -43,6 +44,7 @@
 import java.util.function.BiFunction;
 
 class POJOActivityTaskHandler implements ActivityTaskHandler {
+  private static final RateLimiter metricsRateLimiter = RateLimiter.create(1);
 
   private final DataConverter dataConverter;
   private final ScheduledExecutorService heartbeatExecutor;
@@ -186,6 +188,13 @@ public Result handle(
           metricsScope,
           isLocalActivity);
     }
+    if (metricsRateLimiter.tryAcquire(1)) {
+        if (isLocalActivity) {
+            metricsScope.gauge(MetricsType.LOCAL_ACTIVITY_ACTIVE_THREAD_COUNT).update(Thread.activeCount());
+        } else {
+            metricsScope.gauge(MetricsType.ACTIVITY_ACTIVE_THREAD_COUNT).update(Thread.activeCount());
+        }
+    }
     return activity.execute(activityTask, metricsScope);
   }