temporal: Emit accurate Ruby worker slot metrics

lib/temporal/activity/poller.rb

@@ -129,6 +129,14 @@ def thread_pool
             pool_name: 'activity_task_poller',
             namespace: namespace,
             task_queue: task_queue
+          },
+          worker_metrics_tags: {
+            namespace: namespace,
+            task_queue: task_queue,
+            worker_type: 'ActivityWorker',
+            # DogStatsD gauges from multiple worker processes otherwise share
+            # one metric context and overwrite each other.
+            worker_id: config.for_connection.identity.to_s.strip
           }
         )
       end

lib/temporal/metric_keys.rb

@@ -12,5 +12,7 @@ module MetricKeys
     WORKFLOW_TASK_EXECUTION_FAILED = 'workflow_task.execution_failed'.freeze
 
     THREAD_POOL_AVAILABLE_THREADS = 'thread_pool.available_threads'.freeze
+    WORKER_TASK_SLOTS_AVAILABLE = 'worker_task_slots_available'.freeze
+    WORKER_TASK_SLOTS_USED = 'worker_task_slots_used'.freeze
   end
 end

lib/temporal/thread_pool.rb

@@ -9,11 +9,21 @@
 #
 module Temporal
   class ThreadPool
+    DEFAULT_METRICS_REPORT_INTERVAL_SECONDS = 10
+
     attr_reader :size
 
-    def initialize(size, config, metrics_tags)
+    def initialize(
+      size,
+      config,
+      metrics_tags,
+      worker_metrics_tags: nil,
+      metrics_report_interval_seconds: DEFAULT_METRICS_REPORT_INTERVAL_SECONDS
+    )
       @size = size
       @metrics_tags = metrics_tags
+      @worker_metrics_tags = worker_metrics_tags
+      @metrics_report_interval_seconds = metrics_report_interval_seconds
       @queue = Queue.new
       @mutex = Mutex.new
       @config = config
@@ -22,6 +32,11 @@ def initialize(size, config, metrics_tags)
       @pool = Array.new(size) do |_i|
         Thread.new { poll }
       end
+
+      # Task transitions can be shorter than a metrics aggregation interval. A
+      # periodic snapshot makes this a stable saturation signal and reports idle
+      # workers even when they have not received a task recently.
+      start_metrics_reporter if worker_metrics_tags
     end
 
     def report_metrics
@@ -44,6 +59,8 @@ def schedule(&block)
     end
 
     def shutdown
+      stop_metrics_reporter
+
       size.times do
         schedule { throw EXIT_SYMBOL }
       end
@@ -55,6 +72,50 @@ def shutdown
 
     EXIT_SYMBOL = :exit
 
+    def report_worker_slot_metrics
+      available_threads = @mutex.synchronize { @available_threads }
+
+      Temporal.metrics.gauge(
+        Temporal::MetricKeys::WORKER_TASK_SLOTS_AVAILABLE,
+        available_threads,
+        @worker_metrics_tags
+      )
+      Temporal.metrics.gauge(
+        Temporal::MetricKeys::WORKER_TASK_SLOTS_USED,
+        size - available_threads,
+        @worker_metrics_tags
+      )
+    end
+
+    def start_metrics_reporter
+      @metrics_reporter_mutex = Mutex.new
+      @metrics_reporter_condition = ConditionVariable.new
+      @metrics_reporter_shutdown = false
+
+      report_worker_slot_metrics
+      @metrics_reporter_thread = Thread.new do
+        loop do
+          shutting_down = @metrics_reporter_mutex.synchronize do
+            @metrics_reporter_condition.wait(@metrics_reporter_mutex, @metrics_report_interval_seconds)
+            @metrics_reporter_shutdown
+          end
+          break if shutting_down
+
+          report_worker_slot_metrics
+        end
+      end
+    end
+
+    def stop_metrics_reporter
+      return unless @metrics_reporter_thread
+
+      @metrics_reporter_mutex.synchronize do
+        @metrics_reporter_shutdown = true
+        @metrics_reporter_condition.signal
+      end
+      @metrics_reporter_thread.join
+    end
+
     def poll
       Thread.current.abort_on_exception = true
 

lib/temporal/workflow/poller.rb

@@ -125,6 +125,14 @@ def thread_pool
             pool_name: 'workflow_task_poller',
             namespace: namespace,
             task_queue: task_queue
+          },
+          worker_metrics_tags: {
+            namespace: namespace,
+            task_queue: task_queue,
+            worker_type: 'WorkflowWorker',
+            # DogStatsD gauges from multiple worker processes otherwise share
+            # one metric context and overwrite each other.
+            worker_id: config.for_connection.identity.to_s.strip
           }
         )
       end

spec/unit/lib/temporal/thread_pool_spec.rb

@@ -9,6 +9,14 @@
   let(:size) { 2 }
   let(:tags) { { foo: 'bar', bat: 'baz' } }
   let(:thread_pool) { described_class.new(size, config, tags) }
+  let(:worker_metrics_tags) do
+    {
+      namespace: 'test-namespace',
+      task_queue: 'test-task-queue',
+      worker_type: 'ActivityWorker',
+      worker_id: 'worker-1'
+    }
+  end
 
   describe '#new' do
     it 'executes one task on a thread and exits' do
@@ -71,5 +79,57 @@
         )
         .at_least(:once)
     end
+
+    it 'periodically reports available and used task slots for an individual worker' do
+      pool = described_class.new(
+        size,
+        config,
+        tags,
+        worker_metrics_tags: worker_metrics_tags,
+        metrics_report_interval_seconds: 0.01
+      )
+
+      sleep 0.02
+      pool.shutdown
+
+      expect(Temporal.metrics)
+        .to have_received(:gauge)
+        .with(Temporal::MetricKeys::WORKER_TASK_SLOTS_AVAILABLE, size, worker_metrics_tags)
+        .at_least(:twice)
+      expect(Temporal.metrics)
+        .to have_received(:gauge)
+        .with(Temporal::MetricKeys::WORKER_TASK_SLOTS_USED, 0, worker_metrics_tags)
+        .at_least(:twice)
+    end
+
+    it 'reports task slot usage while a task is running' do
+      task_started = Queue.new
+      finish_task = Queue.new
+      pool = described_class.new(
+        size,
+        config,
+        tags,
+        worker_metrics_tags: worker_metrics_tags,
+        metrics_report_interval_seconds: 0.01
+      )
+
+      pool.schedule do
+        task_started << true
+        finish_task.pop
+      end
+      task_started.pop
+      sleep 0.02
+      finish_task << true
+      pool.shutdown
+
+      expect(Temporal.metrics)
+        .to have_received(:gauge)
+        .with(Temporal::MetricKeys::WORKER_TASK_SLOTS_AVAILABLE, size - 1, worker_metrics_tags)
+        .at_least(:once)
+      expect(Temporal.metrics)
+        .to have_received(:gauge)
+        .with(Temporal::MetricKeys::WORKER_TASK_SLOTS_USED, 1, worker_metrics_tags)
+        .at_least(:once)
+    end
   end
 end