diff --git a/lib/async/service/supervisor/utilization_monitor.rb b/lib/async/service/supervisor/utilization_monitor.rb index f8c1472..596ed7d 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_by_worker.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_by_worker + @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..7f67c6d 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_by_worker + + 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