From 53c1fc7779f6f7ce4c58cbd1c39702f84816b40d Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Sat, 1 Aug 2026 00:53:54 +1200 Subject: [PATCH 1/2] Expose per-worker utilization snapshots --- .../service/supervisor/utilization_monitor.rb | 63 +++++++++++-------- releases.md | 4 ++ test/async/service/utilization_monitor.rb | 20 ++++++ 3 files changed, 62 insertions(+), 25 deletions(-) diff --git a/lib/async/service/supervisor/utilization_monitor.rb b/lib/async/service/supervisor/utilization_monitor.rb index f8c1472..a3ef643 100644 --- a/lib/async/service/supervisor/utilization_monitor.rb +++ b/lib/async/service/supervisor/utilization_monitor.rb @@ -281,35 +281,48 @@ def self.monitor_type # # @returns [Hash] Hash mapping service names to aggregated utilization metrics. def sample - @guard.synchronize do - aggregated = {} + aggregated = {} + + sample_workers.each_value do |worker| + service_name = worker[:state][:name] || "unknown" - @workers.each do |worker_id, supervisor_controller| - service_name = supervisor_controller.state[:name] || "unknown" - - data = @allocator.read(worker_id) - next unless data - - # Initialize service aggregation if needed - aggregated[service_name] ||= {} - - # Sum up all numeric fields - data.each do |key, value| - if value.is_a?(Numeric) - aggregated[service_name][key] ||= 0 - aggregated[service_name][key] += value - else - # For non-numeric values, we could handle differently - # For now, just store the last value - aggregated[service_name][key] = value - end + data = worker[:utilization] + + # Initialize service aggregation if needed + aggregated[service_name] ||= {} + + # Sum up all numeric fields + data.each do |key, value| + if value.is_a?(Numeric) + aggregated[service_name][key] ||= 0 + aggregated[service_name][key] += value + else + # For non-numeric values, we could handle differently + # For now, just store the last value + aggregated[service_name][key] = value end - - # Count workers per service (for utilization denominator) - aggregated[service_name][:worker_count] = (aggregated[service_name][:worker_count] || 0) + 1 end - aggregated + # Count workers per service (for utilization denominator) + aggregated[service_name][:worker_count] = (aggregated[service_name][:worker_count] || 0) + 1 + end + + aggregated + end + + # Sample utilization data for each registered worker. + # + # @returns [Hash] An immutable hash keyed by worker ID, with supervisor state and utilization values. + def sample_workers + @guard.synchronize do + @workers.each_with_object({}) do |(worker_id, supervisor_controller), workers| + if utilization = @allocator.read(worker_id) + workers[worker_id] = { + state: supervisor_controller.state.dup.freeze, + utilization: utilization.freeze, + }.freeze + end + end.freeze end end diff --git a/releases.md b/releases.md index c7198e9..52be2fa 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,9 @@ # Releases +## Unreleased + + - Add per-worker snapshots to `Async::Service::Supervisor::UtilizationMonitor`. + ## v0.18.0 - Allow supervised services to merge additional worker state during preparation and worker construction. diff --git a/test/async/service/utilization_monitor.rb b/test/async/service/utilization_monitor.rb index af98e73..19ec5b5 100644 --- a/test/async/service/utilization_monitor.rb +++ b/test/async/service/utilization_monitor.rb @@ -142,6 +142,26 @@ ) end + it "can sample utilization data by worker" do + monitor.register(supervisor_controller) + + worker_registry.metric(:connections_total).set(100) + worker_registry.metric(:connections_active).set(5) + + workers = monitor.sample_workers + + expect(workers).to be == { + 1 => { + state: {name: "test_service"}, + utilization: {connections_total: 100, connections_active: 5}, + } + } + expect(workers).to be(:frozen?) + expect(workers[1]).to be(:frozen?) + expect(workers[1][:state]).to be(:frozen?) + expect(workers[1][:utilization]).to be(:frozen?) + end + it "aggregates metrics from multiple workers" do # Create second worker registry2 = Async::Utilization::Registry.new From 7e0506469149a2922efe4902326184ec9e8dc31c Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Sat, 1 Aug 2026 10:23:16 +1200 Subject: [PATCH 2/2] Rename per-worker utilization sample --- lib/async/service/supervisor/utilization_monitor.rb | 4 ++-- test/async/service/utilization_monitor.rb | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/lib/async/service/supervisor/utilization_monitor.rb b/lib/async/service/supervisor/utilization_monitor.rb index a3ef643..596ed7d 100644 --- a/lib/async/service/supervisor/utilization_monitor.rb +++ b/lib/async/service/supervisor/utilization_monitor.rb @@ -283,7 +283,7 @@ def self.monitor_type def sample aggregated = {} - sample_workers.each_value do |worker| + sample_by_worker.each_value do |worker| service_name = worker[:state][:name] || "unknown" data = worker[:utilization] @@ -313,7 +313,7 @@ def sample # Sample utilization data for each registered worker. # # @returns [Hash] An immutable hash keyed by worker ID, with supervisor state and utilization values. - def sample_workers + def sample_by_worker @guard.synchronize do @workers.each_with_object({}) do |(worker_id, supervisor_controller), workers| if utilization = @allocator.read(worker_id) diff --git a/test/async/service/utilization_monitor.rb b/test/async/service/utilization_monitor.rb index 19ec5b5..7f67c6d 100644 --- a/test/async/service/utilization_monitor.rb +++ b/test/async/service/utilization_monitor.rb @@ -148,7 +148,7 @@ worker_registry.metric(:connections_total).set(100) worker_registry.metric(:connections_active).set(5) - workers = monitor.sample_workers + workers = monitor.sample_by_worker expect(workers).to be == { 1 => {