Merge pull request coinbase#103 from stripe-private-oss-forks/jeffsch…

…oner/idle-metrics

Thread pool and poller metrics

lib/temporal/activity/poller.rb

@@ -72,6 +72,12 @@ def poll_loop
 
           task = poll_for_task
           last_poll_time = Time.now
+
+          Temporal.metrics.increment(
+            Temporal::MetricKeys::ACTIVITY_POLLER_POLL_COMPLETED,
+            metrics_tags.merge(received_task: !task.nil?.to_s)
+          )
+
           next unless task&.activity_type
 
           thread_pool.schedule { process(task) }
@@ -98,7 +104,7 @@ def process(task)
       end
 
       def thread_pool
-        @thread_pool ||= ThreadPool.new(options[:thread_pool_size])
+        @thread_pool ||= ThreadPool.new(options[:thread_pool_size], { pool_name: 'activity_task_poller' })
       end
     end
   end

lib/temporal/metric_keys.rb

@@ -1,12 +1,17 @@
 module Temporal
   class MetricKeys
     ACTIVITY_POLLER_TIME_SINCE_LAST_POLL = 'activity_poller.time_since_last_poll'.freeze
+    ACTIVITY_POLLER_POLL_COMPLETED = 'activity_poller.poll_completed'.freeze
     ACTIVITY_TASK_QUEUE_TIME = 'activity_task.queue_time'.freeze
     ACTIVITY_TASK_LATENCY = 'activity_task.latency'.freeze
 
     WORKFLOW_POLLER_TIME_SINCE_LAST_POLL = 'workflow_poller.time_since_last_poll'.freeze
+    WORKFLOW_POLLER_POLL_COMPLETED = 'workflow_poller.poll_completed'.freeze
     WORKFLOW_TASK_QUEUE_TIME = 'workflow_task.queue_time'.freeze
     WORKFLOW_TASK_LATENCY = 'workflow_task.latency'.freeze
     WORKFLOW_TASK_EXECUTION_FAILED = 'workflow_task.execution_failed'.freeze
+
+    THREAD_POOL_AVAILABLE_THREADS = 'thread_pool.available_threads'.freeze
+    THREAD_POOL_QUEUE_SIZE = 'thread_pool.queue_size'.freeze
   end
 end

lib/temporal/thread_pool.rb

@@ -9,8 +9,9 @@ module Temporal
   class ThreadPool
     attr_reader :size
 
-    def initialize(size)
+    def initialize(size, metrics_tags)
       @size = size
+      @metrics_tags = metrics_tags
       @queue = Queue.new
       @mutex = Mutex.new
       @availability = ConditionVariable.new
@@ -20,6 +21,11 @@ def initialize(size)
       end
     end
 
+    def report_metrics
+      Temporal.metrics.gauge(Temporal::MetricKeys::THREAD_POOL_AVAILABLE_THREADS, @available_threads, @metrics_tags)
+      Temporal.metrics.gauge(Temporal::MetricKeys::THREAD_POOL_QUEUE_SIZE, @queue.size, @metrics_tags)
+    end
+
     def wait_for_available_threads
       @mutex.synchronize do
         while @available_threads <= 0
@@ -33,6 +39,8 @@ def schedule(&block)
         @available_threads -= 1
         @queue << block
       end
+
+      report_metrics
     end
 
     def shutdown
@@ -56,6 +64,8 @@ def poll
             @available_threads += 1
             @availability.signal
           end
+
+          report_metrics
         end
       end
     end

lib/temporal/workflow/poller.rb

@@ -73,6 +73,12 @@ def poll_loop
 
           task = poll_for_task
           last_poll_time = Time.now
+
+          Temporal.metrics.increment(
+            Temporal::MetricKeys::WORKFLOW_POLLER_POLL_COMPLETED,
+            metrics_tags.merge(received_task: !task.nil?.to_s)
+          )
+
           next unless task&.workflow_type
 
           thread_pool.schedule { process(task) }
@@ -98,7 +104,7 @@ def process(task)
       end
 
       def thread_pool
-        @thread_pool ||= ThreadPool.new(options[:thread_pool_size])
+        @thread_pool ||= ThreadPool.new(options[:thread_pool_size], { pool_name: 'workflow_task_poller' })
       end
 
       def binary_checksum