diff --git a/lib/temporal/activity/poller.rb b/lib/temporal/activity/poller.rb index 859fb688..589eeaae 100644 --- a/lib/temporal/activity/poller.rb +++ b/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 diff --git a/lib/temporal/metric_keys.rb b/lib/temporal/metric_keys.rb index e945f0b6..a99f2b35 100644 --- a/lib/temporal/metric_keys.rb +++ b/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 diff --git a/lib/temporal/thread_pool.rb b/lib/temporal/thread_pool.rb index 3febbf82..b9e20adb 100644 --- a/lib/temporal/thread_pool.rb +++ b/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 diff --git a/lib/temporal/workflow/poller.rb b/lib/temporal/workflow/poller.rb index 198f4502..812dd1c3 100644 --- a/lib/temporal/workflow/poller.rb +++ b/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 diff --git a/spec/unit/lib/temporal/thread_pool_spec.rb b/spec/unit/lib/temporal/thread_pool_spec.rb index 5de5b03a..a42ae98f 100644 --- a/spec/unit/lib/temporal/thread_pool_spec.rb +++ b/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